From 16e299162a010d774b96e2e0351240add4ee7602 Mon Sep 17 00:00:00 2001 From: Dennis Date: Mon, 26 Oct 2026 02:52:04 -0700 Subject: [PATCH 1/3] Fail only the faulted file when an HTTP transfer breaks Since e2c1e61 the HTTP backend reads response.raw directly, which bypasses Requests' error wrapping: a reset, truncated or stalled body raised a urllib3 ProtocolError or ReadTimeoutError. Neither is a RequestException, so the error escaped fetch, _run_transfers treated it as an interrupt that cancelled every other transfer, and the CLI printed a traceback. Catch urllib3's errors in fetch and report them as the usual per-file DownloadError; the CLI's user-error list gets the same types as a backstop. Port the fault-injecting loopback server, the chaos helpers and the network fault tests from ci/test-automation so this stays covered: 29 tests over real sockets, including the former #54 xfail, which now passes. Counts are taken from the download plan because a published manifest is catalogued and fetched as well. Co-Authored-By: Claude Fable 5.1 --- mod/hallmark/cli.py | 5 +- mod/hallmark/transport/http.py | 18 +- pyproject.toml | 3 + test/chaos.py | 188 ++++++++++++++++++ test/conftest.py | 42 +++++ test/http_server.py | 335 ++++++++++++++++++++++++++++++++- test/survey.py | 170 +++++++++++++++++ test/test_chaos_network.py | 186 ++++++++++++++++++ 8 files changed, 940 insertions(+), 7 deletions(-) create mode 100644 test/chaos.py create mode 100644 test/survey.py create mode 100644 test/test_chaos_network.py diff --git a/mod/hallmark/cli.py b/mod/hallmark/cli.py index a1b021c..cfbb8ed 100644 --- a/mod/hallmark/cli.py +++ b/mod/hallmark/cli.py @@ -19,9 +19,11 @@ from pathlib import Path import click +import requests import yaml from click import ClickException from git.exc import GitError +from urllib3.exceptions import HTTPError as Urllib3Error from . import Repo from .fmt_detection import detect_fmt @@ -175,7 +177,8 @@ def _run_download(repo, plan, *, max_workers): # Errors a user causes or can fix, such as a typo in config.yml, a missing # file, a full disk or a git failure. Every command reports them in one line; # anything else is a bug and keeps its traceback. -_USER_ERRORS = (HallmarkError, GitError, OSError, yaml.YAMLError, ValueError) +_USER_ERRORS = (HallmarkError, GitError, OSError, yaml.YAMLError, ValueError, + requests.RequestException, Urllib3Error) class _HallmarkGroup(click.Group): diff --git a/mod/hallmark/transport/http.py b/mod/hallmark/transport/http.py index bb7bed8..054115d 100644 --- a/mod/hallmark/transport/http.py +++ b/mod/hallmark/transport/http.py @@ -14,6 +14,8 @@ import requests from requests.auth import AuthBase +from urllib3.exceptions import (HTTPError as Urllib3Error, ProtocolError, + ReadTimeoutError) from ..utils import REMOTE_REQUEST_TIMEOUT from .base import (DownloadError, RemoteConfigurationError, RemoteEntry, @@ -151,11 +153,18 @@ def _open_file_response(self, path): ) def _download_error(self, exc): - # Requests exceptions can embed credentials in their URL or response body. """Describe an HTTP failure without exposing response contents.""" + # Requests exceptions can embed credentials in their URL or response body. status = getattr(getattr(exc, "response", None), "status_code", None) cls = RemoteObjectMissing if status == 404 else DownloadError - detail = f"HTTP {status}" if isinstance(status, int) else type(exc).__name__ + if isinstance(status, int): + detail = f"HTTP {status}" + elif isinstance(exc, ReadTimeoutError): + detail = "the connection stalled" + elif isinstance(exc, ProtocolError): + detail = "the connection was reset or the body was truncated" + else: + detail = type(exc).__name__ if status in {401, 403}: detail += (f"; check auth profile {self.context.remote.auth!r}" if self.auth is not None else @@ -224,8 +233,11 @@ def fetch(self, relative_path, destination, *, chunk_size=8192): handle.write(chunk) if self.context.on_bytes is not None: self.context.on_bytes(len(chunk)) - except requests.RequestException as exc: + # Reading response.raw directly bypasses Requests' error wrapping, so + # a reset, truncated or stalled transfer arrives as a urllib3 error. + except (requests.RequestException, Urllib3Error) as exc: raise self._download_error(exc) from None + def read_text(self, relative_path, limit): """Read metadata within the source root and byte limit.""" content = bytearray() diff --git a/pyproject.toml b/pyproject.toml index cc336e1..ebd223f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,6 +69,9 @@ addopts = "--strict-markers" markers = [ "ssh_client: exercise the actual local OpenSSH SFTP command parser", "ssh_integration: disposable loopback sshd (HALLMARK_RUN_SSH_TESTS=1 required)", + "chaos: fault injection, crash consistency and seeded random operations", + "survey: end-to-end survey workflows against loopback HTTP and SSH servers", + "survey_live: tiny downloads from public archives (HALLMARK_RUN_LIVE_TESTS=1 required)", ] [tool.coverage.run] diff --git a/test/chaos.py b/test/chaos.py new file mode 100644 index 0000000..fc15745 --- /dev/null +++ b/test/chaos.py @@ -0,0 +1,188 @@ +"""Helpers for fault-injection ("chaos monkey") tests. + +``check_invariants`` describes what must hold for a repository after any +operation, failed or not. Problems are grouped by severity: + +* corruption: the repository cannot be opened, an object does not match + its checksum, a committed file has no object, or Git reports damage; +* debris: temporary files that a finished operation left behind. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import errno +import hashlib +from io import StringIO +import os +import pathlib +from pathlib import Path +import subprocess + +import pandas as pd +import pytest + +from hallmark import Repo + +STATE_FILES = ("config.yml", "meta.yml", "data.tsv") +DEBRIS_PATTERNS = ("**/.*.part", "**/.*.tmp", "**/.hallmark-checkout-*") + + +@dataclass +class Violations: + """Invariant violations found in a repository.""" + corruption: list = field(default_factory=list) + debris: list = field(default_factory=list) + + def __bool__(self): + return bool(self.corruption or self.debris) + + def __str__(self): + lines = [f"corruption: {item}" for item in self.corruption] + lines += [f"debris: {item}" for item in self.debris] + return "\n".join(lines) or "no violations" + + +def skip_if_root(): + """Skip permission tests when running as root, which ignores modes.""" + if hasattr(os, "geteuid") and os.geteuid() == 0: + pytest.skip("root ignores file permissions") + + +def fail_nth_replace(monkeypatch, nth, *, code=errno.ENOSPC, match=lambda target: True): + """Make the ``nth`` matching ``Path.replace`` call raise ``OSError(code)``. + + Patch the method on the class: on Python 3.9 and 3.10, pathlib calls a + reference to ``os.replace`` captured at import, so patching ``os`` misses it. + + Returns: + list: Targets of the matching calls, in order, for assertions. + """ + original = pathlib.Path.replace + calls = [] + + def replace(self, target): + if match(Path(target)): + calls.append(Path(target)) + if len(calls) == nth: + raise OSError(code, os.strerror(code), str(target)) + return original(self, target) + + monkeypatch.setattr(pathlib.Path, "replace", replace) + return calls + + +class _FailingWriter: + """File proxy whose writes fail once ``limit`` bytes have been written.""" + + def __init__(self, handle, limit, code): + self._handle, self._limit, self._code = handle, limit, code + self._written = 0 + + def write(self, data): + if self._written + len(data) > self._limit: + raise OSError(self._code, os.strerror(self._code)) + self._written += len(data) + return self._handle.write(data) + + def __getattr__(self, name): + return getattr(self._handle, name) + + def __enter__(self): + return self + + def __exit__(self, *exc): + return self._handle.__exit__(*exc) + + +def fail_writes_after(monkeypatch, limit, *, code=errno.ENOSPC, + match=lambda path: True): + """Make binary writes to matching paths fail after ``limit`` bytes.""" + original = pathlib.Path.open + + def open_(self, mode="r", *args, **kwargs): + handle = original(self, mode, *args, **kwargs) + if "w" in mode and "b" in mode and match(self): + return _FailingWriter(handle, limit, code) + return handle + + monkeypatch.setattr(pathlib.Path, "open", open_) + + +def _git(dothm, *args): + return subprocess.run(["git", "-C", str(dothm), *args], check=True, + capture_output=True, text=True).stdout + + +def snapshot_hm(repo_path): + """Capture a repository's tracked state: state files, index and refs.""" + repo_path = Path(repo_path) + dothm = repo_path if repo_path.name.endswith(".hm") else repo_path / ".hm" + return { + "files": {name: (dothm / name).read_bytes() for name in STATE_FILES + if (dothm / name).is_file()}, + "index": _git(dothm, "ls-files", "--stage"), + "refs": _git(dothm, "for-each-ref", "--format=%(refname) %(objectname)"), + "head": subprocess.run(["git", "-C", str(dothm), "symbolic-ref", "-q", "HEAD"], + capture_output=True, text=True).stdout.strip() + or _git(dothm, "rev-parse", "HEAD").strip(), + } + + +def check_invariants(repo_path, *, fsck=True): + """Check a repository for corruption and leftover temporary files. + + Args: + repo_path: Worktree or bare ``.hm`` path. + fsck (bool): Also run ``git fsck``, which is slower. + + Returns: + Violations: Empty (falsy) when every invariant holds. + """ + found = Violations() + try: + repo = Repo(repo_path) + repo.status() + except Exception as exc: + found.corruption.append( + f"repository does not open: {type(exc).__name__}: {exc}") + return found + + objects = repo.objects.root + if objects.is_dir(): + for path in objects.glob("??/*"): + if path.name.startswith("."): + found.debris.append(str(path)) + continue + digest = hashlib.sha1(path.read_bytes()).hexdigest() + if digest != path.parent.name + path.name: + found.corruption.append(f"object {path} has checksum {digest}") + + dothm = Path(repo.dothm.path) + for head in repo.dothm.heads: + try: + text = _git(dothm, "show", f"{head.name}:data.tsv") + except subprocess.CalledProcessError: + continue + frame = pd.read_csv(StringIO(text), sep="\t", dtype=str, + keep_default_na=False) + if "sha1" in frame.columns: + for sha1 in frame.sha1: + if sha1 and not repo.objects.contains(sha1): + found.corruption.append( + f"branch {head.name} references missing object {sha1}") + + if fsck: + result = subprocess.run(["git", "-C", str(dothm), "fsck", "--no-dangling"], + capture_output=True, text=True) + if result.returncode != 0: + found.corruption.append(f"git fsck: {result.stdout}{result.stderr}") + + roots = {Path(repo_path)} + if repo.worktree is not None: + roots.add(Path(repo.worktree)) + for root in roots: + for pattern in DEBRIS_PATTERNS: + found.debris.extend(str(path) for path in root.glob(pattern)) + found.debris = sorted(set(found.debris)) + return found diff --git a/test/conftest.py b/test/conftest.py index ca5f752..fa727cd 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,11 +1,14 @@ from pathlib import Path +from urllib.parse import urlsplit import pytest +import requests from hallmark import ParaFrame, Repo from hallmark.repo.dothm import Dothm from hallmark.remote.download import _select_remote_config, execute_download_plan from hallmark.remote.plan import DownloadItem, DownloadPlan +from http_server import DatasetHTTPServer @pytest.hookimpl(hookwrapper=True) @@ -158,3 +161,42 @@ def download_selection(repo, output, selected, max_workers=4, show_progress=Fals backend_options=remote.get("backend_options") or {}) return execute_download_plan(repo, plan, approved=True, max_workers=max_workers, show_progress=show_progress) + + +@pytest.fixture +def http_server(monkeypatch): + """Start loopback HTTP servers for directories; stop them at teardown. + + Call the fixture value with a directory and an optional listing style. + Proxies are disabled so requests reach 127.0.0.1 directly. + """ + for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", + "http_proxy", "https_proxy", "all_proxy"): + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("NO_PROXY", "127.0.0.1,localhost") + servers = [] + + def start(root, listing_style="apache"): + server = DatasetHTTPServer(root, listing_style=listing_style).start() + servers.append(server) + return server + + try: + yield start + finally: + for server in servers: + server.close() + + +@pytest.fixture +def loopback_only(monkeypatch): + """Fail any HTTP request that would leave this machine.""" + send = requests.adapters.HTTPAdapter.send + + def guarded(self, request, *args, **kwargs): + host = urlsplit(request.url).hostname + if host not in {"127.0.0.1", "localhost", "::1"}: + raise AssertionError(f"Test attempted a non-loopback request to {host}") + return send(self, request, *args, **kwargs) + + monkeypatch.setattr(requests.adapters.HTTPAdapter, "send", guarded) diff --git a/test/http_server.py b/test/http_server.py index a4df579..96295ae 100644 --- a/test/http_server.py +++ b/test/http_server.py @@ -1,8 +1,337 @@ -"""A small loopback HTTP server that serves files, indexes and redirects.""" +"""Loopback HTTP servers for tests. -import threading +``DatasetHTTPServer`` serves a directory with scripted faults; +``LocalServer`` serves an in-memory set of files with indexes, redirects +and basic authentication. + +Unlike ``mock_server.MockServer``, this is a real server: requests, +urllib3 and Hallmark's HTTP backend exercise their actual network code, +including status handling, short reads, resets and timeouts. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import fnmatch +import html from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from urllib.parse import unquote, urlsplit +import os +from pathlib import Path +import socket +import struct +import threading +import time +from urllib.parse import quote, unquote, urlsplit + + +@dataclass(frozen=True) +class Status: + """Respond with an HTTP status and an empty body.""" + code: int + headers: tuple = () + + +@dataclass(frozen=True) +class ResetAfter: + """Announce the full length, send ``nbytes``, then reset the connection.""" + nbytes: int + + +@dataclass(frozen=True) +class ShortBody: + """Announce the full length, send ``nbytes``, then close cleanly.""" + nbytes: int + + +@dataclass(frozen=True) +class NoLengthTruncate: + """Omit Content-Length, send ``nbytes``, then close cleanly.""" + nbytes: int + + +@dataclass(frozen=True) +class BitFlip: + """Serve the file with one byte inverted at ``offset``.""" + offset: int = 0 + + +@dataclass(frozen=True) +class Stall: + """Wait before responding, or after the headers when ``after_headers``.""" + seconds: float + after_headers: bool = False + + +@dataclass(frozen=True) +class SlowDrip: + """Send the body in ``chunk``-byte pieces ``delay`` seconds apart.""" + chunk: int = 1024 + delay: float = 0.05 + + +@dataclass(frozen=True) +class Redirect: + """Redirect to ``location`` (an absolute or server-relative URL).""" + location: str + code: int = 302 + + +@dataclass(frozen=True) +class NotAListing: + """Serve a directory as an ordinary page without listing markers.""" + + +@dataclass +class _Rule: + pattern: str + fault: object + method: str + remaining: int | None + + +@dataclass +class FaultPlan: + """Faults to apply to matching request paths, consumed per request. + + Paths are relative to the served root, without a leading slash; + directories end with ``/``. Rules are keyed by path rather than by a + global request count so concurrent download workers stay deterministic. + """ + _rules: list = field(default_factory=list) + _lock: threading.Lock = field(default_factory=threading.Lock) + + def add(self, pattern, fault, *, times=1, method="GET"): + """Apply ``fault`` to the next ``times`` matches (None: every match).""" + with self._lock: + self._rules.append(_Rule(pattern, fault, method, times)) + + def clear(self): + with self._lock: + self._rules.clear() + + def take(self, method, path): + with self._lock: + for rule in self._rules: + if (rule.method in {method, "*"} and rule.remaining != 0 + and fnmatch.fnmatchcase(path, rule.pattern)): + if rule.remaining is not None: + rule.remaining -= 1 + return rule.fault + return None + + +_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", + "Jul", "Aug", "Sep", "Oct", "Nov", "Dec") + + +def _listing(style, request_path, children): + """Render ``children`` as (name, is_directory, size) in a server's style.""" + title = html.escape(request_path) + if style == "python": + items = "".join( + f'
  • ' + f'{html.escape(name)}{"/" if is_dir else ""}
    • \n' + for name, is_dir, _ in children) + return (f'\n' + f"Directory listing for {title}\n\n" + f"

    Directory listing for {title}

    \n
    \n\n
    \n\n\n") + if style == "apache": + stamp = time.gmtime(1759600000) + date = (f"{stamp.tm_mday:02d}-{_MONTHS[stamp.tm_mon - 1]}-{stamp.tm_year} " + f"{stamp.tm_hour:02d}:{stamp.tm_min:02d}") + rows = "".join( + f'' + f'{html.escape(name)}{"/" if is_dir else ""}' + f' {date} {"-" if is_dir else size}\n' + for name, is_dir, size in children) + return (f"Index of {title}\n" + f"

    Index of {title}

    "
+                'Name  Last modified'
+                '  Size\n
    '
+                'Parent Directory\n'
+                f"{rows}
    \n") + if style == "cyverse": + rows = "".join( + f'' + f'' + f"{html.escape(name)}" + f'{"" if is_dir else size}' + for name, is_dir, size in children) + return f"{rows}
    " + raise ValueError(f"Unknown listing style {style!r}") + + +class _Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + # Idle keep-alive connections end so the server can shut down promptly. + timeout = 5 + + def log_message(self, format, *args): + pass + + def do_HEAD(self): + self._respond(head=True) + + def do_GET(self): + self._respond(head=False) + + def _respond(self, head): + server = self.server.owner + request_path = unquote(urlsplit(self.path).path, errors="strict") + relative = request_path.lstrip("/") + with server._lock: + server.requests.append((self.command, relative)) + fault = server.faults.take(self.command, relative) + if isinstance(fault, Status): + return self._send_empty(fault.code, dict(fault.headers)) + if isinstance(fault, Redirect): + return self._send_empty(fault.code, {"Location": fault.location}) + if isinstance(fault, Stall) and not fault.after_headers: + server.release.wait(fault.seconds) + + target = (server.root / relative).resolve() + if server.root.resolve() not in (target, *target.parents): + return self._send_empty(404) + if target.is_dir(): + if not request_path.endswith("/"): + return self._send_empty(301, {"Location": request_path + "/"}) + if isinstance(fault, NotAListing): + body = b"

    Welcome

    " + else: + children = sorted( + (child.name, child.is_dir(), + None if child.is_dir() else child.stat().st_size) + for child in target.iterdir()) + body = _listing(server.listing_style, request_path, + children).encode("utf-8") + return self._send_body(body, "text/html; charset=utf-8", head, fault) + if not target.is_file(): + return self._send_empty(404) + body = target.read_bytes() + if isinstance(fault, BitFlip) and body: + index = min(fault.offset, len(body) - 1) + body = body[:index] + bytes([body[index] ^ 0xFF]) + body[index + 1:] + return self._send_body(body, "application/octet-stream", head, fault) + + def _send_empty(self, code, headers=None): + self.send_response(code) + for name, value in (headers or {}).items(): + self.send_header(name, value) + self.send_header("Content-Length", "0") + self.end_headers() + + def _send_body(self, body, content_type, head, fault): + server = self.server.owner + self.send_response(200) + self.send_header("Content-Type", content_type) + if isinstance(fault, NoLengthTruncate): + self.send_header("Connection", "close") + self.close_connection = True + else: + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if head: + return + if isinstance(fault, Stall) and fault.after_headers: + self.wfile.flush() + server.release.wait(fault.seconds) + if isinstance(fault, (ResetAfter, ShortBody, NoLengthTruncate)): + self.wfile.write(body[:fault.nbytes]) + self.wfile.flush() + self.close_connection = True + if isinstance(fault, ResetAfter): + # A zero linger timeout makes close() send RST instead of FIN. + self.connection.setsockopt( + socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0)) + return + if isinstance(fault, SlowDrip): + for start in range(0, len(body), fault.chunk): + self.wfile.write(body[start:start + fault.chunk]) + self.wfile.flush() + if server.release.wait(fault.delay): + break + return + self.wfile.write(body) + + +class _Server(ThreadingHTTPServer): + daemon_threads = True + + def handle_error(self, request, client_address): + # Clients that abandon faulted transfers are expected; stay quiet. + pass + + +class DatasetHTTPServer: + """A loopback HTTP server for a directory, with faults and a request log. + + Args: + root: Directory to serve. + listing_style: ``"python"`` (http.server), ``"apache"`` (with sizes) + or ``"cyverse"`` (CyVerse WebDAV table rows with sizes). + """ + + def __init__(self, root, listing_style="python"): + self.root = Path(root) + self.root.mkdir(parents=True, exist_ok=True) + self.listing_style = listing_style + self.faults = FaultPlan() + self.requests = [] + # Set at shutdown so stalled and dripping responses end promptly. + self.release = threading.Event() + self._lock = threading.Lock() + self._httpd = _Server(("127.0.0.1", 0), _Handler) + self._httpd.owner = self + self._thread = threading.Thread(target=self._httpd.serve_forever, + kwargs={"poll_interval": 0.05}, daemon=True) + + @property + def port(self): + return self._httpd.server_address[1] + + def url(self, path=""): + """Return the URL for ``path`` relative to the served root.""" + return f"http://127.0.0.1:{self.port}/" + quote(path, safe="/") + + def gets(self, prefix=""): + """Paths requested with GET, in order, optionally under ``prefix``.""" + with self._lock: + return [path for method, path in self.requests + if method == "GET" and path.startswith(prefix)] + + def payload_gets(self): + """GET requests for files rather than directory listings.""" + return [path for path in self.gets() + if path and not path.endswith("/") + and (self.root / path).is_file()] + + def start(self): + self._thread.start() + return self + + def close(self): + self.release.set() + self._httpd.shutdown() + self._httpd.server_close() + self._thread.join(timeout=5) + + def __enter__(self): + return self.start() + + def __exit__(self, *exc): + self.close() + + +def write_file(root, relative, data): + """Write ``data`` (bytes or text) at ``root/relative``, creating folders.""" + path = Path(root) / relative + path.parent.mkdir(parents=True, exist_ok=True) + if isinstance(data, str): + data = data.encode("utf-8") + path.write_bytes(data) + os.utime(path, (1759600000, 1759600000)) + return path class LocalServer: diff --git a/test/survey.py b/test/survey.py new file mode 100644 index 0000000..99fdc4f --- /dev/null +++ b/test/survey.py @@ -0,0 +1,170 @@ +"""Synthetic survey data releases and a CLI helper for workflow tests. + +The layouts follow real releases, scaled down to a few kilobytes: +DESI DR1 healpix directories with sha256 manifests, EHT UVFITS files on +CyVerse, a simulation export and a telescope tree. Contents are +deterministic for a seed so failures reproduce. +""" + +from __future__ import annotations + +import hashlib +import random + +from click.testing import CliRunner + +from hallmark.cli import hallmark +from hallmark.utils import use_working_directory +from http_server import write_file + +DESI_DARK = "public/dr1/spectro/redux/iron/healpix/main/dark" +EHT_UVFITS = "EHTC_FirstM87Results_Apr2019/uvfits" + + +def _fits_like(rng, size): + """Bytes that start like a FITS header, padded to ``size``.""" + header = b"SIMPLE = T".ljust(80) + b"END".ljust(80) + return header + rng.randbytes(max(0, size - len(header))) + + +def _write_manifest(root, directory, names, files, name): + """Write a sha256sum-style manifest for ``names`` in ``directory``.""" + lines = "".join( + f"{hashlib.sha256(files[f'{directory}/{n}']).hexdigest()} {n}\n" + for n in names) + write_file(root, f"{directory}/{name}", lines) + + +def make_desi_like_release(root, pixels=((230, 23040), (230, 23041)), seed=0): + """Write a DESI-like healpix release under ``root``. + + Each pixel directory has redrock and qso_mgii FITS-like files, an + exposure CSV and a sha256 manifest named like the real release. + + Returns: + dict: Release-relative path to contents, excluding manifests. + """ + rng = random.Random(seed) + files = {} + for group, pixel in pixels: + directory = f"{DESI_DARK}/{group}/{pixel}" + names = [f"redrock-main-dark-{pixel}.fits", + f"qso_mgii-main-dark-{pixel}.fits", + f"hpixexp-main-dark-{pixel}.csv"] + for name in names: + if name.endswith(".csv"): + data = f"EXPID,NIGHT\n{rng.randrange(10**5)},20210514\n".encode() + else: + data = _fits_like(rng, 2880 + rng.randrange(2880)) + write_file(root, f"{directory}/{name}", data) + files[f"{directory}/{name}"] = data + _write_manifest(root, directory, names, files, + f"redux_iron_healpix_main_dark_{group}_{pixel}.sha256sum") + return files + + +def make_eht_like_release(root, seed=0): + """Write EHT-like UVFITS files without a checksum manifest. + + Returns: + dict: Release-relative path to contents. + """ + rng = random.Random(seed) + files = {} + for day in (95, 96, 100, 101): + for band in ("lo", "hi"): + path = (f"{EHT_UVFITS}/SR1_M87_2017_{day:03d}_{band}" + "_hops_netcal_StokesI.uvfits") + files[path] = _fits_like(rng, 2880 + rng.randrange(1440)) + write_file(root, path, files[path]) + return files + + +def make_lab_export(root, runs=2, frames=3, seed=0): + """Write a simulation export laid out as ``run{run}/frame{frame}.h5``. + + Returns: + dict: Export-relative path to contents. + """ + rng = random.Random(seed) + files = {} + for run in range(1, runs + 1): + for frame in range(frames): + path = f"run{run}/frame{frame}.h5" + files[path] = b"\x89HDF\r\n\x1a\n" + rng.randbytes(512) + write_file(root, path, files[path]) + return files + + +def make_telescope_tree(root, sites=("ALMA", "SMA"), years=(2024, 2025), + days=(95, 96), seed=0): + """Write observations laid out as ``{site}/{year:d}/{day:d}.fits``. + + Returns: + dict: Worktree-relative path to contents. + """ + rng = random.Random(seed) + files = {} + for site in sites: + for year in years: + for day in days: + path = f"{site}/{year}/{day}.fits" + files[path] = _fits_like(rng, 2880) + write_file(root, path, files[path]) + return files + + +def apply_dr_update(root, files, pixel=(230, 23042), seed=1): + """Reprocess one redrock file and add a pixel, updating manifests. + + Returns: + tuple: The changed path and the paths of the new pixel's files. + """ + rng = random.Random(seed) + changed = sorted(path for path in files if "/redrock-" in path)[0] + files[changed] = _fits_like(rng, len(files[changed])) + write_file(root, changed, files[changed]) + directory = changed.rsplit("/", 1)[0] + group, number = directory.split("/")[-2:] + names = sorted(path.rsplit("/", 1)[1] for path in files + if path.rsplit("/", 1)[0] == directory) + _write_manifest(root, directory, names, files, + f"redux_iron_healpix_main_dark_{group}_{number}.sha256sum") + added = make_desi_like_release(root, pixels=(pixel,), seed=seed) + files.update(added) + return changed, sorted(added) + + +def apply_silent_corruption(root, path): + """Change a published file without updating its manifest.""" + target = root / path + data = bytearray(target.read_bytes()) + data[-1] ^= 0xFF + target.write_bytes(bytes(data)) + + +class Hm: + """Run ``hm`` commands in-process through Click's test runner.""" + + def __init__(self): + self.runner = CliRunner() + + def __call__(self, *args, cwd, input=None, expect=0): + """Run ``hm ARGS`` in ``cwd`` and check its exit code. + + Args: + cwd: Directory to run in. + input (str, optional): Text for prompts, such as ``"y\\n"``. + expect (int | None): Required exit code, or None to skip the check. + + Returns: + click.testing.Result: The completed invocation. + """ + with use_working_directory(cwd): + result = self.runner.invoke(hallmark, [str(arg) for arg in args], + input=input) + if expect is not None and result.exit_code != expect: + raise AssertionError( + f"hm {' '.join(map(str, args))} exited {result.exit_code}, " + f"expected {expect}:\n{result.output}") from result.exception + return result diff --git a/test/test_chaos_network.py b/test/test_chaos_network.py new file mode 100644 index 0000000..55e973c --- /dev/null +++ b/test/test_chaos_network.py @@ -0,0 +1,186 @@ +"""Network faults during remote scans and downloads. + +A real loopback HTTP server misbehaves on chosen paths. Each failed file +must leave no partial destination or temporary file, other files must +still arrive intact, and a failed scan must leave the catalog unchanged. +""" + +import hashlib +import os + +import pytest + +from chaos import check_invariants, snapshot_hm +from hallmark import Repo +import hallmark.transport.http as http_transport +from http_server import (BitFlip, NoLengthTruncate, NotAListing, Redirect, + ResetAfter, ShortBody, SlowDrip, Stall, Status, write_file) +from survey import Hm + +pytestmark = [pytest.mark.chaos, pytest.mark.usefixtures("loopback_only")] + +FILES = {f"rel/f{i}.bin": os.urandom(40000) for i in range(4)} + +# Faults that the HTTP transport detects without any checksum. +TRANSPORT_FAULTS = { + "reset": ResetAfter(1000), + "short-body": ShortBody(1000), + "500": Status(500), + "503": Status(503), + "429": Status(429, (("Retry-After", "1"),)), + "404": Status(404), + "stall-before-headers": Stall(5), + "stall-mid-body": Stall(5, after_headers=True), +} +# Faults that only a published checksum can detect. +CONTENT_FAULTS = { + "bit-flip": BitFlip(100), + "unannounced-truncation": NoLengthTruncate(1000), +} + + +@pytest.fixture(autouse=True) +def fast_timeouts(monkeypatch): + monkeypatch.setattr(http_transport, "REMOTE_REQUEST_TIMEOUT", (2, 0.5)) + + +@pytest.fixture +def release(tmp_path, http_server): + """Serve FILES; ``publish_checksums`` adds a sha256 manifest.""" + www = tmp_path / "www" + for path, data in FILES.items(): + write_file(www, path, data) + + def serve(publish_checksums=True, listing_style="apache"): + if publish_checksums: + write_file(www, "rel/sha256sum.txt", "".join( + f"{hashlib.sha256(data).hexdigest()} {path.split('/')[1]}\n" + for path, data in FILES.items())) + return http_server(www, listing_style=listing_style) + + return serve + + +def _catalog_repo(path, server): + repo = Repo.init(path) + repo.add(server.url("rel/") + "{name}.bin") + repo.commit("Catalog release") + return repo + + +def _download_all(repo, max_workers=2): + """Download every planned file; return the result and the plan's size.""" + plan = repo.plan_download(all_files=True) + result = repo.download(plan, approved=True, max_workers=max_workers) + return result, len(plan.items) + + +def _assert_only_faulted_file_failed(repo, outcome, faulted, server): + # A published manifest is catalogued and downloaded as well, so count + # successes against the plan rather than against FILES. + result, planned = outcome + assert result["failed"] == 1, result + assert result["succeeded"] == planned - 1, result + for path, data in FILES.items(): + destination = repo.worktree / path.split("/", 1)[1] + if path == faulted: + assert not destination.exists() + else: + assert destination.read_bytes() == data + errors = " ".join(map(str, dict(result["errors"]).values()) + if isinstance(result["errors"], dict) else result["errors"]) + assert server.url("rel/") not in errors + problems = check_invariants(repo.worktree) + assert not problems, problems + + +@pytest.mark.parametrize("publish_checksums", [True, False], + ids=["checksums", "no-checksums"]) +@pytest.mark.parametrize("fault", TRANSPORT_FAULTS.values(), ids=TRANSPORT_FAULTS) +def test_transport_fault_fails_only_that_file(tmp_path, release, fault, + publish_checksums): + server = release(publish_checksums) + repo = _catalog_repo(tmp_path / "repo", server) + server.faults.add("rel/f1.bin", fault) + result = _download_all(repo) + _assert_only_faulted_file_failed(repo, result, "rel/f1.bin", server) + + +@pytest.mark.parametrize("fault", CONTENT_FAULTS.values(), ids=CONTENT_FAULTS) +def test_published_checksum_rejects_damaged_content(tmp_path, release, fault): + server = release(publish_checksums=True) + repo = _catalog_repo(tmp_path / "repo", server) + server.faults.add("rel/f2.bin", fault) + result = _download_all(repo) + _assert_only_faulted_file_failed(repo, result, "rel/f2.bin", server) + + +def test_catalogued_size_rejects_unannounced_truncation(tmp_path, release): + server = release(publish_checksums=False, listing_style="apache") + repo = _catalog_repo(tmp_path / "repo", server) + server.faults.add("rel/f3.bin", NoLengthTruncate(1000)) + result = _download_all(repo) + _assert_only_faulted_file_failed(repo, result, "rel/f3.bin", server) + + +def test_concurrent_faults_fail_independently(tmp_path, release): + server = release(publish_checksums=True) + repo = _catalog_repo(tmp_path / "repo", server) + server.faults.add("rel/f0.bin", ResetAfter(10)) + server.faults.add("rel/f2.bin", BitFlip(0)) + server.faults.add("rel/f3.bin", SlowDrip(chunk=4096, delay=0.01)) + result, planned = _download_all(repo, max_workers=4) + assert (result["succeeded"], result["failed"]) == (planned - 2, 2), result + assert (repo.worktree / "f3.bin").read_bytes() == FILES["rel/f3.bin"] + assert not check_invariants(repo.worktree) + + +def test_rerun_after_transient_failure_completes(tmp_path, release): + server = release(publish_checksums=True) + repo = _catalog_repo(tmp_path / "repo", server) + server.faults.add("rel/f1.bin", Status(503), times=1) + assert _download_all(repo)[0]["failed"] == 1 + plan = repo.plan_download(file_paths=["f1.bin"]) + assert repo.download(plan, approved=True)["succeeded"] == 1 + for path, data in FILES.items(): + assert (repo.worktree / path.split("/", 1)[1]).read_bytes() == data + assert not check_invariants(repo.worktree) + + +SCAN_FAULTS = { + "listing-500": ("rel/sub/", Status(500)), + "listing-reset": ("rel/sub/", ResetAfter(10)), + "listing-stall": ("rel/sub/", Stall(5)), + "not-a-listing": ("rel/sub/", NotAListing()), + "redirect-off-host": ("rel/sub/", Redirect("http://127.0.0.2:9/elsewhere/")), + "redirect-outside-root": ("rel/sub/", Redirect("/other/")), + "manifest-500": ("rel/sub/sha256sum.txt", Status(500)), + "manifest-reset": ("rel/sub/sha256sum.txt", ResetAfter(5)), +} + + +@pytest.mark.parametrize("pattern, fault", SCAN_FAULTS.values(), ids=SCAN_FAULTS) +def test_failed_rescan_keeps_catalog_and_reports_cleanly( + tmp_path, http_server, monkeypatch, pattern, fault): + www = tmp_path / "www" + for name in ("top.bin", "sub/a.bin", "sub/b.bin"): + write_file(www, f"rel/{name}", os.urandom(64)) + write_file(www, "rel/sub/sha256sum.txt", "".join( + f"{hashlib.sha256((www / 'rel/sub' / n).read_bytes()).hexdigest()} {n}\n" + for n in ("a.bin", "b.bin"))) + server = http_server(www) + monkeypatch.delenv("HALLMARK_AUTH_FILE", raising=False) + hm = Hm() + repo = tmp_path / "repo" + hm("init", repo, cwd=tmp_path) + url = server.url("rel/") + "{name}.bin" + hm("add", url, cwd=repo) + hm("commit", "-m", "Catalog release", cwd=repo) + before = snapshot_hm(repo) + + server.faults.add(pattern, fault) + result = hm("add", url, cwd=repo, expect=None) + assert result.exit_code != 0, result.output + assert "Traceback" not in result.output + assert snapshot_hm(repo) == before + assert not check_invariants(repo) From 06b0bdd5990c24e8535d22b704e55385e2a7da52 Mon Sep 17 00:00:00 2001 From: Dennis Date: Mon, 26 Oct 2026 03:01:24 -0700 Subject: [PATCH 2/3] Port the crash, storage and monkey chaos tests The fault-injection suite that found #53 to #58 lived only on ci/test-automation and had drifted from the code. Bring over the helpers, the kill-at-a-point child process, and the three test modules under the chaos marker, adapted to the current layout: branches are created with hm branch before checkout, partial downloads live in .hm/objects/tmp (now counted as debris after a finished operation), download counts come from the plan because a published manifest is catalogued too, and the storage tests use distinct file names so one temporary file can be targeted. The former #58 xfail passes and becomes a regression test; #55, #56 and #57 keep strict xfails so their fixes flip them. The nightly workflow reruns the suite with fresh seeds and more steps from the default branch, and on pull requests that touch it. Locally: 50 passed and 5 xfailed in 38 s; both monkeys hold for three seeds at 150 steps. Co-Authored-By: Claude Fable 5.1 --- .github/workflows/nightly.yml | 85 ++++++++++++ test/chaos.py | 3 +- test/chaos_child.py | 60 +++++++++ test/test_chaos_crash.py | 187 +++++++++++++++++++++++++ test/test_chaos_monkey.py | 247 ++++++++++++++++++++++++++++++++++ test/test_chaos_storage.py | 129 ++++++++++++++++++ 6 files changed, 710 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/nightly.yml create mode 100644 test/chaos_child.py create mode 100644 test/test_chaos_crash.py create mode 100644 test/test_chaos_monkey.py create mode 100644 test/test_chaos_storage.py diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml new file mode 100644 index 0000000..d2702df --- /dev/null +++ b/.github/workflows/nightly.yml @@ -0,0 +1,85 @@ +# Runs the chaos tests with many more random operations and fresh seeds. + +name: Nightly + +on: + schedule: + - cron: "23 6 * * *" + workflow_dispatch: + inputs: + steps: + description: Random operations per seed + default: "1000" + seeds: + description: Comma-separated seeds (blank derives them from the run) + default: "" + # Scheduled and manual runs start only from the default branch. + pull_request: + paths: + - .github/workflows/nightly.yml + - test/chaos.py + - test/chaos_child.py + - test/http_server.py + - test/survey.py + - test/test_chaos_*.py + +permissions: + contents: read + +concurrency: + group: nightly-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + chaos: + runs-on: ${{ matrix.os }} + timeout-minutes: 45 + strategy: + fail-fast: false + matrix: + include: + - os: ubuntu-24.04 + python-version: "3.9" + lane: 1 + - os: ubuntu-24.04 + python-version: "3.14" + lane: 2 + - os: macos-latest + python-version: "3.14" + lane: 3 + steps: + - uses: actions/checkout@v7 + with: + persist-credentials: false + - uses: actions/setup-python@v7 + with: + python-version: ${{ matrix.python-version }} + - name: Install Hallmark and test tools + run: | + python -m pip install --upgrade pip + python -m pip install -e . --group dev + - name: Choose seeds and steps + env: + INPUT_SEEDS: ${{ inputs.seeds }} + INPUT_STEPS: ${{ inputs.steps }} + RUN_ID: ${{ github.run_id }} + LANE: ${{ matrix.lane }} + EVENT: ${{ github.event_name }} + run: | + seeds="${INPUT_SEEDS:-${RUN_ID}${LANE}1,${RUN_ID}${LANE}2,${RUN_ID}${LANE}3}" + if [ "$EVENT" = pull_request ]; then default_steps=300; else default_steps=1000; fi + steps="${INPUT_STEPS:-$default_steps}" + [[ "$seeds" =~ ^[0-9,]+$ && "$steps" =~ ^[0-9]+$ ]] || { echo "Invalid seeds or steps"; exit 1; } + echo "HALLMARK_CHAOS_SEED=$seeds" >> "$GITHUB_ENV" + echo "HALLMARK_CHAOS_STEPS=$steps" >> "$GITHUB_ENV" + echo "Chaos seeds: \`$seeds\`, steps per seed: $steps" >> "$GITHUB_STEP_SUMMARY" + - name: Run the chaos tests + run: python -m pytest -m chaos -ra --timeout=1800 --basetemp="$RUNNER_TEMP/pytest" + - name: Keep failed repositories for inspection + if: failure() + uses: actions/upload-artifact@v7 + with: + name: chaos-${{ runner.os }}-py${{ matrix.python-version }} + path: ${{ runner.temp }}/pytest + include-hidden-files: true + retention-days: 14 diff --git a/test/chaos.py b/test/chaos.py index fc15745..b0d384f 100644 --- a/test/chaos.py +++ b/test/chaos.py @@ -25,7 +25,8 @@ from hallmark import Repo STATE_FILES = ("config.yml", "meta.yml", "data.tsv") -DEBRIS_PATTERNS = ("**/.*.part", "**/.*.tmp", "**/.hallmark-checkout-*") +DEBRIS_PATTERNS = ("**/.*.part", "**/.*.tmp", "**/.hallmark-checkout-*", + "**/objects/tmp/*") @dataclass diff --git a/test/chaos_child.py b/test/chaos_child.py new file mode 100644 index 0000000..3c9d956 --- /dev/null +++ b/test/chaos_child.py @@ -0,0 +1,60 @@ +"""Run ``hm`` in a process that kills itself at a chosen point. + +Used by test_chaos_crash.py; pytest does not collect this module. +``HALLMARK_CHAOS_KILL`` holds a JSON object: + +* ``target``: ``"module:Class.method"`` or ``"module:function"`` to wrap; +* ``nth``: which matching call triggers the kill (1-based); +* ``when``: ``"before"`` or ``"after"`` the wrapped call runs; +* ``match``: optional text that must appear in the call's arguments. + +The process sends itself SIGKILL, so no cleanup handlers run, as in a +power cut or an out-of-memory kill. Arguments are passed to ``hm``, or +``--api CODE`` runs Python with ``Repo`` imported, in the current folder. +""" + +import importlib +import json +import os +import signal +import sys + + +def _install(spec): + module_name, _, qualname = spec["target"].partition(":") + owner = importlib.import_module(module_name) + *parents, attribute = qualname.split(".") + for name in parents: + owner = getattr(owner, name) + original = getattr(owner, attribute) + count = 0 + + def wrapper(*args, **kwargs): + nonlocal count + if spec.get("match", "") in " ".join(map(str, args)): + count += 1 + if count == spec["nth"] and spec.get("when", "before") == "before": + os.kill(os.getpid(), signal.SIGKILL) + result = original(*args, **kwargs) + if count == spec["nth"] and spec.get("when") == "after": + os.kill(os.getpid(), signal.SIGKILL) + return result + return original(*args, **kwargs) + + setattr(owner, attribute, wrapper) + + +def main(argv): + spec = os.environ.get("HALLMARK_CHAOS_KILL") + if spec: + _install(json.loads(spec)) + if argv[:1] == ["--api"]: + from hallmark import Repo + exec(argv[1], {"Repo": Repo}) + return 0 + from hallmark.cli import hallmark + return hallmark.main(args=argv, prog_name="hm") + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/test/test_chaos_crash.py b/test/test_chaos_crash.py new file mode 100644 index 0000000..f5460bb --- /dev/null +++ b/test/test_chaos_crash.py @@ -0,0 +1,187 @@ +"""Crash consistency: hm is killed with SIGKILL partway through a command. + +A killed command may leave temporary files, but never a corrupt +repository, and running the same command again must finish the job. +""" + +import hashlib +import json +import os +from pathlib import Path +import signal +import subprocess +import sys +import time + +import pytest + +from chaos import check_invariants, snapshot_hm +from hallmark import Repo +from http_server import SlowDrip, write_file +from survey import make_telescope_tree + +pytestmark = [ + pytest.mark.chaos, + pytest.mark.usefixtures("loopback_only"), + pytest.mark.skipif(not hasattr(signal, "SIGKILL"), reason="needs SIGKILL"), +] + +CHILD = str(Path(__file__).with_name("chaos_child.py")) +FMT = "{site}/{year:d}/{day:d}.fits" + + +def _run(cwd, args, kill=None, input=None): + env = dict(os.environ) + env.pop("HALLMARK_CHAOS_KILL", None) + if kill is not None: + env["HALLMARK_CHAOS_KILL"] = json.dumps(kill) + return subprocess.run([sys.executable, CHILD, *args], cwd=cwd, env=env, + input=input, capture_output=True, text=True, timeout=120) + + +def _killed(cwd, args, kill, input=None): + result = _run(cwd, args, kill, input) + # Otherwise the kill point was never reached and the test proves nothing. + assert result.returncode == -signal.SIGKILL, result.stdout + result.stderr + problems = check_invariants(cwd) + assert not problems.corruption, problems + return result + + +def _rerun(cwd, args, input=None): + result = _run(cwd, args, input=input) + assert result.returncode == 0, result.stdout + result.stderr + problems = check_invariants(cwd) + assert not problems, problems + return result + + +def _replace(nth, match, when="after"): + return {"target": "pathlib:Path.replace", "nth": nth, "when": when, + "match": match} + + +@pytest.fixture +def telescope(tmp_path): + repo = Repo.init(tmp_path / "obs") + files = make_telescope_tree(repo.worktree) + return repo, files + + +@pytest.mark.parametrize("nth", [1, 2, 3]) +def test_killed_add_recovers_on_rerun(tmp_path, telescope, nth): + repo, _ = telescope + reference = Repo.init(tmp_path / "reference") + make_telescope_tree(reference.worktree) + reference.add(FMT) + + _killed(repo.worktree, ["add", FMT], _replace(nth, "/.hm/")) + _rerun(repo.worktree, ["add", FMT]) + assert snapshot_hm(repo.worktree)["files"] == \ + snapshot_hm(reference.worktree)["files"] + + +@pytest.mark.parametrize("kill", [ + _replace(1, "/objects/"), + _replace(5, "/objects/"), + {"target": "git.index.base:IndexFile.commit", "nth": 1, "when": "before"}, +], ids=["first-object", "fifth-object", "before-git-commit"]) +def test_killed_commit_recovers_on_rerun(telescope, kill): + repo, files = telescope + repo.add(FMT) + before = snapshot_hm(repo.worktree)["refs"] + _killed(repo.worktree, ["commit", "-m", "Ingest"], kill) + assert snapshot_hm(repo.worktree)["refs"] == before + _rerun(repo.worktree, ["commit", "-m", "Ingest"]) + repo = Repo(repo.worktree) + assert "Ingest" in repo.log() + assert all(repo.objects.contains(hashlib.sha1(data).hexdigest()) + for data in files.values()) + + +@pytest.fixture +def two_branches(telescope): + """Commit main, then branch exp with two files changed; return to main.""" + repo, files = telescope + repo.add(FMT) + repo.commit("main data") + assert _run(repo.worktree, ["branch", "exp"]).returncode == 0 + assert _run(repo.worktree, ["checkout", "exp"]).returncode == 0 + for path in ("ALMA/2024/95.fits", "SMA/2025/96.fits"): + (repo.worktree / path).write_bytes(b"exp version of " + path.encode()) + exp = Repo(repo.worktree) + exp.add(FMT) + exp.commit("exp data") + assert _run(repo.worktree, ["checkout", "main"]).returncode == 0 + return repo, files + + +@pytest.mark.parametrize("nth", [ + pytest.param(1, marks=pytest.mark.xfail(strict=True, reason=( + "Bug: a checkout killed after moving a file to its backup folder " + "cannot be rerun; the file is reported missing (#56)"))), + pytest.param(3, marks=pytest.mark.xfail(strict=True, reason=( + "Bug: a checkout killed after moving a file to its backup folder " + "cannot be rerun; the file is reported missing (#56)"))), +]) +def test_killed_checkout_can_be_completed(two_branches, nth): + repo, _ = two_branches + _killed(repo.worktree, ["checkout", "exp"], _replace(nth, ".fits")) + _rerun(repo.worktree, ["checkout", "exp"]) + assert (repo.worktree / "SMA/2025/96.fits").read_bytes() == \ + b"exp version of SMA/2025/96.fits" + + +@pytest.mark.xfail(strict=True, reason=( + "Bug: add_worktree returns early when the destination already has .hm, " + "so a rerun after a crash leaves the worktree without its files (#57)")) +def test_killed_add_worktree_is_completed_by_rerun(telescope): + repo, files = telescope + repo.add(FMT) + repo.commit("main data") + code = 'Repo(".").add_worktree("obs2")' + # Killed after Git created the linked worktree, before restoring files. + _killed(repo.worktree, ["--api", code], { + "target": "hallmark.repo.objects:Objects.restore", "nth": 1, + "when": "before"}) + _rerun(repo.worktree, ["--api", code]) + second = repo.worktree.parent / "obs2" + for path, data in files.items(): + assert (second / path).read_bytes() == data + + +def test_killed_download_keeps_partial_data_separate(tmp_path, http_server): + www = tmp_path / "www" + data = os.urandom(400_000) + write_file(www, "rel/big.bin", data) + write_file(www, "rel/sha256sum.txt", + f"{hashlib.sha256(data).hexdigest()} big.bin\n") + server = http_server(www) + repo = Repo.init(tmp_path / "repo") + repo.add(server.url("rel/") + "{name}.bin") + repo.commit("Catalog") + server.faults.add("rel/big.bin", SlowDrip(chunk=4096, delay=0.02), times=1) + + env = dict(os.environ) + env.pop("HALLMARK_CHAOS_KILL", None) + child = subprocess.Popen([sys.executable, CHILD, "download", "--all"], + cwd=repo.worktree, env=env, stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + child.stdin.write(b"y\n") + child.stdin.close() + deadline = time.monotonic() + 30 + partials = repo.objects.tmp.glob + while not any(path.stat().st_size for path in partials(".big.bin.*")): + assert child.poll() is None and time.monotonic() < deadline, \ + "download finished or stalled before it could be interrupted" + time.sleep(0.01) + child.kill() + child.wait(timeout=10) + + # The partial transfer stays in the store's temporary folder, never at + # the destination; the rerun completes and verifies the file. + assert not (repo.worktree / "big.bin").exists() + assert list(partials(".big.bin.*")) + assert not check_invariants(repo.worktree).corruption + _run(repo.worktree, ["download", "--all"], input="y\n") + assert (repo.worktree / "big.bin").read_bytes() == data diff --git a/test/test_chaos_monkey.py b/test/test_chaos_monkey.py new file mode 100644 index 0000000..82ec916 --- /dev/null +++ b/test/test_chaos_monkey.py @@ -0,0 +1,247 @@ +"""Seeded random operation sequences checked against repository invariants. + +A "monkey" edits files, stages, commits, switches branches and downloads +while faults are injected at random. It does not model every rule; +instead it checks properties that must always hold: + +* operations succeed or raise a documented error; +* a failed operation changes neither the repository nor the worktree; +* after a checkout, every tracked file matches its catalogued checksum; +* the repository is never corrupt, and successful operations leave no + temporary files; +* a downloaded file always equals a version the server published. + +Set HALLMARK_CHAOS_SEED (comma-separated) and HALLMARK_CHAOS_STEPS to +explore further; nightly CI runs many more steps with fresh seeds. +""" + +import errno +import hashlib +import os +import random + +import pytest + +from chaos import check_invariants, fail_nth_replace, snapshot_hm +from hallmark import Repo +from hallmark.error import CheckoutError +import hallmark.transport.http as http_transport +from hallmark.transport.base import DownloadError +from hallmark.repo.config import row_to_path +from http_server import (BitFlip, NoLengthTruncate, ResetAfter, ShortBody, Status, + write_file) + +pytestmark = [pytest.mark.chaos, pytest.mark.usefixtures("loopback_only")] + +SEEDS = [int(seed) for seed in + os.environ.get("HALLMARK_CHAOS_SEED", "101,202,303").split(",") if seed] +STEPS = int(os.environ.get("HALLMARK_CHAOS_STEPS", "40")) +FAULT_RATE = float(os.environ.get("HALLMARK_CHAOS_FAULT_RATE", "0.2")) +FMT = "{site}/{year:d}/{day:d}.fits" +# Known bug the monkey must not trip over until it is fixed: +# * Dothm.save_state is not atomic when a pattern change rewrites +# config.yml (#55, test_chaos_storage.py), so the pattern never changes +# here and the add that first sets it runs without faults. +EXPECTED_ERRORS = (CheckoutError, DownloadError, FileNotFoundError, OSError, + RuntimeError, ValueError) + + +class Monkey: + """Seeded operation log with a reproducible failure report.""" + + def __init__(self, seed, nodeid): + self.seed, self.nodeid = seed, nodeid + self.rng = random.Random(seed) + self.log = [] + + def fail(self, step, message): + recent = "\n".join(f" {line}" for line in self.log[-30:]) + pytest.fail( + f"seed {self.seed}, step {step}: {message}\nLast operations:\n{recent}\n" + f"Reproduce: HALLMARK_CHAOS_SEED={self.seed} " + f"HALLMARK_CHAOS_STEPS={step + 1} pytest '{self.nodeid}'", + pytrace=False) + + +def _worktree_files(root): + return {path.relative_to(root).as_posix(): path.read_bytes() + for path in root.rglob("*.fits") if ".hm" not in path.parts} + + +def _tracked_files_match_catalog(repo): + """Return paths whose bytes differ from the current branch's catalog.""" + fmt = repo.state.config["data"][0]["fmt"] + wrong = [] + for _, row in repo.state.data.iterrows(): + path = repo.worktree / row_to_path(row, fmt) + if (not path.is_file() + or hashlib.sha1(path.read_bytes()).hexdigest() != row["sha1"]): + wrong.append(str(path.relative_to(repo.worktree))) + return wrong + + +@pytest.mark.parametrize("seed", SEEDS) +def test_local_monkey_keeps_repository_consistent(tmp_path, monkeypatch, request, + seed): + monkey = Monkey(seed, request.node.nodeid) + rng = monkey.rng + root = tmp_path / "obs" + Repo.init(root) + branches = {"main"} + patterned = committed = False + + for step in range(STEPS): + repo = Repo(root) + name = rng.choice(["write", "write", "delete", "add", "add", "commit", + "commit", "checkout"]) + if name == "checkout" and not committed: + name = "commit" + path = f"{rng.choice(['ALMA', 'SMA'])}/{rng.choice([2024, 2025])}/" \ + f"{rng.randrange(1, 4)}.fits" + target = rng.choice(sorted(branches) + [f"b{step}"]) + fault = None + if name in {"add", "commit", "checkout"} and rng.random() < FAULT_RATE: + fault = (rng.randrange(1, 4), rng.choice([errno.ENOSPC, errno.EIO])) + if name == "add" and not patterned: + fault = None + monkey.log.append(f"{step}: {name} path={path} target={target} " + f"fault={fault}") + + if name == "write": + write_file(root, path, rng.randbytes(rng.randrange(1, 200))) + continue + if name == "delete": + files = sorted(_worktree_files(root)) + if files: + (root / rng.choice(files)).unlink() + continue + + before, before_files = snapshot_hm(root), _worktree_files(root) + if fault is not None: + fail_nth_replace(monkeypatch, fault[0], code=fault[1]) + try: + if name == "add": + repo.add(FMT if not patterned else ".") + patterned = True + elif name == "commit": + committed = repo.commit(f"step {step}") or committed + else: + repo.checkout(target) + except EXPECTED_ERRORS as exc: + monkeypatch.undo() + monkey.log[-1] += f" -> {type(exc).__name__}: {exc}" + after = snapshot_hm(root) + if name == "commit": + # Objects stored before the failure are harmless extras. + unchanged = after["refs"] == before["refs"] and \ + after["files"] == before["files"] + else: + unchanged = after == before + if not unchanged: + monkey.fail(step, f"failed {name} changed repository state") + if _worktree_files(root) != before_files and name != "commit": + monkey.fail(step, f"failed {name} changed worktree files") + problems = check_invariants(root, fsck=False) + if problems.corruption: + monkey.fail(step, str(problems)) + continue + except Exception as exc: + monkeypatch.undo() + monkey.fail(step, f"undocumented {type(exc).__name__}: {exc}") + monkeypatch.undo() + monkey.log[-1] += " -> ok" + if name == "checkout": + branches.add(target) + wrong = _tracked_files_match_catalog(Repo(root)) + if wrong: + monkey.fail(step, f"checkout left wrong tracked files: {wrong}") + problems = check_invariants(root, fsck=step % 10 == 9) + if problems: + monkey.fail(step, str(problems)) + problems = check_invariants(root) + if problems.corruption: + monkey.fail(STEPS, str(problems)) + + +def test_checkout_before_first_commit_reports_a_documented_error(tmp_path): + # Issue #58: this used to surface a raw GitCommandError. + repo = Repo.init(tmp_path / "obs") + try: + repo.checkout("main") + except EXPECTED_ERRORS: + pass + + +@pytest.mark.parametrize("seed", SEEDS) +def test_remote_monkey_never_lands_unpublished_bytes(tmp_path, monkeypatch, + http_server, request, seed): + monkeypatch.setattr(http_transport, "REMOTE_REQUEST_TIMEOUT", (2, 0.5)) + monkey = Monkey(seed, request.node.nodeid) + rng = monkey.rng + www = tmp_path / "www" + names = [f"d{i}/f{j}.bin" for i in range(2) for j in range(3)] + published = {} + + def write_manifest(directory): + write_file(www, f"rel/{directory}/sha256sum.txt", "".join( + f"{hashlib.sha256((www / 'rel' / n).read_bytes()).hexdigest()} " + f"{n.split('/')[1]}\n" for n in names if n.startswith(directory + "/"))) + + def publish(name, stale=False): + data = rng.randbytes(rng.randrange(1000, 20000)) + write_file(www, f"rel/{name}", data) + published.setdefault(name, set()).add(hashlib.sha256(data).hexdigest()) + if not stale: + write_manifest(name.split("/")[0]) + + for name in names: + publish(name, stale=True) + for directory in ("d0", "d1"): + write_manifest(directory) + server = http_server(www) + url = server.url("rel/") + "{d}/{f}.bin" + repo = Repo.init(tmp_path / "repo") + repo.add(url) + repo.commit("Catalog") + faults = [ResetAfter(100), ShortBody(100), Status(500), Status(404), + BitFlip(7), NoLengthTruncate(100)] + + for step in range(STEPS): + name = rng.choice(["publish", "publish-stale", "rescan", "commit", + "download", "download", "delete"]) + target = rng.choice(names) + monkey.log.append(f"{step}: {name} {target}") + try: + if name == "publish": + publish(target) + elif name == "publish-stale": + publish(target, stale=True) + elif name == "rescan": + repo.add(url) + elif name == "commit": + repo.commit(f"step {step}") + elif name == "delete": + (repo.worktree / target).unlink(missing_ok=True) + else: + if rng.random() < 0.5: + server.faults.add(f"rel/{target}", rng.choice(faults)) + result = repo.download(repo.plan_download(all_files=True), + approved=True, max_workers=3) + monkey.log[-1] += f" -> {result['succeeded']} ok, " \ + f"{result['failed']} failed" + except EXPECTED_ERRORS as exc: + monkey.log[-1] += f" -> {type(exc).__name__}: {exc}" + except Exception as exc: + monkey.fail(step, f"undocumented {type(exc).__name__}: {exc}") + finally: + server.faults.clear() + repo = Repo(repo.worktree) + + for name_ in names: + path = repo.worktree / name_ + if path.is_file() and hashlib.sha256(path.read_bytes()).hexdigest() \ + not in published[name_]: + monkey.fail(step, f"{name_} holds bytes the server never published") + problems = check_invariants(repo.worktree, fsck=False) + if problems.corruption or (name.startswith("download") and problems.debris): + monkey.fail(step, str(problems)) diff --git a/test/test_chaos_storage.py b/test/test_chaos_storage.py new file mode 100644 index 0000000..3672518 --- /dev/null +++ b/test/test_chaos_storage.py @@ -0,0 +1,129 @@ +"""Local storage faults: full disks, I/O errors and read-only folders. + +Writes fail partway through staging, committing and downloading. Each +failure must surface as an error and leave the repository consistent: +no corrupt objects, no partial files and no half-updated state. +""" + +import errno +import hashlib +import os + +import pytest + +from chaos import (check_invariants, fail_nth_replace, fail_writes_after, + skip_if_root, snapshot_hm) +from hallmark import Repo +from http_server import write_file +from survey import make_telescope_tree + +pytestmark = [pytest.mark.chaos, pytest.mark.usefixtures("loopback_only")] + +FMT = "{site}/{year:d}/{day:d}.fits" + + +def _in_dothm(target): + return ".hm" in target.parts + + +@pytest.mark.parametrize("nth", [ + 1, + pytest.param(2, marks=pytest.mark.xfail(strict=True, reason=( + "Bug: Dothm.save_state writes config.yml, meta.yml and data.tsv " + "separately, so a failed later write leaves mixed generations (#55)"))), + pytest.param(3, marks=pytest.mark.xfail(strict=True, reason=( + "Bug: Dothm.save_state writes config.yml, meta.yml and data.tsv " + "separately, so a failed later write leaves mixed generations (#55)"))), +]) +def test_failed_state_write_keeps_one_generation(tmp_path, monkeypatch, nth): + reference = Repo.init(tmp_path / "reference") + make_telescope_tree(reference.worktree) + reference.add(FMT) + new = snapshot_hm(reference.worktree)["files"] + + repo = Repo.init(tmp_path / "repo") + make_telescope_tree(repo.worktree) + old = snapshot_hm(repo.worktree) + calls = fail_nth_replace(monkeypatch, nth, match=_in_dothm) + with pytest.raises(OSError): + repo.add(FMT) + monkeypatch.undo() + assert len(calls) == nth + + after = snapshot_hm(repo.worktree) + generations = {"old" if after["files"][name] == old["files"][name] else + "new" if after["files"][name] == new[name] else "other" + for name in after["files"]} + assert generations in ({"old"}, {"new"}), generations + assert after["index"] == old["index"] + assert not check_invariants(repo.worktree) + + +@pytest.mark.parametrize("code", [errno.ENOSPC, errno.EIO]) +@pytest.mark.parametrize("nth", [1, 3]) +def test_failed_object_store_keeps_commit_retryable(tmp_path, monkeypatch, nth, code): + repo = Repo.init(tmp_path / "repo") + files = make_telescope_tree(repo.worktree) + repo.add(FMT) + before = snapshot_hm(repo.worktree) + + objects = repo.objects.root + calls = fail_nth_replace(monkeypatch, nth, code=code, + match=lambda target: objects in target.parents) + with pytest.raises(OSError): + repo.commit("Ingest") + monkeypatch.undo() + assert len(calls) == nth + assert snapshot_hm(repo.worktree)["refs"] == before["refs"] + assert not check_invariants(repo.worktree) + + repo = Repo(repo.worktree) + repo.commit("Ingest") + assert all(repo.objects.contains(hashlib.sha1(data).hexdigest()) + for data in files.values()) + assert not check_invariants(repo.worktree) + + +@pytest.fixture +def served_release(tmp_path, http_server): + www = tmp_path / "www" + files = {f"rel/d{i}/f{i}.bin": os.urandom(20000) for i in range(3)} + for path, data in files.items(): + write_file(www, path, data) + write_file(www, "rel/sha256sum.txt", "".join( + f"{hashlib.sha256(data).hexdigest()} {path[4:]}\n" + for path, data in files.items())) + server = http_server(www) + repo = Repo.init(tmp_path / "repo") + repo.add(server.url("rel/") + "d{i:d}/f{j:d}.bin") + repo.commit("Catalog release") + return repo, files + + +def test_disk_full_mid_download_fails_only_that_file(monkeypatch, served_release): + repo, files = served_release + fail_writes_after(monkeypatch, 5000, + match=lambda path: path.name.startswith(".f1.bin.")) + plan = repo.plan_download(all_files=True) + result = repo.download(plan, approved=True) + monkeypatch.undo() + assert (result["succeeded"], result["failed"]) == (len(plan.items) - 1, 1), result + assert not (repo.worktree / "d1/f1.bin").exists() + assert (repo.worktree / "d0/f0.bin").read_bytes() == files["rel/d0/f0.bin"] + assert not check_invariants(repo.worktree) + + +def test_read_only_output_folder_fails_only_its_files(served_release): + skip_if_root() + repo, files = served_release + locked = repo.worktree / "d2" + locked.mkdir() + locked.chmod(0o555) + plan = repo.plan_download(all_files=True) + try: + result = repo.download(plan, approved=True) + finally: + locked.chmod(0o755) + assert (result["succeeded"], result["failed"]) == (len(plan.items) - 1, 1), result + assert not (locked / "f2.bin").exists() + assert not check_invariants(repo.worktree) From 32378500a5632f18dc196ed7bf792133a2f8ccda Mon Sep 17 00:00:00 2001 From: Dennis Date: Wed, 28 Oct 2026 01:12:09 -0700 Subject: [PATCH 3/3] Refuse an HTTP body shorter than its Content-Length on urllib3 1.x urllib3 1.x, which requests 2.26 still installs, does not check a body against its Content-Length unless asked: a connection closed early simply ends the body. On the lowest-dependencies CI job a checksum manifest cut off after five bytes was read as a manifest with no entries, so a rescan succeeded and silently dropped every published checksum; a download with no known size or checksum would likewise have stored a truncated file. urllib3 2 refuses such a body by default. HttpBackend now sets enforce_content_length on each urllib3 response it opens, for file transfers and for metadata (listings, manifests, catalogue files and WebDAV PROPFIND), so a short body fails as it does on urllib3 2. Responses that are not urllib3's, as test doubles are, are left alone. A manifest that ends cleanly but early joins the rescan faults. Locally: full suite with SSH tests on Python 3.9 with the oldest allowed dependencies (requests 2.26.0, urllib3 1.26.20) 1547 passed, 4 skipped, 5 xfailed; on Python 3.13 with current ones 1548 passed, 3 skipped, 5 xfailed; without this change both manifest faults fail on the oldest dependencies; ruff clean. Co-Authored-By: Claude Opus 5.5 --- mod/hallmark/transport/http.py | 13 ++++++++++--- test/test_chaos_network.py | 1 + 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/mod/hallmark/transport/http.py b/mod/hallmark/transport/http.py index 054115d..2940834 100644 --- a/mod/hallmark/transport/http.py +++ b/mod/hallmark/transport/http.py @@ -16,6 +16,7 @@ from requests.auth import AuthBase from urllib3.exceptions import (HTTPError as Urllib3Error, ProtocolError, ReadTimeoutError) +from urllib3.response import HTTPResponse as Urllib3Response from ..utils import REMOTE_REQUEST_TIMEOUT from .base import (DownloadError, RemoteConfigurationError, RemoteEntry, @@ -25,6 +26,12 @@ from .index import _parse_directory_listing, _redirected_directory +def _refuse_short_body(response): + """Make urllib3 1.x refuse a body shorter than its Content-Length, as 2.x does.""" + if isinstance(response.raw, Urllib3Response): + response.raw.enforce_content_length = True + return response + def _declared_charset(content_type): """ @@ -144,13 +151,13 @@ def _open_file_response(self, path): Asking for the identity encoding keeps servers from compressing a file on the fly, so its checksums still match. """ - return self.context.session().get( + return _refuse_short_body(self.context.session().get( self.context.remote.file_url(path), stream=True, timeout=REMOTE_REQUEST_TIMEOUT, headers={"Accept-Encoding": "identity"}, **self._auth_option(), - ) + )) def _download_error(self, exc): """Describe an HTTP failure without exposing response contents.""" @@ -211,7 +218,7 @@ def endpoint(parsed): response = self.context.session().request( method, url, headers=headers, data=data, **options) if response.status_code not in {301, 302, 303, 307, 308}: - return response, url + return _refuse_short_body(response), url location = response.headers.get("Location") response.close() if not isinstance(location, str) or not location: diff --git a/test/test_chaos_network.py b/test/test_chaos_network.py index 55e973c..a2ffc41 100644 --- a/test/test_chaos_network.py +++ b/test/test_chaos_network.py @@ -156,6 +156,7 @@ def test_rerun_after_transient_failure_completes(tmp_path, release): "redirect-outside-root": ("rel/sub/", Redirect("/other/")), "manifest-500": ("rel/sub/sha256sum.txt", Status(500)), "manifest-reset": ("rel/sub/sha256sum.txt", ResetAfter(5)), + "manifest-short-body": ("rel/sub/sha256sum.txt", ShortBody(5)), }