| 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() |
|
|
