36 lines
1.2 KiB
Python
36 lines
1.2 KiB
Python
from linkerhand_calibration.storage import (
|
|
append_jsonl,
|
|
atomic_write_json,
|
|
completed_scan_keys,
|
|
load_json,
|
|
load_jsonl,
|
|
)
|
|
|
|
|
|
def test_jsonl_checkpoint_and_resume_keys(tmp_path) -> None:
|
|
raw_path = tmp_path / "raw_samples.jsonl"
|
|
record = {
|
|
"kind": "sample",
|
|
"phase": "root",
|
|
"cycle": 0,
|
|
"direction": "decreasing",
|
|
"command_u8": 255,
|
|
}
|
|
append_jsonl(raw_path, record)
|
|
append_jsonl(raw_path, {"kind": "validation", "command_u8": 10})
|
|
loaded = load_jsonl(raw_path)
|
|
assert loaded[0] == record
|
|
assert completed_scan_keys(loaded) == {("root", 0, "decreasing", 255)}
|
|
|
|
checkpoint_path = tmp_path / "checkpoint.json"
|
|
atomic_write_json(checkpoint_path, {"state": "PAUSED", "records": 1})
|
|
assert load_json(checkpoint_path) == {"state": "PAUSED", "records": 1}
|
|
|
|
|
|
def test_resume_ignores_only_a_truncated_final_jsonl_record(tmp_path) -> None:
|
|
raw_path = tmp_path / "raw_samples.jsonl"
|
|
append_jsonl(raw_path, {"kind": "sample", "phase": "root"})
|
|
with raw_path.open("a", encoding="utf-8") as stream:
|
|
stream.write('{"kind":"sample"\n\n')
|
|
assert load_jsonl(raw_path) == [{"kind": "sample", "phase": "root"}]
|