Coverage for enderchest/enderchest.py: 99%
182 statements
« prev ^ index » next coverage.py v7.5.0, created at 2024-05-04 01:41 +0000
1"""Specification and configuration of an EnderChest"""
3from dataclasses import dataclass
4from pathlib import Path
5from socket import gethostname
6from typing import Any, Iterable
7from urllib.parse import ParseResult, urlparse
9from . import config as cfg
10from . import filesystem as fs
11from . import instance as i
12from . import sync
13from .loggers import CRAFT_LOGGER, GATHER_LOGGER, INVENTORY_LOGGER
14from .sync import abspath_from_uri
16_DEFAULTS = (
17 ("offer_to_update_symlink_allowlist", True),
18 ("sync_confirm_wait", 5),
19 ("place_after_open", True),
20 ("do_not_sync", ("EnderChest/enderchest.cfg", "EnderChest/.*", ".DS_Store")),
21 (
22 "shulker_box_folders",
23 (
24 "config",
25 "mods",
26 "resourcepacks",
27 "saves",
28 "shaderpacks",
29 ),
30 ),
31 ("standard_link_folders", ()),
32 (
33 "global_link_folders",
34 (
35 "backups",
36 "cachedImages",
37 "crash-reports",
38 "logs",
39 "replay_recordings",
40 "screenshots",
41 "schematics",
42 "config/litematica", # still worth having in case max_depth>2
43 ".bobby",
44 ),
45 ),
46)
49@dataclass(init=False, repr=False, eq=False)
50class EnderChest:
51 """Configuration of an EnderChest
53 Parameters
54 ----------
55 uri : URI or Path
56 The "address" of this EnderChest, ideally as it can be accessed from other
57 EnderChest installations, including both the path to where
58 the EnderChest folder can be found (that is, the parent of the
59 EnderChest folder itself, aka the "minecraft_root"), its net location
60 including credentials, and the protocol that should be used to perform
61 the syncing. All that being said, if just a path is provided, the
62 constructor will try to figure out the rest.
63 name : str, optional
64 A unique name to give to this EnderChest installation. If None is
65 provided, this will be taken from the hostname of the supplied URI.
66 instances : list-like of InstanceSpec, optional
67 The list of instances to register with this EnderChest installation
68 remotes : list-like of URI, or (URI, str) tuples
69 A list of other installations that this EnderChest should be aware of
70 (for syncing purposes). When a (URI, str) tuple is provided, the
71 second value will be used as the name/alias of the remote.
73 Attributes
74 ----------
75 name : str
76 The unique name of this EnderChest installation. This is most commonly
77 the computer's hostname, but one can configure multiple EnderChests
78 to coexist on the same system (either for the sake of having a "cold"
79 backup or for multi-user systems).
80 uri : str
81 The complete URI of this instance
82 root : Path
83 The path to this EnderChest folder
84 instances : list-like of InstanceSpec
85 The instances registered with this EnderChest
86 remotes : list-like of (ParseResult, str) pairs
87 The other EnderChest installations this EnderChest is aware of, paired
88 with their aliases
89 offer_to_update_symlink_allowlist : bool
90 By default, EnderChest will offer to create or update `allowed_symlinks.txt`
91 on any 1.20+ instances that do not already blanket allow links into
92 EnderChest. **EnderChest will never modify that or any other Minecraft
93 file without your express consent.** If you would prefer to edit these
94 files yourself (or simply not symlink your world saves), change this
95 parameter to False.
96 sync_confirm_wait : bool or int
97 The default behavior when syncing EnderChests is to first perform a dry
98 run of every sync operation and then wait 5 seconds before proceeding with the
99 real sync. The idea is to give the user time to interrupt the sync if
100 the dry run looks wrong. This can be changed by either raising or lowering
101 the value of confirm, by disabling the dry-run-first behavior entirely
102 (`confirm=False`) or by requiring that the user explicitly confirms
103 the sync (`confirm=True`). This default behavior can also be overridden
104 when actually calling the sync commands.
105 place_after_open: bool
106 By default, EnderChest will follow up any `enderchest open` operation
107 with an `enderchest place` to refresh any changed symlinks. This
108 functionality can be disabled by setting this parameter to False.
109 do_not_sync : list of str
110 Glob patterns of files that should not be synced between EnderChest
111 installations. By default, this list comprises `EnderChest/enderchest.cfg`,
112 any top-level folders starting with a "." (like .git) and
113 `.DS_Store` (for all you mac gamers).
114 shulker_box_folders : list of str
115 The folders that will be created inside each new shulker box
116 standard_link_folders : list of str
117 The default set of "link folders" when crafting a new shulker box
118 global_link_folders : list of str
119 The "global" set of "link folders," offered as a suggestion when
120 crafting a new shulker box
121 """
123 name: str
124 _uri: ParseResult
125 _instances: list[i.InstanceSpec]
126 _remotes: dict[str, ParseResult]
127 offer_to_update_symlink_allowlist: bool
128 sync_confirm_wait: bool | int
129 place_after_open: bool
130 do_not_sync: list[str]
131 shulker_box_folders: list[str]
132 standard_link_folders: list[str]
133 global_link_folders: list[str]
135 def __init__(
136 self,
137 uri: str | ParseResult | Path,
138 name: str | None = None,
139 remotes: (
140 Iterable[str | ParseResult | tuple[str, str] | tuple[ParseResult, str]]
141 | None
142 ) = None,
143 instances: Iterable[i.InstanceSpec] | None = None,
144 ):
145 for setting, value in _DEFAULTS:
146 setattr(self, setting, list(value) if isinstance(value, tuple) else value)
148 try:
149 if isinstance(uri, ParseResult):
150 self._uri = uri
151 elif isinstance(uri, Path):
152 self._uri = urlparse(uri.absolute().as_uri())
153 else:
154 self._uri = urlparse(uri)
155 except AttributeError as parse_problem: # pragma: no cover
156 raise ValueError(f"{uri} is not a valid URI") from parse_problem
158 if not self._uri.netloc:
159 self._uri = self._uri._replace(netloc=sync.get_default_netloc())
160 if not self._uri.scheme:
161 self._uri = self._uri._replace(scheme=sync.DEFAULT_PROTOCOL)
163 self.name = name or self._uri.hostname or gethostname()
165 self._instances = []
166 self._remotes = {}
168 for instance in instances or ():
169 self.register_instance(instance)
171 for remote in remotes or ():
172 if isinstance(remote, (str, ParseResult)):
173 self.register_remote(remote)
174 else:
175 self.register_remote(*remote)
177 @property
178 def uri(self) -> str:
179 return self._uri.geturl()
181 def __repr__(self) -> str:
182 return f"EnderChest({self.uri, self.name})"
184 @property
185 def root(self) -> Path:
186 return fs.ender_chest_folder(abspath_from_uri(self._uri), check_exists=False)
188 @property
189 def instances(self) -> tuple[i.InstanceSpec, ...]:
190 return tuple(self._instances)
192 def register_instance(self, instance: i.InstanceSpec) -> i.InstanceSpec:
193 """Register a new Minecraft installation
195 Parameters
196 ----------
197 instance : InstanceSpec
198 The instance to register
200 Returns
201 -------
202 InstanceSpec
203 The spec of the instance as it was actually registered (in case the
204 name changed or somesuch)
206 Notes
207 -----
208 - If the instance's name is already assigned to a registered instance,
209 this method will choose a new one
210 - If this instance shares a path with an existing instance, it will
211 replace that instance
212 """
213 matching_instances: list[i.InstanceSpec] = []
214 for old_instance in self._instances:
215 if i.equals(abspath_from_uri(self._uri), instance, old_instance):
216 matching_instances.append(old_instance)
217 self._instances.remove(old_instance)
219 instance = i.merge(*matching_instances, instance)
221 name = instance.name
222 counter = 0
223 taken_names = {old_instance.name for old_instance in self._instances}
224 while True:
225 if name not in taken_names:
226 break
227 counter += 1
228 name = f"{instance.name}.{counter}"
230 GATHER_LOGGER.debug(f"Registering instance {name} at {instance.root}")
231 self._instances.append(instance._replace(name=name))
232 return self._instances[-1]
234 @property
235 def remotes(self) -> tuple[tuple[ParseResult, str], ...]:
236 return tuple((remote, alias) for alias, remote in self._remotes.items())
238 def register_remote(
239 self, remote: str | ParseResult, alias: str | None = None
240 ) -> None:
241 """Register a new remote EnderChest installation (or update an existing
242 registry)
244 Parameters
245 ----------
246 remote : URI
247 The URI of the remote
248 alias : str, optional
249 an alias to give to this remote. If None is provided, the URI's hostname
250 will be used.
252 Raises
253 ------
254 ValueError
255 If the provided remote is invalid
256 """
257 try:
258 remote = remote if isinstance(remote, ParseResult) else urlparse(remote)
259 alias = alias or remote.hostname
260 if not alias: # pragma: no cover
261 raise AttributeError(f"{remote.geturl()} has no hostname")
262 GATHER_LOGGER.debug("Registering remote %s (%s)", remote.geturl(), alias)
263 self._remotes[alias] = remote
264 except AttributeError as parse_problem: # pragma: no cover
265 raise ValueError(f"{remote} is not a valid URI") from parse_problem
267 @classmethod
268 def from_cfg(cls, config_file: Path) -> "EnderChest":
269 """Parse an EnderChest from its config file
271 Parameters
272 ----------
273 config_file : Path
274 The path to the config file
276 Returns
277 -------
278 EnderChest
279 The resulting EnderChest
281 Raises
282 ------
283 ValueError
284 If the config file at that location cannot be parsed
285 FileNotFoundError
286 If there is no config file at the specified location
287 """
288 INVENTORY_LOGGER.debug("Reading config file from %s", config_file)
289 config = cfg.read_cfg(config_file)
291 # All I'm gonna say is that Windows pathing is the worst
292 path = urlparse(config_file.absolute().parent.parent.as_uri()).path
294 instances: list[i.InstanceSpec] = []
295 remotes: list[str | tuple[str, str]] = []
297 requires_rewrite = False
299 scheme: str | None = None
300 netloc: str | None = None
301 name: str | None = None
302 sync_confirm_wait: str | None = None
303 place_after_open: bool | None = None
304 offer_to_update_symlink_allowlist: bool = True
305 do_not_sync: list[str] | None = None
306 folder_defaults: dict[str, list[str] | None] = {
307 "shulker_box_folders": None,
308 "standard_link_folders": None,
309 "global_link_folders": None,
310 }
312 for section in config.sections():
313 if section == "properties":
314 scheme = config[section].get("sync-protocol")
315 netloc = config[section].get("address")
316 name = config[section].get("name")
317 sync_confirm_wait = config[section].get("sync-confirmation-time")
318 place_after_open = config[section].getboolean("place-after-open")
319 offer_to_update_symlink_allowlist = config[section].getboolean(
320 "offer-to-update-symlink-allowlist", True
321 )
322 if "do-not-sync" in config[section].keys():
323 do_not_sync = cfg.parse_ini_list(
324 config[section]["do-not-sync"] or ""
325 )
326 for setting in folder_defaults.keys():
327 setting_key = setting.replace("_", "-")
328 if setting_key in config[section].keys():
329 folder_defaults[setting] = cfg.parse_ini_list(
330 config[section][setting_key] or ""
331 )
332 elif section == "remotes":
333 for remote in config[section].items():
334 if remote[1] is None:
335 raise ValueError("All remotes must have an alias specified")
336 remotes.append((remote[1], remote[0]))
337 else:
338 # TODO: flag requires_rewrite if instance was normalized
339 instances.append(i.InstanceSpec.from_cfg(config[section]))
341 scheme = scheme or sync.DEFAULT_PROTOCOL
342 netloc = netloc or sync.get_default_netloc()
343 uri = ParseResult(
344 scheme=scheme, netloc=netloc, path=path, params="", query="", fragment=""
345 )
347 ender_chest = EnderChest(uri, name, remotes, instances)
348 if sync_confirm_wait is not None:
349 match sync_confirm_wait.lower():
350 case "true" | "prompt" | "yes" | "confirm":
351 ender_chest.sync_confirm_wait = True
352 case "false" | "no" | "skip":
353 ender_chest.sync_confirm_wait = False
354 case _:
355 try:
356 ender_chest.sync_confirm_wait = int(sync_confirm_wait)
357 except ValueError as bad_input:
358 raise ValueError(
359 "Invalid value for sync-confirmation-time:"
360 f" {sync_confirm_wait}"
361 ) from bad_input
362 if place_after_open is None:
363 INVENTORY_LOGGER.warning(
364 "This EnderChest does not have a value set for place-after-open."
365 "\nIt is being set to False for now. To enable this functionality,"
366 "\nedit the value in %s",
367 config_file,
368 )
369 ender_chest.place_after_open = False
370 requires_rewrite = True
371 else:
372 ender_chest.place_after_open = place_after_open
374 ender_chest.offer_to_update_symlink_allowlist = (
375 offer_to_update_symlink_allowlist
376 )
378 if do_not_sync is not None:
379 ender_chest.do_not_sync = do_not_sync
380 chest_cfg_exclusion = "/".join(
381 (fs.ENDER_CHEST_FOLDER_NAME, fs.ENDER_CHEST_CONFIG_NAME)
382 )
383 if chest_cfg_exclusion not in do_not_sync:
384 INVENTORY_LOGGER.warning(
385 "This EnderChest was not configured to exclude the EnderChest"
386 " config file from sync operations."
387 "\nThat is being fixed now."
388 )
389 ender_chest.do_not_sync.insert(0, chest_cfg_exclusion)
390 requires_rewrite = True
391 for setting in folder_defaults.keys():
392 if folder_defaults[setting] is None:
393 folder_defaults[setting] = dict(_DEFAULTS)[setting] # type: ignore
394 # requires_rewrite = True # though I'm considering it
395 setattr(ender_chest, setting, folder_defaults[setting])
397 if requires_rewrite:
398 ender_chest.write_to_cfg(config_file)
399 return cls.from_cfg(config_file)
400 return ender_chest
402 def write_to_cfg(self, config_file: Path | None = None) -> str:
403 """Write this EnderChest's configuration to INI
405 Parameters
406 ----------
407 config_file : Path, optional
408 The path to the config file, assuming you'd like to write the
409 contents to file
411 Returns
412 -------
413 str
414 An INI-syntax rendering of this EnderChest's config
416 Notes
417 -----
418 The "root" attribute is ignored for this method
419 """
420 properties: dict[str, Any] = {
421 "name": self.name,
422 "address": self._uri.netloc,
423 "sync-protocol": self._uri.scheme,
424 }
425 if self.sync_confirm_wait is True:
426 properties["sync-confirmation-time"] = "prompt"
427 else:
428 properties["sync-confirmation-time"] = self.sync_confirm_wait
430 for setting, _ in _DEFAULTS:
431 if setting == "sync_confirm_wait":
432 continue # already did this one
433 setting_key = setting.replace("_", "-")
434 properties[setting_key] = getattr(self, setting)
436 remotes: dict[str, str] = {name: uri.geturl() for uri, name in self.remotes}
438 instances: dict[str, dict[str, Any]] = {}
440 for instance in self.instances:
441 instances[instance.name] = {
442 "root": instance.root,
443 "minecraft-version": instance.minecraft_versions,
444 "modloader": instance.modloader,
445 "groups": instance.groups_,
446 "tags": instance.tags_,
447 }
449 config = cfg.dumps(
450 fs.ENDER_CHEST_CONFIG_NAME, properties, remotes=remotes, **instances
451 )
453 if config_file:
454 CRAFT_LOGGER.debug("Writing configuration file to %s", config_file)
455 config_file.write_text(config)
456 return config
459def create_ender_chest(minecraft_root: Path, ender_chest: EnderChest) -> None:
460 """Create an EnderChest based on the provided configuration
462 Parameters
463 ----------
464 minecraft_root : Path
465 The root directory that your minecraft stuff is in (or, at least, the
466 one inside which you want to create your EnderChest)
467 ender_chest : EnderChest
468 The spec of the chest to create
470 Notes
471 -----
472 - The "root" attribute of the EnderChest config will be ignored--instead
473 the EnderChest will be created at <minecraft_root>/EnderChest
474 - This method does not check to see if there is already an EnderChest set
475 up at the specified location--if one exists, its config will
476 be overwritten
477 """
478 root = fs.ender_chest_folder(minecraft_root, check_exists=False)
479 root.mkdir(exist_ok=True)
481 config_path = fs.ender_chest_config(minecraft_root, check_exists=False)
482 ender_chest.write_to_cfg(config_path)
483 CRAFT_LOGGER.info(f"EnderChest configuration written to {config_path}")