" + '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 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_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_network.py b/test/test_chaos_network.py new file mode 100644 index 0000000..a2ffc41 --- /dev/null +++ b/test/test_chaos_network.py @@ -0,0 +1,187 @@ +"""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)), + "manifest-short-body": ("rel/sub/sha256sum.txt", ShortBody(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) 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)