Coverage for enderchest/remote.py: 86%
85 statements
« prev ^ index » next coverage.py v7.5.0, created at 2024-05-04 01:41 +0000
1"""Higher-level functionality around synchronizing with different EnderCherts"""
3import logging
4from pathlib import Path
5from time import sleep
6from typing import Sequence
7from urllib.parse import ParseResult, urlparse
9from . import filesystem as fs
10from . import inventory, place
11from .enderchest import EnderChest
12from .loggers import IMPORTANT, SYNC_LOGGER
13from .prompt import confirm
14from .sync import abspath_from_uri, pull, push, remote_file, render_remote
17def load_remote_ender_chest(uri: str | ParseResult) -> EnderChest:
18 """Load an EnderChest configuration from another machine
20 Parameters
21 ----------
22 uri : URI
23 The URI to the remote Minecraft root
25 Returns
26 -------
27 EnderChest
28 The remote EnderChest configuration
30 Raises
31 ------
32 ValueError
33 If the provided URI is invalid
34 RuntimeError
35 If the config from the remote EnderChest could not be parsed
36 """
37 try:
38 uri = uri if isinstance(uri, ParseResult) else urlparse(uri)
39 except AttributeError as bad_uri:
40 raise ValueError(f"{uri} is not a valid URI") from bad_uri
42 remote_root = Path(uri.path)
43 remote_config_path = fs.ender_chest_config(remote_root, check_exists=False)
44 uri = uri._replace(path=remote_config_path.as_posix())
46 with remote_file(uri) as remote_config:
47 try:
48 return EnderChest.from_cfg(remote_config)
49 except ValueError as bad_chest:
50 raise RuntimeError(
51 "The remote EnderChest config downloaded"
52 f"from {uri.geturl()} could not be parsed."
53 ) from bad_chest
56def fetch_remotes_from_a_remote_ender_chest(
57 uri: str | ParseResult,
58) -> list[tuple[ParseResult, str]]:
59 """Grab the list of EnderChests registered with the specified remote EnderChest
61 Parameters
62 ----------
63 uri : URI
64 The URI to the remote Minecraft root
66 Returns
67 -------
68 list of (URI, str) tuples
69 The URIs of the remote EnderChests, paired with their aliases
71 Raises
72 ------
73 RuntimeError
74 If the remote list could not be pulled
75 """
76 remote_chest = load_remote_ender_chest(uri)
77 remotes: list[tuple[ParseResult, str]] = [
78 (urlparse(uri) if isinstance(uri, str) else uri, remote_chest.name)
79 ]
81 remotes.extend(remote_chest.remotes)
82 SYNC_LOGGER.info(
83 "Loaded the following remotes:\n %s",
84 "\n".join(f" - {render_remote(alias, uri)}" for uri, alias in remotes),
85 )
87 if len(set(alias for _, alias in remotes)) != len(remotes):
88 raise RuntimeError(
89 f"There are duplicates aliases in the list of remotes pulled from {uri}"
90 )
91 return remotes
94def sync_with_remotes(
95 minecraft_root: Path,
96 pull_or_push: str,
97 dry_run: bool = False,
98 sync_confirm_wait: bool | int | None = None,
99 **sync_kwargs,
100) -> None:
101 """Pull changes from or push changes to remote EnderChests
103 Parameters
104 ----------
105 minecraft_root : Path
106 The root directory that your minecraft stuff (or, at least, the one
107 that's the parent of your EnderChest folder). This will be used to
108 construct relative paths.
109 pull_or_push : str
110 "pull" or "push"
111 dry_run: bool, optional
112 Perform a dry run of the sync operation, reporting the operations\
113 that will be performed but not actually carrying them out
114 sync_confirm_wait : bool or int, optional
115 The default behavior when syncing EnderChests is to first perform a dry
116 run of every sync operation and then wait 5 seconds before proceeding with the
117 real sync. The idea is to give the user time to interrupt the sync if
118 the dry run looks wrong. This can be changed by either raising or lowering
119 the value of confirm, by disabling the dry-run-first behavior entirely
120 (`confirm=False`) or by requiring that the user explicitly confirms
121 the sync (`confirm=True`). This default behavior can also be modified
122 in the EnderChest config. This parameter will be ignored when performing
123 a dry run.
124 sync_kwargs
125 Any additional arguments that should be passed into the syncing
126 operation
128 Notes
129 -----
130 - When pulling changes, this method will try each remote in the order they
131 are configured and stop once it has successfully pulled from a remote.
133 This method will attempt to push local changes to *every* remote
134 """
135 if pull_or_push not in ("pull", "push"):
136 raise ValueError(
137 'Invalid choice for sync operation. Choices are "pull" and "push"'
138 )
139 try:
140 if sync_confirm_wait is None:
141 sync_confirm_wait = inventory.load_ender_chest(
142 minecraft_root
143 ).sync_confirm_wait
144 this_chest = inventory.load_ender_chest(minecraft_root)
146 # I know this is redundant, but we want those logs
147 remotes = inventory.load_ender_chest_remotes(
148 minecraft_root, log_level=logging.DEBUG
149 )
150 except (FileNotFoundError, ValueError) as bad_chest:
151 SYNC_LOGGER.error(
152 f"Could not load EnderChest from {minecraft_root}:\n {bad_chest}"
153 )
154 return
155 if not remotes:
156 SYNC_LOGGER.error("EnderChest has no remotes. Aborting.")
157 return # kinda unnecessary
159 synced_somewhere = False
160 exclusions: Sequence[str] = sync_kwargs.pop("exclude", None) or ()
161 for remote_uri, alias in remotes:
162 if dry_run:
163 runs: tuple[bool, ...] = (True,)
164 elif sync_confirm_wait is False or sync_confirm_wait <= 0:
165 runs = (False,)
166 else:
167 runs = (True, False)
168 for do_dry_run in runs:
169 if dry_run:
170 prefix = "Simulating an attempt"
171 else:
172 prefix = "Attempting"
173 try:
174 remote_chest = load_remote_ender_chest(remote_uri)
175 if pull_or_push == "pull":
176 SYNC_LOGGER.log(
177 IMPORTANT,
178 f"{prefix} to pull changes from %s",
179 render_remote(alias, remote_uri),
180 )
181 remote_chest_folder = remote_uri._replace(
182 path=urlparse(
183 (
184 fs.ender_chest_folder(
185 abspath_from_uri(remote_uri),
186 check_exists=False,
187 )
188 ).as_uri()
189 ).path
190 )
191 pull(
192 remote_chest_folder,
193 minecraft_root,
194 exclude={
195 *this_chest.do_not_sync,
196 *remote_chest.do_not_sync,
197 *exclusions,
198 },
199 dry_run=do_dry_run,
200 **sync_kwargs,
201 )
202 else:
203 SYNC_LOGGER.log(
204 IMPORTANT,
205 f"{prefix} to push changes"
206 f" to {render_remote(alias, remote_uri)}",
207 )
208 local_chest = fs.ender_chest_folder(minecraft_root)
209 push(
210 local_chest,
211 remote_uri,
212 exclude={
213 *this_chest.do_not_sync,
214 *remote_chest.do_not_sync,
215 *exclusions,
216 },
217 dry_run=do_dry_run,
218 **sync_kwargs,
219 )
220 except (
221 FileNotFoundError,
222 ValueError,
223 NotImplementedError,
224 TimeoutError,
225 RuntimeError,
226 ) as sync_fail:
227 SYNC_LOGGER.warning(
228 f"Could not sync changes with {render_remote(alias, remote_uri)}:"
229 f"\n {sync_fail}"
230 )
231 break
232 if do_dry_run == runs[-1]:
233 continue
234 if sync_confirm_wait is True:
235 if not confirm(default=True):
236 SYNC_LOGGER.error("Aborting")
237 return
238 else:
239 SYNC_LOGGER.debug(f"Waiting for {sync_confirm_wait} seconds")
240 sleep(sync_confirm_wait)
241 else:
242 synced_somewhere = True
243 if pull_or_push == "pull":
244 if this_chest.place_after_open and not dry_run:
245 place.place_ender_chest(
246 minecraft_root,
247 keep_broken_links=False,
248 keep_stale_links=False,
249 error_handling="abort",
250 relative=False,
251 )
252 break
253 if not synced_somewhere:
254 SYNC_LOGGER.error("Could not sync with any remote EnderChests")