" + 'Name Last modified' + ' Size\n\n") + if style == "cyverse": + rows = "".join( + f'
' + 'Parent Directory\n' + f"{rows}
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 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_crash.py b/test/test_chaos_crash.py new file mode 100644 index 0000000..b98da14 --- /dev/null +++ b/test/test_chaos_crash.py @@ -0,0 +1,184 @@ +"""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, ["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 + parts = repo.worktree.glob + while not any(path.stat().st_size for path in parts(".big.bin.*.part")): + 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) + + assert not (repo.worktree / "big.bin").exists() + assert list(repo.worktree.glob(".big.bin.*.part")) + 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..4351014 --- /dev/null +++ b/test/test_chaos_monkey.py @@ -0,0 +1,254 @@ +"""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 bugs the monkey must not trip over until they are fixed: +# * Dothm.save_state is not atomic when a pattern change rewrites +# config.yml (test_chaos_storage.py), so the pattern never changes here +# and the add that first sets it runs without faults. +# * A truncated body without Content-Length is accepted when no checksum +# is published (test_chaos_network.py), so every file has a checksum. +# * Checkout before the first commit raises a raw GitCommandError +# (test_checkout_before_first_commit_reports_a_documented_error), so the +# monkey commits once before switching branches. +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)) + + +@pytest.mark.xfail(strict=True, reason=( + "Bug: before the first commit, checkout raises GitCommandError because " + "init leaves data.tsv uncommitted (#58)")) +def test_checkout_before_first_commit_reports_a_documented_error(tmp_path): + 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_network.py b/test/test_chaos_network.py new file mode 100644 index 0000000..a7e8e3e --- /dev/null +++ b/test/test_chaos_network.py @@ -0,0 +1,205 @@ +"""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): + plan = repo.plan_download(all_files=True) + return repo.download(plan, approved=True, max_workers=max_workers) + + +def _assert_only_faulted_file_failed(repo, result, faulted, server): + assert result["failed"] == 1, result + assert result["succeeded"] == len(FILES) - 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) + + +@pytest.mark.xfail(strict=True, reason=( + "Bug: a body truncated without Content-Length is accepted when the " + "catalog records the file size but no checksum (#54)")) +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 = _download_all(repo, max_workers=4) + assert (result["succeeded"], result["failed"]) == (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)["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) + + +@pytest.mark.ssh_integration +def test_unreadable_remote_folder_stops_ssh_scan(tmp_path, ssh_server): + if hasattr(os, "geteuid") and os.geteuid() == 0: + pytest.skip("root can read folders without permission") + root = ssh_server["root"] + for name in ("a/1.h5", "b/2.h5"): + write_file(root, name, os.urandom(32)) + repo = Repo.init(tmp_path / "repo") + repo.add(ssh_server["url"] + "{group}/{n:d}.h5") + repo.commit("Catalog export") + before = snapshot_hm(repo.worktree) + (root / "b").chmod(0) + try: + with pytest.raises(Exception) as raised: + repo.add(ssh_server["url"] + "{group}/{n:d}.h5") + finally: + (root / "b").chmod(0o755) + assert "Traceback" not in str(raised.value) + assert snapshot_hm(repo.worktree) == before diff --git a/test/test_chaos_storage.py b/test/test_chaos_storage.py new file mode 100644 index 0000000..473ecd5 --- /dev/null +++ b/test/test_chaos_storage.py @@ -0,0 +1,128 @@ +"""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.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.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(".f.bin.") + and path.parent.name == "d1") + result = repo.download(repo.plan_download(all_files=True), approved=True) + monkeypatch.undo() + assert (result["succeeded"], result["failed"]) == (2, 1), result + assert not (repo.worktree / "d1/f.bin").exists() + assert (repo.worktree / "d0/f.bin").read_bytes() == files["rel/d0/f.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) + try: + result = repo.download(repo.plan_download(all_files=True), approved=True) + finally: + locked.chmod(0o755) + assert (result["succeeded"], result["failed"]) == (2, 1), result + assert not (locked / "f.bin").exists() + assert not check_invariants(repo.worktree) diff --git a/test/test_cli.py b/test/test_cli.py index df11e45..8083ffd 100644 --- a/test/test_cli.py +++ b/test/test_cli.py @@ -15,7 +15,7 @@ import os import requests -import importlib +import importlib.metadata import pytest import pandas as pd from pathlib import Path @@ -30,6 +30,10 @@ from hallmark.remote.plan import DownloadItem, DownloadPlan from hallmark.utils import use_working_directory +_CLICK_RUNNER_SIGNALS_EOF = tuple( + int(part) for part in importlib.metadata.version("click").split(".")[:2] +) >= (8, 2) + cli_module = importlib.import_module("hallmark.cli") # files to create for testing, with a variety of a and i values to test regex encoding @@ -915,6 +919,10 @@ def capture_get(url, **kwargs): assert "1 file(s); 4 bytes" in result.output assert "Download these files? [y/N]" in result.output assert "Source: https://example.test/data/" in result.output + # Click 8.1 (the newest for Python 3.9) reads exhausted CliRunner input + # as an empty line instead of end of input, so "" acts like "\n" there. + if answer == "" and not _CLICK_RUNNER_SIGNALS_EOF: + answer = "\n" if answer == "y\n": assert result.exit_code == 0, result.output assert requests_made == ["https://example.test/data/tiny.fits"] diff --git a/test/test_ssh_integration.py b/test/test_ssh_integration.py index c5b272b..319e665 100644 --- a/test/test_ssh_integration.py +++ b/test/test_ssh_integration.py @@ -4,13 +4,8 @@ from dataclasses import replace import getpass import hashlib -import os from pathlib import Path -import shutil -import socket -import subprocess import time -from urllib.parse import quote from click.testing import CliRunner import pytest @@ -28,105 +23,6 @@ pytestmark = pytest.mark.ssh_integration -@pytest.fixture -def ssh_server(tmp_path, monkeypatch, request): - """Start a loopback SSH server with temporary keys and configuration.""" - if os.environ.get("HALLMARK_RUN_SSH_TESTS") != "1": - pytest.skip("Set HALLMARK_RUN_SSH_TESTS=1 for disposable loopback SSH tests") - sshd = shutil.which("sshd") or "/usr/sbin/sshd" - if not Path(sshd).exists(): - pytest.fail("Opted-in SSH tests require openssh-server") - for name in ("host", "client", "second", "untrusted"): - subprocess.run( - ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(tmp_path / name)], - check=True, - ) - authorized = tmp_path / "authorized" - authorized.write_text( - (tmp_path / "client.pub").read_text() + (tmp_path / "second.pub").read_text() - ) - with socket.socket() as sock: - sock.bind(("127.0.0.1", 0)) - port = sock.getsockname()[1] - config = tmp_path / "sshd_config" - config.write_text(f"""ListenAddress 127.0.0.1 -Port {port} -HostKey {tmp_path}/host -PidFile {tmp_path}/sshd.pid -AuthorizedKeysFile {authorized} -StrictModes no -UsePAM no -PasswordAuthentication no -KbdInteractiveAuthentication no -PermitRootLogin prohibit-password -MaxSessions 4 -Subsystem sftp internal-sftp -""") - mode = getattr(request, "param", None) - if mode == "sftp-only": - config.write_text(config.read_text() + "ForceCommand internal-sftp\n") - elif mode == "no-sftp": - config.write_text( - config.read_text().replace("Subsystem sftp internal-sftp\n", "") - ) - elif mode == "max-one": - config.write_text(config.read_text().replace("MaxSessions 4", "MaxSessions 1")) - log = (tmp_path / "sshd.log").open("wb") - daemon = subprocess.Popen( - [sshd, "-D", "-e", "-f", str(config)], stderr=log, start_new_session=True - ) - known = tmp_path / "known_hosts" - known.write_text(f"[127.0.0.1]:{port} " + (tmp_path / "host.pub").read_text()) - client_config = tmp_path / "ssh_config" - client_config.write_text(f"""Host hm-test - HostName 127.0.0.1 -Host * - User {getpass.getuser()} - Port {port} - IdentityFile {tmp_path}/client - IdentityAgent none - IdentitiesOnly yes - UserKnownHostsFile {known} - GlobalKnownHostsFile /dev/null -""") - original = SshBackend._ssh_options - monkeypatch.setattr( - SshBackend, - "_ssh_options", - lambda self, **kw: ["-F", str(client_config)] + original(self, **kw), - ) - monkeypatch.delenv("HALLMARK_AUTH_FILE", raising=False) - monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg")) - root = tmp_path / "export" - root.mkdir() - try: - deadline = time.monotonic() + 5 - while True: - if daemon.poll() is not None: - log.flush() - pytest.fail((tmp_path / "sshd.log").read_text()) - try: - with socket.create_connection(("127.0.0.1", port), timeout=0.1): - break - except OSError: - if time.monotonic() >= deadline: - pytest.fail("Local SSH test server did not start") - time.sleep(0.05) - yield { - "root": root, - "port": port, - "known": known, - "base": tmp_path, - "config": client_config, - "daemon": daemon, - "url": "ssh://hm-test" + quote(str(root), safe="/") + "/", - } - finally: - daemon.terminate() - daemon.wait(timeout=5) - log.close() - - @pytest.mark.parametrize("scheme", ["ssh", "sftp"]) def test_parallel_exact_files_and_master_cleanup(ssh_server, tmp_path, scheme): root = ssh_server["root"] diff --git a/test/test_survey.py b/test/test_survey.py new file mode 100644 index 0000000..9ba290a --- /dev/null +++ b/test/test_survey.py @@ -0,0 +1,305 @@ +"""End-to-end survey workflows against loopback HTTP and SSH servers. + +Scenarios follow doc/usecase.rst and the public-data examples in +doc/private_data.rst, using scaled-down synthetic releases. No request +leaves the machine. +""" + +import hashlib +import shutil + +import pandas as pd +import pytest + +from hallmark import ParaFrame, Repo +from survey import (DESI_DARK, EHT_UVFITS, Hm, apply_dr_update, + apply_silent_corruption, make_desi_like_release, + make_eht_like_release, make_lab_export, make_telescope_tree) + +pytestmark = [pytest.mark.survey, pytest.mark.usefixtures("loopback_only")] + +TELESCOPE_FMT = "{site}/{year:d}/{day:d}.fits" +DESI_PATTERN = "{group:d}/{pixel:d}/redrock-main-dark-{p:d}.fits" + + +@pytest.fixture +def hm(tmp_path, monkeypatch): + """Run hm with no local authentication profiles.""" + monkeypatch.delenv("HALLMARK_AUTH_FILE", raising=False) + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg")) + return Hm() + + +@pytest.fixture +def desi(tmp_path, http_server): + """A DESI-like release served with Apache-style listings.""" + www = tmp_path / "www" + files = make_desi_like_release(www) + return {"www": www, "files": files, "server": http_server(www)} + + +def _catalog(repo_path): + return pd.read_csv(repo_path / ".hm" / "data.tsv", sep="\t", dtype=str, + keep_default_na=False) + + +def _assert_downloaded(worktree, files, prefix=""): + for path, data in files.items(): + relative = path[len(prefix):] if prefix else path + assert (worktree / relative).read_bytes() == data, relative + + +# 1. CLI: standard repository ingest +def test_telescope_ingest_commit_and_update(tmp_path, hm): + obs = tmp_path / "obs" + hm("init", obs, cwd=tmp_path) + files = make_telescope_tree(obs) + added = hm("add", TELESCOPE_FMT, cwd=obs) + assert all(path in added.output for path in files) + hm("commit", "-m", "Initial observation ingest", cwd=obs) + assert "SMA/2025/96.fits" not in hm("status", cwd=obs).output + + (obs / "SMA/2025/96.fits").write_bytes(b"recalibrated") + assert "SMA/2025/96.fits" in hm("status", cwd=obs).output + hm("add", ".", cwd=obs) + hm("commit", "-m", "Recalibrate SMA day 96", cwd=obs) + + log = hm("log", cwd=obs).output + assert log.index("Recalibrate SMA day 96") < log.index("Initial observation ingest") + info = hm("info", cwd=obs).output + assert str(obs.resolve()) in info or str(obs) in info + # Local catalogs record the checksum and the pattern's fields per file. + rows = _catalog(obs).set_index(["site", "year", "day"]) + assert rows.loc[("SMA", "2025", "96"), "sha1"] == hashlib.sha1( + b"recalibrated").hexdigest() + assert len(rows) == len(files) + + +# 2. CLI: bare repository with explicit downloads +@pytest.mark.ssh_integration +def test_bare_simulation_catalog_downloads_to_explicit_output( + tmp_path, hm, ssh_server): + files = make_lab_export(ssh_server["root"]) + hm("init", "sim.hm", cwd=tmp_path) + sim = tmp_path / "sim.hm" + hm("add", ssh_server["url"], "--fmt", "run{run:d}/frame{frame:d}.h5", cwd=sim) + hm("commit", "-m", "Record remote catalog", cwd=sim) + assert "sim.hm" in hm("info", cwd=sim).output + + preview = hm("download", "--all", "--output", "../outputs", "--dry-run", cwd=sim) + assert f"{len(files)} file(s)" in preview.output + assert not (tmp_path / "outputs").exists() + hm("download", "--all", "--output", "../outputs", cwd=sim, input="y\n") + _assert_downloaded(tmp_path / "outputs", files) + + # The downloaded files can become a standard repository of their own. + outputs = tmp_path / "outputs" + hm("init", outputs, cwd=tmp_path) + hm("add", "run{run:d}/frame{frame:d}.h5", cwd=outputs) + hm("commit", "-m", "Version the downloaded frames", cwd=outputs) + assert len(_catalog(outputs)) == len(files) + + +# 3. Python: branch-isolated analysis with multiple worktrees +def test_second_worktree_keeps_observations_isolated(tmp_path): + repo = Repo.init(tmp_path / "obs") + make_telescope_tree(repo.worktree) + repo.add(TELESCOPE_FMT) + repo.commit("Initial observation ingest") + + assert repo.add_worktree("obs2") + second = Repo(tmp_path / "obs2") + make_telescope_tree(second.worktree, sites=("APEX",), seed=7) + second.add(TELESCOPE_FMT) + second.commit("Observation ingest on branch obs2") + + assert second.dothm.active_branch.name == "obs2" + assert repo.dothm.active_branch.name != "obs2" + assert not (repo.worktree / "APEX").exists() + assert "APEX" not in set(repo.state.data.site) + assert "APEX" in set(second.state.data.site) + + +# 4. Python: programmatic repository updates +def test_nightly_programmatic_ingest_keeps_every_version(tmp_path): + repo = Repo.init(tmp_path / "obs") + versions = set() + for night, day in enumerate((95, 96, 97)): + make_telescope_tree(repo.worktree, sites=("ALMA",), years=(2025,), + days=(day,), seed=night) + # Each night also reprocesses the first observation. + (repo.worktree / "ALMA/2025/95.fits").write_bytes(f"pass {night}".encode()) + repo.add(TELESCOPE_FMT) + repo.commit(f"Nightly ingest {night}") + versions.update(repo.state.data.sha1) + assert len(repo.state.data) == 3 + assert all(repo.objects.contains(sha1) for sha1 in versions) + assert all(f"Nightly ingest {night}" in repo.log() for night in range(3)) + + +# 5. Python: in-memory state workflows +def test_paraframe_selects_files_without_a_repository(tmp_path): + data = tmp_path / "data" + make_telescope_tree(data) + selected = ParaFrame.parse(TELESCOPE_FMT, base_path=data).filter(year=2025) + assert sorted(selected.path) == sorted( + path for path in (str(p.relative_to(data)) for p in data.rglob("*.fits")) + if "/2025/" in path) + assert not (data / ".hm").exists() + + +# 6. CLI: private lab data over SSH +@pytest.mark.ssh_integration +def test_private_lab_data_from_named_ssh_remote(tmp_path, hm, ssh_server): + files = make_lab_export(ssh_server["root"], runs=1) + lab = tmp_path / "lab" + hm("init", lab, cwd=tmp_path) + hm("add", ssh_server["url"], "--fmt", "run{run:d}/frame{frame:d}.h5", cwd=lab) + hm("commit", "-m", "Catalog lab export", cwd=lab) + hm("set-config", "--remote-name", "campus", "--remote-url", ssh_server["url"], + cwd=lab) + preview = hm("download", "--remote", "campus", "--all", "--dry-run", cwd=lab) + assert "Source: ssh://hm-test" in preview.output + assert not (lab / "run1").exists() + hm("download", "--remote", "campus", "--all", cwd=lab, input="y\n") + _assert_downloaded(lab, files) + + +def test_desi_like_pixel_records_published_checksums(tmp_path, hm, desi): + pixel_dir = f"{DESI_DARK}/230/23040/" + repo = tmp_path / "desi" + hm("init", repo, cwd=tmp_path) + hm("add", desi["server"].url(pixel_dir), "--filter", "redrock-*.fits", cwd=repo) + rows = _catalog(repo) + assert list(rows.path) == ["redrock-main-dark-23040.fits"] + data = desi["files"][pixel_dir + "redrock-main-dark-23040.fits"] + assert rows.loc[0, "checksum_algorithm"] == "sha256" + assert rows.loc[0, "checksum"] == hashlib.sha256(data).hexdigest() + assert rows.loc[0, "size_bytes"] == str(len(data)) + hm("commit", "-m", "Catalog redrock file", cwd=repo) + hm("download", "--all", cwd=repo, input="y\n") + assert (repo / "redrock-main-dark-23040.fits").read_bytes() == data + + +def test_url_pattern_add_catalogs_without_payload_requests(tmp_path, hm, desi): + server = desi["server"] + repo = tmp_path / "desi" + hm("init", repo, cwd=tmp_path) + hm("add", server.url(DESI_DARK + "/") + DESI_PATTERN, cwd=repo) + rows = _catalog(repo) + assert list(rows.pixel) == ["23040", "23041"] + assert set(rows.group) == {"230"} + payloads = [path for path in server.payload_gets() + if not path.endswith(".sha256sum")] + assert payloads == [] + + hm("commit", "-m", "Catalog redrock files", cwd=repo) + hm("download", "--all", "--filter", "230/23041/*", cwd=repo, input="y\n") + assert server.payload_gets()[-1].endswith("redrock-main-dark-23041.fits") + assert not (repo / "230/23040").exists() + hm("download", "--all", "--fmt", "{group:d}/23040/{name}", "--output", "../out", + cwd=repo, input="y\n") + _assert_downloaded(tmp_path / "out", { + path: data for path, data in desi["files"].items() + if path.endswith("redrock-main-dark-23040.fits")}, prefix=DESI_DARK + "/") + + +def test_eht_like_cyverse_listing_records_sizes(tmp_path, hm, http_server): + www = tmp_path / "www" + files = make_eht_like_release(www) + server = http_server(www, listing_style="cyverse") + repo = tmp_path / "eht" + hm("init", repo, cwd=tmp_path) + hm("add", server.url(EHT_UVFITS + "/"), "--filter", "SR1_M87_2017_095_lo_*", + cwd=repo) + rows = _catalog(repo) + name = "SR1_M87_2017_095_lo_hops_netcal_StokesI.uvfits" + assert list(rows.path) == [name] + assert rows.loc[0, "size_bytes"] == str(len(files[f"{EHT_UVFITS}/{name}"])) + hm("commit", "-m", "Catalog M87 day 95", cwd=repo) + hm("download", "--all", cwd=repo, input="y\n") + assert (repo / name).read_bytes().startswith(b"SIMPLE") + + +def test_data_release_update_rescans_and_refreshes(tmp_path, hm, desi): + server, www, files = desi["server"], desi["www"], desi["files"] + url = server.url(DESI_DARK + "/") + DESI_PATTERN + repo = tmp_path / "desi" + hm("init", repo, cwd=tmp_path) + hm("add", url, cwd=repo) + hm("commit", "-m", "DR1 catalog", cwd=repo) + hm("download", "--all", cwd=repo, input="y\n") + + changed, added = apply_dr_update(www, files) + hm("add", url, cwd=repo) + rows = _catalog(repo).set_index("path") + relative = changed[len(DESI_DARK) + 1:] + assert rows.loc[relative, "checksum"] == hashlib.sha256(files[changed]).hexdigest() + assert "230/23042/redrock-main-dark-23042.fits" in rows.index + hm("commit", "-m", "DR1 reprocessing", cwd=repo) + + # Replacing a file with different contents means deleting it first. + (repo / relative).unlink() + hm("download", "--all", cwd=repo, input="y\n") + _assert_downloaded(repo, {path: data for path, data in files.items() + if path.endswith(".fits") and "redrock" in path}, + prefix=DESI_DARK + "/") + + +def test_silently_corrupted_release_file_is_rejected(tmp_path, hm, desi): + server, www, files = desi["server"], desi["www"], desi["files"] + repo = tmp_path / "desi" + hm("init", repo, cwd=tmp_path) + hm("add", server.url(DESI_DARK + "/") + DESI_PATTERN, cwd=repo) + hm("commit", "-m", "DR1 catalog", cwd=repo) + corrupted = f"{DESI_DARK}/230/23041/redrock-main-dark-23041.fits" + apply_silent_corruption(www, corrupted) + + result = hm("download", "--all", cwd=repo, input="y\n", expect=1) + assert "1 file" in result.output + assert not (repo / "230/23041/redrock-main-dark-23041.fits").exists() + assert not list(repo.rglob("*.part")) + intact = f"{DESI_DARK}/230/23040/redrock-main-dark-23040.fits" + assert (repo / "230/23040/redrock-main-dark-23040.fits").read_bytes() == \ + files[intact] + + +def test_clone_git_catalog_prompts_then_downloads(tmp_path, hm, desi): + source = tmp_path / "desi" + hm("init", source, cwd=tmp_path) + hm("add", desi["server"].url(DESI_DARK + "/") + DESI_PATTERN, cwd=source) + hm("commit", "-m", "DR1 catalog", cwd=source) + + declined = hm("clone", source, "later", cwd=tmp_path, input="n\n") + assert "2 file(s)" in declined.output + assert "Download these files? [y/N]" in declined.output + assert not (tmp_path / "later/230").exists() + assert len(_catalog(tmp_path / "later")) == 2 + + hm("clone", source, "copy", cwd=tmp_path, input="y\n") + _assert_downloaded(tmp_path / "copy", { + path: data for path, data in desi["files"].items() + if "redrock" in path}, prefix=DESI_DARK + "/") + assert "DR1 catalog" in hm("log", cwd=tmp_path / "copy").output + + hm("clone", source, "catalog-only.hm", "--no-download", cwd=tmp_path) + assert (tmp_path / "catalog-only.hm/data.tsv").is_file() + hm("clone", source, "refused.hm", cwd=tmp_path, expect=1) + assert not (tmp_path / "refused.hm").exists() + + +def test_clone_published_catalog_snapshot_over_http(tmp_path, hm, desi, http_server): + source = tmp_path / "desi" + hm("init", source, cwd=tmp_path) + hm("add", desi["server"].url(DESI_DARK + "/") + DESI_PATTERN, cwd=source) + hm("commit", "-m", "DR1 catalog", cwd=source) + published = tmp_path / "published" / "catalogs" / "desi" + published.mkdir(parents=True) + for name in ("config.yml", "meta.yml", "data.tsv"): + shutil.copy(source / ".hm" / name, published / name) + catalogs = http_server(tmp_path / "published") + + hm("clone", catalogs.url("catalogs/desi/"), "copy", cwd=tmp_path, input="y\n") + _assert_downloaded(tmp_path / "copy", { + path: data for path, data in desi["files"].items() + if "redrock" in path}, prefix=DESI_DARK + "/") diff --git a/test/test_survey_live.py b/test/test_survey_live.py new file mode 100644 index 0000000..d8bba9b --- /dev/null +++ b/test/test_survey_live.py @@ -0,0 +1,147 @@ +"""Tiny end-to-end downloads from real public survey archives. + +These follow the public-collection examples in doc/private_data.rst +against DESI DR1 and the EHT M87 release on CyVerse. They need network +access, so they only run with HALLMARK_RUN_LIVE_TESTS=1 (nightly in CI). +Every plan is checked before approval so a run downloads at most a few +megabytes. CyVerse lists sizes such as "554K" that Hallmark does not +record, so its downloads are limited to single, exactly named files. +""" + +import hashlib +import os + +import pytest +import requests + +from hallmark import Repo +from hallmark.transport.base import DownloadError + +pytestmark = [ + pytest.mark.survey_live, + pytest.mark.skipif(os.environ.get("HALLMARK_RUN_LIVE_TESTS") != "1", + reason="Set HALLMARK_RUN_LIVE_TESTS=1 to reach public archives"), +] + +DESI = ("https://data.desi.lbl.gov/public/dr1/spectro/redux/iron/" + "healpix/main/dark/230/23040/") +EHT = ("https://data.cyverse.org/dav-anon/iplant/commons/" + "cyverse_curated/EHTC_FirstM87Results_Apr2019/uvfits/") +MAX_TARGET_BYTES = 2_000_000 +MAX_RUN_BYTES = 10_000_000 +_downloaded = [] + + +@pytest.fixture(scope="module") +def archive_available(): + """Skip, rather than fail, when an archive itself is down or overloaded. + + The probe uses plain Requests with Hallmark's timeouts, so a failure + after a successful probe points at Hallmark rather than the archive. + """ + checked = {} + + def check(url): + if url not in checked: + try: + status = requests.get(url, timeout=(10, 30)).status_code + checked[url] = (status < 500, f"HTTP {status}") + except requests.RequestException as exc: + checked[url] = (False, type(exc).__name__) + available, detail = checked[url] + if not available: + pytest.skip(f"{url} is unavailable ({detail})") + + return check + + +def _archive_failure(error): + """Whether an error means the archive timed out or failed on its side.""" + return any(sign in str(error) + for sign in ("ReadTimeout", "ConnectTimeout", "HTTP 5")) + + +def _scan(repo, url, **options): + """Catalog ``url``, skipping when the archive times out or errors.""" + try: + return repo.add(url, **options) + except DownloadError as exc: + if _archive_failure(exc): + pytest.skip(f"The archive failed during the scan: {exc}") + raise + + +def _download(repo, plan): + """Download a plan, skipping when the archive times out or errors.""" + result = repo.download(plan, approved=True) + errors = " ".join(map(str, result["errors"])) + if result["failed"] and all(map(_archive_failure, result["errors"])): + pytest.skip(f"The archive failed during the download: {errors}") + return result + + +def _approve_small(plan, expected_files, sizes_listed=True): + """Refuse to download anything unexpectedly large. + + Without listed sizes, only a single file the test names exactly is + allowed, and its size is checked after the download. + """ + assert plan.file_count == expected_files, plan.summary() + if sizes_listed: + assert plan.unknown_size_count == 0, plan.summary() + else: + assert plan.file_count == 1, plan.summary() + assert plan.known_bytes <= MAX_TARGET_BYTES, plan.summary() + assert sum(_downloaded) + plan.known_bytes <= MAX_RUN_BYTES + _downloaded.append(plan.known_bytes or MAX_TARGET_BYTES) + + +def test_desi_dr1_files_match_published_checksums(tmp_path, archive_available): + archive_available(DESI) + repo = Repo.init(tmp_path / "desi") + _scan(repo, DESI, filter=["redrock-main-dark-23040.fits", + "hpixexp-main-dark-23040.csv"]) + rows = repo.state.data.set_index("path") + assert set(rows.checksum_algorithm) == {"sha256"} + repo.commit("Record remote catalog") + + plan = repo.plan_download(all_files=True) + _approve_small(plan, expected_files=2) + result = _download(repo, plan) + assert (result["succeeded"], result["failed"]) == (2, 0), result + for path, row in rows.iterrows(): + data = (repo.worktree / path).read_bytes() + assert hashlib.sha256(data).hexdigest() == row.checksum + assert len(data) == int(row.size_bytes) + + +def test_eht_m87_uvfits_downloads_from_cyverse(tmp_path, archive_available): + archive_available(EHT) + name = "SR1_M87_2017_095_lo_hops_netcal_StokesI.uvfits" + repo = Repo.init(tmp_path / "eht") + _scan(repo, EHT, filter=name) + assert list(repo.state.data.path) == [name] + repo.commit("Record remote catalog") + + plan = repo.plan_download(all_files=True) + _approve_small(plan, expected_files=1, sizes_listed=False) + result = _download(repo, plan) + assert (result["succeeded"], result["failed"]) == (1, 0), result + data = (repo.worktree / name).read_bytes() + assert data.startswith(b"SIMPLE") + assert len(data) <= MAX_TARGET_BYTES + + +def test_eht_url_pattern_catalogs_every_day_and_band(tmp_path, archive_available): + archive_available(EHT) + repo = Repo.init(tmp_path / "eht") + _scan(repo, EHT + "SR1_M87_2017_{day:d}_{band}_hops_netcal_StokesI.uvfits") + rows = repo.state.data + assert set(rows.band) == {"lo", "hi"} + assert {95, 96, 100, 101} <= {int(day) for day in rows.day} + repo.commit("Record remote catalog") + + plan = repo.plan_download(all_files=True, filter="*_100_lo_*") + _approve_small(plan, expected_files=1, sizes_listed=False) + result = _download(repo, plan) + assert (result["succeeded"], result["failed"]) == (1, 0), result