File size: 3,671 Bytes
ed3aeeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | from __future__ import annotations
import hashlib
import json
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPT = REPO_ROOT / "scripts" / "stages" / "publish_artifacts.py"
SCHEMA = REPO_ROOT / "schemas" / "artifact_publish_plan.schema.json"
def digest(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
class PublishArtifactsTests(unittest.TestCase):
def run_cli(self, root: Path, mode: str, *extra: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[
sys.executable, str(SCRIPT), "--mode", mode, "--repo-root", str(root),
"--plan", "plan.json", "--schema", "schema.json",
"--manifest", "canonical/manifest.json", *extra,
],
capture_output=True,
text=True,
check=False,
)
def fixture(self, root: Path) -> bytes:
payload = b"immutable-smoke-artifact\x00\x01\n"
(root / "evidence").mkdir()
(root / "evidence" / "artifact.mlir").write_bytes(payload)
shutil.copyfile(SCHEMA, root / "schema.json")
plan = {
"schema_version": "1.0",
"model_id": "TEST01",
"source_evidence_root": "evidence",
"toolchain": {"name": "test", "version": "1"},
"source_inputs": [
{"role": "source_model", "path": "evidence/artifact.mlir", "sha256": digest(payload)}
],
"assertions": {"weights_modified": False},
"artifacts": [
{
"artifact_id": "test-mlir", "variant": "fp32", "ir_stage": "onnx",
"source": "evidence/artifact.mlir", "destination": "canonical/onnx.mlir",
"sha256": digest(payload), "status": "PASS", "failure_code": None,
}
],
}
(root / "plan.json").write_text(json.dumps(plan), encoding="utf-8")
return payload
def test_publish_reuse_verify_and_overwrite_refusal(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
payload = self.fixture(root)
first = self.run_cli(root, "publish")
self.assertEqual(first.returncode, 0, first.stderr)
self.assertEqual((root / "canonical" / "onnx.mlir").read_bytes(), payload)
first_manifest = (root / "canonical" / "manifest.json").read_bytes()
second = self.run_cli(root, "publish")
self.assertEqual(second.returncode, 0, second.stderr)
self.assertIn('"copied_count": 0', second.stdout)
self.assertEqual((root / "canonical" / "manifest.json").read_bytes(), first_manifest)
verified = self.run_cli(root, "verify", "--report", "canonical/verification.json")
self.assertEqual(verified.returncode, 0, verified.stderr)
report = json.loads((root / "canonical" / "verification.json").read_text(encoding="utf-8"))
self.assertEqual(report["status"], "PASS")
self.assertTrue(report["checksum_match"])
(root / "canonical" / "onnx.mlir").write_bytes(b"tampered")
refused = self.run_cli(root, "publish")
self.assertEqual(refused.returncode, 1)
self.assertIn("refusing to overwrite different destination", refused.stderr)
failed_verify = self.run_cli(root, "verify", "--report", "canonical/verification.json")
self.assertEqual(failed_verify.returncode, 1)
if __name__ == "__main__":
unittest.main()
|