diff --git a/CHANGELOG.md b/CHANGELOG.md index 607c6d9..0eaecfe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,60 @@ # Changelog +## [0.3.1] - 2026-05-21 + +### Added +- `pos3` console-script entry point with `ls`, `download`, `upload` subcommands + ([#10](https://github.com/Positronic-Robotics/pos3/issues/10)). + - `pos3 ls [-r] [--profile NAME]` lists objects, one full `s3://` URL + per line on stdout. + - `pos3 download [--local PATH] [--delete] [--exclude PATTERN]... [--profile NAME]` + prints only the resulting local path to stdout; progress and logs go to + stderr, so `data_dir=$(pos3 download s3://bucket/dataset/)` is safe. + - `pos3 upload [--local PATH] [--delete] [--exclude PATTERN]... [--profile NAME]` + is one-shot (no background loop or interval). Source defaults to the cache + path `pos3 download` would have produced; errors if the source doesn't + exist. + - `--delete` defaults OFF for both `download` and `upload` (the Python API + defaults to `True`; the CLI is more conservative for interactive use). + - `--profile` is supported alongside the URL form `s3://@bucket/...`; + the URL form wins on conflict, matching the Python precedence. + - `-n` / `--dry-run` on `download` and `upload` prints the planned per-file + actions to stdout (in `aws s3 sync --dryrun` style) and performs no + transfers, no deletes, and no directory creation. + - `download` and `upload` require an `s3://` URL. The Python API's + local-path passthrough still works in code; the CLI rejects non-S3 + inputs with a clear error so a typo can't silently succeed. `ls` is + unchanged and still accepts both forms. +- `pos3.TransferPlan` dataclass plus module-level `pos3.plan_download(remote, ...)` + and `pos3.plan_upload(remote, ...)` wrappers: compute the set of + `(source, destination)` copies and target deletes a real call would + perform, without performing any of them. Same calling pattern as + `pos3.download` / `pos3.upload` — use them inside a `with pos3.mirror():` + block. The CLI's `-n` / `--dry-run` is implemented on top of these. +- `TransferError` and `TransferPlan` are now in `pos3.__all__`. + +### Changed +- Per-object transfer failures now raise `pos3.TransferError` instead of + being logged and swallowed. Previously, if any worker in a download or + upload batch failed, the error was sent to the logger and the call + returned normally — `data_dir = pos3.download(...)` could return a path + to a partial cache, and `pos3 download` exited 0 after a failed S3 GET. + Both now propagate. The new exception exposes `.operation` and + `.failures` (list of the underlying per-worker exceptions). The CLI + catches it and exits 1 with the failure on stderr. **Background + interval syncs** (`upload(..., interval=N)`) are best-effort: a + `TransferError` from one tick is logged and the daemon continues so + the next interval can retry. Only the final sync on context exit and + any one-shot call propagate. **Cleanup on error** — when the + `mirror()` body is unwinding with an exception, a `TransferError` from + the `sync_on_error=True` cleanup sync is logged but swallowed, so the + original application exception remains the visible cause. +- `pos3.mirror()` no longer creates the cache root directory eagerly on + context entry. The leaf directory is still created on demand when a + file is actually downloaded, so the visible behavior of `download()` is + unchanged. Dry-run and `plan_*` paths are now genuinely side-effect + free on the local filesystem. + ## [0.3.0] - 2026-05-19 ### Added diff --git a/README.md b/README.md index d207e67..e3d5477 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,62 @@ Lists files/objects in a directory or S3 prefix. **Returns**: List of full S3 URLs or local paths. +## CLI + +`pos3` ships a small command-line interface for the most common one-shot +operations. After `uv pip install pos3` (or `pip install pos3`), `pos3` is on +your `$PATH`: + +```bash +# List objects (one full s3:// URL per line on stdout) +pos3 ls s3://bucket/dataset/ +pos3 ls -r s3://bucket/dataset/ + +# Download an S3 prefix or object into the cache (or a custom --local path). +# Only the resulting local path is written to stdout — progress and logs go +# to stderr — so it's safe to capture in a shell variable: +data_dir=$(pos3 download s3://bucket/dataset/) + +# One-shot upload. Source defaults to the same cache path `pos3 download` +# would have produced; --local overrides. Errors if the source doesn't exist. +pos3 upload s3://bucket/results/ --local ./out/ + +# Preview what download/upload would do, without touching anything. +pos3 download -n s3://bucket/dataset/ --local ./data/ --delete +pos3 upload -n s3://bucket/results/ --local ./out/ +``` + +All three subcommands accept `--profile NAME`. The URL form +`s3://@bucket/...` takes precedence over `--profile` on conflict +(matching the Python API). + +`--delete` defaults to **OFF** for both `download` and `upload`, even though +the Python API defaults to `True`. CLI defaults are conservative for +interactive shell use; pass `--delete` explicitly to mirror file removals. + +`-n` / `--dry-run` is accepted on `download` and `upload` (not `ls`). It +prints per-file plan lines to stdout in `aws s3 sync --dryrun` style and +performs no transfers, no deletes, and no local directory creation. + +The same plan is available from Python via `pos3.plan_download` and +`pos3.plan_upload`, each returning a `pos3.TransferPlan` with +`to_copy: list[tuple[str, str]]` and `to_delete: list[str]`: + +```python +with pos3.mirror(): + plan = pos3.plan_download("s3://bucket/dataset/", local="./data/") + for src, dst in plan.to_copy: + print(f"would download {src} → {dst}") +``` + +`download` and `upload` require an `s3://` URL; non-S3 inputs are rejected +with a non-zero exit. `ls` still accepts both `s3://` prefixes and local +paths, matching the Python API. + +The CLI is one-shot only — no background sync, no `pos3 sync` subcommand. +Use the Python `pos3.mirror()` context manager when you need an interval-based +loop or bi-directional sync over a job's lifetime. + ## Comparison with Libraries Why use `pos3` instead of other Python libraries? diff --git a/pos3/__init__.py b/pos3/__init__.py index 6892179..121256b 100644 --- a/pos3/__init__.py +++ b/pos3/__init__.py @@ -80,12 +80,56 @@ def _make_s3_key(prefix: str, info: FileInfo) -> str: return key +class TransferError(Exception): + """Raised when one or more S3 transfer or delete workers failed. + + ``failures`` carries the per-worker exceptions; the message names the + operation (Download / Upload / Delete) and the failure count. + """ + + def __init__(self, operation: str, failures: list[BaseException]): + super().__init__(f"{operation}: {len(failures)} worker(s) failed: {failures[0]}") + self.operation = operation + self.failures = failures + + +@dataclass(frozen=True) +class TransferPlan: + """The set of operations a download/upload *would* perform, if executed. + + Returned by :meth:`_Mirror.plan_download` and :meth:`_Mirror.plan_upload`. + + ``to_copy`` is a list of ``(source, destination)`` pairs as strings — + each pair is one ``s3://bucket/key`` URL and one local filesystem path, + direction depending on whether the plan is for download or upload. + ``to_delete`` is a list of destination URLs / paths that ``delete=True`` + on the corresponding call would remove. Directory-only entries from + the underlying scan are filtered out so the plan only reflects file + operations. + """ + + to_copy: list[tuple[str, str]] + to_delete: list[str] + + def _process_futures(futures, operation: str) -> None: + """Drain ``futures`` and raise :class:`TransferError` if any worker failed. + + Drains all futures before raising, so a 1-of-N failure does not cancel + the rest of the batch (preserves best-effort completion). Callers that + need to keep going across failures wrap this in ``try / except`` — the + only such caller today is :meth:`_Mirror._background_worker`, which + treats interval syncs as retry-next-tick. + """ + failures: list[BaseException] = [] for future in futures: try: future.result() except Exception as exc: logger.error("%s failed: %s", operation, exc) + failures.append(exc) + if failures: + raise TransferError(operation, failures) @dataclass(frozen=True) @@ -210,7 +254,7 @@ def __eq__(self, other): @dataclass class _UploadRegistration: - remote: str + remote: str # normalized form — used as the registration / dedup key local_path: Path interval: int | None delete: bool @@ -218,6 +262,11 @@ class _UploadRegistration: exclude: list[str] | None profile: Profile | None = None last_sync: float = 0.0 + # The user's original URL, preserving trailing-slash intent. _sync_uploads + # parses it raw so `s3://bucket/data/` skips head_object('data') in + # _list_s3_objects and scans the directory as the user meant. NOT + # compared in __eq__ — only the normalized form determines identity. + raw_remote: str = "" def __eq__(self, other): if not isinstance(other, _UploadRegistration): @@ -241,8 +290,11 @@ def __eq__(self, other): class _Mirror: def __init__(self, options: _Options): self.options = options + # cache_root is resolved eagerly but NOT created — the per-file + # workers (_put_locally) mkdir the leaf directory on demand. Keeping + # the constructor side-effect free lets dry-run / planning paths + # construct a Mirror without mutating the filesystem. self.cache_root = Path(self.options.cache_root).expanduser().resolve() - self.cache_root.mkdir(parents=True, exist_ok=True) self._default_profile = options.default_profile self._clients: dict[Profile | None, Any] = {} @@ -318,6 +370,7 @@ def download( Raises: FileNotFoundError: If remote is a local path that does not exist. ValueError: If download registration conflicts with an existing download or upload or parameters differ. + TransferError: If any object failed to download (was previously logged and swallowed). """ effective_profile = self._effective_profile(profile, remote) @@ -351,7 +404,9 @@ def download( if need_download: try: - self._perform_download(normalized, local_path, delete, exclude, effective_profile) + # Pass the raw URL — _perform_download preserves trailing + # slashes for scanning while normalizing for output keys. + self._perform_download(remote, local_path, delete, exclude, effective_profile) except Exception as exc: registration.error = exc registration.ready.set() @@ -394,6 +449,8 @@ def upload( Raises: ValueError: If upload registration conflicts with an existing download or upload or parameters differ. + TransferError: Raised from the mirror context exit (or background sync) if any object failed to upload + or delete (was previously logged and swallowed). """ effective_profile = self._effective_profile(profile, remote) @@ -418,6 +475,7 @@ def upload( exclude=exclude, profile=effective_profile, last_sync=0, + raw_remote=remote, ) with self._lock: @@ -435,6 +493,109 @@ def upload( return local_path + def plan_download( + self, + remote: str, + local: str | Path | None = None, + exclude: list[str] | None = None, + profile: str | Profile | None = None, + ) -> TransferPlan: + """Compute the :class:`TransferPlan` ``download()`` would execute. + + Reads from S3 and the local filesystem but performs no transfers, + no deletes, and no directory creation. Raises ``ValueError`` if + ``remote`` is not an ``s3://`` URL. + """ + if not _is_s3_path(remote): + raise ValueError(f"plan_download requires an s3:// URL, got: {remote}") + effective_profile = self._effective_profile(profile, remote) + local_path = ( + self.options.cache_path_for(remote, effective_profile) + if local is None + else Path(local).expanduser().resolve() + ) + # Two prefixes on purpose: the raw one preserves the user's + # trailing-slash intent so _list_s3_objects skips its exact-key + # head_object probe (matters when both `data` and `data/...` exist); + # the normalized one feeds _make_s3_key so reconstructed output keys + # don't get a double slash. _scan_s3 strips len(scan_prefix) then + # lstrip("/"), so the same FileInfo.relative_path drops out either + # way. + scan_bucket, scan_prefix = _parse_s3_url(remote) + bucket, out_prefix = _parse_s3_url(_normalize_s3_url(remote)) + to_copy, to_delete = _compute_sync_diff( + _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, effective_profile), exclude), + _filter_fileinfo(_scan_local(local_path), exclude), + ) + copies: list[tuple[str, str]] = [] + for info in to_copy: + if info.is_dir: + continue + s3_key = _make_s3_key(out_prefix, info) + dst = local_path / info.relative_path if info.relative_path else local_path + copies.append((f"s3://{bucket}/{s3_key}", str(dst))) + deletes: list[str] = [] + for info in to_delete: + if info.is_dir: + continue + target = local_path / info.relative_path if info.relative_path else local_path + deletes.append(str(target)) + return TransferPlan(to_copy=copies, to_delete=deletes) + + def plan_upload( + self, + remote: str, + local: str | Path | None = None, + exclude: list[str] | None = None, + profile: str | Profile | None = None, + ) -> TransferPlan: + """Compute the :class:`TransferPlan` ``upload()`` would execute. + + Reads from S3 and the local filesystem but performs no transfers, + no deletes, and no directory creation. Raises ``ValueError`` if + ``remote`` is not an ``s3://`` URL. A missing local source yields + an empty ``to_copy``. + """ + if not _is_s3_path(remote): + raise ValueError(f"plan_upload requires an s3:// URL, got: {remote}") + effective_profile = self._effective_profile(profile, remote) + source = ( + self.options.cache_path_for(remote, effective_profile) + if local is None + else Path(local).expanduser().resolve() + ) + # Mirror real upload() behavior: _sync_uploads skips any registration + # whose local_path does not exist (does nothing — not even deletes). + # Without this short-circuit, _scan_local yields nothing while + # _scan_s3 yields remote objects, so _compute_sync_diff would + # falsely report every remote object as "would delete" — a plan the + # real call would never execute. + if not source.exists(): + return TransferPlan(to_copy=[], to_delete=[]) + # Two prefixes on purpose — see plan_download for the rationale. The + # raw prefix preserves trailing-slash intent for the S3 scan; the + # normalized one builds collision-free output keys. + scan_bucket, scan_prefix = _parse_s3_url(remote) + bucket, out_prefix = _parse_s3_url(_normalize_s3_url(remote)) + to_copy, to_delete = _compute_sync_diff( + _filter_fileinfo(_scan_local(source), exclude), + _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, effective_profile), exclude), + ) + copies: list[tuple[str, str]] = [] + for info in to_copy: + if info.is_dir: + continue + s3_key = _make_s3_key(out_prefix, info) + src = source / info.relative_path if info.relative_path else source + copies.append((str(src), f"s3://{bucket}/{s3_key}")) + deletes: list[str] = [] + for info in to_delete: + if info.is_dir: + continue + s3_key = _make_s3_key(out_prefix, info) + deletes.append(f"s3://{bucket}/{s3_key}") + return TransferPlan(to_copy=copies, to_delete=deletes) + def sync( self, remote: str, @@ -462,23 +623,40 @@ def ls(self, prefix: str, recursive: bool = False, profile: str | Profile | None effective_profile = self._effective_profile(profile, prefix) if _is_s3_path(prefix): - normalized = _normalize_s3_url(prefix) - bucket, key = _parse_s3_url(normalized) - # Ensure directory-like listing by appending '/' to avoid spurious prefix matches - if key: - key = key + "/" + # Use _parse_s3_url directly, NOT _normalize_s3_url: the latter + # strips trailing slashes, but `s3://bucket/data/` vs + # `s3://bucket/data` carries user intent here — the trailing + # slash means "treat as a directory prefix". _list_s3_objects + # respects that ("if key and not key.endswith('/')") so we + # preserve it. Otherwise a key collision (an exact object named + # `data` plus a `data/` directory) would silently hide the + # directory contents. + bucket, key = _parse_s3_url(prefix) + # Don't force a trailing "/" here. _list_s3_objects probes the + # key as an exact object first (via head_object) and only falls + # back to directory listing with a trailing "/" on 404. Forcing + # the slash here suppresses the exact-key probe, so + # `pos3 ls s3://bucket/results.json` would return nothing for an + # existing object. The "droid/recovery" vs "droid/recovery_towels" + # spurious-prefix case is already covered there. items = [] for info in self._scan_s3(bucket, key, effective_profile): - if info.relative_path: - # Skip nested items if not recursive - if not recursive and "/" in info.relative_path: - continue - # Reconstruct the full S3 key - if key: - s3_key = key.rstrip("/") + "/" + info.relative_path - else: - s3_key = info.relative_path - items.append(f"s3://{bucket}/{s3_key}") + if not info.relative_path: + # Empty relative_path means either the root directory + # marker (skip) or that ``key`` was the exact object key + # — yield the input URL as a single-line result. + if not info.is_dir: + items.append(f"s3://{bucket}/{key}") + continue + # Skip nested items if not recursive + if not recursive and "/" in info.relative_path: + continue + # Reconstruct the full S3 key + if key: + s3_key = key.rstrip("/") + "/" + info.relative_path + else: + s3_key = info.relative_path + items.append(f"s3://{bucket}/{s3_key}") return items else: display_path = Path(prefix).expanduser() @@ -526,14 +704,43 @@ def _background_worker(self) -> None: registration.last_sync = now due.append(registration) - self._sync_uploads(due) + try: + self._sync_uploads(due) + except Exception as exc: + # Background interval syncs are best-effort: a TransferError + # (or anything else) here must not kill the daemon thread — + # the next tick will retry. _final_sync still propagates on + # context exit so one-shot callers see definitive failures. + logger.error("Background sync iteration failed: %s", exc) def _final_sync(self, had_error: bool = False) -> None: with self._lock: uploads = list(self._uploads.values()) if had_error: + # The mirror() context is already unwinding with the caller's + # exception. Anything that goes wrong in cleanup must be + # logged and swallowed so the original exception stays the + # visible cause. The exception type can be: + # - TransferError (worker put/delete failure, after + # _sync_uploads built futures) + # - ClientError / BotoCoreError (_scan_s3 → head_object / + # paginate, BEFORE any worker future exists) + # - OSError on the local fs, etc. + # Catching broad Exception here is the right call: the + # `had_error=True` path is fundamentally "best-effort, do no + # more harm." The clean-exit path (had_error=False) still + # propagates so one-shot callers see definitive failures. uploads = [u for u in uploads if u.sync_on_error] - self._sync_uploads(uploads) + try: + self._sync_uploads(uploads) + except Exception as exc: + logger.error( + "Cleanup sync after error failed; original exception preserved: %s: %s", + type(exc).__name__, + exc, + ) + else: + self._sync_uploads(uploads) def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None: tasks: list[tuple[str, Path, bool, list[str] | None, Profile | None]] = [] @@ -541,7 +748,11 @@ def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None: if registration.local_path.exists(): tasks.append( ( - registration.remote, + # raw_remote preserves the user's trailing slash so + # _list_s3_objects skips its exact-key head_object + # probe; out-key building below normalizes to avoid + # double slashes. + registration.raw_remote or registration.remote, registration.local_path, registration.delete, registration.exclude, @@ -558,19 +769,20 @@ def _sync_uploads(self, registrations: Iterable[_UploadRegistration]) -> None: for remote, local_path, delete, exclude, profile in tasks: logger.debug("Syncing upload: %s from %s (delete=%s)", remote, local_path, delete) - bucket, prefix = _parse_s3_url(remote) + scan_bucket, scan_prefix = _parse_s3_url(remote) + bucket, out_prefix = _parse_s3_url(_normalize_s3_url(remote)) to_copy, to_delete = _compute_sync_diff( _filter_fileinfo(_scan_local(local_path), exclude), - _filter_fileinfo(self._scan_s3(bucket, prefix, profile), exclude), + _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, profile), exclude), ) for info in to_copy: - s3_key = _make_s3_key(prefix, info) + s3_key = _make_s3_key(out_prefix, info) to_put.append((info, local_path, bucket, s3_key, profile)) total_bytes += info.size for info in to_delete if delete else []: - s3_key = _make_s3_key(prefix, info) + s3_key = _make_s3_key(out_prefix, info) to_remove.append((bucket, s3_key, profile)) if to_put: @@ -608,16 +820,21 @@ def _perform_download( exclude: list[str] | None, profile: Profile | None = None, ) -> None: - bucket, prefix = _parse_s3_url(remote) + # Dual-prefix: raw remote preserves the user's trailing slash so + # _list_s3_objects skips its exact-key head_object probe (matters + # when both `data` and `data/` exist); normalized remote feeds + # _make_s3_key so output keys don't get double slashes. + scan_bucket, scan_prefix = _parse_s3_url(remote) + bucket, out_prefix = _parse_s3_url(_normalize_s3_url(remote)) logger.debug( "Performing download: s3://%s/%s to %s (delete=%s)", - bucket, - prefix, + scan_bucket, + scan_prefix, local_path, delete, ) to_copy, to_delete = _compute_sync_diff( - _filter_fileinfo(self._scan_s3(bucket, prefix, profile), exclude), + _filter_fileinfo(self._scan_s3(scan_bucket, scan_prefix, profile), exclude), _filter_fileinfo(_scan_local(local_path), exclude), ) @@ -626,7 +843,7 @@ def _perform_download( total_bytes = 0 for info in to_copy: - s3_key = _make_s3_key(prefix, info) + s3_key = _make_s3_key(out_prefix, info) to_put.append((info, bucket, s3_key, local_path)) total_bytes += info.size @@ -691,6 +908,13 @@ def _scan_s3(self, bucket: str, prefix: str, profile: Profile | None = None) -> logger.debug("Scanning S3: s3://%s/%s", bucket, prefix) seen_dirs: set[str] = set() has_content = False + # When the prefix matches an exact object (head_object hit in + # _list_s3_objects), we emit a file FileInfo with relative_path="". + # In that case the root-dir marker below would collide with it in + # _compute_sync_diff's dict-by-relative_path and silently win, + # causing download() to mkdir the local path instead of fetching + # the object. Track this to suppress the redundant marker. + has_root_file = False for obj in self._list_s3_objects(bucket, prefix, profile): has_content = True @@ -703,6 +927,8 @@ def _scan_s3(self, bucket: str, prefix: str, profile: Profile | None = None) -> yield FileInfo(relative_path=relative, size=0, is_dir=True) seen_dirs.add(relative) else: + if relative == "": + has_root_file = True yield FileInfo(relative_path=relative, size=obj["Size"], is_dir=False) if "/" in relative: @@ -713,7 +939,7 @@ def _scan_s3(self, bucket: str, prefix: str, profile: Profile | None = None) -> yield FileInfo(relative_path=dir_path, size=0, is_dir=True) seen_dirs.add(dir_path) - if has_content: + if has_content and not has_root_file: yield FileInfo( relative_path="", size=0, is_dir=True ) # Yield root directory marker for symmetry with _scan_local @@ -952,4 +1178,52 @@ def ls(prefix: str, recursive: bool = False, profile: str | Profile | None = Non return mirror_obj.ls(prefix, recursive, profile) -__all__ = ["mirror", "with_mirror", "download", "upload", "sync", "ls", "register_profile", "Profile", "_parse_s3_url"] +def plan_download( + remote: str, + local: str | Path | None = None, + exclude: list[str] | None = None, + profile: str | Profile | None = None, +) -> TransferPlan: + """Return the :class:`TransferPlan` ``download()`` would execute. + + Side-effect free: reads from S3 and the local filesystem but performs + no transfers, deletes, or directory creation. Raises ``ValueError`` if + ``remote`` is not an ``s3://`` URL. + """ + mirror_obj = _require_active_mirror() + return mirror_obj.plan_download(remote, local, exclude, profile) + + +def plan_upload( + remote: str, + local: str | Path | None = None, + exclude: list[str] | None = None, + profile: str | Profile | None = None, +) -> TransferPlan: + """Return the :class:`TransferPlan` ``upload()`` would execute. + + Side-effect free: reads from S3 and the local filesystem but performs + no transfers, deletes, or directory creation. Raises ``ValueError`` if + ``remote`` is not an ``s3://`` URL. Returns an empty plan when the + local source does not exist (matches real upload() behavior, which + skips registrations with a missing local_path). + """ + mirror_obj = _require_active_mirror() + return mirror_obj.plan_upload(remote, local, exclude, profile) + + +__all__ = [ + "mirror", + "with_mirror", + "download", + "upload", + "sync", + "ls", + "plan_download", + "plan_upload", + "register_profile", + "Profile", + "TransferError", + "TransferPlan", + "_parse_s3_url", +] diff --git a/pos3/cli.py b/pos3/cli.py new file mode 100644 index 0000000..7378496 --- /dev/null +++ b/pos3/cli.py @@ -0,0 +1,234 @@ +"""Command-line interface for pos3. + +Three subcommands — ``ls``, ``download``, ``upload`` — each running inside a +short-lived ``pos3.mirror()`` context. ``upload`` is one-shot: no background +interval, no sync loop. + +CLI defaults intentionally differ from the Python API: ``--delete`` defaults +to OFF for both ``download`` and ``upload``, because the API's ``True`` +default is too destructive for interactive shell use. ``download`` writes +only the resulting local path to stdout so the output is safe to capture in +``$(pos3 download ...)``; progress bars and logs go to stderr. + +``--dry-run`` / ``-n`` (download and upload only) prints the planned +per-file actions to stdout in ``aws s3 sync --dryrun`` style and performs +no transfers and no deletes. (The cache root directory is initialized the +same way it is for any pos3 invocation.) +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +from botocore.exceptions import BotoCoreError, ClientError + +from . import ( + TransferError, + _is_s3_path, + _require_active_mirror, + download, + ls, + mirror, + plan_download, + plan_upload, + upload, +) +from .profiles import _resolve_profile, _url_profile + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="pos3", description="pos3 command-line interface.") + subparsers = parser.add_subparsers(dest="command", required=True) + + def add_profile(p: argparse.ArgumentParser) -> None: + p.add_argument( + "--profile", + metavar="NAME", + help="Named pos3 profile. Overridden by the URL form s3://@bucket/key.", + ) + + def add_transfer_args(p: argparse.ArgumentParser, local_help: str) -> None: + p.add_argument("--local", metavar="PATH", help=local_help) + p.add_argument( + "--delete", + action="store_true", + help="Delete files not present at the other end. Defaults to OFF in the CLI.", + ) + p.add_argument( + "--exclude", + metavar="PATTERN", + action="append", + default=None, + help="Glob pattern to skip. May be passed multiple times.", + ) + p.add_argument( + "-n", + "--dry-run", + action="store_true", + help="Print the planned actions and exit without transferring or deleting anything.", + ) + add_profile(p) + + p_ls = subparsers.add_parser("ls", help="List objects under a prefix.") + p_ls.add_argument("prefix", help="S3 prefix (s3://bucket/key) or local path.") + p_ls.add_argument("-r", "--recursive", action="store_true", help="List subdirectories recursively.") + add_profile(p_ls) + + p_dl = subparsers.add_parser("download", help="Download an S3 prefix or object to a local path.") + p_dl.add_argument("url", help="Source S3 URL (s3://bucket/key).") + add_transfer_args(p_dl, local_help="Destination path. Defaults to the cache path.") + + p_up = subparsers.add_parser("upload", help="One-shot upload of a local path to S3.") + p_up.add_argument("url", help="Destination S3 URL (s3://bucket/key).") + add_transfer_args(p_up, local_help="Source path. Defaults to the cache path.") + + return parser + + +def _cmd_ls(args: argparse.Namespace) -> int: + with mirror(show_progress=False): + for item in ls(args.prefix, recursive=args.recursive, profile=args.profile): + print(item) + return 0 + + +def _cmd_download(args: argparse.Namespace) -> int: + if not _is_s3_path(args.url): + print( + f"pos3 download: url must be an s3:// URL, got: {args.url}", + file=sys.stderr, + ) + return 1 + if args.dry_run: + with mirror(show_progress=False): + _print_download_plan(args) + return 0 + with mirror(show_progress=True): + local_path = download( + args.url, + local=args.local, + delete=args.delete, + exclude=args.exclude, + profile=args.profile, + ) + print(str(local_path)) + return 0 + + +def _resolve_upload_source(args: argparse.Namespace) -> Path | None: + """Resolve the local source path, mirroring the precedence used by upload(). + + Prints an error and returns None if the source path does not exist. + """ + mirror_obj = _require_active_mirror() + if args.local: + source = Path(args.local).expanduser().resolve() + else: + # Resolve the same profile precedence (URL > --profile > context default) + # the upload() call will use, so the cache path we check matches the one + # upload() would target. + url_profile = _url_profile(args.url) + profile = url_profile if url_profile is not None else args.profile + effective_profile = _resolve_profile(profile) or mirror_obj.options.default_profile + source = mirror_obj.options.cache_path_for(args.url, effective_profile) + if not source.exists(): + print(f"pos3 upload: source path does not exist: {source}", file=sys.stderr) + return None + return source + + +def _cmd_upload(args: argparse.Namespace) -> int: + if not _is_s3_path(args.url): + print( + f"pos3 upload: url must be an s3:// URL, got: {args.url}", + file=sys.stderr, + ) + return 1 + with mirror(show_progress=not args.dry_run): + source = _resolve_upload_source(args) + if source is None: + return 1 + if args.dry_run: + _print_upload_plan(args, source) + return 0 + upload( + args.url, + local=source, + interval=None, + delete=args.delete, + exclude=args.exclude, + profile=args.profile, + ) + return 0 + + +def _print_download_plan(args: argparse.Namespace) -> None: + plan = plan_download( + args.url, + local=args.local, + exclude=args.exclude, + profile=args.profile, + ) + for src, dst in plan.to_copy: + print(f"download: {src} to {dst}") + if args.delete: + for target in plan.to_delete: + print(f"delete: {target}") + + +def _print_upload_plan(args: argparse.Namespace, source: Path) -> None: + plan = plan_upload( + args.url, + local=str(source), + exclude=args.exclude, + profile=args.profile, + ) + for src, dst in plan.to_copy: + print(f"upload: {src} to {dst}") + if args.delete: + for target in plan.to_delete: + print(f"delete: {target}") + + +_COMMANDS = { + "ls": _cmd_ls, + "download": _cmd_download, + "upload": _cmd_upload, +} + + +def main(argv: list[str] | None = None) -> int: + parser = _build_parser() + # Use parse_known_args so we can re-emit the error through the SUBPARSER + # for the chosen command. argparse's default routes "unrecognized + # arguments" to the top-level parser, which prints + # usage: pos3 [-h] {ls,download,upload} ... + # — useless when the user typo'd a subcommand flag (e.g. `--dry_run`) + # because they can't see the flags the subcommand actually exposes. + namespace, leftover = parser.parse_known_args(argv) + if leftover: + subparsers_action = next( + a for a in parser._actions if isinstance(a, argparse._SubParsersAction) + ) + sub = subparsers_action.choices.get(namespace.command, parser) + sub.error(f"unrecognized arguments: {' '.join(leftover)}") + args = namespace + try: + return _COMMANDS[args.command](args) + except (ValueError, TransferError) as exc: + print(f"pos3 {args.command}: {exc}", file=sys.stderr) + return 1 + except (BotoCoreError, ClientError) as exc: + # boto3/botocore failures from S3 calls outside _process_futures — + # access denied, missing bucket, expired creds, throttling, etc. + # _scan_s3 (called by ls and the pre-transfer scan in download / + # upload / plan_*) re-raises non-404 ClientErrors directly, so they + # would otherwise escape main() and surface as a Python traceback. + print(f"pos3 {args.command}: {type(exc).__name__}: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/pyproject.toml b/pyproject.toml index 9421ebf..4d6627e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pos3" -version = "0.3.0" +version = "0.3.1" description = "S3 Simple Sync - Make using S3 as simple as using local files" readme = "README.md" requires-python = ">=3.11" @@ -27,6 +27,9 @@ dependencies = [ Homepage = "https://github.com/Positronic-Robotics/pos3" Repository = "https://github.com/Positronic-Robotics/pos3" +[project.scripts] +pos3 = "pos3.cli:main" + [project.optional-dependencies] dev = [ "pytest>=7.0", diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..bcdbd34 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,569 @@ +import tempfile +from pathlib import Path +from unittest.mock import Mock, patch + +import pytest +from botocore.exceptions import ClientError + +from pos3.cli import main + +BOTO3_PATCH_TARGET = "pos3.profiles.boto3.client" + + +def _make_404_error(*_args, **_kwargs): + raise ClientError({"Error": {"Code": "404"}}, "head_object") + + +def _setup_s3_mock(mock_boto_client, paginate_return_value=None): + mock_s3 = Mock() + mock_boto_client.return_value = mock_s3 + mock_s3.head_object.side_effect = _make_404_error + mock_paginator = Mock() + mock_s3.get_paginator.return_value = mock_paginator + mock_paginator.paginate.return_value = paginate_return_value or [{"Contents": []}] + return mock_s3 + + +class TestCliLs: + @patch(BOTO3_PATCH_TARGET) + def test_ls_prints_full_s3_urls(self, mock_boto_client, capsys): + paginate = [ + { + "Contents": [ + {"Key": "data/file.txt", "Size": 5}, + {"Key": "data/sub/nested.txt", "Size": 10}, + ] + } + ] + _setup_s3_mock(mock_boto_client, paginate) + + rc = main(["ls", "s3://bucket/data"]) + + captured = capsys.readouterr() + assert rc == 0 + lines = captured.out.strip().splitlines() + assert "s3://bucket/data/file.txt" in lines + # Non-recursive excludes nested items + assert "s3://bucket/data/sub/nested.txt" not in lines + + @patch(BOTO3_PATCH_TARGET) + def test_ls_recursive(self, mock_boto_client, capsys): + paginate = [{"Contents": [{"Key": "data/sub/nested.txt", "Size": 10}]}] + _setup_s3_mock(mock_boto_client, paginate) + + rc = main(["ls", "-r", "s3://bucket/data"]) + + captured = capsys.readouterr() + assert rc == 0 + assert "s3://bucket/data/sub/nested.txt" in captured.out + + def test_ls_local_path(self, capsys): + with tempfile.TemporaryDirectory() as tmpdir: + base = Path(tmpdir) + (base / "file.txt").write_text("x") + + rc = main(["ls", str(base)]) + + captured = capsys.readouterr() + assert rc == 0 + assert str(base / "file.txt") in captured.out + + @patch(BOTO3_PATCH_TARGET) + def test_ls_trailing_slash_forces_directory_listing(self, mock_boto_client, capsys): + """`pos3 ls s3://bucket/data/` must list the data/ contents even if + an object exactly named 'data' also exists. ls() used to strip the + trailing slash via _normalize_s3_url, letting head_object('data') + win and hide the directory contents.""" + mock_s3 = _setup_s3_mock( + mock_boto_client, + [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}], + ) + # Pretend an object exactly named 'data' ALSO exists — without the + # fix, head_object('data') would return this and the directory + # listing would be skipped. + mock_s3.head_object.side_effect = None + mock_s3.head_object.return_value = {"ContentLength": 100} + + rc = main(["ls", "s3://bucket/data/"]) + + captured = capsys.readouterr() + assert rc == 0 + lines = captured.out.strip().splitlines() + # Should be the directory contents, not the exact-key 'data' object. + assert lines == ["s3://bucket/data/file.txt"] + + @patch(BOTO3_PATCH_TARGET) + def test_ls_single_object(self, mock_boto_client, capsys): + """`pos3 ls s3://bucket/file.json` on an exact object key must return + the object URL, not an empty list. ls() used to force a trailing + slash, suppressing the head_object exact-key probe.""" + mock_s3 = _setup_s3_mock(mock_boto_client) + # Override the default 404: this key IS an exact S3 object. + mock_s3.head_object.side_effect = None + mock_s3.head_object.return_value = {"ContentLength": 42} + + rc = main(["ls", "s3://bucket/results.json"]) + + captured = capsys.readouterr() + assert rc == 0 + assert captured.out.strip() == "s3://bucket/results.json" + + +class TestCliDownload: + @patch(BOTO3_PATCH_TARGET) + def test_download_prints_only_local_path_to_stdout(self, mock_boto_client, capsys): + paginate = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}] + _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "dst" + rc = main(["download", "s3://bucket/data", "--local", str(local_dir)]) + + captured = capsys.readouterr() + assert rc == 0 + # Exactly one line on stdout: the resulting local path. + lines = captured.out.strip().splitlines() + assert len(lines) == 1 + assert lines[0] == str(local_dir.resolve()) + + @patch(BOTO3_PATCH_TARGET) + def test_download_default_does_not_delete(self, mock_boto_client): + """CLI default for --delete is OFF; orphan local files survive.""" + paginate = [{"Contents": [{"Key": "data/file1.txt", "Size": 5}]}] + _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "data" + local_dir.mkdir() + (local_dir / "file1.txt").write_bytes(b"12345") + orphan = local_dir / "orphan.txt" + orphan.write_text("orphan") + + rc = main(["download", "s3://bucket/data", "--local", str(local_dir)]) + + assert rc == 0 + assert orphan.exists() + + @patch(BOTO3_PATCH_TARGET) + def test_download_delete_flag_removes_orphans(self, mock_boto_client): + paginate = [{"Contents": [{"Key": "data/file1.txt", "Size": 5}]}] + _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "data" + local_dir.mkdir() + (local_dir / "file1.txt").write_bytes(b"12345") + orphan = local_dir / "orphan.txt" + orphan.write_text("orphan") + + rc = main(["download", "s3://bucket/data", "--local", str(local_dir), "--delete"]) + + assert rc == 0 + assert not orphan.exists() + + @patch(BOTO3_PATCH_TARGET) + def test_download_exclude_multiple_patterns(self, mock_boto_client): + paginate = [ + { + "Contents": [ + {"Key": "data/file.txt", "Size": 5}, + {"Key": "data/file.log", "Size": 10}, + {"Key": "data/file.tmp", "Size": 10}, + ] + } + ] + mock_s3 = _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "dst" + rc = main( + [ + "download", + "s3://bucket/data", + "--local", + str(local_dir), + "--exclude", + "*.log", + "--exclude", + "*.tmp", + ] + ) + + assert rc == 0 + assert mock_s3.download_file.call_count == 1 + downloaded_key = mock_s3.download_file.call_args_list[0][0][1] + assert "file.txt" in downloaded_key + + @patch(BOTO3_PATCH_TARGET) + def test_download_single_object_calls_download_file(self, mock_boto_client, capsys, tmp_path): + """`pos3 download s3://bucket/results.json` must actually fetch the + object, not silently mkdir the destination and exit 0.""" + mock_s3 = _setup_s3_mock(mock_boto_client) + mock_s3.head_object.side_effect = None + mock_s3.head_object.return_value = {"ContentLength": 42, "Size": 42} + + local = tmp_path / "results.json" + rc = main(["download", "s3://bucket/results.json", "--local", str(local)]) + + assert rc == 0 + assert mock_s3.download_file.call_count == 1 + # Stdout still emits the local path on success. + captured = capsys.readouterr() + assert captured.out.strip() == str(local) + + @patch(BOTO3_PATCH_TARGET) + def test_download_unknown_profile_returns_error(self, mock_boto_client, capsys): + _setup_s3_mock(mock_boto_client) + + rc = main(["download", "s3://bucket/data", "--profile", "no-such-profile"]) + + captured = capsys.readouterr() + assert rc == 1 + assert "Unknown profile" in captured.err + + +class TestCliUrlValidation: + @patch(BOTO3_PATCH_TARGET) + def test_download_rejects_non_s3_url(self, mock_boto_client, capsys): + mock_s3 = _setup_s3_mock(mock_boto_client) + + rc = main(["download", "bucket/data", "--local", "/tmp/ignored"]) + + captured = capsys.readouterr() + assert rc == 1 + assert "must be an s3:// URL" in captured.err + # Nothing should have happened on stdout, nothing on the wire. + assert captured.out == "" + mock_s3.download_file.assert_not_called() + + @patch(BOTO3_PATCH_TARGET) + def test_upload_rejects_non_s3_url(self, mock_boto_client, capsys, tmp_path): + mock_s3 = _setup_s3_mock(mock_boto_client) + src = tmp_path / "src" + src.mkdir() + (src / "file.txt").write_text("x") + + rc = main(["upload", str(tmp_path / "dst"), "--local", str(src)]) + + captured = capsys.readouterr() + assert rc == 1 + assert "must be an s3:// URL" in captured.err + mock_s3.upload_file.assert_not_called() + # The bogus "destination" must not have been mkdir'd as a side effect. + assert not (tmp_path / "dst").exists() + + +class TestCliUpload: + @patch(BOTO3_PATCH_TARGET) + def test_upload_errors_when_source_missing(self, mock_boto_client, capsys): + _setup_s3_mock(mock_boto_client) + + with tempfile.TemporaryDirectory() as tmpdir: + missing = Path(tmpdir) / "does-not-exist" + rc = main(["upload", "s3://bucket/data", "--local", str(missing)]) + + captured = capsys.readouterr() + assert rc == 1 + assert "does not exist" in captured.err + # Nothing should have been written to S3. + mock_boto_client.return_value.upload_file.assert_not_called() + + @patch(BOTO3_PATCH_TARGET) + def test_upload_uploads_existing_local_source(self, mock_boto_client, capsys): + mock_s3 = _setup_s3_mock(mock_boto_client) + + with tempfile.TemporaryDirectory() as tmpdir: + src = Path(tmpdir) / "src" + src.mkdir() + (src / "file.txt").write_text("content") + + rc = main(["upload", "s3://bucket/data", "--local", str(src)]) + + captured = capsys.readouterr() + assert rc == 0 + assert mock_s3.upload_file.call_count >= 1 + # upload's success path is silent on stdout: no path, no progress. + # Progress bars and logs go to stderr. This is the counterpart to + # download's "exactly one line = local path" contract. + assert captured.out == "" + + @patch(BOTO3_PATCH_TARGET) + def test_upload_default_source_is_cache_path(self, mock_boto_client): + """When --local is omitted, source defaults to the same cache path pos3 download + would have produced.""" + mock_s3 = _setup_s3_mock(mock_boto_client) + + with tempfile.TemporaryDirectory() as tmpdir: + cache_root = Path(tmpdir) + cache_path = cache_root / "_" / "bucket" / "data" + cache_path.mkdir(parents=True) + (cache_path / "file.txt").write_text("hi") + + # Override the CLI's mirror() to anchor cache_root at our tmpdir, so + # the no-`--local` path resolves under it. + import pos3 + + real_mirror = pos3.mirror + + def fixed_mirror(**_kwargs): + return real_mirror(cache_root=str(cache_root), show_progress=False) + + with patch("pos3.cli.mirror", side_effect=fixed_mirror): + rc = main(["upload", "s3://bucket/data"]) + + assert rc == 0 + assert mock_s3.upload_file.call_count >= 1 + + @patch(BOTO3_PATCH_TARGET) + def test_upload_default_does_not_delete(self, mock_boto_client): + """CLI default for --delete is OFF; remote orphans survive.""" + # S3 has remote_only.txt; local has file.txt. Without --delete, remote should stay. + paginate = [{"Contents": [{"Key": "data/remote_only.txt", "Size": 5}]}] + mock_s3 = _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + src = Path(tmpdir) / "src" + src.mkdir() + (src / "file.txt").write_text("content") + + rc = main(["upload", "s3://bucket/data", "--local", str(src)]) + + assert rc == 0 + # No deletes should have been issued. + assert mock_s3.delete_object.call_count == 0 + + @patch(BOTO3_PATCH_TARGET) + def test_upload_delete_flag_removes_remote_orphans(self, mock_boto_client): + paginate = [{"Contents": [{"Key": "data/remote_only.txt", "Size": 5}]}] + mock_s3 = _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + src = Path(tmpdir) / "src" + src.mkdir() + (src / "file.txt").write_text("content") + + rc = main(["upload", "s3://bucket/data", "--local", str(src), "--delete"]) + + assert rc == 0 + assert mock_s3.delete_object.call_count >= 1 + + +class TestCliTransferFailures: + @patch(BOTO3_PATCH_TARGET) + def test_download_returns_nonzero_when_worker_fails(self, mock_boto_client, capsys, tmp_path): + paginate = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}] + mock_s3 = _setup_s3_mock(mock_boto_client, paginate) + mock_s3.download_file.side_effect = RuntimeError("boom") + + local_dir = tmp_path / "dst" + rc = main(["download", "s3://bucket/data", "--local", str(local_dir)]) + + captured = capsys.readouterr() + assert rc == 1 + # The success-path stdout (the local cache path) MUST NOT be printed + # on failure, since `data_dir=$(pos3 download …)` would treat the + # path as valid and downstream reads would silently use a partial cache. + assert captured.out == "" + assert "Download" in captured.err + assert "boom" in captured.err + + @patch(BOTO3_PATCH_TARGET) + def test_upload_returns_nonzero_when_worker_fails(self, mock_boto_client, capsys, tmp_path): + mock_s3 = _setup_s3_mock(mock_boto_client) + mock_s3.upload_file.side_effect = RuntimeError("boom") + + src = tmp_path / "src" + src.mkdir() + (src / "file.txt").write_text("content") + + rc = main(["upload", "s3://bucket/data", "--local", str(src)]) + + captured = capsys.readouterr() + assert rc == 1 + assert "Upload" in captured.err + assert "boom" in captured.err + + +class TestCliDryRun: + @patch(BOTO3_PATCH_TARGET) + def test_download_dry_run_prints_plan_and_does_not_transfer(self, mock_boto_client, capsys): + paginate = [ + { + "Contents": [ + {"Key": "data/file.txt", "Size": 5}, + {"Key": "data/sub/nested.txt", "Size": 7}, + ] + } + ] + mock_s3 = _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "dst" + rc = main(["download", "-n", "s3://bucket/data", "--local", str(local_dir)]) + + captured = capsys.readouterr() + assert rc == 0 + # No actual download was performed. + mock_s3.download_file.assert_not_called() + out_lines = captured.out.strip().splitlines() + # Two files planned, no extra trailing local-path line. + copy_lines = [line for line in out_lines if line.startswith("download:")] + assert len(copy_lines) == 2 + assert any("s3://bucket/data/file.txt" in line for line in copy_lines) + assert any("s3://bucket/data/sub/nested.txt" in line for line in copy_lines) + assert all(" to " in line for line in copy_lines) + # No delete lines without --delete. + assert not any(line.startswith("delete:") for line in out_lines) + + @patch(BOTO3_PATCH_TARGET) + def test_download_dry_run_with_delete_emits_delete_lines(self, mock_boto_client, capsys): + paginate = [{"Contents": [{"Key": "data/keep.txt", "Size": 5}]}] + mock_s3 = _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "data" + local_dir.mkdir() + (local_dir / "keep.txt").write_bytes(b"12345") + orphan = local_dir / "orphan.txt" + orphan.write_text("orphan") + + rc = main( + ["download", "-n", "s3://bucket/data", "--local", str(local_dir), "--delete"] + ) + + assert rc == 0 + # Dry-run must not touch the filesystem. + assert orphan.exists() + + captured = capsys.readouterr() + delete_lines = [ + line for line in captured.out.splitlines() if line.startswith("delete:") + ] + assert any(str(orphan) in line for line in delete_lines) + mock_s3.download_file.assert_not_called() + + @patch(BOTO3_PATCH_TARGET) + def test_upload_dry_run_prints_plan_and_does_not_transfer(self, mock_boto_client, capsys): + mock_s3 = _setup_s3_mock(mock_boto_client) + + with tempfile.TemporaryDirectory() as tmpdir: + src = Path(tmpdir) / "src" + src.mkdir() + (src / "file.txt").write_text("content") + + rc = main(["upload", "-n", "s3://bucket/data", "--local", str(src)]) + + captured = capsys.readouterr() + assert rc == 0 + mock_s3.upload_file.assert_not_called() + upload_lines = [ + line for line in captured.out.splitlines() if line.startswith("upload:") + ] + assert len(upload_lines) == 1 + assert "s3://bucket/data/file.txt" in upload_lines[0] + assert str(src / "file.txt") in upload_lines[0] + + @patch(BOTO3_PATCH_TARGET) + def test_upload_dry_run_with_delete_emits_remote_delete_lines(self, mock_boto_client, capsys): + paginate = [{"Contents": [{"Key": "data/remote_only.txt", "Size": 5}]}] + mock_s3 = _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + src = Path(tmpdir) / "src" + src.mkdir() + (src / "file.txt").write_text("content") + + rc = main( + ["upload", "-n", "s3://bucket/data", "--local", str(src), "--delete"] + ) + + captured = capsys.readouterr() + assert rc == 0 + mock_s3.upload_file.assert_not_called() + mock_s3.delete_object.assert_not_called() + delete_lines = [ + line for line in captured.out.splitlines() if line.startswith("delete:") + ] + assert any("s3://bucket/data/remote_only.txt" in line for line in delete_lines) + + def test_dry_run_not_accepted_on_ls(self): + with pytest.raises(SystemExit) as exc: + main(["ls", "-n", "s3://bucket/data"]) + assert exc.value.code != 0 + + +class TestCliEntry: + def test_no_subcommand_exits_with_error(self, capsys): + with pytest.raises(SystemExit) as exc: + main([]) + assert exc.value.code != 0 + + def test_unknown_flag_uses_subcommand_usage(self, capsys): + """Typoing a subcommand flag (e.g. `--dry_run` instead of `--dry-run`) + must produce the SUBCOMMAND's usage, not the top-level one, so the + user can see which flags actually exist for what they ran.""" + with pytest.raises(SystemExit) as exc: + main(["download", "s3://bucket/key", "--dry_run"]) + assert exc.value.code == 2 + captured = capsys.readouterr() + # The error must address the subcommand, not the top-level parser. + assert "pos3 download" in captured.err + # And it must show download's actual flags so the user can spot `-n`. + assert "[-n]" in captured.err or "--dry-run" in captured.err + # Sanity: NOT the top-level usage. + assert "{ls,download,upload}" not in captured.err + + +class TestCliBotoErrors: + @patch(BOTO3_PATCH_TARGET) + def test_ls_handles_client_error(self, mock_boto_client, capsys): + """A non-404 ClientError from _list_s3_objects (e.g. 403 access + denied) must produce the `pos3 ls: ...` error + exit 1, not a + Python traceback.""" + mock_s3 = Mock() + mock_boto_client.return_value = mock_s3 + mock_s3.head_object.side_effect = ClientError( + {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject" + ) + + rc = main(["ls", "s3://bucket/key"]) + + captured = capsys.readouterr() + assert rc == 1 + assert "pos3 ls:" in captured.err + assert "ClientError" in captured.err + assert "403" in captured.err + + @patch(BOTO3_PATCH_TARGET) + def test_download_handles_client_error_during_scan(self, mock_boto_client, capsys, tmp_path): + """ClientError raised from _scan_s3 (the pre-transfer scan inside + Mirror.download) must be caught by main(), not escape as a + traceback.""" + mock_s3 = Mock() + mock_boto_client.return_value = mock_s3 + mock_s3.head_object.side_effect = ClientError( + {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject" + ) + + rc = main(["download", "s3://bucket/data", "--local", str(tmp_path / "dst")]) + + captured = capsys.readouterr() + assert rc == 1 + assert "pos3 download:" in captured.err + assert captured.out == "" + + @patch(BOTO3_PATCH_TARGET) + def test_dry_run_handles_client_error_during_plan(self, mock_boto_client, capsys, tmp_path): + """plan_download propagates ClientError too — it goes through + _scan_s3 the same way.""" + mock_s3 = Mock() + mock_boto_client.return_value = mock_s3 + mock_s3.head_object.side_effect = ClientError( + {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject" + ) + + rc = main(["download", "-n", "s3://bucket/data", "--local", str(tmp_path / "dst")]) + + captured = capsys.readouterr() + assert rc == 1 + assert "pos3 download:" in captured.err diff --git a/tests/test_s3.py b/tests/test_s3.py index 98662a7..10736a8 100644 --- a/tests/test_s3.py +++ b/tests/test_s3.py @@ -243,6 +243,35 @@ def test_background_sync_uploads_repeatedly(self, mock_boto_client): assert mock_s3.upload_file.call_count >= 2 + @patch(BOTO3_PATCH_TARGET) + def test_background_worker_survives_transfer_error(self, mock_boto_client): + """A TransferError from one interval-sync iteration must not kill the + daemon thread; subsequent ticks should keep retrying.""" + mock_s3 = _setup_s3_mock(mock_boto_client) + + call_count = {"n": 0} + + def upload_side_effect(*_args, **_kwargs): + call_count["n"] += 1 + if call_count["n"] == 1: + raise RuntimeError("transient") + # Subsequent calls succeed (no return value needed). + + mock_s3.upload_file.side_effect = upload_side_effect + + with tempfile.TemporaryDirectory() as tmpdir: + output = Path(tmpdir) / "output" + output.mkdir() + (output / "data.txt").write_text("content") + + with s3.mirror(cache_root=tmpdir, show_progress=False): + s3.upload("s3://bucket/output", local=output, interval=1) + time.sleep(2.5) + + # If the worker had died after the first failure, count would stay at 1. + # Surviving means at least one more attempt happened on a later tick. + assert call_count["n"] >= 2 + @patch(BOTO3_PATCH_TARGET) def test_upload_no_sync_on_error(self, mock_boto_client): """Test that uploads with sync_on_error=False don't sync when context exits with error.""" @@ -686,6 +715,384 @@ def test_sync_conflicts(self, mock_boto_client): s3.sync("s3://bucket/data", interval=None) +class TestPlanPublicAPI: + """The plan API is callable via the public pos3.plan_download / + pos3.plan_upload module-level wrappers, the same way pos3.download / + pos3.upload are. Users following the README must not need to reach + into pos3._require_active_mirror().""" + + def test_plan_download_and_upload_are_module_level(self): + # Sanity: they exist on the package surface. + assert callable(s3.plan_download) + assert callable(s3.plan_upload) + # And in __all__ so `from pos3 import *` includes them. + assert "plan_download" in s3.__all__ + assert "plan_upload" in s3.__all__ + assert "TransferPlan" in s3.__all__ + assert "TransferError" in s3.__all__ + + @patch(BOTO3_PATCH_TARGET) + def test_plan_download_via_public_wrapper(self, mock_boto_client): + paginate = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}] + _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "dst" + with s3.mirror(cache_root=tmpdir, show_progress=False): + plan = s3.plan_download("s3://bucket/data", local=str(local_dir)) + + assert isinstance(plan, s3.TransferPlan) + sources = [src for src, _ in plan.to_copy] + assert "s3://bucket/data/file.txt" in sources + + +class TestPlan: + """plan_download / plan_upload return what a real call would do, without + transferring, deleting, or creating directories.""" + + @patch(BOTO3_PATCH_TARGET) + def test_plan_download_lists_files_to_copy(self, mock_boto_client): + paginate = [ + { + "Contents": [ + {"Key": "data/file.txt", "Size": 5}, + {"Key": "data/sub/nested.txt", "Size": 7}, + ] + } + ] + mock_s3 = _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "dst" + with s3.mirror(cache_root=tmpdir, show_progress=False) as _: + from pos3 import _require_active_mirror + + plan = _require_active_mirror().plan_download( + "s3://bucket/data", local=str(local_dir) + ) + + sources = [src for src, _ in plan.to_copy] + assert "s3://bucket/data/file.txt" in sources + assert "s3://bucket/data/sub/nested.txt" in sources + # No real transfer happened. + mock_s3.download_file.assert_not_called() + # Directory entries are filtered out — file-level only. + assert all(not src.endswith("/") for src in sources) + + @patch(BOTO3_PATCH_TARGET) + def test_plan_download_lists_orphans_in_to_delete(self, mock_boto_client): + paginate = [{"Contents": [{"Key": "data/keep.txt", "Size": 5}]}] + _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "data" + local_dir.mkdir() + (local_dir / "keep.txt").write_bytes(b"12345") + orphan = local_dir / "orphan.txt" + orphan.write_text("x") + + with s3.mirror(cache_root=tmpdir, show_progress=False): + from pos3 import _require_active_mirror + + plan = _require_active_mirror().plan_download( + "s3://bucket/data", local=str(local_dir) + ) + + # Dry-plan is read-only. + assert orphan.exists() + + assert str(orphan) in plan.to_delete + + @patch(BOTO3_PATCH_TARGET) + def test_plan_upload_lists_files_to_copy(self, mock_boto_client): + mock_s3 = _setup_s3_mock(mock_boto_client) + + with tempfile.TemporaryDirectory() as tmpdir: + src = Path(tmpdir) / "src" + src.mkdir() + (src / "file.txt").write_text("content") + + with s3.mirror(cache_root=tmpdir, show_progress=False): + from pos3 import _require_active_mirror + + plan = _require_active_mirror().plan_upload( + "s3://bucket/data", local=str(src) + ) + + destinations = [dst for _, dst in plan.to_copy] + assert destinations == ["s3://bucket/data/file.txt"] + mock_s3.upload_file.assert_not_called() + + def test_plan_download_rejects_non_s3_url(self): + with tempfile.TemporaryDirectory() as tmpdir: + with s3.mirror(cache_root=tmpdir, show_progress=False): + from pos3 import _require_active_mirror + + with pytest.raises(ValueError, match="s3:// URL"): + _require_active_mirror().plan_download("/local/path") + + def test_plan_upload_rejects_non_s3_url(self): + with tempfile.TemporaryDirectory() as tmpdir: + with s3.mirror(cache_root=tmpdir, show_progress=False): + from pos3 import _require_active_mirror + + with pytest.raises(ValueError, match="s3:// URL"): + _require_active_mirror().plan_upload("/local/path", local=tmpdir) + + @patch(BOTO3_PATCH_TARGET) + def test_plan_download_normalizes_trailing_slash_in_url(self, mock_boto_client): + """A trailing slash on the input URL must not produce s3://bucket/data//file.txt + in the plan — Mirror.download normalizes before parsing and plan_* must agree.""" + paginate = [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}] + _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + local_dir = Path(tmpdir) / "dst" + with s3.mirror(cache_root=tmpdir, show_progress=False): + from pos3 import _require_active_mirror + + plan = _require_active_mirror().plan_download( + "s3://bucket/data/", local=str(local_dir) + ) + + sources = [src for src, _ in plan.to_copy] + assert sources == ["s3://bucket/data/file.txt"] + assert all("//" not in src.replace("s3://", "") for src in sources) + + @patch(BOTO3_PATCH_TARGET) + def test_plan_download_trailing_slash_forces_directory_listing(self, mock_boto_client): + """`pos3 download -n s3://bucket/data/` must plan the directory + contents even if an exact object `data` also exists. Pre-fix, + plan_download normalized away the slash, head_object('data') won, + and the plan reported the exact-object copy instead of `data/*`.""" + mock_s3 = _setup_s3_mock( + mock_boto_client, + [{"Contents": [{"Key": "data/file.txt", "Size": 5}]}], + ) + # Exact 'data' object ALSO exists — without preserving the slash, + # head_object('data') would win and shadow the directory contents. + mock_s3.head_object.side_effect = None + mock_s3.head_object.return_value = {"ContentLength": 100} + + with tempfile.TemporaryDirectory() as tmpdir: + with s3.mirror(cache_root=tmpdir, show_progress=False): + from pos3 import _require_active_mirror + + plan = _require_active_mirror().plan_download( + "s3://bucket/data/", local=str(Path(tmpdir) / "dst") + ) + + sources = [src for src, _ in plan.to_copy] + assert sources == ["s3://bucket/data/file.txt"] + + @patch(BOTO3_PATCH_TARGET) + def test_plan_upload_normalizes_trailing_slash_in_url(self, mock_boto_client): + paginate = [{"Contents": [{"Key": "data/orphan.txt", "Size": 5}]}] + _setup_s3_mock(mock_boto_client, paginate) + + with tempfile.TemporaryDirectory() as tmpdir: + src = Path(tmpdir) / "src" + src.mkdir() + (src / "file.txt").write_text("content") + + with s3.mirror(cache_root=tmpdir, show_progress=False): + from pos3 import _require_active_mirror + + plan = _require_active_mirror().plan_upload( + "s3://bucket/data/", local=str(src) + ) + + destinations = [dst for _, dst in plan.to_copy] + assert destinations == ["s3://bucket/data/file.txt"] + # And the delete list, which also goes through _make_s3_key for upload. + assert plan.to_delete == ["s3://bucket/data/orphan.txt"] + + +class TestTrailingSlashRealTransfers: + """The real (non-dry-run) download() and upload() paths must preserve + the user's trailing-slash intent. Mirror.download / _sync_uploads used + to normalize the URL before scanning, letting head_object('data') win + over the requested data/ directory listing.""" + + @patch(BOTO3_PATCH_TARGET) + def test_download_trailing_slash_transfers_directory_contents(self, mock_boto_client): + mock_s3 = Mock() + mock_boto_client.return_value = mock_s3 + # Both: an exact 'data' object AND objects under 'data/' exist. + mock_s3.head_object.return_value = {"ContentLength": 100} + mock_paginator = Mock() + mock_s3.get_paginator.return_value = mock_paginator + mock_paginator.paginate.return_value = [ + {"Contents": [{"Key": "data/file.txt", "Size": 5}]} + ] + + with tempfile.TemporaryDirectory() as tmpdir: + local = Path(tmpdir) / "dst" + with s3.mirror(cache_root=tmpdir, show_progress=False): + s3.download("s3://bucket/data/", local=str(local), delete=False) + + # The directory content must be the one downloaded, not the exact + # 'data' object that head_object would have picked up. + keys = [c[0][1] for c in mock_s3.download_file.call_args_list] + assert keys == ["data/file.txt"] + assert "data" not in keys + + @patch(BOTO3_PATCH_TARGET) + def test_upload_trailing_slash_scans_directory_for_delete(self, mock_boto_client): + mock_s3 = Mock() + mock_boto_client.return_value = mock_s3 + # Exact 'data' object AND an orphan under 'data/' both exist. + mock_s3.head_object.return_value = {"ContentLength": 100} + mock_paginator = Mock() + mock_s3.get_paginator.return_value = mock_paginator + mock_paginator.paginate.return_value = [ + {"Contents": [{"Key": "data/orphan.txt", "Size": 5}]} + ] + + with tempfile.TemporaryDirectory() as tmpdir: + source = Path(tmpdir) / "src" + source.mkdir() + (source / "file.txt").write_text("x") + + with s3.mirror(cache_root=tmpdir, show_progress=False): + s3.upload( + "s3://bucket/data/", + local=str(source), + interval=None, + delete=True, + ) + + # The directory orphan must be the one deleted, not the exact + # 'data' object. + deleted_keys = [c[1]["Key"] for c in mock_s3.delete_object.call_args_list] + assert "data/orphan.txt" in deleted_keys + assert "data" not in deleted_keys + + @patch(BOTO3_PATCH_TARGET) + def test_plan_upload_empty_when_source_missing(self, mock_boto_client): + """Real _sync_uploads skips registrations whose local_path doesn't + exist (no transfers, no deletes). plan_upload must mirror that + instead of reporting every remote object as 'would delete'.""" + _setup_s3_mock( + mock_boto_client, + [{"Contents": [{"Key": "data/orphan.txt", "Size": 5}]}], + ) + + with tempfile.TemporaryDirectory() as tmpdir: + missing = Path(tmpdir) / "does-not-exist" + with s3.mirror(cache_root=tmpdir, show_progress=False): + from pos3 import _require_active_mirror + + plan = _require_active_mirror().plan_upload( + "s3://bucket/data", local=str(missing) + ) + + assert plan.to_copy == [] + assert plan.to_delete == [] + + +class TestSingleObjectDownload: + """Downloading an exact S3 object key (not a prefix) must actually fetch + the file. _scan_s3 used to emit a root directory marker that collided + with the file in _compute_sync_diff's dict, causing download() to mkdir + the local path instead of calling download_file.""" + + @patch(BOTO3_PATCH_TARGET) + def test_download_single_object_calls_download_file(self, mock_boto_client): + mock_s3 = Mock() + mock_boto_client.return_value = mock_s3 + # head_object 200 → _list_s3_objects yields the exact object. + mock_s3.head_object.return_value = {"ContentLength": 42, "Size": 42} + mock_paginator = Mock() + mock_s3.get_paginator.return_value = mock_paginator + mock_paginator.paginate.return_value = [{"Contents": []}] + + with tempfile.TemporaryDirectory() as tmpdir: + local = Path(tmpdir) / "results.json" + with s3.mirror(cache_root=tmpdir, show_progress=False): + s3.download("s3://bucket/results.json", local=str(local)) + + assert mock_s3.download_file.call_count == 1 + args = mock_s3.download_file.call_args[0] + assert args[0] == "bucket" + assert args[1] == "results.json" + assert args[2] == str(local) + + +class TestMirrorConstructorIsSideEffectFree: + def test_constructing_mirror_does_not_create_cache_root(self): + """Constructing a Mirror (entering pos3.mirror()) must not mkdir the + cache root — dry-run and planning paths rely on this.""" + with tempfile.TemporaryDirectory() as tmpdir: + cache_root = Path(tmpdir) / "nested" / "cache" / "root" + assert not cache_root.exists() + + with s3.mirror(cache_root=str(cache_root), show_progress=False): + # Just entering the context must not have created cache_root. + assert not cache_root.exists() + + +class TestFinalSyncPreservesOriginalException: + @patch(BOTO3_PATCH_TARGET) + def test_app_exception_survives_failed_cleanup_sync(self, mock_boto_client): + """When the mirror body raises AND a sync_on_error upload's cleanup + sync also fails, the user must see the app exception, not the + TransferError from cleanup.""" + mock_s3 = _setup_s3_mock(mock_boto_client) + mock_s3.upload_file.side_effect = RuntimeError("cleanup upload failed") + + class AppError(Exception): + pass + + with tempfile.TemporaryDirectory() as tmpdir: + output = Path(tmpdir) / "output" + output.mkdir() + (output / "data.txt").write_text("content") + + with pytest.raises(AppError, match="the real failure"): + with s3.mirror(cache_root=tmpdir, show_progress=False): + s3.upload( + "s3://bucket/output", + local=output, + interval=None, + sync_on_error=True, + ) + raise AppError("the real failure") + + @patch(BOTO3_PATCH_TARGET) + def test_app_exception_survives_scan_client_error_in_cleanup(self, mock_boto_client): + """Cleanup-time _scan_s3 can raise ClientError (e.g. 403) BEFORE + any worker future is created — that path bypasses TransferError. + The cleanup catch must be broad enough to preserve the app + exception in this case too.""" + mock_s3 = Mock() + mock_boto_client.return_value = mock_s3 + # _scan_s3 → _list_s3_objects calls head_object first. A 403 here + # propagates through the scan iterator, not through a worker future, + # so the previous TransferError-only catch would have unmasked it. + mock_s3.head_object.side_effect = ClientError( + {"Error": {"Code": "403", "Message": "Forbidden"}}, "HeadObject" + ) + + class AppError(Exception): + pass + + with tempfile.TemporaryDirectory() as tmpdir: + output = Path(tmpdir) / "output" + output.mkdir() + (output / "data.txt").write_text("content") + + with pytest.raises(AppError, match="the real failure"): + with s3.mirror(cache_root=tmpdir, show_progress=False): + s3.upload( + "s3://bucket/output", + local=output, + interval=None, + sync_on_error=True, + ) + raise AppError("the real failure") + + class TestLs: def test_ls_local_non_recursive(self): """Test non-recursive listing excludes nested items."""