From abf949f0fbcf75ad65abfd41ca032df3426f0f5d Mon Sep 17 00:00:00 2001 From: mecattaf Date: Wed, 16 Sep 2026 23:05:17 +0200 Subject: [PATCH] Land a working lane into a preservation packet and prove it (CNA-M07) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `land` is the tool the nightly ~/today sweep and every one-time landing lane (loose ~/*.md, dated dirs, home canon, Downloads, ~/sept7 prose, root relocations) run to move working files into notes without trusting that the copy worked. land copy ... --dest copy-then-verify, writes the packet land verify recompute every hash and count in it land diff the rsync -rn --checksum equivalence check The manifest is the shape built by hand for the 2026-09-16 orchestration-day packet: a header with file count, total bytes and exclusions carrying reasons, rows of {source, preserved, bytes, sha256, mtime, source_session}. That packet verifies unchanged (697 rows == 697 files, sha256 all match). Older packets without mtime or source_session warn rather than fail. The copy keeps cp -p semantics and relative structure, never follows a symlink (each becomes a kind: symlink row with its target), renames a nested .git to dot-git and records renamed_from with the source-side path, and leaves out any file over the 95 MiB ceiling with its reason in the README. Sources are only ever read. The secret guard runs over the whole plan before the first byte is written, so an abort leaves no half-packet: names matching *.env, *token*, *secret*, *credential*, id_ed25519* and *.age (overridable with --allow-secret-names), and text content matching a GitHub token, an OpenAI-style key, an AWS access key id or a PEM private key block (not overridable). It names the file and never the value. checks.land builds its own fixture lane in $TMPDIR — nested .git, symlink, 3-byte file, excluded glob, sparse file over the ceiling — and pins that a flipped byte or a stray file fails verify, that the sources are untouched afterwards, and that a count fault is not reported as a hash fault. Co-Authored-By: Claude Opus 5 --- flake.nix | 29 ++ overlays/default.nix | 5 + pkgs/land/default.nix | 44 ++ pkgs/land/land.py | 868 ++++++++++++++++++++++++++++++++++++++++ tests/land/test_land.py | 385 ++++++++++++++++++ 5 files changed, 1331 insertions(+) create mode 100644 pkgs/land/default.nix create mode 100644 pkgs/land/land.py create mode 100644 tests/land/test_land.py diff --git a/flake.nix b/flake.nix index 96c26c0eb..7fe5447b3 100644 --- a/flake.nix +++ b/flake.nix @@ -654,6 +654,7 @@ crm dcal fleet-status + land local-ai-monthly # `nix build .#local-models-prune` — the ONLY verb on this fleet # that deletes a working copy. Exposed so the guard suite can be @@ -2623,6 +2624,34 @@ touch "$out" ''; + # land (CNA-M07): the landing tool that carries a working lane into the + # notes repository. The suite builds its own fixture lane in $TMPDIR -- + # nested .git, symlink, 3-byte file, excluded glob, sparse file over the + # 95 MiB ceiling -- and drives copy/verify/diff over it. rsync is in the + # build inputs so the diff cross-check is exercised here and not only on + # a host that happens to have it; the sparse file costs no store space. + # What is pinned: a packet's row count equals `find -type f`, a flipped + # byte or a stray file FAILS verify, the sources are byte-identical + # afterwards, and the secret guard aborts before writing anything while + # printing the file's name and never the matched value. + land = + pkgs.runCommand "land" + { + nativeBuildInputs = [ + pkgs.python3 + pkgs.rsync + ]; + } + '' + set -euo pipefail + export HOME="$TMPDIR/home" + export PYTHONDONTWRITEBYTECODE=1 + export LAND_PY=${pkgs.land}/share/land/land.py + mkdir -p "$HOME" + python3 -m unittest discover -s ${./tests/land} -p 'test_*.py' -v + touch "$out" + ''; + # seats: one capacity oracle across every seat on this box. Hermetic — # SEATS_NO_NETWORK=1 and a home tree the test builds itself, because # every fact the program reports is relative to now and a checked-in diff --git a/overlays/default.nix b/overlays/default.nix index 43e1b7875..1141fb9f4 100644 --- a/overlays/default.nix +++ b/overlays/default.nix @@ -121,4 +121,9 @@ final: prev: { # Huion Note X10 offline-note extractor over BLE, pinned by commit, thin # strokes. Only hosts/client/huion.nix consumes it. See pkgs/huion-notes.nix. huion-notes = final.callPackage ../pkgs/huion-notes.nix { }; + + # land (CNA-M07): copy-then-verify for preservation packets — manifest, + # README, sha256 over every row. Used by the nightly ~/today sweep and by + # every one-time landing lane. Stdlib Python; see pkgs/land. + land = final.callPackage ../pkgs/land { }; } diff --git a/pkgs/land/default.nix b/pkgs/land/default.nix new file mode 100644 index 000000000..e933a1f30 --- /dev/null +++ b/pkgs/land/default.nix @@ -0,0 +1,44 @@ +{ + lib, + stdenvNoCC, + python3, + makeWrapper, + rsync, + coreutils, +}: +# land (CNA-M07): copy-then-verify for preservation packets. One stdlib Python +# file, one entry point. `land copy` writes a packet (the tree plus +# preservation-.json and README.md), `land verify` recomputes every hash +# and count in it, `land diff` proves a source and its copy are the same bytes +# before anybody removes a source. +# +# rsync is a SECOND OPINION, never the authority: a packet legitimately differs +# from its source in two recorded ways (the dot-git rename and the exclusions), +# so `land diff` always hashes and only adds the rsync -rn --checksum pass when +# neither applies. Suffix, not prefix, on PATH — the host's own rsync is fine, +# this one only guarantees the command exists. +stdenvNoCC.mkDerivation { + pname = "land"; + version = "1"; + src = ./.; + nativeBuildInputs = [ makeWrapper ]; + dontBuild = true; + installPhase = '' + runHook preInstall + install -Dm0644 land.py $out/share/land/land.py + makeWrapper ${python3.interpreter} $out/bin/land \ + --add-flags "$out/share/land/land.py" \ + --argv0 land \ + --suffix PATH : ${ + lib.makeBinPath [ + rsync + coreutils + ] + } + runHook postInstall + ''; + meta = { + description = "Copy-then-verify landing tool for preservation packets: manifest, README, hashes"; + mainProgram = "land"; + }; +} diff --git a/pkgs/land/land.py b/pkgs/land/land.py new file mode 100644 index 000000000..908d92bde --- /dev/null +++ b/pkgs/land/land.py @@ -0,0 +1,868 @@ +#!/usr/bin/env python3 +"""land — copy-then-verify landing tool for preservation packets (CNA-M07). + +A "lane" is one pile of working files that has to reach the notes repository +intact: the nightly ``~/today`` sweep, the loose ``~/*.md``, a dated directory, +Downloads, a root relocation. `land copy` writes a *packet*: the copied tree +plus two artifacts at its root, ``preservation-.json`` and +``README.md``. `land verify` re-derives every claim the manifest makes, and +`land diff` proves the source and the copy are the same bytes before anybody +removes a source. + +The manifest shape is the one built by hand for +notes/references/continuity/2026-09-16-orchestration-day — header with +file_count, total_bytes and exclusions carrying reasons, rows of +{source, preserved, bytes, sha256, mtime, source_session}. Older packets omit +mtime and source_session; verify warns about those rows, it does not fail them. + +Rules the copy obeys, all of them because a packet is evidence and not a backup: + * sources are never modified, moved or removed — land only reads them; + * ``cp -p`` semantics (mtime and mode preserved), relative structure kept; + * a regular file over the 95 MiB ceiling is left out, with its reason, and + named in the README under "Left out on purpose"; + * a nested ``.git`` directory or gitfile is renamed to ``dot-git`` in the + copy (notes never tracks a nested repository) and the row records the + packet-relative path it would have had, as ``renamed_from``; + * symlinks are never followed — each becomes a row with ``kind: symlink`` + and its ``target``, and the link is recreated verbatim in the copy; + * the secret guard runs over the whole plan BEFORE the first byte is + written, so an abort leaves no half-packet behind. It names the file it + tripped on and never the value it matched. +""" + +from __future__ import annotations + +import argparse +import fnmatch +import hashlib +import json +import os +import re +import shutil +import subprocess +import sys +from datetime import date, datetime +from pathlib import Path, PurePosixPath + +TOOL = "land 1" +MANIFEST_GLOB = "preservation-*.json" +README_NAME = "README.md" + +# GitHub refuses a blob over 100 MiB; the ceiling sits under it with room for +# the packet's own artifacts. Apparent size (st_size), so a sparse file is +# judged by what a copy would cost, not by the blocks it occupies today. +MAX_FILE_BYTES = 95 * 1024 * 1024 + +READ_CHUNK = 1 << 20 + +SECRET_NAME_GLOBS = ( + "*.env", + "*token*", + "*secret*", + "*credential*", + "id_ed25519*", + "*.age", +) + +SECRET_CONTENT_PATTERNS = ( + ("GitHub personal access token", re.compile(rb"ghp_[A-Za-z0-9]{20,}")), + ("OpenAI-style API key", re.compile(rb"sk-[A-Za-z0-9]{20,}")), + ("AWS access key id", re.compile(rb"AKIA[0-9A-Z]{16}")), + ("PEM private key block", re.compile(rb"-----BEGIN .*PRIVATE KEY")), +) + + +class LandError(Exception): + """Anything that must stop the run with a named cause and no partial packet.""" + + +# -------------------------------------------------------------------------- +# small helpers + + +def iso_mtime(timestamp: float) -> str: + """Local time with offset, seconds resolution: 2026-09-16T11:45:53+02:00.""" + return datetime.fromtimestamp(timestamp).astimezone().isoformat(timespec="seconds") + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(READ_CHUNK), b""): + digest.update(chunk) + return digest.hexdigest() + + +def sha256_bytes(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def human_bytes(count: int) -> str: + return f"{count:,}" + + +CEILING_TEXT = f"95 MiB ({human_bytes(MAX_FILE_BYTES)} bytes)" + + +def looks_binary(path: Path) -> bool: + with open(path, "rb") as handle: + return b"\0" in handle.read(8192) + + +def scan_for_secrets(path: Path) -> str | None: + """Return the NAME of the first pattern that matches, never the match.""" + if looks_binary(path): + return None + tail = b"" + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(READ_CHUNK), b""): + window = tail + chunk + for name, pattern in SECRET_CONTENT_PATTERNS: + if pattern.search(window): + return name + tail = window[-256:] + return None + + +def secret_name_hit(name: str) -> str | None: + lowered = name.lower() + for glob in SECRET_NAME_GLOBS: + if fnmatch.fnmatch(lowered, glob): + return glob + return None + + +# -------------------------------------------------------------------------- +# planning + + +class Item: + """One planned row. `rel` is the packet-relative preserved path.""" + + __slots__ = ("source", "rel", "kind", "renamed_from", "size", "mtime", "mode", "target") + + def __init__(self, source: Path, rel: PurePosixPath, kind: str) -> None: + self.source = source + self.rel = rel + self.kind = kind + self.renamed_from: str | None = None + self.size = 0 + self.mtime = 0.0 + self.mode = 0o644 + self.target: str | None = None + + +def excluded_by(rel: PurePosixPath, globs: tuple[str, ...]) -> str | None: + text = str(rel) + for glob in globs: + if fnmatch.fnmatch(text, glob) or fnmatch.fnmatch(rel.name, glob): + return glob + return None + + +def plan_lane( + sources: list[Path], excludes: tuple[str, ...] +) -> tuple[list[Item], list[dict[str, str]]]: + """Walk the sources without following a single symlink. Deterministic order.""" + items: list[Item] = [] + excluded: list[dict[str, str]] = [] + + def note_exclusion(path: Path, reason: str) -> None: + excluded.append({"path": str(path), "reason": reason}) + + def take(source: Path, rel: PurePosixPath, renamed_from: str | None) -> None: + stat_result = os.lstat(source) + if os.path.islink(source): + item = Item(source, rel, "symlink") + item.target = os.readlink(source) + item.size = stat_result.st_size + item.mtime = stat_result.st_mtime + item.renamed_from = renamed_from + items.append(item) + return + if not os.path.isfile(source): + note_exclusion(source, "not a regular file, directory or symlink") + return + if stat_result.st_size > MAX_FILE_BYTES: + note_exclusion( + source, + f"{human_bytes(stat_result.st_size)} bytes, over the " + f"{CEILING_TEXT} per-file ceiling", + ) + return + item = Item(source, rel, "file") + item.size = stat_result.st_size + item.mtime = stat_result.st_mtime + item.mode = stat_result.st_mode + item.renamed_from = renamed_from + items.append(item) + + def walk( + directory: Path, + rel_dir: PurePosixPath, + source_rel_dir: PurePosixPath, + renamed_above: bool, + ) -> None: + """rel_dir is the path inside the packet, source_rel_dir the one at the source. + + The two diverge the moment a `.git` is renamed, and `renamed_from` has to + name the SOURCE side — that is the whole point of recording it. + """ + try: + entries = sorted(os.scandir(directory), key=lambda e: e.name) + except PermissionError as exc: + note_exclusion(directory, f"unreadable: {exc.strerror}") + return + files = [e for e in entries if not e.is_dir(follow_symlinks=False)] + directories = [e for e in entries if e.is_dir(follow_symlinks=False)] + for entry in files + directories: + name = entry.name + renamed = name == ".git" + rel = rel_dir / ("dot-git" if renamed else name) + source_rel = source_rel_dir / name + glob = excluded_by(rel, excludes) or excluded_by(source_rel, excludes) + if glob: + note_exclusion(Path(entry.path), f"matched --exclude {glob}") + continue + renamed_from = str(source_rel) if (renamed or renamed_above) else None + if entry.is_dir(follow_symlinks=False): + walk(Path(entry.path), rel, source_rel, renamed or renamed_above) + else: + take(Path(entry.path), rel, renamed_from) + + for source in sources: + if not os.path.lexists(source): + raise LandError(f"source does not exist: {source}") + name = source.name + renamed = name == ".git" + rel = PurePosixPath("dot-git" if renamed else name) + glob = excluded_by(rel, excludes) + if glob: + note_exclusion(source, f"matched --exclude {glob}") + continue + source_rel = PurePosixPath(name) + if os.path.isdir(source) and not os.path.islink(source): + walk(source, rel, source_rel, renamed) + else: + take(source, rel, str(source_rel) if renamed else None) + + return items, excluded + + +def guard_secrets(items: list[Item], allow_names: bool) -> None: + """Abort the whole run on the first hit. Names only, never values.""" + offences: list[str] = [] + for item in items: + if item.kind != "file": + continue + if not allow_names: + glob = secret_name_hit(item.source.name) + if glob: + offences.append(f"{item.source}: name matches {glob}") + continue + try: + hit = scan_for_secrets(item.source) + except OSError as exc: + raise LandError(f"cannot read {item.source}: {exc.strerror}") from exc + if hit: + offences.append(f"{item.source}: contains what looks like a {hit}") + if not offences: + return + lines = ["land copy: refusing to write a packet, the secret guard tripped:"] + lines += [f" {offence}" for offence in offences] + lines.append("Nothing was written. Move or redact the file, exclude it with") + lines.append("--exclude, or pass --allow-secret-names for a name-only hit.") + raise LandError("\n".join(lines)) + + +# -------------------------------------------------------------------------- +# copy + + +def copy_lane( + items: list[Item], dest: Path, source_session: str | None +) -> list[dict[str, object]]: + """cp -p, then hash BOTH sides. A copy that did not land is an error here.""" + rows: list[dict[str, object]] = [] + for item in items: + target = dest / item.rel + target.parent.mkdir(parents=True, exist_ok=True) + row: dict[str, object] = { + "source": str(item.source), + "preserved": str(target), + } + if item.kind == "symlink": + if target.is_symlink() or target.exists(): + target.unlink() + os.symlink(item.target, target) + row["kind"] = "symlink" + row["target"] = item.target + row["bytes"] = item.size + row["sha256"] = sha256_bytes(os.fsencode(item.target)) + else: + before = sha256_file(item.source) + shutil.copy2(item.source, target, follow_symlinks=False) + after = sha256_file(target) + if before != after: + raise LandError( + f"copy did not land intact: {item.source} -> {target} " + "(source changed under the copy, or the destination is faulty)" + ) + row["bytes"] = item.size + row["sha256"] = after + row["mtime"] = iso_mtime(item.mtime) + if source_session: + row["source_session"] = source_session + if item.renamed_from: + row["renamed_from"] = item.renamed_from + rows.append(row) + return rows + + +def common_source_root(sources: list[Path]) -> str: + parents = [str(source.parent) for source in sources] + try: + return os.path.commonpath(parents) + except ValueError: + return "" + + +# -------------------------------------------------------------------------- +# packet reading + + +def find_manifest(packet: Path) -> Path: + matches = sorted(packet.glob(MANIFEST_GLOB)) + if not matches: + raise LandError(f"no {MANIFEST_GLOB} in {packet}") + if len(matches) > 1: + names = ", ".join(m.name for m in matches) + raise LandError(f"{packet} holds more than one manifest: {names}") + return matches[0] + + +def load_manifest(packet: Path) -> tuple[Path, dict]: + path = find_manifest(packet) + try: + with open(path, "rb") as handle: + return path, json.load(handle) + except json.JSONDecodeError as exc: + raise LandError(f"{path} is not valid JSON: {exc}") from exc + + +def resolve_preserved(packet: Path, manifest: dict, preserved: str) -> Path: + """Absolute path if it is still there, else re-root it on this packet dir. + + A packet that was moved (or is being checked out of the repository at a + different path) must still verify — the manifest's preserved_root is what + makes the row relative again. + """ + candidate = Path(preserved) + if candidate.exists() or candidate.is_symlink(): + return candidate + root = manifest.get("preserved_root") + if root: + try: + return packet / Path(preserved).relative_to(root) + except ValueError: + pass + return candidate + + +def packet_file_count(packet: Path, manifest_name: str) -> int: + """`find -type f`, minus the packet's own two artifacts.""" + count = 0 + for root, directories, files in os.walk(packet, followlinks=False): + directories[:] = [d for d in directories if not os.path.islink(os.path.join(root, d))] + for name in files: + full = os.path.join(root, name) + if os.path.islink(full): + continue + if Path(root) == packet and name in {manifest_name, README_NAME}: + continue + count += 1 + return count + + +# -------------------------------------------------------------------------- +# verify + + +def verify_packet(packet: Path, quiet: bool = False) -> tuple[bool, list[str], dict[str, int]]: + manifest_path, manifest = load_manifest(packet) + rows = manifest.get("files") + if not isinstance(rows, list): + raise LandError(f"{manifest_path} has no files[] array") + + failures: list[str] = [] + warnings: list[str] = [] + # Row failures and packet-shape failures are kept apart so the receipt can + # say which one actually happened: a manifest that covers only part of its + # directory (every hash right, the count wrong) is a different fault from a + # tampered file, and reporting the first as "sha256 MISMATCH" is a lie. + file_rows = 0 + symlink_rows = 0 + total_bytes = 0 + + for index, row in enumerate(rows): + preserved = row.get("preserved") + if not preserved: + failures.append(f"row {index} has no preserved path") + continue + path = resolve_preserved(packet, manifest, preserved) + expected = row.get("sha256") + kind = row.get("kind", "file") + if "mtime" not in row: + warnings.append(f"{preserved}: no mtime recorded (pre-land packet)") + if "source_session" not in row: + warnings.append(f"{preserved}: no source_session recorded (pre-land packet)") + if kind == "symlink": + symlink_rows += 1 + if not os.path.islink(path): + failures.append(f"{preserved}: recorded as a symlink, is not one") + continue + target = os.readlink(path) + if row.get("target") is not None and target != row["target"]: + failures.append(f"{preserved}: symlink target changed") + continue + if expected and sha256_bytes(os.fsencode(target)) != expected: + failures.append(f"{preserved}: symlink target does not match sha256") + continue + file_rows += 1 + if not path.is_file() or path.is_symlink(): + failures.append(f"{preserved}: missing from the packet") + continue + size = path.stat().st_size + if row.get("bytes") is not None and size != row["bytes"]: + failures.append(f"{preserved}: {size} bytes, manifest says {row['bytes']}") + continue + total_bytes += size + if not expected: + failures.append(f"{preserved}: no sha256 in the manifest") + continue + if sha256_file(path) != expected: + failures.append(f"{preserved}: sha256 does not match") + + rows_before_shape_checks = len(failures) + found = packet_file_count(packet, manifest_path.name) + if found != file_rows: + failures.append( + f"count mismatch: {file_rows} manifest file rows, " + f"{found} files under the packet (find -type f, minus the two artifacts)" + ) + declared = manifest.get("file_count") + if declared is not None and declared != file_rows: + failures.append(f"header file_count {declared} != {file_rows} file rows") + declared_bytes = manifest.get("total_bytes") + if declared_bytes is not None and not failures and declared_bytes != total_bytes: + failures.append(f"header total_bytes {declared_bytes} != {total_bytes} bytes on disk") + + counts = { + "rows": len(rows), + "files": file_rows, + "symlinks": symlink_rows, + "found": found, + "bytes": total_bytes, + } + row_failures = rows_before_shape_checks + if not quiet: + for warning in warnings[:10]: + print(f"land verify: warning: {warning}", file=sys.stderr) + if len(warnings) > 10: + print( + f"land verify: warning: ... and {len(warnings) - 10} more rows " + "without mtime/source_session", + file=sys.stderr, + ) + for failure in failures: + print(f"land verify: FAIL: {failure}", file=sys.stderr) + state = "OK" if not failures else "FAILED" + symlink_note = f", {symlink_rows} symlinks" if symlink_rows else "" + print( + f"land verify {state} {packet} {file_rows} file rows == {found} files" + f"{symlink_note} {human_bytes(counts['bytes'])} bytes " + f"sha256 {'all match' if not row_failures else 'MISMATCH'}" + f"{f' {len(warnings)} warnings' if warnings else ''}" + ) + return (not failures), warnings, counts + + +# -------------------------------------------------------------------------- +# diff + + +def rsync_available() -> bool: + return shutil.which("rsync") is not None + + +def rsync_differences(source: Path, target: Path) -> list[str]: + command = ["rsync", "-rn", "--checksum", "-i", f"{source}/", f"{target}/"] + result = subprocess.run(command, capture_output=True, text=True, check=False) + if result.returncode != 0: + raise LandError(f"rsync failed: {result.stderr.strip()}") + return [ + line + for line in result.stdout.splitlines() + if line.strip() and not line.startswith(".") + ] + + +def diff_source(packet: Path, source: Path) -> tuple[bool, list[str], str]: + """Equivalence between one source and its copy inside the packet. + + Hashing is authoritative because a packet legitimately differs from its + source in two recorded ways — the dot-git rename and the exclusions. rsync + runs as a second opinion only when neither applies to this source. + """ + _, manifest = load_manifest(packet) + rows = manifest.get("files", []) + source = Path(os.path.abspath(source)) + prefix = str(source) + os.sep + relevant = [ + row + for row in rows + if row.get("source") == str(source) or str(row.get("source", "")).startswith(prefix) + ] + if not relevant: + raise LandError(f"{packet} has no rows under {source}") + + problems: list[str] = [] + renamed = any(row.get("renamed_from") for row in relevant) + excluded_paths = [entry["path"] for entry in manifest.get("excluded", [])] + excluded_here = [ + path + for path in excluded_paths + if path == str(source) or path.startswith(prefix) or path.rstrip("/").startswith(prefix) + ] + + seen: set[str] = set() + for row in relevant: + origin = Path(row["source"]) + seen.add(str(origin)) + preserved = resolve_preserved(packet, manifest, row["preserved"]) + if row.get("kind") == "symlink": + if not os.path.islink(origin): + problems.append(f"{origin}: source is no longer a symlink") + elif os.readlink(origin) != row.get("target"): + problems.append(f"{origin}: symlink target differs from the packet") + continue + if not origin.is_file() or origin.is_symlink(): + problems.append(f"{origin}: source file is gone") + continue + if sha256_file(origin) != row.get("sha256"): + problems.append(f"{origin}: source bytes differ from the packet copy") + continue + if not preserved.is_file(): + problems.append(f"{row['preserved']}: missing from the packet") + + if source.is_dir() and not source.is_symlink(): + for root, directories, files in os.walk(source, followlinks=False): + directories[:] = [ + d for d in directories if not os.path.islink(os.path.join(root, d)) + ] + for name in files: + full = os.path.join(root, name) + if full in seen: + continue + if os.path.islink(full): + continue + if any(full == p or full.startswith(p.rstrip("/") + os.sep) for p in excluded_paths): + continue + problems.append(f"{full}: present in the source, absent from the packet") + + method = "sha256 over every manifest row" + if not problems and not renamed and not excluded_here and rsync_available(): + target = packet / source.name + if target.is_dir(): + lines = rsync_differences(source, target) + method = f"sha256 + rsync -rn --checksum ({len(lines)} transfer lines)" + problems += [f"rsync would transfer: {line}" for line in lines] + elif renamed or excluded_here: + method = "sha256 over every manifest row (rsync skipped: renames/exclusions recorded)" + elif not rsync_available(): + method = "sha256 over every manifest row (rsync not on PATH)" + return (not problems), problems, method + + +# -------------------------------------------------------------------------- +# README + + +def top_level_entries(dest: Path, manifest_name: str) -> list[tuple[str, bool]]: + entries = [] + for entry in sorted(os.scandir(dest), key=lambda e: e.name): + if entry.name in {manifest_name, README_NAME}: + continue + entries.append((entry.name, entry.is_dir(follow_symlinks=False))) + return entries + + +def render_readme( + title: str, + sources: list[Path], + manifest: dict, + manifest_name: str, + dest: Path, + verification: list[tuple[str, str, str]], + renames: list[dict[str, object]], + symlinks: list[dict[str, object]], +) -> str: + lines: list[str] = [f"# {title}", ""] + lines.append( + f"A packet landed by `{TOOL}` on {manifest['date']}: " + f"{manifest['file_count']} files, {human_bytes(manifest['total_bytes'])} bytes, " + "copied then verified. The sources were not modified, moved or removed." + ) + lines += ["", "## Sources", ""] + for source in sources: + lines.append(f"- `{source}`") + if manifest.get("source_session"): + lines += ["", f"Source session: `{manifest['source_session']}`."] + lines += ["", "## What is preserved", "", "| Entry | Kind |", "|---|---|"] + for name, is_dir in top_level_entries(dest, manifest_name): + lines.append(f"| `{name}` | {'directory' if is_dir else 'file'} |") + lines += [ + "", + f"{manifest['file_count']} files, {human_bytes(manifest['total_bytes'])} bytes.", + f"Manifest: [{manifest_name}]({manifest_name}) — source path, preserved path,", + "bytes, SHA-256 and mtime for every copied file, each row tagged with the source session.", + "", + "## Left out on purpose", + "", + ] + if manifest["excluded"]: + for entry in manifest["excluded"]: + lines.append(f"- `{entry['path']}` — {entry['reason']}") + else: + lines.append( + f"Nothing. No file exceeded the {CEILING_TEXT} per-file ceiling " + "and no exclusion was given." + ) + if renames: + lines += [ + "", + "## Nested repositories", + "", + f"{len(renames)} copied paths sat under a nested `.git`. Notes never tracks a nested", + "repository, so each was renamed to `dot-git` in this copy; contents are unchanged and", + "every affected manifest row carries `renamed_from` with the path it had at the source.", + "The sources keep their original names.", + ] + if symlinks: + lines += [ + "", + "## Symlinks", + "", + f"{len(symlinks)} symlinks were recorded and recreated, never followed:", + "", + ] + for row in symlinks[:20]: + lines.append(f"- `{row['source']}` -> `{row['target']}`") + if len(symlinks) > 20: + lines.append(f"- ... and {len(symlinks) - 20} more, all in the manifest") + lines += ["", "## Verification", "", "Run after the copy, against the preserved tree:", "", "```"] + width = max(len(check) for check, _, _ in verification) + 2 + result_width = max(len(result) for _, result, _ in verification) + 2 + for check, result, status in verification: + lines.append(f"{check.ljust(width)}{result.ljust(result_width)}{status}".rstrip()) + lines += ["```", ""] + lines.append( + "Reproduce with `land verify ` and " + "`land diff `." + ) + lines.append("") + return "\n".join(lines) + + +# -------------------------------------------------------------------------- +# commands + + +def cmd_copy(args: argparse.Namespace) -> int: + sources = [Path(os.path.abspath(source)) for source in args.sources] + dest = Path(os.path.abspath(args.dest)) + excludes = tuple(args.exclude or ()) + + if dest.exists(): + existing = [ + entry.name + for entry in os.scandir(dest) + if entry.is_file(follow_symlinks=False) or entry.is_dir(follow_symlinks=False) + ] + if existing and not args.force: + raise LandError( + f"{dest} is not empty ({len(existing)} entries). A packet owns its " + "directory: verify counts manifest rows against every file under it. " + "Pick a fresh directory, or pass --force if you meant to add to this one." + ) + for source in sources: + if dest == source or str(dest).startswith(str(source) + os.sep): + raise LandError(f"destination {dest} sits inside source {source}") + + items, excluded = plan_lane(sources, excludes) + if not items: + raise LandError("nothing to copy: the plan is empty") + guard_secrets(items, args.allow_secret_names) + + dest.mkdir(parents=True, exist_ok=True) + rows = copy_lane(items, dest, args.source_session) + + file_rows = [row for row in rows if row.get("kind") != "symlink"] + symlink_rows = [row for row in rows if row.get("kind") == "symlink"] + manifest_name = f"preservation-{date.today().isoformat()}.json" + manifest: dict[str, object] = { + "date": date.today().isoformat(), + "packet": dest.name, + "source_root": common_source_root(sources), + "preserved_root": str(dest), + "source_session": args.source_session, + "file_count": len(file_rows), + "total_bytes": sum(int(row["bytes"]) for row in file_rows), + "symlink_count": len(symlink_rows), + "max_file_bytes": MAX_FILE_BYTES, + "tool": TOOL, + "excluded": excluded, + "files": rows, + } + manifest_path = dest / manifest_name + with open(manifest_path, "w", encoding="utf-8") as handle: + json.dump(manifest, handle, indent=1) + handle.write("\n") + + ok, _, counts = verify_packet(dest, quiet=True) + verification = [ + ( + "manifest rows == find -type f count", + f"{counts['files']} == {counts['found']}", + "exit 0" if counts["files"] == counts["found"] else "MISMATCH", + ), + ( + "sha256 recomputed over all manifest rows", + f"{counts['files']} files {'OK' if ok else 'FAILED'}", + "exit 0" if ok else "exit 1", + ), + ] + diff_ok = True + for source in sources: + source_ok, problems, method = diff_source(dest, source) + diff_ok = diff_ok and source_ok + verification.append( + ( + f"equivalence, {source.name}", + "no differences" if source_ok else f"{len(problems)} differences", + "exit 0" if source_ok else "exit 1", + ) + ) + if not source_ok: + for problem in problems[:20]: + print(f"land copy: diff: {problem}", file=sys.stderr) + if verification: + method_line = method if sources else "" + if method_line: + verification.append(("equivalence method", method_line, "")) + + renames = [row for row in rows if row.get("renamed_from")] + readme = render_readme( + args.readme_title or dest.name, + sources, + manifest, + manifest_name, + dest, + verification, + renames, + symlink_rows, + ) + readme_path = dest / README_NAME + if readme_path.exists() and not args.force: + raise LandError( + f"{readme_path} already exists; land wrote the copy and the manifest but " + "will not overwrite a README. Move it aside and rerun, or pass --force." + ) + readme_path.write_text(readme, encoding="utf-8") + + print( + f"land copy {'OK' if ok and diff_ok else 'FAILED'} {dest} " + f"{len(file_rows)} files" + f"{f', {len(symlink_rows)} symlinks' if symlink_rows else ''}" + f"{f', {len(renames)} dot-git renames' if renames else ''} " + f"{human_bytes(int(manifest['total_bytes']))} bytes " + f"{len(excluded)} left out -> {manifest_name}" + ) + return 0 if (ok and diff_ok) else 1 + + +def cmd_verify(args: argparse.Namespace) -> int: + packet = Path(os.path.abspath(args.packet)) + if not packet.is_dir(): + raise LandError(f"not a directory: {packet}") + ok, _, _ = verify_packet(packet) + return 0 if ok else 1 + + +def cmd_diff(args: argparse.Namespace) -> int: + packet = Path(os.path.abspath(args.packet)) + source = Path(os.path.abspath(args.source)) + ok, problems, method = diff_source(packet, source) + for problem in problems[:50]: + print(f"land diff: {problem}", file=sys.stderr) + if len(problems) > 50: + print(f"land diff: ... and {len(problems) - 50} more", file=sys.stderr) + print( + f"land diff {'OK' if ok else 'FAILED'} {source} == {packet} " + f"{'no differences' if ok else str(len(problems)) + ' differences'} ({method})" + ) + return 0 if ok else 1 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="land", + description="Copy a lane into a preservation packet, then prove the copy.", + ) + sub = parser.add_subparsers(dest="command", required=True) + + copy = sub.add_parser("copy", help="copy sources into a packet and write its manifest") + copy.add_argument("sources", nargs="+", help="files or directories to preserve") + copy.add_argument("--dest", required=True, help="the packet directory to write") + copy.add_argument( + "--exclude", + action="append", + metavar="GLOB", + help="skip paths matching this glob (packet-relative path or basename); repeatable", + ) + copy.add_argument("--source-session", help="session id stamped on every row") + copy.add_argument("--readme-title", help="title line for the packet README") + copy.add_argument( + "--allow-secret-names", + action="store_true", + help="copy files whose NAME looks like a secret; content hits still abort", + ) + copy.add_argument( + "--force", + action="store_true", + help="write into a non-empty destination and overwrite an existing README", + ) + copy.set_defaults(func=cmd_copy) + + verify = sub.add_parser("verify", help="recompute every hash and count in a packet") + verify.add_argument("packet", help="the packet directory") + verify.set_defaults(func=cmd_verify) + + diff = sub.add_parser("diff", help="prove a source and its copy are the same bytes") + diff.add_argument("source", help="the original source path") + diff.add_argument("packet", help="the packet directory") + diff.set_defaults(func=cmd_diff) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + return int(args.func(args)) + except LandError as exc: + print(f"land: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/land/test_land.py b/tests/land/test_land.py new file mode 100644 index 000000000..218d0cc46 --- /dev/null +++ b/tests/land/test_land.py @@ -0,0 +1,385 @@ +"""checks.land — the copy-then-verify landing tool (CNA-M07). + +Hermetic: every lane is built in a temp directory by the test itself, because +the tool's whole claim is about bytes it copied a second ago. The fixture lane +carries the five cases that have actually cost time on a real sweep — a nested +`.git`, a symlink, a tiny file, an excluded glob and a file over the per-file +ceiling — plus the two guards that must abort before anything is written. +""" + +from __future__ import annotations + +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = Path(os.environ.get("LAND_PY", REPO_ROOT / "pkgs/land/land.py")) + +spec = importlib.util.spec_from_file_location("land", SCRIPT) +land = importlib.util.module_from_spec(spec) +spec.loader.exec_module(land) + +BIG_BYTES = 100 * 1024 * 1024 # over the 95 MiB ceiling; sparse, costs no disk + + +def run(*argv: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(SCRIPT), *argv], + capture_output=True, + text=True, + check=False, + ) + + +def build_lane(root: Path) -> Path: + """The fixture lane. Returns its path.""" + lane = root / "lane" + (lane / "repo" / ".git").mkdir(parents=True) + (lane / "scratch").mkdir(parents=True) + (lane / "notes.md").write_text("abc") # the 3-byte file + (lane / "repo" / "README.md").write_text("a nested repository\n") + (lane / "repo" / ".git" / "config").write_text("[core]\n\trepositoryformatversion = 0\n") + (lane / "repo" / ".git" / "HEAD").write_text("ref: refs/heads/main\n") + (lane / "scratch" / "noise.log").write_text("excluded by glob\n") + os.symlink("notes.md", lane / "link.md") + with open(lane / "big.bin", "wb") as handle: + handle.truncate(BIG_BYTES) + return lane + + +def manifest_of(packet: Path) -> dict: + return json.loads(land.find_manifest(packet).read_text()) + + +def tree_digest(root: Path) -> list[tuple[str, int, float, str]]: + """lstat-visible fingerprint of a tree: proof the sources were not touched.""" + out = [] + for path in sorted(root.rglob("*")): + stat_result = path.lstat() + kind = "link" if path.is_symlink() else ("dir" if path.is_dir() else "file") + out.append((str(path.relative_to(root)), stat_result.st_size, stat_result.st_mtime, kind)) + return out + + +class LandFixtureLane(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + self.lane = build_lane(self.root) + self.packet = self.root / "packet" + self.before = tree_digest(self.lane) + self.copy = run( + "copy", + str(self.lane), + "--dest", + str(self.packet), + "--exclude", + "*.log", + "--source-session", + "test-session-1", + "--readme-title", + "Fixture lane", + ) + + def tearDown(self) -> None: + self.tmp.cleanup() + + def test_copy_succeeds(self) -> None: + self.assertEqual(self.copy.returncode, 0, self.copy.stderr) + self.assertIn("land copy OK", self.copy.stdout) + + def test_manifest_row_count_equals_find_type_f(self) -> None: + manifest = manifest_of(self.packet) + files = [row for row in manifest["files"] if row.get("kind") != "symlink"] + found = [ + path + for path in self.packet.rglob("*") + if path.is_file() and not path.is_symlink() and path.parent != self.packet + ] + found += [ + path + for path in self.packet.iterdir() + if path.is_file() + and not path.is_symlink() + and path.name != "README.md" + and not path.name.startswith("preservation-") + ] + self.assertEqual(manifest["file_count"], len(files)) + self.assertEqual(manifest["file_count"], len(found)) + # notes.md, repo/README.md, repo/dot-git/{config,HEAD} + self.assertEqual(manifest["file_count"], 4) + + def test_dot_git_rename_recorded(self) -> None: + manifest = manifest_of(self.packet) + renamed = [row for row in manifest["files"] if row.get("renamed_from")] + self.assertEqual(len(renamed), 2) + self.assertTrue((self.packet / "lane/repo/dot-git/config").is_file()) + self.assertFalse((self.packet / "lane/repo/.git").exists()) + for row in renamed: + self.assertIn("dot-git", row["preserved"]) + self.assertIn(".git", row["renamed_from"]) + # the source keeps its own name + self.assertTrue((self.lane / "repo" / ".git" / "config").is_file()) + + def test_symlink_row_not_followed(self) -> None: + manifest = manifest_of(self.packet) + links = [row for row in manifest["files"] if row.get("kind") == "symlink"] + self.assertEqual(len(links), 1) + self.assertEqual(links[0]["target"], "notes.md") + self.assertTrue(os.path.islink(self.packet / "lane/link.md")) + self.assertEqual(manifest["symlink_count"], 1) + + def test_three_byte_file_lands_with_mode_and_mtime(self) -> None: + manifest = manifest_of(self.packet) + row = next(r for r in manifest["files"] if r["source"].endswith("notes.md")) + self.assertEqual(row["bytes"], 3) + self.assertEqual(row["source_session"], "test-session-1") + self.assertIn("mtime", row) + source = self.lane / "notes.md" + preserved = self.packet / "lane/notes.md" + self.assertEqual(int(source.stat().st_mtime), int(preserved.stat().st_mtime)) + self.assertEqual(source.stat().st_mode, preserved.stat().st_mode) + + def test_exclusions_carry_reasons(self) -> None: + manifest = manifest_of(self.packet) + reasons = {entry["path"]: entry["reason"] for entry in manifest["excluded"]} + log = str(self.lane / "scratch/noise.log") + big = str(self.lane / "big.bin") + self.assertIn(log, reasons) + self.assertIn("--exclude *.log", reasons[log]) + self.assertIn(big, reasons) + self.assertIn("ceiling", reasons[big]) + self.assertFalse((self.packet / "lane/big.bin").exists()) + + def test_readme_states_counts_exclusions_and_verification(self) -> None: + readme = (self.packet / "README.md").read_text() + self.assertIn("# Fixture lane", readme) + self.assertIn("big.bin", readme) + self.assertIn("## Verification", readme) + self.assertIn("manifest rows == find -type f count", readme) + self.assertIn("4 == 4", readme) + self.assertIn("dot-git", readme) + + def test_verify_and_diff_exit_zero(self) -> None: + verify = run("verify", str(self.packet)) + self.assertEqual(verify.returncode, 0, verify.stderr) + self.assertIn("land verify OK", verify.stdout) + diff = run("diff", str(self.lane), str(self.packet)) + self.assertEqual(diff.returncode, 0, diff.stderr) + self.assertIn("no differences", diff.stdout) + + def test_sources_were_not_touched(self) -> None: + self.assertEqual(self.before, tree_digest(self.lane)) + + def test_verify_fails_on_a_flipped_byte(self) -> None: + target = self.packet / "lane/notes.md" + target.write_text("abd") + verify = run("verify", str(self.packet)) + self.assertEqual(verify.returncode, 1) + self.assertIn("sha256 does not match", verify.stderr) + self.assertIn("sha256 MISMATCH", verify.stdout) + + def test_verify_fails_on_a_stray_file(self) -> None: + (self.packet / "lane" / "stray.md").write_text("not in the manifest\n") + verify = run("verify", str(self.packet)) + self.assertEqual(verify.returncode, 1) + self.assertIn("count mismatch", verify.stderr) + # every hash still matched: the receipt must not blame sha256 for a count + self.assertIn("sha256 all match", verify.stdout) + self.assertIn("land verify FAILED", verify.stdout) + + def test_verify_fails_on_a_missing_file(self) -> None: + (self.packet / "lane/notes.md").unlink() + verify = run("verify", str(self.packet)) + self.assertEqual(verify.returncode, 1) + self.assertIn("missing from the packet", verify.stderr) + + def test_diff_fails_when_the_source_drifts(self) -> None: + (self.lane / "notes.md").write_text("abcd") + diff = run("diff", str(self.lane), str(self.packet)) + self.assertEqual(diff.returncode, 1) + self.assertIn("source bytes differ", diff.stderr) + + def test_diff_fails_on_a_source_file_absent_from_the_packet(self) -> None: + (self.lane / "late.md").write_text("written after the copy\n") + diff = run("diff", str(self.lane), str(self.packet)) + self.assertEqual(diff.returncode, 1) + self.assertIn("absent from the packet", diff.stderr) + + def test_verify_survives_a_moved_packet(self) -> None: + moved = self.root / "moved-packet" + os.rename(self.packet, moved) + verify = run("verify", str(moved)) + self.assertEqual(verify.returncode, 0, verify.stderr) + + +class LandGuards(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + self.lane = self.root / "lane" + self.lane.mkdir() + (self.lane / "keep.md").write_text("ordinary prose\n") + self.packet = self.root / "packet" + + def tearDown(self) -> None: + self.tmp.cleanup() + + def test_secret_name_aborts_before_writing(self) -> None: + (self.lane / "deploy.env").write_text("PORT=8080\n") + result = run("copy", str(self.lane), "--dest", str(self.packet)) + self.assertEqual(result.returncode, 2) + self.assertIn("deploy.env", result.stderr) + self.assertIn("*.env", result.stderr) + self.assertFalse(self.packet.exists()) + + def test_allow_secret_names_lets_a_name_through(self) -> None: + (self.lane / "deploy.env").write_text("PORT=8080\n") + result = run( + "copy", str(self.lane), "--dest", str(self.packet), "--allow-secret-names" + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((self.packet / "lane/deploy.env").is_file()) + + def test_secret_content_aborts_and_never_prints_the_value(self) -> None: + value = "ghp_" + "A1b2C3d4E5f6G7h8I9j0KL" + (self.lane / "runbook.md").write_text(f"export GH_TOKEN={value}\n") + result = run( + "copy", str(self.lane), "--dest", str(self.packet), "--allow-secret-names" + ) + self.assertEqual(result.returncode, 2) + self.assertIn("runbook.md", result.stderr) + self.assertIn("GitHub personal access token", result.stderr) + self.assertNotIn(value, result.stderr + result.stdout) + self.assertFalse(self.packet.exists()) + + def test_private_key_block_aborts(self) -> None: + (self.lane / "notes.md").write_text("-----BEGIN OPENSSH PRIVATE KEY-----\n") + result = run("copy", str(self.lane), "--dest", str(self.packet)) + self.assertEqual(result.returncode, 2) + self.assertIn("PEM private key block", result.stderr) + + def test_binary_file_is_not_content_scanned(self) -> None: + (self.lane / "image.bin").write_bytes(b"\0\0ghp_" + b"A" * 30) + result = run("copy", str(self.lane), "--dest", str(self.packet)) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_gitfile_is_renamed_like_a_git_directory(self) -> None: + (self.lane / "submodule").mkdir() + (self.lane / "submodule" / ".git").write_text("gitdir: ../.git/modules/x\n") + result = run("copy", str(self.lane), "--dest", str(self.packet)) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((self.packet / "lane/submodule/dot-git").is_file()) + row = next( + r + for r in manifest_of(self.packet)["files"] + if r["preserved"].endswith("submodule/dot-git") + ) + self.assertTrue(row["renamed_from"].endswith("submodule/.git")) + + def test_non_empty_destination_is_refused(self) -> None: + self.packet.mkdir() + (self.packet / "already-here.md").write_text("prior contents\n") + result = run("copy", str(self.lane), "--dest", str(self.packet)) + self.assertEqual(result.returncode, 2) + self.assertIn("is not empty", result.stderr) + + def test_destination_inside_source_is_refused(self) -> None: + result = run("copy", str(self.lane), "--dest", str(self.lane / "inner")) + self.assertEqual(result.returncode, 2) + self.assertIn("sits inside source", result.stderr) + + +class LandLegacyManifest(unittest.TestCase): + """A pre-land packet has no mtime and no source_session. Warn, do not fail.""" + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.packet = Path(self.tmp.name) / "nyu-style" + self.packet.mkdir() + body = "the delivered deck\n" + (self.packet / "deck.md").write_text(body) + manifest = { + "date": "2026-09-16", + "source_session": "codex:01a0a46d", + "files": [ + { + "source": "/home/tom/decks/nyu-2026-09-15/deck.md", + "preserved": str(self.packet / "deck.md"), + "bytes": len(body), + "sha256": land.sha256_bytes(body.encode()), + } + ], + } + (self.packet / "preservation-2026-09-16.json").write_text(json.dumps(manifest, indent=1)) + + def tearDown(self) -> None: + self.tmp.cleanup() + + def test_missing_mtime_warns_and_still_exits_zero(self) -> None: + result = run("verify", str(self.packet)) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("no mtime recorded", result.stderr) + self.assertIn("no source_session recorded", result.stderr) + self.assertIn("land verify OK", result.stdout) + self.assertIn("2 warnings", result.stdout) + + +class LandCleanLane(unittest.TestCase): + """No renames, no exclusions: the one shape where rsync can be a second opinion.""" + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + root = Path(self.tmp.name) + self.lane = root / "clean" + (self.lane / "sub").mkdir(parents=True) + (self.lane / "a.md").write_text("one\n") + (self.lane / "sub" / "b.md").write_text("two\n") + self.packet = root / "packet" + self.copy = run("copy", str(self.lane), "--dest", str(self.packet)) + + def tearDown(self) -> None: + self.tmp.cleanup() + + @unittest.skipUnless(land.rsync_available(), "rsync is not on PATH") + def test_diff_cross_checks_with_rsync(self) -> None: + diff = run("diff", str(self.lane), str(self.packet)) + self.assertEqual(diff.returncode, 0, diff.stderr) + self.assertIn("rsync -rn --checksum (0 transfer lines)", diff.stdout) + + @unittest.skipUnless(land.rsync_available(), "rsync is not on PATH") + def test_rsync_sees_a_planted_difference(self) -> None: + (self.packet / "clean/a.md").write_text("one\n") + os.utime(self.packet / "clean/a.md", (0, 0)) + (self.lane / "sub" / "c.md").write_text("three\n") + lines = land.rsync_differences(self.lane, self.packet / "clean") + self.assertTrue(any("c.md" in line for line in lines), lines) + + def test_readme_names_the_ceiling_in_mib(self) -> None: + self.assertIn("95 MiB", (self.packet / "README.md").read_text()) + + +class LandUnits(unittest.TestCase): + def test_secret_name_globs(self) -> None: + self.assertEqual(land.secret_name_hit("deploy.env"), "*.env") + self.assertEqual(land.secret_name_hit("GITHUB_TOKEN.md"), "*token*") + self.assertEqual(land.secret_name_hit("id_ed25519.pub"), "id_ed25519*") + self.assertEqual(land.secret_name_hit("backup.age"), "*.age") + self.assertIsNone(land.secret_name_hit("README.md")) + + def test_mtime_is_iso_with_offset(self) -> None: + stamp = land.iso_mtime(1758016000.0) + self.assertRegex(stamp, r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}[+-]\d{2}:\d{2}$") + + def test_ceiling_is_95_mib(self) -> None: + self.assertEqual(land.MAX_FILE_BYTES, 95 * 1024 * 1024) + + +if __name__ == "__main__": + unittest.main(verbosity=2)