mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-08-12 10:33:33 +00:00
366 lines
16 KiB
Python
366 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Deterministic tests for the Pulse Intelligence aggregate release gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import hashlib
|
|
import re
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import pulse_intelligence_gate as gate
|
|
|
|
|
|
SHA = "1" * 40
|
|
NEXT_SHA = "2" * 40
|
|
ANCESTOR_SHA = "3" * 40
|
|
MOBILE_SHA = "4" * 40
|
|
|
|
|
|
def write_json(path: Path, payload: object) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def create_rg12_artifact(root: Path, audited_sha: str) -> dict[str, str]:
|
|
action_ids = ["approve-action", "deny-action"]
|
|
environment = {
|
|
"core_sha": audited_sha,
|
|
"mobile_sha": MOBILE_SHA,
|
|
"action_ids": action_ids,
|
|
"revoked_barrier_action_id": "revoked-barrier",
|
|
"temporary_token_record_id": "temporary-token",
|
|
"credential_material_in_bundle": False,
|
|
}
|
|
write_json(root / "environment.json", environment)
|
|
write_json(
|
|
root / "correlation.json",
|
|
{
|
|
"core_sha": audited_sha,
|
|
"mobile_sha": MOBILE_SHA,
|
|
"action_ids": action_ids,
|
|
"token_record_id": "temporary-token",
|
|
"target_type": "test",
|
|
"executor": "none",
|
|
"physical_proof_sequence": list(gate.RG12_PROOF_NAMES),
|
|
},
|
|
)
|
|
|
|
def decision(action_id: str, state: str, outcome: str) -> dict[str, object]:
|
|
return {
|
|
"actionId": action_id,
|
|
"state": state,
|
|
"execute_endpoint_called": False,
|
|
"approval": {"outcome": outcome},
|
|
"audit": {"request": {"params": {"target_type": "test", "executor": "none"}}},
|
|
}
|
|
|
|
write_json(
|
|
root / "canonical-action-api.json",
|
|
{
|
|
"approve_http_status": 200,
|
|
"deny_http_status": 200,
|
|
"matching_pending_after_count": 0,
|
|
"approve": decision(action_ids[0], "approved", "approved"),
|
|
"deny": decision(action_ids[1], "rejected", "rejected"),
|
|
},
|
|
)
|
|
write_json(
|
|
root / "action-store-observations.json",
|
|
{
|
|
"records": [
|
|
{"id": action_ids[0], "state": "approved", "target_type": "test", "executor": "none"},
|
|
{"id": action_ids[1], "state": "rejected", "target_type": "test", "executor": "none"},
|
|
{"id": "revoked-barrier", "state": "pending_approval", "target_type": "test", "executor": "none"},
|
|
],
|
|
"dispatch_attempts": 0,
|
|
"dispatch_outbox": 0,
|
|
"dispatch_receipts": 0,
|
|
},
|
|
)
|
|
barrier = {"id": "revoked-barrier", "state": "pending_approval", "decision_revision": 0}
|
|
write_json(
|
|
root / "revocation-negative-barrier.json",
|
|
{
|
|
"token_delete_http_status": 204,
|
|
"post_revoke_pending_http_status": 401,
|
|
"post_revoke_decision_http_status": 401,
|
|
"admin_bypass_enabled": False,
|
|
"before": [barrier],
|
|
"after": [barrier],
|
|
"triple_zero": {field: 0 for field in gate.TRIPLE_ZERO_FIELDS},
|
|
"transport_residue": {"dispatch_outbox": 0, "dispatch_receipts": 0},
|
|
},
|
|
)
|
|
write_json(
|
|
root / "cleanup.json",
|
|
{
|
|
"relay_after": [
|
|
{"surface": "instances", "temp_count": 0},
|
|
{"surface": "device_tokens", "temp_count": 0},
|
|
{"surface": "connection_log", "temp_count": 0},
|
|
{"surface": "original_preserved", "temp_count": 1},
|
|
],
|
|
"local_action_store_after": [[{"temp_action_audits": 0}], [{"temp_events": 0}]],
|
|
"temporary_api_token_present_after": False,
|
|
"local_environment_restored_from_pre_run_backup": True,
|
|
"temporary_license_activation_files_present_after": False,
|
|
"ipad_local_pairing_removed": True,
|
|
},
|
|
)
|
|
|
|
digest_records: list[dict[str, object]] = []
|
|
for proof_name in gate.RG12_PROOF_NAMES:
|
|
proof_dir = root / "mobile-proofs" / proof_name
|
|
write_json(
|
|
proof_dir / "test-summary.json",
|
|
{"result": "Passed", "totalTestCount": 1, "failedTests": []},
|
|
)
|
|
(proof_dir / "attachments").mkdir(parents=True)
|
|
(proof_dir / "attachments" / "proof.png").write_bytes(b"png")
|
|
source = root.parent / f"{proof_name}.xcresult"
|
|
source.mkdir()
|
|
(source / "member.bin").write_bytes(proof_name.encode())
|
|
member_digest = hashlib.sha256(proof_name.encode()).hexdigest()
|
|
line = f"{member_digest} member.bin\n"
|
|
digest_records.append(
|
|
{
|
|
"proof": proof_name,
|
|
"source": str(source),
|
|
"file_count": 1,
|
|
"aggregate_sha256": hashlib.sha256(line.encode()).hexdigest(),
|
|
"members": [{"path": "member.bin", "sha256": member_digest}],
|
|
}
|
|
)
|
|
write_json(root / "xcresult-digests.json", digest_records)
|
|
files = sorted(candidate for candidate in root.rglob("*") if candidate.is_file())
|
|
lines = [f"{hashlib.sha256(candidate.read_bytes()).hexdigest()} {candidate.relative_to(root)}" for candidate in files]
|
|
manifest = root / "SHA256SUMS"
|
|
manifest.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return {
|
|
"contract": gate.RG12_ARTIFACT_CONTRACT,
|
|
"path": str(root),
|
|
"sha256sums_sha256": hashlib.sha256(manifest.read_bytes()).hexdigest(),
|
|
"mobile_git_sha": MOBILE_SHA,
|
|
}
|
|
|
|
|
|
def valid_matrix() -> dict[str, object]:
|
|
gates: list[dict[str, object]] = []
|
|
for gate_id in gate.GATE_IDS:
|
|
gates.append(
|
|
{
|
|
"id": gate_id,
|
|
"title": f"Test gate {gate_id}",
|
|
"owner": gate.APPROVED_OWNERS[gate_id],
|
|
"required_evidence_tier": "integration",
|
|
"requires_triple_zero": True,
|
|
"allows_ancestor_provenance": gate_id == "RG-08",
|
|
"non_mutating_proof_commands": [
|
|
{"id": f"{gate_id.lower()}-proof", "argv": ["go", "test", "./internal/example"]}
|
|
],
|
|
"mutation_gated_commands": [],
|
|
}
|
|
)
|
|
return {
|
|
"schema_version": 3,
|
|
"matrix_id": "pulse-intelligence-rg-01-rg-12",
|
|
"program": "pulse-intelligence",
|
|
"result_semantics": {
|
|
"PASS": "External evidence satisfies the gate at the audited Git SHA.",
|
|
"FAIL": "External evidence bound to the audited Git SHA records a failed proof or an invalid binding.",
|
|
"NOT_RUN": "No qualifying external evidence exists for the gate at the audited Git SHA.",
|
|
},
|
|
"evidence_tiers": list(gate.EVIDENCE_TIERS),
|
|
"triple_zero_semantics": dict(gate.TRIPLE_ZERO_SEMANTICS),
|
|
"gates": gates,
|
|
}
|
|
|
|
|
|
def evidence_for(matrix: dict[str, object], audited_sha: str) -> dict[str, object]:
|
|
records: list[dict[str, object]] = []
|
|
kind_for_tier = {
|
|
"unit": "source",
|
|
"integration": "automated-test",
|
|
"browser": "browser-artifact",
|
|
"real-lab": "real-lab-artifact",
|
|
"physical-device": "physical-device-artifact",
|
|
"live-relay": "live-relay-artifact",
|
|
}
|
|
for item in matrix["gates"]:
|
|
record: dict[str, object] = {
|
|
"id": f"{item['id'].lower()}-evidence",
|
|
"gate_id": item["id"],
|
|
"result": "PASS",
|
|
"tier": item["required_evidence_tier"],
|
|
"kind": kind_for_tier[item["required_evidence_tier"]],
|
|
"git_sha": audited_sha,
|
|
"command_id": item["non_mutating_proof_commands"][0]["id"],
|
|
}
|
|
if item["requires_triple_zero"]:
|
|
record["triple_zero"] = {
|
|
"persistence_writes": 0,
|
|
"dispatch_attempts": 0,
|
|
"external_mutations": 0,
|
|
}
|
|
records.append(record)
|
|
return {
|
|
"schema_version": 2,
|
|
"matrix_id": matrix["matrix_id"],
|
|
"audited_git_sha": audited_sha,
|
|
"records": records,
|
|
}
|
|
|
|
|
|
class PulseIntelligenceGateTest(unittest.TestCase):
|
|
def test_malformed_matrix_is_rejected(self) -> None:
|
|
payload = valid_matrix()
|
|
del payload["result_semantics"]
|
|
with self.assertRaisesRegex(gate.MatrixError, "missing fields: result_semantics"):
|
|
gate.validate_matrix(payload)
|
|
|
|
def test_duplicate_gate_is_rejected(self) -> None:
|
|
payload = valid_matrix()
|
|
payload["gates"][1]["id"] = "RG-01"
|
|
payload["gates"][1]["owner"] = "task-01"
|
|
with self.assertRaisesRegex(gate.MatrixError, "duplicate gate ids"):
|
|
gate.validate_matrix(payload)
|
|
|
|
def test_wrong_owner_is_rejected(self) -> None:
|
|
payload = valid_matrix()
|
|
payload["gates"][4]["owner"] = "task-12"
|
|
with self.assertRaisesRegex(gate.MatrixError, "RG-05 owner must be 'task-05'"):
|
|
gate.validate_matrix(payload)
|
|
|
|
def test_over_tier_evidence_is_rejected(self) -> None:
|
|
matrix = gate.validate_matrix(valid_matrix())
|
|
payload = evidence_for(matrix, SHA)
|
|
payload["records"][0]["kind"] = "mock"
|
|
payload["records"][0]["tier"] = "real-lab"
|
|
with self.assertRaisesRegex(gate.MatrixError, "overstates 'mock' as 'real-lab'"):
|
|
gate.validate_evidence(payload, matrix)
|
|
|
|
def test_external_exact_sha_evidence_can_certify_go(self) -> None:
|
|
matrix = gate.validate_matrix(valid_matrix())
|
|
evidence = gate.validate_evidence(evidence_for(matrix, SHA), matrix)
|
|
evaluation = gate.evaluate_matrix(matrix, SHA, evidence)
|
|
self.assertEqual(evaluation.verdict, "GO")
|
|
self.assertEqual(set(evaluation.gate_results.values()), {"PASS"})
|
|
|
|
def test_committing_matrix_does_not_inherently_stale_it(self) -> None:
|
|
matrix = gate.validate_matrix(valid_matrix())
|
|
self.assertNotIn("evaluation_git_sha", matrix)
|
|
for audited_sha in (SHA, NEXT_SHA):
|
|
with self.subTest(audited_sha=audited_sha):
|
|
evidence = gate.validate_evidence(evidence_for(matrix, audited_sha), matrix)
|
|
self.assertEqual(gate.evaluate_matrix(matrix, audited_sha, evidence).verdict, "GO")
|
|
|
|
def test_ancestor_evidence_is_visible_but_non_qualifying(self) -> None:
|
|
matrix = gate.validate_matrix(valid_matrix())
|
|
payload = evidence_for(matrix, SHA)
|
|
payload["records"][7]["git_sha"] = ANCESTOR_SHA
|
|
evidence = gate.validate_evidence(payload, matrix)
|
|
evaluation = gate.evaluate_matrix(
|
|
matrix,
|
|
SHA,
|
|
evidence,
|
|
is_ancestor=lambda older, newer: (older, newer) == (ANCESTOR_SHA, SHA),
|
|
)
|
|
self.assertEqual(evaluation.gate_results["RG-08"], "NOT_RUN")
|
|
self.assertEqual(evaluation.verdict, "NO-GO")
|
|
self.assertTrue(any("RG-08 provenance" in detail and "does not certify" in detail for detail in evaluation.details))
|
|
|
|
def test_wrong_sha_external_evidence_fails(self) -> None:
|
|
matrix = gate.validate_matrix(valid_matrix())
|
|
payload = evidence_for(matrix, SHA)
|
|
payload["records"][0]["git_sha"] = NEXT_SHA
|
|
evidence = gate.validate_evidence(payload, matrix)
|
|
evaluation = gate.evaluate_matrix(matrix, SHA, evidence, is_ancestor=lambda _older, _newer: False)
|
|
self.assertEqual(evaluation.gate_results["RG-01"], "FAIL")
|
|
self.assertTrue(any("bound to wrong SHA" in detail for detail in evaluation.details))
|
|
|
|
def test_wrong_external_document_binding_fails(self) -> None:
|
|
matrix = gate.validate_matrix(valid_matrix())
|
|
evidence = gate.validate_evidence(evidence_for(matrix, NEXT_SHA), matrix)
|
|
evaluation = gate.evaluate_matrix(matrix, SHA, evidence)
|
|
self.assertEqual(set(evaluation.gate_results.values()), {"FAIL"})
|
|
self.assertIn("not audited SHA", evaluation.details[0])
|
|
|
|
def test_missing_external_evidence_is_honest_not_run(self) -> None:
|
|
matrix = gate.validate_matrix(valid_matrix())
|
|
evaluation = gate.evaluate_matrix(matrix, SHA, None)
|
|
self.assertEqual(evaluation.verdict, "NO-GO")
|
|
self.assertEqual(set(evaluation.gate_results.values()), {"NOT_RUN"})
|
|
self.assertEqual(evaluation.details, ())
|
|
|
|
def test_missing_triple_zero_fails_applicable_gate(self) -> None:
|
|
matrix = gate.validate_matrix(valid_matrix())
|
|
payload = evidence_for(matrix, SHA)
|
|
del payload["records"][0]["triple_zero"]
|
|
evidence = gate.validate_evidence(payload, matrix)
|
|
evaluation = gate.evaluate_matrix(matrix, SHA, evidence)
|
|
self.assertEqual(evaluation.gate_results["RG-01"], "FAIL")
|
|
self.assertTrue(any("missing required triple-zero" in detail for detail in evaluation.details))
|
|
|
|
def test_live_relay_rg12_requires_semantically_valid_sealed_artifact(self) -> None:
|
|
payload = valid_matrix()
|
|
rg12 = payload["gates"][11]
|
|
rg12["required_evidence_tier"] = "live-relay"
|
|
rg12["artifact_contract"] = gate.RG12_ARTIFACT_CONTRACT
|
|
matrix = gate.validate_matrix(payload)
|
|
evidence_payload = evidence_for(matrix, SHA)
|
|
evidence_payload["records"][11]["tier"] = "live-relay"
|
|
evidence_payload["records"][11]["kind"] = "live-relay-artifact"
|
|
|
|
missing = gate.validate_evidence(evidence_payload, matrix)
|
|
missing_evaluation = gate.evaluate_matrix(matrix, SHA, missing)
|
|
self.assertEqual(missing_evaluation.gate_results["RG-12"], "FAIL")
|
|
self.assertTrue(any("missing the required sealed artifact descriptor" in detail for detail in missing_evaluation.details))
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
artifact = create_rg12_artifact(Path(temp_dir) / "sealed", SHA)
|
|
evidence_payload["records"][11]["artifact"] = artifact
|
|
evidence = gate.validate_evidence(evidence_payload, matrix)
|
|
evaluation = gate.evaluate_matrix(matrix, SHA, evidence)
|
|
self.assertEqual(evaluation.gate_results["RG-12"], "PASS")
|
|
self.assertEqual(evaluation.verdict, "GO")
|
|
|
|
def test_rg12_artifact_checksum_tampering_fails_closed(self) -> None:
|
|
payload = valid_matrix()
|
|
rg12 = payload["gates"][11]
|
|
rg12["required_evidence_tier"] = "live-relay"
|
|
rg12["artifact_contract"] = gate.RG12_ARTIFACT_CONTRACT
|
|
matrix = gate.validate_matrix(payload)
|
|
evidence_payload = evidence_for(matrix, SHA)
|
|
evidence_payload["records"][11]["tier"] = "live-relay"
|
|
evidence_payload["records"][11]["kind"] = "live-relay-artifact"
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir) / "sealed"
|
|
evidence_payload["records"][11]["artifact"] = create_rg12_artifact(root, SHA)
|
|
(root / "cleanup.json").write_text("{}\n", encoding="utf-8")
|
|
evidence = gate.validate_evidence(evidence_payload, matrix)
|
|
evaluation = gate.evaluate_matrix(matrix, SHA, evidence)
|
|
self.assertEqual(evaluation.gate_results["RG-12"], "FAIL")
|
|
self.assertTrue(any("checksum mismatch" in detail for detail in evaluation.details))
|
|
|
|
def test_repository_matrix_schema_is_deterministic_and_requirement_only(self) -> None:
|
|
matrix_path = Path(__file__).resolve().parents[2] / "docs/release-control/v6/internal/pulse-intelligence-release-gate.json"
|
|
matrix_text = matrix_path.read_text(encoding="utf-8")
|
|
raw = json.loads(matrix_text)
|
|
first = gate.validate_matrix(raw)
|
|
second = gate.validate_matrix(json.loads(json.dumps(raw, sort_keys=True)))
|
|
self.assertEqual(first, second)
|
|
self.assertNotIn("evaluation_git_sha", first)
|
|
self.assertNotIn("overall_result", first)
|
|
self.assertEqual(first["triple_zero_semantics"], gate.TRIPLE_ZERO_SEMANTICS)
|
|
self.assertIsNone(re.search(r"\b[0-9a-f]{40}\b", matrix_text))
|
|
for item in first["gates"]:
|
|
self.assertNotIn("result", item)
|
|
self.assertNotIn("evidence", item)
|
|
self.assertEqual(first["gates"][11]["artifact_contract"], gate.RG12_ARTIFACT_CONTRACT)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|