"""Receipt artifact ownership for the bounded K1 experiment adapter, no HTTP.""" import importlib.util from pathlib import Path import pytest from k1link.reconstruction.map_version import sha256 def adapter(): pytest.importorskip("scipy") # The packaging command uses the optional correction extra. path = Path(__file__).parents[1] / "experiments/package_recorded_map_version.py" spec = importlib.util.spec_from_file_location("package_recorded_map_version", path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module.RecordedParent @pytest.mark.parametrize("changed", [None, "raw", "metadata", "origin", "clock"]) def test_native_artifact_ids_match_plugin_contract(tmp_path, monkeypatch, changed): raw = tmp_path / "mqtt.raw.k1mqtt" metadata = tmp_path / "mqtt.metadata.jsonl" origin = tmp_path / "mqtt.timeline.origin.json" raw.write_bytes(b"synthetic raw") metadata.write_bytes(b"synthetic per-message receipt index") origin.write_bytes(b"synthetic clock origin") current = tmp_path / "mqtt.timeline.json" current.write_bytes(b"synthetic frozen clock") clock_digest = sha256(current) frozen = tmp_path / f"mqtt.timeline.session-{clock_digest}.json" frozen.write_bytes(current.read_bytes()) current.write_bytes(b"later mutable clock is not the source session snapshot") source = dict( generation="a" * 64, source_digests={ "raw-transport-primary": sha256(raw), "raw-transport-index": sha256(metadata), "raw-transport-clock-origin": sha256(origin), "raw-transport-clock": clock_digest, }, ) parent = adapter()(raw, "source-a", "a" * 64) monkeypatch.setattr(parent, "get", lambda _: source) if changed: {"raw": raw, "metadata": metadata, "origin": origin, "clock": frozen}[changed].write_bytes( b"changed" ) with pytest.raises(ValueError): parent.verify("source-a", "a" * 64) else: assert parent.verify("source-a", "a" * 64) == source with pytest.raises(ValueError): parent.verify("source-a", "b" * 64)