From 15a44d1fb3a5d457e4e8bf87ffe46d3795481a8a Mon Sep 17 00:00:00 2001 From: Sagar Kharal Date: Mon, 7 Sep 2026 14:55:37 +0530 Subject: [PATCH] feat(snapshot): verify delivered snapshots against integrity receipts --- docs/how-to/export-dataset-snapshot.md | 37 ++-- src/hflow/__init__.py | 2 + src/hflow/cli.py | 48 +++++ src/hflow/snapshot.py | 130 ++++++++++++ src/hflow/verification.py | 75 +++++++ tests/test_snapshot_verify.py | 276 +++++++++++++++++++++++++ 6 files changed, 554 insertions(+), 14 deletions(-) create mode 100644 src/hflow/verification.py create mode 100644 tests/test_snapshot_verify.py diff --git a/docs/how-to/export-dataset-snapshot.md b/docs/how-to/export-dataset-snapshot.md index c3f814d9..76918812 100644 --- a/docs/how-to/export-dataset-snapshot.md +++ b/docs/how-to/export-dataset-snapshot.md @@ -44,15 +44,15 @@ The observable result is: ```text dataset-snapshot/ -├── format.json -├── samples.parquet -├── measurements.parquet -├── observations.parquet -├── media.parquet -├── check_runs.parquet -├── tags.parquet -├── intervals.parquet -└── assets/ # copy mode only, when artifacts exist +|-- format.json +|-- samples.parquet +|-- measurements.parquet +|-- observations.parquet +|-- media.parquet +|-- check_runs.parquet +|-- tags.parquet +|-- intervals.parquet +`-- assets/ # copy mode only, when artifacts exist ``` ## Format contract @@ -79,11 +79,20 @@ external readers; receipts live only under `integrity`: Copy mode re-reads each copied asset once after the copy to compute its hash (a second full read of media bytes on export). -This is a receipt, not a verify command: export does not re-read the -destination after transfer, and there is no public `verify_dataset_snapshot` -API or CLI yet. Older HFlow overwrite checks only `format` and -`format_version`, and readers that only consume the string `tables` map keep -working; the `integrity` key is purely additive under format version `1`. +Verify the delivery after transfer with +`hflow verify snapshot ` or +`verify_dataset_snapshot()`: every table and copied asset named +in the receipt is re-read and compared by size and sha256, and every finding +is returned in one report built from the shared verification types in +`hflow.verification` (a `VerificationReport` with status `ok`, `damaged`, +or `unverifiable`). Recorded paths are joined only onto the handed +directory, so a copied delivery verifies in place. Exit `0` clean, +`1` damaged, `3` unverifiable, `2` unreadable input. A pre-#401 +`format.json` with no `integrity` key is reported as `no-receipt`, +unverifiable, not corrupt. Older HFlow overwrite checks only +`format` and `format_version`, and readers that only consume the string +`tables` map keep working; the `integrity` key is purely additive under +format version `1`. The receipt travels unsigned inside the `format.json` it describes, so it catches corruption and accidental loss, not tampering: anyone who can edit a diff --git a/src/hflow/__init__.py b/src/hflow/__init__.py index 708bacc5..ff84bf9c 100644 --- a/src/hflow/__init__.py +++ b/src/hflow/__init__.py @@ -72,6 +72,7 @@ RetainedDatasetSnapshotBackup, SnapshotMediaMode, export_dataset_snapshot, + verify_dataset_snapshot, ) from hflow.steps import ( RUN_PROFILES, @@ -208,5 +209,6 @@ "step_version_from_contract", "testing", "to_grid", + "verify_dataset_snapshot", "write_canonical_episode", ] diff --git a/src/hflow/cli.py b/src/hflow/cli.py index b5b1f976..4e6e38de 100644 --- a/src/hflow/cli.py +++ b/src/hflow/cli.py @@ -377,6 +377,31 @@ def _build_parser() -> argparse.ArgumentParser: help="atomically replace an existing export directory", ) + verify_parser = subparsers.add_parser( + "verify", + help="verify a delivery against its receipt", + description=( + "Group commands for verifying deliveries against their recorded " + "receipts. Use `snapshot` to re-check a delivered dataset snapshot " + "directory against the integrity receipt inside its format.json." + ), + ) + verify_subparsers = verify_parser.add_subparsers(dest="verify_command", required=True) + verify_snapshot_parser = verify_subparsers.add_parser( + "snapshot", + help="verify a delivered dataset snapshot against its integrity receipt", + description=( + "Re-reads every table and copied asset named in the snapshot's " + "integrity receipt and reports bytes changed, files missing, and " + "size mismatches. Exit 0 clean, 1 damaged, 2 unreadable input, " + "3 no receipt (unverifiable)." + ), + ) + verify_snapshot_parser.add_argument( + "directory", + help="the delivered dataset snapshot directory to verify", + ) + stale_parser = subparsers.add_parser( "stale", help="list episodes whose latest cataloged run predates the current pipeline version", @@ -1472,6 +1497,25 @@ def _command_export_snapshot(arguments: argparse.Namespace) -> int: return 0 +def _command_verify_snapshot(arguments: argparse.Namespace) -> int: + from hflow.snapshot import verify_dataset_snapshot + from hflow.verification import exit_code_for + + try: + report = verify_dataset_snapshot(Path(arguments.directory)) + except (ValueError, FileNotFoundError, NotADirectoryError, OSError) as error: + print(f"verify snapshot: {error}", file=sys.stderr) + return 2 + if report.ok: + print("verified: every receipted file matches its size and sha256") + return exit_code_for(report) + print(f"not verified: {len(report.findings)} finding(s)") + for finding in report.findings: + print(f" [{finding.reason}] {finding.uri}") + print(f" {finding.detail}") + return exit_code_for(report) + + def _command_doctor(arguments: argparse.Namespace) -> int: # Findings, not exceptions, across files as well: an unreadable path is a # finding about the corpus, reported in place, so a batch run never loses @@ -1644,6 +1688,10 @@ def main(argv: list[str] | None = None) -> int: if arguments.export_command == "snapshot": return _command_export_snapshot(arguments) raise AssertionError(f"unhandled export command {arguments.export_command!r}") + if arguments.command == "verify": + if arguments.verify_command == "snapshot": + return _command_verify_snapshot(arguments) + raise AssertionError(f"unhandled verify command {arguments.verify_command!r}") if arguments.command == "stale": return _command_stale(arguments) if arguments.command == "doctor": diff --git a/src/hflow/snapshot.py b/src/hflow/snapshot.py index 19f8fd94..bab40e6d 100644 --- a/src/hflow/snapshot.py +++ b/src/hflow/snapshot.py @@ -15,6 +15,7 @@ from dataclasses import dataclass from enum import StrEnum from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING from uuid import uuid4 import duckdb @@ -24,6 +25,9 @@ from hflow.curation import open_catalog_connection from hflow.storage import StorageRoot, fetch_uri +if TYPE_CHECKING: + from hflow.verification import VerificationReport + DATASET_SNAPSHOT_FORMAT_NAME = "hflow-dataset-snapshot" DATASET_SNAPSHOT_FORMAT_VERSION = "1" @@ -797,3 +801,129 @@ def export_dataset_snapshot( media_mode=resolved_media_mode, retained_backup=retained_backup, ) + + +def verify_dataset_snapshot( + snapshot_directory: Path | str, +) -> "VerificationReport": + """Check a delivered dataset snapshot against the receipt inside it. + + Re-reads every table and copied asset named in the ``integrity`` block of + ``format.json`` and compares sizes and SHA-256 digests, resolving every + recorded path strictly relative to the handed snapshot directory. A + recorded path is never read as absolute, so a snapshot moved or copied + to another root verifies in place. Returns every finding in one report + instead of raising at the first mismatch, because a partial transfer + usually damages more than one file. + + Findings use the shared reason vocabulary from ``hflow.verification`` + (#432 contract, #454 shape), each meaning something different: + + - ``content-id-mismatch``: the sha256 of the file under this root does + not match the sha256 recorded in the receipt. + - ``missing``: named in the receipt, absent under this root. + - ``size-mismatch``: size differs first; the hash read is skipped. + - ``no-receipt``: a valid pre-#401 ``format.json`` with no + ``integrity`` key. That snapshot is unverifiable, not corrupt. + + Unreadable input (a missing directory, a missing or unparsable + ``format.json``) raises instead: that is not a finding about a delivered + snapshot, it is the wrong input entirely. + + Extra files under ``assets/`` that the receipt does not name are ignored: + the receipt covers what was exported, not everything a recipient may add. + + This catches corruption, truncation, and partial transfer. It is not a + tamper defence: the receipt travels unsigned inside the same + ``format.json`` it describes, so anyone who can rewrite a table can + recompute the hashes to match. + """ + from hflow.verification import ( + REASON_CONTENT_ID_MISMATCH, + REASON_MISSING, + REASON_NO_RECEIPT, + REASON_SIZE_MISMATCH, + VerificationFinding, + VerificationReport, + VerificationStatus, + ) + + resolved_directory = Path(snapshot_directory) + findings: list[VerificationFinding] = [] + + if not resolved_directory.is_dir(): + raise NotADirectoryError(f"snapshot directory does not exist: {resolved_directory}") + marker_path = resolved_directory / "format.json" + if not marker_path.is_file(): + raise FileNotFoundError(f"no format.json in {resolved_directory}; not a dataset snapshot") + try: + format_marker = json.loads(marker_path.read_text()) + except (json.JSONDecodeError, UnicodeDecodeError) as error: + raise ValueError(f"format.json is unreadable: {error}") from error + + integrity = format_marker.get("integrity") + if not isinstance(integrity, dict): + # A valid v1 snapshot from before #401: verifiable nothing, corrupt nothing. + return VerificationReport( + status=VerificationStatus.UNVERIFIABLE, + findings=[ + VerificationFinding( + uri="format.json", + reason=REASON_NO_RECEIPT, + detail=( + "format.json carries no integrity receipt " + "(pre-#401 snapshot); the delivery is unverifiable, not corrupt" + ), + ), + ], + ) + + receipt_entries: list[dict[str, str | int]] = [ + *integrity.get("tables", {}).values(), + *integrity.get("assets", []), + ] + for entry in receipt_entries: + relative_path = str(entry["path"]) + delivered_path = resolved_directory / relative_path + if not delivered_path.is_file(): + findings.append( + VerificationFinding( + uri=relative_path, + reason=REASON_MISSING, + detail=( + f"receipted file missing at {relative_path!r} " + "under the verified snapshot root" + ), + ) + ) + continue + delivered_size = delivered_path.stat().st_size + receipt_size = int(entry["size_bytes"]) + if delivered_size != receipt_size: + findings.append( + VerificationFinding( + uri=relative_path, + reason=REASON_SIZE_MISMATCH, + detail=( + f"size under the verified root {delivered_size} bytes " + f"!= receipt {receipt_size} bytes" + ), + ) + ) + continue + delivered_sha256 = _sha256_hex(delivered_path) + if delivered_sha256 != entry["sha256"]: + findings.append( + VerificationFinding( + uri=relative_path, + reason=REASON_CONTENT_ID_MISMATCH, + detail=( + f"sha256 under the verified root {delivered_sha256!r} " + f"!= receipt sha256 {entry['sha256']!r}" + ), + ) + ) + + if findings: + return VerificationReport(status=VerificationStatus.DAMAGED, findings=findings) + return VerificationReport(status=VerificationStatus.OK) diff --git a/src/hflow/verification.py b/src/hflow/verification.py new file mode 100644 index 00000000..6e0308a7 --- /dev/null +++ b/src/hflow/verification.py @@ -0,0 +1,75 @@ +"""Shared verification types for every ``hflow verify ...`` command. + +``VerificationReport`` is the common surface for delivery verifiers: a +verifier reads the receipt a delivery carries, compares it to the bytes +under the root being verified, and returns one report. Files nobody +listed are ignored. The LeRobot import verifier (#454) and the dataset +snapshot verifier (#428) both return this shape, so one CLI and one +exit-code mapping cover every delivered artifact. + +Reasons are fixed strings so callers can branch on them. +``exit_code_for`` maps a report to the verify-family exit codes: 0 clean, +1 damaged, 3 unverifiable. Exit 2 (unreadable input) is raised as an +exception before a report exists. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import StrEnum + +REASON_MISSING = "missing" +REASON_SIZE_MISMATCH = "size-mismatch" +REASON_CONTENT_ID_MISMATCH = "content-id-mismatch" +# #428: a readable snapshot format.json that carries no integrity receipt. +REASON_NO_RECEIPT = "no-receipt" + + +class VerificationStatus(StrEnum): + """Outcome of a verify that successfully read its receipt format.""" + + OK = "ok" + DAMAGED = "damaged" + UNVERIFIABLE = "unverifiable" + + +@dataclass(frozen=True) +class VerificationFinding: + """One claimed object that does not match its receipt.""" + + uri: str + reason: str + detail: str + + +@dataclass(frozen=True) +class VerificationReport: + """Result of verifying a delivered artifact against its receipts. + + ``.ok`` is True when every claimed object still matches, including the + empty-claim case (a readable receipt that lists nothing). ``status`` + distinguishes clean, damaged, and unverifiable (no receipt) so CLI exit + codes stay distinct. + """ + + status: VerificationStatus + findings: list[VerificationFinding] = field(default_factory=list) + + @property + def ok(self) -> bool: + return self.status is VerificationStatus.OK + + +def exit_code_for(report: VerificationReport) -> int: + """Map a readable verification outcome to the verify-family exit code. + + ``0`` clean, ``1`` damaged, ``3`` unverifiable. Exit ``2`` (unreadable + input) is raised as an exception before a report exists. + """ + if report.status is VerificationStatus.OK: + return 0 + if report.status is VerificationStatus.DAMAGED: + return 1 + if report.status is VerificationStatus.UNVERIFIABLE: + return 3 + raise AssertionError(f"unhandled verification status {report.status!r}") diff --git a/tests/test_snapshot_verify.py b/tests/test_snapshot_verify.py new file mode 100644 index 00000000..5adda878 --- /dev/null +++ b/tests/test_snapshot_verify.py @@ -0,0 +1,276 @@ +"""Snapshot verification: the delivery checked against its receipt (#428). + +Every test drives the REAL pipeline: a catalog is built, a snapshot is +exported through the public API, the delivery is damaged surgically, and +verify_dataset_snapshot must report exactly the damage -- nothing more, +nothing less. +""" + +import json +import shutil +from pathlib import Path + +import pytest +from test_dataset_snapshot import _append_snapshot_episode + +import hflow +from hflow.catalog import Catalog +from hflow.cli import main as cli_main +from hflow.snapshot import verify_dataset_snapshot + + +def _export_two_episode_snapshot(tmp_path: Path, media_mode: str) -> tuple[Path, dict]: + catalog = Catalog(tmp_path / "catalog") + selected_episode_id, _ = _append_snapshot_episode( + catalog, tmp_path, name="fold-shirt", score=0.75, with_media=(media_mode == "copy") + ) + _append_snapshot_episode(catalog, tmp_path, name="pour-water", score=0.25, with_media=False) + manifest = tmp_path / "manifest.parquet" + hflow.curate( + catalog.location, + f"SELECT episode_id FROM episodes WHERE episode_id = '{selected_episode_id}'", + output=manifest, + ) + output_directory = tmp_path / "dataset-snapshot" + hflow.export_dataset_snapshot( + catalog.location, output_directory, manifest=manifest, media_mode=media_mode + ) + marker = json.loads((output_directory / "format.json").read_text()) + return output_directory, marker + + +def _rewrite_format_without_integrity(output_directory: Path) -> None: + marker_path = output_directory / "format.json" + marker = json.loads(marker_path.read_text()) + marker.pop("integrity", None) + marker_path.write_text(json.dumps(marker, indent=2)) + + +def test_clean_snapshot_verifies_clean_in_references_mode(tmp_path: Path) -> None: + output_directory, _ = _export_two_episode_snapshot(tmp_path, "references") + report = verify_dataset_snapshot(output_directory) + assert report.ok + assert report.findings == [] + + +def test_clean_snapshot_verifies_clean_in_copy_mode(tmp_path: Path) -> None: + output_directory, _ = _export_two_episode_snapshot(tmp_path, "copy") + report = verify_dataset_snapshot(output_directory) + assert report.ok + assert report.findings == [] + + +def test_bytes_changed_reports_content_mismatch_alone(tmp_path: Path) -> None: + """Same size, different bytes: the receipt must call this a content + mismatch, not a size mismatch, and must not raise.""" + output_directory, marker = _export_two_episode_snapshot(tmp_path, "references") + table_path = output_directory / marker["integrity"]["tables"]["samples"]["path"] + data = bytearray(table_path.read_bytes()) + data[len(data) // 2] ^= 0xFF # same length, different content + table_path.write_bytes(bytes(data)) + + report = verify_dataset_snapshot(output_directory) + + assert not report.ok + assert [f.reason for f in report.findings] == ["content-id-mismatch"] + finding = report.findings[0] + assert finding.uri == marker["integrity"]["tables"]["samples"]["path"] + assert finding.detail + + +def test_missing_file_reports_missing_alone(tmp_path: Path) -> None: + output_directory, marker = _export_two_episode_snapshot(tmp_path, "references") + (output_directory / marker["integrity"]["tables"]["samples"]["path"]).unlink() + + report = verify_dataset_snapshot(output_directory) + + assert not report.ok + assert [f.reason for f in report.findings] == ["missing"] + assert marker["integrity"]["tables"]["samples"]["path"] in report.findings[0].uri + + +def test_truncated_file_reports_size_mismatch_and_skips_the_hash( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Size is the cheap pre-filter: a truncated file is reported by size + alone without spending the hash read.""" + output_directory, marker = _export_two_episode_snapshot(tmp_path, "references") + table_path = output_directory / marker["integrity"]["tables"]["measurements"]["path"] + original = table_path.read_bytes() + table_path.write_bytes(original[: len(original) // 2]) + + hashed: list[Path] = [] + real = hflow.snapshot._sha256_hex + + def spy(path: Path) -> str: + hashed.append(path) + return real(path) + + monkeypatch.setattr(hflow.snapshot, "_sha256_hex", spy) + + report = verify_dataset_snapshot(output_directory) + + assert not report.ok + reasons = [f.reason for f in report.findings] + assert "size-mismatch" in reasons + assert "content-id-mismatch" not in reasons + truncated_uri = marker["integrity"]["tables"]["measurements"]["path"] + assert truncated_uri not in [str(path) for path in hashed] + assert hashed, "spy must have recorded at least one hashed file" + + +def test_copied_asset_damage_is_reported(tmp_path: Path) -> None: + """Copy mode stores media inside the snapshot; the receipt covers it.""" + output_directory, marker = _export_two_episode_snapshot(tmp_path, "copy") + assert marker["integrity"]["assets"], "fixture must include copied assets" + asset_uri = marker["integrity"]["assets"][0]["path"] + asset_path = output_directory / asset_uri + data = bytearray(asset_path.read_bytes()) + data[len(data) // 2] ^= 0xFF + asset_path.write_bytes(bytes(data)) + + report = verify_dataset_snapshot(output_directory) + + assert not report.ok + damaged = [f for f in report.findings if f.uri == asset_uri] + assert damaged and damaged[0].reason == "content-id-mismatch" + + +def test_pre_401_format_json_is_unverifiable_not_corrupt(tmp_path: Path) -> None: + """A valid v1 snapshot without an integrity receipt is unverifiable, not + corrupt: the finding says no_receipt and nothing raises.""" + output_directory, _ = _export_two_episode_snapshot(tmp_path, "references") + _rewrite_format_without_integrity(output_directory) + + report = verify_dataset_snapshot(output_directory) + + assert not report.ok + assert [f.reason for f in report.findings] == ["no-receipt"] + + +def test_extra_files_under_assets_are_ignored(tmp_path: Path) -> None: + """Files the receipt does not name produce no finding and no warning: + unlisted extras are outside the receipt's contract.""" + output_directory, _ = _export_two_episode_snapshot(tmp_path, "copy") + (output_directory / "assets" / "unlisted-extra.bin").write_bytes(b"extra bytes") + + report = verify_dataset_snapshot(output_directory) + + assert report.ok + assert report.findings == [] + + +def test_partial_transfer_reports_every_mismatch_in_one_report(tmp_path: Path) -> None: + """A partial transfer usually damages more than one file: the report + carries every mismatch in one list instead of raising on the first.""" + output_directory, marker = _export_two_episode_snapshot(tmp_path, "references") + samples_path = output_directory / marker["integrity"]["tables"]["samples"]["path"] + data = bytearray(samples_path.read_bytes()) + data[len(data) // 2] ^= 0xFF + samples_path.write_bytes(bytes(data)) + (output_directory / marker["integrity"]["tables"]["tags"]["path"]).unlink() + + report = verify_dataset_snapshot(output_directory) + + assert not report.ok + reasons = sorted(f.reason for f in report.findings) + assert reasons == ["content-id-mismatch", "missing"] + + +def test_verify_snapshot_cli_exit_codes(tmp_path: Path) -> None: + """0 clean, 1 damaged, 3 unverifiable, 2 unreadable -- through the CLI.""" + output_directory, _ = _export_two_episode_snapshot(tmp_path, "references") + argv = ["verify", "snapshot", str(output_directory)] + assert cli_main(argv) == 0 + + marker_path = output_directory / "format.json" + marker = json.loads(marker_path.read_text()) + table_path = output_directory / marker["integrity"]["tables"]["samples"]["path"] + data = bytearray(table_path.read_bytes()) + data[len(data) // 2] ^= 0xFF + table_path.write_bytes(bytes(data)) + assert cli_main(argv) == 1 + + _rewrite_format_without_integrity(output_directory) + assert cli_main(argv) == 3 + + marker_path.write_bytes(b"") + assert cli_main(argv) == 2 + + marker_path.write_bytes(b"\x00\xff\xfe\x01garbage") + assert cli_main(argv) == 2 + + missing_root = str(tmp_path / "does-not-exist") + assert cli_main(["verify", "snapshot", missing_root]) == 2 + + +def test_verify_is_read_only_against_the_delivery(tmp_path: Path) -> None: + """verify must not rewrite format.json, the manifest, the table files, + or the catalog. A re-read after a verify run must return byte-identical + contents and unchanged mtimes.""" + output_directory, _ = _export_two_episode_snapshot(tmp_path, "references") + catalog_path = tmp_path / "catalog" + + def snapshot_everything() -> dict[Path, tuple[bytes, float]]: + snapshot: dict[Path, tuple[bytes, float]] = {} + for path in [*output_directory.rglob("*"), *catalog_path.rglob("*")]: + if path.is_file(): + stat = path.stat() + snapshot[path] = (path.read_bytes(), stat.st_mtime) + return snapshot + + before = snapshot_everything() + report = verify_dataset_snapshot(output_directory) + after = snapshot_everything() + + assert report.ok + assert before == after, "verify rewrote at least one file under the delivery or catalog" + + +def test_moved_root_verifies_from_the_new_root_alone(tmp_path: Path) -> None: + """Kingston's homework (#428): no absolute-path read may survive. + + Export to root A, copy the whole delivery to root B, delete A entirely. + Verify B clean; damage one file under B and verify B again. With A gone, + any read outside the handed root would fail or see nothing, so a clean + and then a damaged report can only come from B's own bytes. + """ + root_a, marker = _export_two_episode_snapshot(tmp_path, "references") + root_b = tmp_path / "moved-delivery" + shutil.copytree(root_a, root_b) + shutil.rmtree(root_a) + + assert not root_a.exists() + clean_report = verify_dataset_snapshot(root_b) + assert clean_report.ok + assert clean_report.findings == [] + + table_path = root_b / marker["integrity"]["tables"]["samples"]["path"] + data = bytearray(table_path.read_bytes()) + data[len(data) // 2] ^= 0xFF # same length, different content + table_path.write_bytes(bytes(data)) + + damaged_report = verify_dataset_snapshot(root_b) + assert not damaged_report.ok + assert [f.reason for f in damaged_report.findings] == ["content-id-mismatch"] + assert damaged_report.findings[0].uri == marker["integrity"]["tables"]["samples"]["path"] + + +def test_damage_is_reported_from_the_verified_root_not_the_export_root( + tmp_path: Path, +) -> None: + """Variant with both roots alive: a damaged copy B must be reported as + damaged, not masked by the still-clean original A. Verification reads + only the root it was handed.""" + root_a, marker = _export_two_episode_snapshot(tmp_path, "references") + root_b = tmp_path / "damaged-delivery" + shutil.copytree(root_a, root_b) + table_path = root_b / marker["integrity"]["tables"]["samples"]["path"] + data = bytearray(table_path.read_bytes()) + data[len(data) // 2] ^= 0xFF + table_path.write_bytes(bytes(data)) + + assert verify_dataset_snapshot(root_a).ok + damaged_report = verify_dataset_snapshot(root_b) + assert not damaged_report.ok + assert [f.reason for f in damaged_report.findings] == ["content-id-mismatch"]