fix: make pools.json freeze atomic + add split manifest
This commit is contained in:
@@ -0,0 +1,63 @@
|
||||
"""pools.json 原子冻结与切分 manifest 的单元测试。
|
||||
|
||||
覆盖:
|
||||
- _atomic_write_json 原子写:内容正确、无残留 tmp、可覆盖已有文件。
|
||||
- write_manifest 溯源写:含 baseline_run_id/diag_fingerprint/seed/pools_sha256
|
||||
等键,且 pools_sha256 与给定 pools.json 内容一致。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
|
||||
from app.harness.pools import _atomic_write_json
|
||||
from app.harness.split_manifest import write_manifest
|
||||
|
||||
|
||||
def test_atomic_write_replaces_and_no_tmp_left(tmp_path):
|
||||
p = tmp_path / "pools.json"
|
||||
_atomic_write_json(p, {"a": 1})
|
||||
assert json.loads(p.read_text())["a"] == 1
|
||||
assert list(tmp_path.glob("*.tmp")) == [] # 无残留 tmp
|
||||
|
||||
|
||||
def test_atomic_write_overwrites_existing(tmp_path):
|
||||
p = tmp_path / "pools.json"
|
||||
_atomic_write_json(p, {"a": 1})
|
||||
_atomic_write_json(p, {"a": 2})
|
||||
assert json.loads(p.read_text())["a"] == 2
|
||||
assert list(tmp_path.glob("*.tmp")) == []
|
||||
|
||||
|
||||
def test_write_manifest_contains_keys_and_matching_sha256(tmp_path):
|
||||
pools_json_text = json.dumps({"split_mode": "global", "test": []}, ensure_ascii=False)
|
||||
manifest_path = tmp_path / "manifest.json"
|
||||
|
||||
result = write_manifest(
|
||||
manifest_path,
|
||||
baseline_run_id="run-123",
|
||||
diag_fingerprint="fp-abc",
|
||||
seed=42,
|
||||
config={"train_ratio": 0.5},
|
||||
pools_json_text=pools_json_text,
|
||||
coverage_report={"visual": {"train": 3, "val": 1}},
|
||||
generated_at="2026-07-15T00:00:00Z",
|
||||
)
|
||||
|
||||
written = json.loads(manifest_path.read_text())
|
||||
for key in (
|
||||
"baseline_run_id",
|
||||
"diag_fingerprint",
|
||||
"seed",
|
||||
"config",
|
||||
"pools_sha256",
|
||||
"coverage_report",
|
||||
"generated_at",
|
||||
):
|
||||
assert key in written
|
||||
|
||||
expected_sha = hashlib.sha256(pools_json_text.encode("utf-8")).hexdigest()
|
||||
assert written["pools_sha256"] == expected_sha
|
||||
assert result == written # 返回值与落盘内容一致
|
||||
assert list(tmp_path.glob("*.tmp")) == []
|
||||
Reference in New Issue
Block a user