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