257 lines
11 KiB
Python
257 lines
11 KiB
Python
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from copy import deepcopy
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
GUARD = ROOT / "scripts" / "dhf_catalog_guard.py"
|
|
|
|
|
|
class DHFCatalogGuardTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.dev = self.root / "dev"
|
|
self.dev.mkdir()
|
|
subprocess.run(["git", "init", "-q", str(self.dev)], check=True)
|
|
self.items = []
|
|
self.assets = []
|
|
for index, value in enumerate((b"source-one", b"source-two"), 1):
|
|
path = f".templates/DHF 文件模板/阶段{index}/模板{index}.docx"
|
|
absolute = self.dev / path
|
|
absolute.parent.mkdir(parents=True)
|
|
absolute.write_bytes(value)
|
|
digest = hashlib.sha256(value).hexdigest()
|
|
self.items.append({
|
|
"source": path,
|
|
"sourceSha256": digest,
|
|
"proposedDocType": f"type-{index}",
|
|
})
|
|
self.assets.append({
|
|
"templateId": f"dhf.logical-{index}",
|
|
"version": "1.0.0",
|
|
"state": "needs-review",
|
|
"source": {"sha256": digest, "format": "docx"},
|
|
"sourceProvenance": {"repository": "Agent-System-Dev", "path": path},
|
|
"applicability": {"docTypes": [f"type-{index}"], "confirmed": False},
|
|
})
|
|
subprocess.run(["git", "-C", str(self.dev), "add", ".templates"], check=True)
|
|
subprocess.run([
|
|
"git", "-C", str(self.dev), "-c", "user.name=Test",
|
|
"-c", "user.email=test@example.invalid", "commit", "-qm", "sources",
|
|
], check=True)
|
|
commit = subprocess.check_output(
|
|
["git", "-C", str(self.dev), "rev-parse", "HEAD"], text=True,
|
|
).strip()
|
|
for asset in self.assets:
|
|
asset["sourceProvenance"]["commit"] = commit
|
|
self.inventory = self.root / "inventory.json"
|
|
self.catalog = self.root / "catalog.json"
|
|
self.save()
|
|
|
|
def save(self):
|
|
self.inventory.write_text(json.dumps({"templates": self.items}))
|
|
self.catalog.write_text(json.dumps({"schemaVersion": "dhf-assets/v1", "assets": self.assets}))
|
|
|
|
def run_guard(self, *extra):
|
|
return subprocess.run([
|
|
sys.executable, str(GUARD), "--dev-root", str(self.dev),
|
|
"--inventory", str(self.inventory), "--catalog", str(self.catalog),
|
|
"--expected-count", "2", *extra,
|
|
], capture_output=True, text=True)
|
|
|
|
def publish_fixture(self, index=0, *, structure=b'{"blocks":[]}', render=b"rendered-docx"):
|
|
asset = self.assets[index]
|
|
number = index + 1
|
|
source = (self.dev / self.items[index]["source"]).read_bytes()
|
|
names = {"source": f"source{number}.docx", "structure": f"structure{number}.json",
|
|
"render": f"render{number}.docx"}
|
|
for label, raw in [("source", source), ("structure", structure), ("render", render)]:
|
|
(self.root / names[label]).write_bytes(raw)
|
|
asset["state"] = "published"
|
|
asset["applicability"].update({"confirmed": True, "condition": "test scope"})
|
|
asset["source"]["path"] = names["source"]
|
|
asset["normalizedDocx"] = {"path": names["source"], "format": "docx",
|
|
"sha256": hashlib.sha256(source).hexdigest()}
|
|
asset["structure"] = {"path": names["structure"], "format": "json",
|
|
"sha256": hashlib.sha256(structure).hexdigest()}
|
|
asset["render"] = {"path": names["render"], "format": "docx",
|
|
"sha256": hashlib.sha256(render).hexdigest()}
|
|
asset["review"] = {
|
|
"decision": "approved", "reviewer": "reviewer", "reviewedAt": "2026-10-14T00:00:00Z",
|
|
"evidence": "review-record", "sourceSha256": self.items[index]["sourceSha256"],
|
|
"normalizedSha256": asset["normalizedDocx"]["sha256"],
|
|
"structureSha256": asset["structure"]["sha256"],
|
|
"renderSha256": asset["render"]["sha256"],
|
|
}
|
|
return asset
|
|
|
|
def test_accepts_matching_immutable_sources_but_does_not_call_them_published(self):
|
|
result = self.run_guard()
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
report = json.loads(result.stdout)
|
|
self.assertEqual(report["matched"], 2)
|
|
self.assertEqual(report["states"], {"needs-review": 2})
|
|
self.assertEqual(report["published"], 0)
|
|
|
|
def test_rejects_changed_checkout_even_when_catalog_matches_historical_commit(self):
|
|
(self.dev / self.items[0]["source"]).write_bytes(b"changed")
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("checkout hash mismatch", result.stderr)
|
|
|
|
def test_rejects_original_in_symlinked_parent_directory(self):
|
|
source = self.dev / self.items[0]["source"]
|
|
original_parent = source.parent
|
|
moved_parent = original_parent.parent / "alias-target"
|
|
original_parent.rename(moved_parent)
|
|
original_parent.symlink_to(moved_parent, target_is_directory=True)
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("source path symlink", result.stderr)
|
|
|
|
def test_rejects_false_publication(self):
|
|
self.assets[0]["state"] = "published"
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset lacks", result.stderr)
|
|
|
|
def test_reports_revised_candidate_type_without_approving_publication(self):
|
|
self.assets[0]["applicability"]["docTypes"] = ["unrelated"]
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
report = json.loads(result.stdout)
|
|
self.assertEqual(report["published"], 0)
|
|
self.assertEqual(report["candidateTypeDifferences"], [{
|
|
"path": self.items[0]["source"],
|
|
"inventoryCandidate": "type-1",
|
|
"catalogCandidates": ["unrelated"],
|
|
"confirmed": False,
|
|
}])
|
|
|
|
def test_publication_gate_does_not_accept_a_complete_inventory_as_delivery(self):
|
|
result = self.run_guard("--require-published")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("not published", result.stderr)
|
|
|
|
def test_rejects_published_asset_whose_review_does_not_bind_bundle_bytes(self):
|
|
asset = self.publish_fixture()
|
|
asset["review"]["structureSha256"] = "0" * 64
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset review hash mismatch", result.stderr)
|
|
|
|
def test_rejects_bundle_file_changed_after_approval(self):
|
|
asset = self.publish_fixture()
|
|
self.save()
|
|
(self.root / asset["render"]["path"]).write_bytes(b"modified-after-review")
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset file hash mismatch", result.stderr)
|
|
|
|
def test_rejects_published_asset_with_unresolved_fidelity_issue(self):
|
|
asset = self.publish_fixture()
|
|
asset["issues"] = [{"id": "fidelity", "status": "open"}]
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset has unresolved issues", result.stderr)
|
|
|
|
def test_publication_flag_never_claims_authoritative_rdm_validation(self):
|
|
for index in range(2):
|
|
self.publish_fixture(index, structure=b"not-a-valid-structure", render=b"not-a-valid-docx")
|
|
self.save()
|
|
result = self.run_guard("--require-published")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("RDM publication validator required", result.stderr)
|
|
|
|
def test_rejects_swapped_logical_ids_against_fixed_identity_baseline(self):
|
|
baseline = self.root / "identity-baseline.json"
|
|
baseline.write_text(json.dumps({"assets": deepcopy(self.assets)}))
|
|
self.assets[0]["templateId"], self.assets[1]["templateId"] = (
|
|
self.assets[1]["templateId"], self.assets[0]["templateId"])
|
|
self.save()
|
|
result = self.run_guard("--identity-baseline", str(baseline))
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("template identity drift", result.stderr)
|
|
|
|
def test_report_binds_exact_catalog_bytes(self):
|
|
result = self.run_guard()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(json.loads(result.stdout)["catalogSha256"],
|
|
hashlib.sha256(self.catalog.read_bytes()).hexdigest())
|
|
|
|
def test_rejects_published_package_through_symlink_directory(self):
|
|
asset = self.publish_fixture()
|
|
real = self.root / "real"
|
|
real.mkdir()
|
|
for key in ["source", "normalizedDocx", "structure", "render"]:
|
|
p = self.root / asset[key]["path"]
|
|
(real / p.name).write_bytes(p.read_bytes())
|
|
asset[key]["path"] = "alias/" + p.name
|
|
(self.root / "alias").symlink_to(real, target_is_directory=True)
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("symlink", result.stderr)
|
|
|
|
def test_rejects_string_false_and_null_applicability(self):
|
|
asset = self.publish_fixture()
|
|
asset["applicability"]["confirmed"] = "false"
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset lacks approved binding", result.stderr)
|
|
|
|
asset["applicability"]["confirmed"] = True
|
|
asset["applicability"]["condition"] = None
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset lacks approved binding", result.stderr)
|
|
|
|
def test_rejects_null_reviewer_and_non_rfc3339_timestamp(self):
|
|
asset = self.publish_fixture()
|
|
asset["review"]["reviewer"] = None
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset lacks reviewer evidence", result.stderr)
|
|
|
|
asset["review"]["reviewer"] = "reviewer"
|
|
asset["review"]["reviewedAt"] = "2026-10-14 00:00:00Z"
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset has invalid review time", result.stderr)
|
|
|
|
def test_rejects_changed_normalized_file_without_named_converter(self):
|
|
asset = self.publish_fixture()
|
|
normalized = b"changed-normalized-docx"
|
|
(self.root / "normalized.docx").write_bytes(normalized)
|
|
asset["normalizedDocx"] = {
|
|
"path": "normalized.docx", "format": "docx", "sha256": hashlib.sha256(normalized).hexdigest(),
|
|
}
|
|
asset["review"]["normalizedSha256"] = asset["normalizedDocx"]["sha256"]
|
|
asset["converter"] = {
|
|
"inputSha256": self.items[0]["sourceSha256"],
|
|
"outputSha256": asset["normalizedDocx"]["sha256"],
|
|
}
|
|
self.save()
|
|
result = self.run_guard()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("published asset lacks conversion provenance", result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|