Release portable deterministic core 1.1.0: frozen formats, offline recovery and regression checks
ecad52f verified | """Regression checks for the portable source release; standard library only.""" | |
| import copy | |
| import hashlib | |
| import json | |
| import os | |
| import random | |
| import socket | |
| import subprocess | |
| import sys | |
| import tempfile | |
| import unittest | |
| from datetime import datetime, timedelta, timezone | |
| from pathlib import Path | |
| from unittest.mock import patch | |
| from demo import evaluate, evaluate_json, example | |
| from matrix_core import ( | |
| CHECKSUM_FORMAT, ENGINE_VERSION, MAX_PAYLOAD_BYTES, WHITESPACE_CODEPOINTS, | |
| DeterministicMatrixCore, checksum_for_date, normalize_payload, parse_instant, verify_result, | |
| ) | |
| from safe_json import strict_loads | |
| from verify_release import verify_release | |
| ROOT = Path(__file__).resolve().parent.parent | |
| FIXED = datetime(2026, 11, 3, 12, tzinfo=timezone.utc) | |
| class LongevityTests(unittest.TestCase): | |
| def setUp(self): | |
| self.core = DeterministicMatrixCore() | |
| def test_published_vectors_and_utc_dates(self): | |
| vectors = json.loads((ROOT / "test_vectors.json").read_text(encoding="utf-8")) | |
| for vector in vectors["vectors"]: | |
| with self.subTest(id=vector["id"]): | |
| result = self.core.enforce_zero_tolerance_matrix(vector["payload"], parse_instant(vector["at"])) | |
| self.assertEqual(result["input_payload"], vector["normalized"]) | |
| self.assertEqual(result["seal_date"], vector["utc_date"]) | |
| self.assertEqual(result["matrix_verification_seal"], vector["sha256"]) | |
| def test_frozen_whitespace_table(self): | |
| self.assertEqual(len(WHITESPACE_CODEPOINTS), 29) | |
| for codepoint in WHITESPACE_CODEPOINTS: | |
| self.assertEqual(normalize_payload(chr(codepoint) + "a" + chr(codepoint) * 3 + "b" + chr(codepoint)), "a b") | |
| def test_non_whitespace_characters_are_preserved(self): | |
| for codepoint in (0, 0x180E, 0x200B, 0x2060, 0xFEFF, 0x1F600): | |
| payload = chr(codepoint) + "text" + chr(codepoint) | |
| self.assertEqual(normalize_payload(payload), payload) | |
| def test_no_implicit_unicode_normalization(self): | |
| self.assertNotEqual(checksum_for_date("é", "2126-11-03"), checksum_for_date("e\u0301", "2126-11-03")) | |
| def test_normalization_idempotence_for_seeded_unicode_inputs(self): | |
| rng = random.Random(36500) | |
| alphabet = ["a", "Č", "😀", "\u200b", "é"] + [chr(c) for c in WHITESPACE_CODEPOINTS] | |
| for _ in range(100): | |
| payload = "".join(rng.choice(alphabet) for _ in range(100)) | |
| clean = normalize_payload(payload) | |
| self.assertEqual(normalize_payload(clean), clean) | |
| if clean: | |
| self.assertEqual(checksum_for_date(payload, "2126-11-03"), checksum_for_date(clean, "2126-11-03")) | |
| def test_lone_surrogates_have_no_checksum(self): | |
| for payload in ("\ud800", "x\udfff", "\ud800\udc00"): | |
| result = self.core.enforce_zero_tolerance_matrix(payload, FIXED) | |
| self.assertFalse(result["valid"]) | |
| self.assertEqual(result["error"], "INVALID_UNICODE") | |
| self.assertNotIn("matrix_verification_seal", result) | |
| def test_raw_utf8_size_limit_before_normalization(self): | |
| for payload in (" " * (MAX_PAYLOAD_BYTES + 1), "😀" * (MAX_PAYLOAD_BYTES // 4 + 1)): | |
| self.assertEqual(self.core.enforce_zero_tolerance_matrix(payload, FIXED)["error"], "PAYLOAD_TOO_LARGE") | |
| self.assertTrue(self.core.enforce_zero_tolerance_matrix("a" * MAX_PAYLOAD_BYTES, FIXED)["valid"]) | |
| def test_future_gregorian_calendar_and_leap_rules(self): | |
| for at in ("2126-11-03T12:00:00Z", "2400-02-29T12:00:00Z", "0001-01-01T00:00:00Z", "9999-12-31T23:59:59Z"): | |
| self.assertTrue(self.core.enforce_zero_tolerance_matrix("x", parse_instant(at))["valid"]) | |
| for at in ("2100-02-29T12:00:00Z", "2200-02-29T12:00:00Z", "0000-01-01T00:00:00Z", "2026-02-30T12:00:00Z"): | |
| with self.assertRaises(ValueError): | |
| parse_instant(at) | |
| def test_timestamp_grammar_is_explicit_and_ascii(self): | |
| for at in ("2026-11-03", "2026-11-03 12:00:00Z", "20261103T120000Z", "2026-11-03T12:00:00-00:00", "2026-11-03T12:00:00+00:60", "2026-11-03T12:00:00+24:00", "2026-11-03T12:00:60Z", "2026-11-03T12:00:00.1234567Z"): | |
| with self.subTest(at=at), self.assertRaises(ValueError): | |
| parse_instant(at) | |
| def test_utc_overflow_is_controlled(self): | |
| for at in ("0001-01-01T00:00:00+01:00", "9999-12-31T23:59:59-01:00"): | |
| with self.assertRaises(ValueError): | |
| parse_instant(at) | |
| def test_valid_export_roundtrips(self): | |
| result = self.core.enforce_zero_tolerance_matrix("Č 😀", FIXED) | |
| self.assertTrue(verify_result(strict_loads(json.dumps(result)))["verified"]) | |
| self.assertFalse(verify_result(result)["metadata_authenticated"]) | |
| def test_legacy_351_export_remains_readable(self): | |
| legacy = self.core.enforce_zero_tolerance_matrix("Evidence before confidence.", FIXED) | |
| for key in ("checksum_format", "normalization_profile", "hash_algorithm"): | |
| legacy.pop(key) | |
| legacy["engine_version"] = "3.5.1-OMEGA" | |
| self.assertTrue(verify_result(legacy)["verified"]) | |
| legacy["engine_version"] = "unknown" | |
| self.assertFalse(verify_result(legacy)["verified"]) | |
| def test_export_mutations_fail_closed(self): | |
| result = self.core.enforce_zero_tolerance_matrix("Evidence before confidence.", FIXED) | |
| mutations = {"valid": 1, "input_payload": "other", "seal_date": "2026-11-04", "timestamp": "2026-11-04T12:00:00Z", "matrix_verification_seal": "0" * 64, "checksum_format": "bodul-matrix-v999", "normalization_profile": "unicode-latest", "hash_algorithm": "md5", "engine_version": ""} | |
| for key, value in mutations.items(): | |
| item = dict(result) | |
| item[key] = value | |
| with self.subTest(key=key): | |
| self.assertFalse(verify_result(item)["verified"]) | |
| item = dict(result, extra="ignored?") | |
| self.assertFalse(verify_result(item)["verified"]) | |
| item = dict(result) | |
| del item["hash_algorithm"] | |
| self.assertFalse(verify_result(item)["verified"]) | |
| def test_export_requires_canonical_payload(self): | |
| result = self.core.enforce_zero_tolerance_matrix("x", FIXED) | |
| result["input_payload"] = " x " | |
| self.assertEqual(verify_result(result)["error"], "NON_CANONICAL_PAYLOAD") | |
| def test_metadata_is_not_authenticated(self): | |
| result = self.core.enforce_zero_tolerance_matrix("x", FIXED) | |
| result["engine_version"] = "different-but-unsealed-metadata" | |
| result["timestamp"] = "2026-11-03T23:59:59Z" | |
| self.assertTrue(verify_result(result)["verified"]) | |
| def test_duplicate_keys_and_nonfinite_numbers_are_rejected(self): | |
| for text in ('{"a":1,"a":2}', '{"a":{"x":1,"x":2}}', '{"x":NaN}', '{"x":Infinity}', '{"x":1e9999}'): | |
| with self.subTest(text=text), self.assertRaises(ValueError): | |
| strict_loads(text) | |
| def test_json_limits_and_surrogates(self): | |
| for text in ('[' * 33 + '0' + ']' * 33, '"\\ud800"', '1' * 129): | |
| with self.assertRaises(ValueError): | |
| strict_loads(text) | |
| self.assertEqual(strict_loads('"\\ud83d\\ude00"'), "😀") | |
| self.assertEqual(strict_loads('{"text":"[[[{{{"}'), {"text": "[[[{{{"}) | |
| def test_malformed_public_inputs_never_approve(self): | |
| for payload in (None, True, [], {}, {"claim": {}}, {"evidence": "wrong"}): | |
| self.assertEqual(evaluate(payload)["verdict"], "BLOCK") | |
| self.assertEqual(evaluate_json('{"claim":1,"claim":2}')["reason"], "invalid_json") | |
| def test_public_boolean_types_are_not_integers(self): | |
| payload = example(at=FIXED) | |
| payload["evidence"][0]["supports"] = 1 | |
| self.assertEqual(evaluate(payload)["verdict"], "BLOCK") | |
| def test_public_timestamps_are_frozen_and_utc_normalized(self): | |
| payload = example(at=FIXED) | |
| other = copy.deepcopy(payload) | |
| other["evaluation_time"] = "2026-11-03T14:00:00+02:00" | |
| other["evidence"][0]["observed_at"] = "2026-11-03T14:00:00+02:00" | |
| self.assertEqual(evaluate(payload), evaluate(other)) | |
| payload["evaluation_time"] = "2026-11-03 12:00:00Z" | |
| self.assertEqual(evaluate(payload)["verdict"], "BLOCK") | |
| def test_public_approval_flags_cannot_authorize_high_risk_claims(self): | |
| payload = example(at=FIXED) | |
| payload["claim"]["risk"] = "high" | |
| payload["human_approved"] = True | |
| self.assertEqual(evaluate(payload)["verdict"], "ESCALATE") | |
| self.assertEqual(evaluate(payload)["reason"], "human_approval_required") | |
| def test_missing_stale_future_and_conflicting_evidence_block(self): | |
| payload = example(at=FIXED) | |
| self.assertEqual(evaluate(payload)["verdict"], "APPROVE") | |
| missing = example("unsupported", at=FIXED) | |
| self.assertEqual(evaluate(missing)["classification"], "UNKNOWN") | |
| for seconds in (-301, 1): | |
| item = copy.deepcopy(payload) | |
| item["evidence"][0]["observed_at"] = (FIXED + timedelta(seconds=seconds)).isoformat() | |
| self.assertEqual(evaluate(item)["verdict"], "BLOCK") | |
| contradictory = dict(payload["evidence"][0], source_id="second", supports=False) | |
| payload["evidence"].append(contradictory) | |
| self.assertEqual(evaluate(payload)["classification"], "CONFLICT") | |
| self.assertEqual(evaluate(payload)["verdict"], "BLOCK") | |
| def test_source_duplicates_do_not_satisfy_independence_quorum(self): | |
| payload = example(at=FIXED) | |
| payload["claim"]["risk"] = "medium" | |
| payload["evidence"].append(dict(payload["evidence"][0], evidence_id="copy")) | |
| self.assertEqual(evaluate(payload)["verdict"], "ESCALATE") | |
| def test_offline_execution_does_not_use_network(self): | |
| with patch.object(socket, "socket", side_effect=AssertionError("Network use forbidden")): | |
| self.assertTrue(self.core.enforce_zero_tolerance_matrix("x", FIXED)["valid"]) | |
| self.assertEqual(evaluate(example(at=FIXED))["verdict"], "APPROVE") | |
| def test_explicit_future_evaluation_repeats(self): | |
| payload = example(at=datetime(2126, 11, 3, tzinfo=timezone.utc)) | |
| expected = evaluate(payload) | |
| self.assertEqual(expected["verdict"], "APPROVE") | |
| for _ in range(25): | |
| self.assertEqual(evaluate(payload), expected) | |
| def test_offline_cli_matrix_and_saved_result(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| folder = Path(directory) | |
| text = folder / "input.txt" | |
| result_file = folder / "result.json" | |
| text.write_text("Evidence before confidence.", encoding="utf-8") | |
| run = subprocess.run([sys.executable, "-B", str(ROOT / "offline_cli.py"), "matrix", str(text), "--at", FIXED.isoformat()], capture_output=True, text=True, encoding="utf-8") | |
| self.assertEqual(run.returncode, 0, run.stderr) | |
| result_file.write_text(run.stdout, encoding="utf-8") | |
| replay = subprocess.run([sys.executable, "-B", str(ROOT / "offline_cli.py"), "verify-matrix", str(result_file)], capture_output=True, text=True, encoding="utf-8") | |
| self.assertEqual(replay.returncode, 0, replay.stderr) | |
| self.assertTrue(json.loads(replay.stdout)["verified"]) | |
| def test_offline_cli_block_has_nonzero_exit(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| path = Path(directory) / "claim.json" | |
| path.write_text(json.dumps(example("unsupported", at=FIXED)), encoding="utf-8") | |
| run = subprocess.run([sys.executable, "-B", str(ROOT / "offline_cli.py"), "claims", str(path)], capture_output=True, text=True, encoding="utf-8") | |
| self.assertEqual(run.returncode, 3) | |
| self.assertEqual(json.loads(run.stdout)["verdict"], "BLOCK") | |
| class ArchiveIntegrityTests(unittest.TestCase): | |
| def make_archive(self, folder): | |
| data = b"portable source\n" | |
| (folder / "source.txt").write_bytes(data) | |
| manifest = {"manifest_format": "bodul-source-archive-v1", "files": {"source.txt": {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest(), "sha512": hashlib.sha512(data).hexdigest()}}} | |
| (folder / "RELEASE_MANIFEST.json").write_text(json.dumps(manifest), encoding="utf-8") | |
| return manifest | |
| def test_clean_archive_and_tampering(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| root = Path(directory) | |
| self.make_archive(root) | |
| self.assertTrue(verify_release(root)["verified"]) | |
| (root / "source.txt").write_bytes(b"modified source\n") | |
| self.assertFalse(verify_release(root)["verified"]) | |
| def test_missing_and_unlisted_files(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| root = Path(directory) | |
| self.make_archive(root) | |
| (root / "extra.txt").write_text("x") | |
| self.assertFalse(verify_release(root)["verified"]) | |
| (root / "extra.txt").unlink() | |
| (root / "source.txt").unlink() | |
| self.assertFalse(verify_release(root)["verified"]) | |
| def test_unknown_format_and_unsafe_paths(self): | |
| with tempfile.TemporaryDirectory() as directory: | |
| root = Path(directory) | |
| manifest = self.make_archive(root) | |
| for path in ("../escape.txt", "/absolute.txt", "drive:path", "foo\\bar", "."): | |
| bad = copy.deepcopy(manifest) | |
| bad["files"][path] = bad["files"].pop("source.txt") | |
| (root / "RELEASE_MANIFEST.json").write_text(json.dumps(bad), encoding="utf-8") | |
| self.assertFalse(verify_release(root)["verified"]) | |
| manifest["manifest_format"] = "future-unknown-format" | |
| (root / "RELEASE_MANIFEST.json").write_text(json.dumps(manifest), encoding="utf-8") | |
| self.assertFalse(verify_release(root)["verified"]) | |
| if __name__ == "__main__": | |
| unittest.main() | |