import json
|
import tempfile
|
import unittest
|
from pathlib import Path
|
|
from manifest_pack_validator import sha256_file, validate_case, write_outputs
|
|
|
class ManifestPackValidatorTest(unittest.TestCase):
|
def _write_case(self, root: Path, case_id: str, file_content: str | None, manifest_path: str, sha256: str) -> Path:
|
case_dir = root / case_id
|
case_dir.mkdir(parents=True)
|
if file_content is not None:
|
(case_dir / "data.txt").write_text(file_content, encoding="utf-8")
|
(case_dir / "manifest.json").write_text(
|
json.dumps(
|
{
|
"package_id": case_id,
|
"files": [{"path": manifest_path, "sha256": sha256}],
|
},
|
ensure_ascii=False,
|
indent=2,
|
),
|
encoding="utf-8",
|
)
|
return case_dir
|
|
def test_validate_case_passes_valid_hash(self) -> None:
|
with tempfile.TemporaryDirectory() as tmp:
|
root = Path(tmp)
|
data_path = root / "valid_pack" / "data.txt"
|
case_dir = self._write_case(root, "valid_pack", "abc", "data.txt", "")
|
expected_hash = sha256_file(data_path)
|
payload = json.loads((case_dir / "manifest.json").read_text(encoding="utf-8"))
|
payload["files"][0]["sha256"] = expected_hash
|
(case_dir / "manifest.json").write_text(json.dumps(payload), encoding="utf-8")
|
|
result, checks = validate_case(case_dir)
|
|
self.assertEqual(result.status, "PASS")
|
self.assertEqual(result.expected_status, "PASS")
|
self.assertEqual(len(checks), 1)
|
self.assertEqual(checks[0].status, "PASS")
|
|
def test_validate_case_fails_missing_file(self) -> None:
|
with tempfile.TemporaryDirectory() as tmp:
|
case_dir = self._write_case(Path(tmp), "invalid_pack_missing_file", None, "missing.txt", "0" * 64)
|
result, checks = validate_case(case_dir)
|
|
self.assertEqual(result.status, "FAIL")
|
self.assertIn("FILE_MISSING", result.error_codes)
|
self.assertEqual(checks[0].error_code, "FILE_MISSING")
|
|
def test_validate_case_fails_hash_mismatch(self) -> None:
|
with tempfile.TemporaryDirectory() as tmp:
|
case_dir = self._write_case(Path(tmp), "invalid_pack_hash", "abc", "data.txt", "0" * 64)
|
result, checks = validate_case(case_dir)
|
|
self.assertEqual(result.status, "FAIL")
|
self.assertIn("SHA256_MISMATCH", result.error_codes)
|
self.assertEqual(checks[0].error_code, "SHA256_MISMATCH")
|
|
def test_validate_case_blocks_path_traversal(self) -> None:
|
with tempfile.TemporaryDirectory() as tmp:
|
case_dir = self._write_case(Path(tmp), "invalid_pack_escape", "abc", "../escape.txt", "0" * 64)
|
result, checks = validate_case(case_dir)
|
|
self.assertEqual(result.status, "FAIL")
|
self.assertIn("PATH_OUTSIDE_CASE", result.error_codes)
|
self.assertEqual(checks[0].error_code, "PATH_OUTSIDE_CASE")
|
|
def test_write_outputs_expected_matrix(self) -> None:
|
with tempfile.TemporaryDirectory() as tmp:
|
root = Path(tmp) / "raw"
|
out = Path(tmp) / "result"
|
root.mkdir()
|
valid_case = self._write_case(root, "valid_pack", "abc", "data.txt", "")
|
payload = json.loads((valid_case / "manifest.json").read_text(encoding="utf-8"))
|
payload["files"][0]["sha256"] = sha256_file(valid_case / "data.txt")
|
(valid_case / "manifest.json").write_text(json.dumps(payload), encoding="utf-8")
|
self._write_case(root, "invalid_pack_missing_file", None, "missing.txt", "0" * 64)
|
self._write_case(root, "invalid_pack_hash", "abc", "data.txt", "0" * 64)
|
|
summary = write_outputs(root, out)
|
|
self.assertTrue(summary["quality_pass"])
|
self.assertEqual(summary["total_cases"], 3)
|
self.assertEqual(summary["pass_count"], 1)
|
self.assertEqual(summary["fail_count"], 2)
|
self.assertEqual(summary["experiment_id"], "raw")
|
self.assertEqual(summary["run_id"], "result")
|
self.assertTrue((out / "readout.md").exists())
|
|
|
if __name__ == "__main__":
|
unittest.main()
|