64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""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")) == []
|