Coverage for enderchest/place.py: 100%
178 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-10-03 20:14 +0000
1"""Symlinking functionality"""
2import fnmatch
3import itertools
4import logging
5import os
6from pathlib import Path
7from typing import Iterable
9from . import filesystem as fs
10from .gather import load_ender_chest, load_ender_chest_instances, load_shulker_boxes
11from .loggers import PLACE_LOGGER
12from .prompt import prompt
13from .shulker_box import ShulkerBox
16def place_ender_chest(
17 minecraft_root: Path,
18 keep_broken_links: bool = False,
19 keep_stale_links: bool = False,
20 error_handling: str = "abort",
21 relative: bool = True,
22 rollback=False,
23) -> None:
24 """Link all instance files and folders to all shulker boxes
26 Parameters
27 ----------
28 minecraft_root : Path
29 The root directory that your minecraft stuff (or, at least, the one
30 that's the parent of your EnderChest folder)
31 keep_broken_links : bool, optional
32 By default, this method will remove any broken links in your instances
33 and servers folders. To disable this behavior, pass in
34 `keep_broken_links=True`.
35 keep_stale_links : bool, optional
36 By default, this method will remove any links into your EnderChest folder
37 that are no longer specified by any shulker box (such as because the
38 instance spec or shulker box configuration changed). To disable this
39 behavior, pass in `keep_stale_links=True`.
40 error_handling : str, optional
41 By default, if a linking failure occurs, this method will terminate
42 immediately (`error_handling=abort`). Alternatively,
43 - pass in `error_handling="ignore"` to continue as if the link failure
44 hadn't occurred
45 - pass in `error_handling="skip"` to abort linking the current instance
46 to the current shulker box but otherwise continue on
47 - pass in `error_handling="skip-instance"` to abort linking the current
48 instance altogether but to otherwise continue on with other instances
49 - pass in `error_handling="skip-shulker-box"` to abort linking to the current
50 shulker box altogether but to otherwise continue on with other boxes
51 - pass in `error_handling="prompt"` to ask what to do on each failure
52 relative : bool, optional
53 By default, links will use relative paths when possible. To use absolute
54 paths instead (see: https://bugs.mojang.com/projects/MC/issues/MC-263046),
55 pass in `relative=False`. See note below.
56 rollback: bool, optional
57 In the future in the event of linking errors passing in `rollback=True`
58 can be used to roll back any changes that have already been applied
59 based on the error-handling method specified.
61 Notes
62 -----
63 - If one of the files or folders being placed is itself a symlink, relative
64 links will be created as *nested* links (a link pointing to the link),
65 whereas in "absolute" mode (`relative=False`), the link that will be
66 placed will point **directly** to the final target.
67 - This can lead to the stale-link cleanup behavior not correctly removing
68 an outdated symlink if the fully resolved target of a link falls outside
69 the EnderChest folder.
70 """
71 if rollback is not False: # pragma: no cover
72 raise NotImplementedError("Rollbacks are not currently supported")
74 try:
75 host = load_ender_chest(minecraft_root).name
76 except (FileNotFoundError, ValueError) as bad_chest:
77 PLACE_LOGGER.error(
78 f"Could not load EnderChest from {minecraft_root}:\n {bad_chest}"
79 )
80 return
82 instances = load_ender_chest_instances(minecraft_root, log_level=logging.DEBUG)
84 shulker_boxes: list[ShulkerBox] = []
86 for shulker_box in load_shulker_boxes(minecraft_root, log_level=logging.DEBUG):
87 if not shulker_box.matches_host(host):
88 PLACE_LOGGER.debug(
89 f"{shulker_box.name} is not intended for linking to this host ({host})"
90 )
91 continue
92 shulker_boxes.append(shulker_box)
94 skip_boxes: list[ShulkerBox] = []
96 def handle_error(shulker_box: ShulkerBox | None) -> str:
97 """Centralized error-handling
99 Parameters
100 ----------
101 shulker_box:
102 The current shulker box (in case it needs to be added to the skip list)
104 Returns
105 -------
106 str
107 Instructions on what to do next. Options are:
108 - retry
109 - return
110 - break
111 - continue
112 - pass
113 """
114 if error_handling == "prompt":
115 proceed_how = (
116 prompt(
117 "How would you like to proceed?"
118 "\n[Q]uit; [R]etry; [C]ontinue; skip linking the rest of this:"
119 "\n[I]nstance, [S]hulker box, shulker/instance [M]atch?",
120 suggestion="R",
121 )
122 .lower()
123 .replace(" ", "")
124 .replace("-", "")
125 .replace("_", "")
126 )
127 match proceed_how:
128 case "" | "r":
129 proceed_how = "retry"
130 case "" | "i" | "instance" | "skipinstance":
131 proceed_how = "skip-instance"
132 case "q" | "quit" | "abort" | "exit" | "stop":
133 proceed_how = "abort"
134 case "c" | "continue" | "ignore":
135 proceed_how = "ignore"
136 case "m" | "match" | "skip":
137 proceed_how = "skip"
138 case "s" | "shulker" | "shulkerbox" | "skipshulker":
139 proceed_how = "skip-shulker"
140 case _:
141 PLACE_LOGGER.error("Invalid selection.")
142 return handle_error(shulker_box)
143 else:
144 proceed_how = error_handling
146 match proceed_how:
147 case "retry":
148 return "retry"
149 case "abort" | "stop" | "quit" | "exit":
150 PLACE_LOGGER.error("Aborting")
151 return "return"
152 case "ignore":
153 PLACE_LOGGER.debug("Ignoring")
154 return "pass"
155 case "skip":
156 PLACE_LOGGER.warning("Skipping the rest of this match")
157 return "continue"
158 case "skip-instance":
159 PLACE_LOGGER.warning("Skipping any more linking from this instance")
161 return "break"
162 case "skip-shulker-box" | "skip-shulkerbox" | "skip-shulker":
163 PLACE_LOGGER.warning("Skipping any more linking into this shulker box")
164 if shulker_box:
165 skip_boxes.append(shulker_box)
166 return "continue"
167 case _:
168 raise ValueError(
169 f"Unrecognized error-handling method: {error_handling}"
170 )
172 for instance in instances:
173 instance_root = (minecraft_root / instance.root.expanduser()).expanduser()
175 handling: str | None = "retry"
176 while handling == "retry":
177 if instance_root.exists():
178 handling = None
179 break
181 PLACE_LOGGER.error(
182 "No minecraft instance exists at"
183 f" {instance_root.expanduser().absolute()}"
184 )
185 handling = handle_error(None)
186 if handling is not None:
187 match handling:
188 case "return":
189 return
190 case "break":
191 break
192 case _: # nothing to link, so might as well skip the rest
193 continue
195 # start by removing all existing symlinks into the EnderChest
196 if not keep_stale_links:
197 for file in instance_root.rglob("*"):
198 if file.is_symlink():
199 if fs.links_into_enderchest(minecraft_root, file):
200 PLACE_LOGGER.debug(
201 f"Removing old link: {file} -> {os.readlink(file)}"
202 )
203 file.unlink()
205 for shulker_box in shulker_boxes:
206 if not shulker_box.matches(instance):
207 continue
208 if shulker_box in skip_boxes:
209 continue
211 box_root = shulker_box.root.expanduser().absolute()
213 PLACE_LOGGER.info(f"Linking {instance.root} to {shulker_box.name}")
215 resources = set(_rglob(box_root, shulker_box.max_link_depth))
217 match_exit = "pass"
218 for link_folder in shulker_box.link_folders:
219 resources -= {box_root / link_folder}
220 resources -= set((box_root / link_folder).rglob("*"))
222 handling = "retry"
223 while handling == "retry":
224 try:
225 link_resource(link_folder, box_root, instance_root, relative)
226 handling = None
227 except OSError:
228 PLACE_LOGGER.error(
229 f"Error linking shulker box {shulker_box.name}"
230 f" to instance {instance.name}:"
231 f"\n {(instance.root / link_folder)} is a"
232 " non-empty directory"
233 )
234 handling = handle_error(shulker_box)
235 if handling is not None:
236 match handling:
237 case "return":
238 return
239 case "break":
240 match_exit = "break"
241 break
242 case "continue":
243 match_exit = "continue"
244 break
245 case "pass":
246 continue # or pass--it's the end of the loop
248 if match_exit not in ("break", "continue"):
249 for resource in resources:
250 resource_path = resource.relative_to(box_root)
251 for pattern in shulker_box.do_not_link:
252 if fnmatch.fnmatchcase(
253 str(resource_path), pattern
254 ) or fnmatch.fnmatchcase(
255 str(resource_path), os.path.join("*", pattern)
256 ):
257 PLACE_LOGGER.debug(
258 "Skipping %s (matches pattern %s)",
259 resource_path,
260 pattern,
261 )
262 break
263 else:
264 handling = "retry"
265 while handling == "retry":
266 try:
267 link_resource(
268 resource_path,
269 box_root,
270 instance_root,
271 relative,
272 )
273 handling = None
274 except OSError:
275 PLACE_LOGGER.error(
276 f"Error linking shulker box {shulker_box.name}"
277 f" to instance {instance.name}:"
278 f"\n {(instance.root / resource_path)}"
279 " already exists"
280 )
281 handling = handle_error(shulker_box)
282 if handling is not None:
283 match handling:
284 case "return":
285 return
286 case "break":
287 match_exit = "break"
288 break
289 case "continue":
290 match_exit = "continue" # technically does nothing
291 break
292 case "pass":
293 continue # or pass--it's the end of the loop
295 # consider this a "finally"
296 if not keep_broken_links:
297 # we clean up as we go, just in case of a failure
298 for file in instance_root.rglob("*"):
299 if not file.exists():
300 PLACE_LOGGER.debug(f"Removing broken link: {file}")
301 file.unlink()
303 if match_exit == "break":
304 break
307def link_resource(
308 resource_path: str | Path,
309 shulker_root: Path,
310 instance_root: Path,
311 relative: bool,
312) -> None:
313 """Create a symlink for the specified resource from an instance's space
314 pointing to the tagged file / folder living inside a shulker box.
316 Parameters
317 ----------
318 resource_path : str or Path
319 Location of the resource relative to the instance's ".minecraft" folder
320 shulker_root : Path
321 The path to the shulker box
322 instance_root : Path
323 The path to the instance's ".minecraft" folder
324 relative : bool
325 If True, the link will be use a relative path if possible. Otherwise,
326 an absolute path will be used, regardless of whether a a relative or
327 absolute path was provided.
329 Raises
330 ------
331 OSError
332 If a file or non-empty directory already exists where you're attempting
333 to place the symlink
335 Notes
336 -----
337 - This method will create any folders that do not exist within an instance
338 - This method will overwrite existing symlinks and empty folders
339 but will not overwrite or delete any actual files.
340 """
341 instance_path = (instance_root / resource_path).expanduser().absolute()
342 instance_path.parent.mkdir(parents=True, exist_ok=True)
344 target: str | Path = (shulker_root / resource_path).expanduser().absolute()
345 if relative:
346 target = os.path.relpath(target, instance_path.parent)
347 else:
348 target = target.resolve() # type: ignore
350 if instance_path.is_symlink():
351 # remove previous symlink in this spot
352 instance_path.unlink()
353 PLACE_LOGGER.debug(f"Removed previous link at {instance_path}")
354 else:
355 try:
356 os.rmdir(instance_path)
357 PLACE_LOGGER.debug(f"Removed empty directory at {instance_path}")
358 except FileNotFoundError:
359 pass # A-OK
361 PLACE_LOGGER.debug(f"Linking {instance_path} to {target}")
362 os.symlink(
363 target,
364 instance_path,
365 target_is_directory=(shulker_root / resource_path).is_dir(),
366 )
369def _rglob(root: Path, max_depth: int) -> Iterable[Path]:
370 """Find all files (and directories* and symlinks) in the path up to the
371 specified depth
373 Parameters
374 ----------
375 root : Path
376 The path to search
377 max_depth : int
378 The maximum number of levels to go
380 Returns
381 -------
382 list-like of paths
383 The files (and directories and symlinks) in the path up to that depth
385 Notes
386 -----
387 - Unlike an actual rglob, this method does not return any directories that
388 are not at the maximum depth
389 - Setting max_depth to 0 (or below) will return all files in the root, but
390 ***be warned*** that because this method follows symlinks, you can very
391 easily find yourself in an infinite loop
392 """
393 top_level = root.iterdir()
394 if max_depth == 1:
395 return top_level
396 return itertools.chain(
397 *(
398 _rglob(path, max_depth - 1) if path.is_dir() else (path,)
399 for path in top_level
400 )
401 )