55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
"""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)
|