97 lines
4.0 KiB
Python
97 lines
4.0 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tools.runtime.report_files import report_log, report_output, resolve_report_input
|
|
from tools.evaluation.report_maintenance import apply_moves, migration_plan, retention_plan
|
|
|
|
|
|
def test_migration_preserves_operational_data_and_report_bytes(tmp_path):
|
|
for name in [
|
|
"documents.json",
|
|
"chunks.json",
|
|
"index_version.json",
|
|
".reindex.lock",
|
|
"zp_index.sqlite",
|
|
]:
|
|
(tmp_path / name).write_text("operational")
|
|
(tmp_path / "old.json").write_bytes(b'{"old":true}\n')
|
|
(tmp_path / "old.pytest.log").write_text("log")
|
|
plan = migration_plan(tmp_path)
|
|
assert {Path(m["source"]).name for m in plan} == {"old.json", "old.pytest.log"}
|
|
journal = apply_moves(plan, data_dir=tmp_path)
|
|
assert len(json.loads(journal.read_text())["moves"]) == 2
|
|
for move in plan:
|
|
assert hashlib.sha256(Path(move["destination"]).read_bytes()).hexdigest() == move["sha256"]
|
|
assert resolve_report_input(Path(move["source"]), data_dir=tmp_path) == Path(
|
|
move["destination"]
|
|
)
|
|
assert (tmp_path / "documents.json").read_text() == "operational"
|
|
assert not migration_plan(tmp_path)
|
|
|
|
|
|
def test_no_overwrites_and_changed_plan_is_rejected_before_any_move(tmp_path):
|
|
for name in ["a.json", "b.json"]:
|
|
(tmp_path / name).write_text("{}")
|
|
plan = migration_plan(tmp_path)
|
|
(tmp_path / "b.json").write_text('{"changed":1}')
|
|
with pytest.raises(RuntimeError):
|
|
apply_moves(plan, data_dir=tmp_path)
|
|
assert (tmp_path / "a.json").exists()
|
|
output = report_output(None, "verification", data_dir=tmp_path)
|
|
output.write_text("{}")
|
|
with pytest.raises(FileExistsError):
|
|
report_output(output, "verification", data_dir=tmp_path)
|
|
for name in ["documents.json", "chunks.json", "index_version.json", "zp_index.sqlite"]:
|
|
with pytest.raises(ValueError):
|
|
report_output(tmp_path / name, "verification", data_dir=tmp_path)
|
|
|
|
|
|
def test_flat_legacy_output_is_redirected_and_unique_default_has_separate_log(tmp_path):
|
|
output = report_output(tmp_path / "old.json", "verification", data_dir=tmp_path)
|
|
assert output == tmp_path / "reports/verification/old.json"
|
|
assert report_log(output, data_dir=tmp_path) == tmp_path / "logs/verification/old.pytest.log"
|
|
assert report_output(None, "verification", data_dir=tmp_path) != report_output(
|
|
None, "verification", data_dir=tmp_path
|
|
)
|
|
assert (
|
|
report_log(tmp_path / "custom/result.json", data_dir=tmp_path).parent
|
|
== tmp_path / "logs/custom"
|
|
)
|
|
|
|
|
|
def test_retention_archives_only_completed_verifications_and_follows_move_chain(tmp_path):
|
|
outputs = []
|
|
for i in range(8):
|
|
output = report_output(tmp_path / f"run{i}.json", "verification", data_dir=tmp_path)
|
|
output.write_text(json.dumps({"passed": bool(i % 2)}))
|
|
os.utime(output, (100 + i, 100 + i))
|
|
report_log(output, data_dir=tmp_path).write_text("log")
|
|
outputs.append(output)
|
|
incomplete = tmp_path / "reports/verification/incomplete.json"
|
|
incomplete.write_text("{}")
|
|
experiment = report_output(None, "rag_verified", data_dir=tmp_path)
|
|
experiment.write_text('{"passed":true}')
|
|
plan = retention_plan(tmp_path, keep=5)
|
|
assert len(plan) == 6
|
|
apply_moves(plan, data_dir=tmp_path)
|
|
assert incomplete.exists() and experiment.exists()
|
|
assert len(list((tmp_path / "reports/verification").glob("*.json"))) == 6
|
|
archived = resolve_report_input(outputs[0], data_dir=tmp_path)
|
|
assert archived.exists() and archived.parent == tmp_path / "reports/archive/verification"
|
|
next_target = tmp_path / "reports/archive/older" / archived.name
|
|
apply_moves(
|
|
[
|
|
{
|
|
"source": str(archived),
|
|
"destination": str(next_target),
|
|
"sha256": hashlib.sha256(archived.read_bytes()).hexdigest(),
|
|
}
|
|
],
|
|
data_dir=tmp_path,
|
|
)
|
|
assert resolve_report_input(outputs[0], data_dir=tmp_path) == next_target
|