93 lines
5.8 KiB
Python
93 lines
5.8 KiB
Python
"""r26b submission accounting controls; no auth, subprocess, sockets, or writes."""
|
|
import copy
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
spec = importlib.util.spec_from_file_location("r26b_fixture", ROOT / "server/tests/PrivateTcmR26bHarnessTest.py")
|
|
fixture = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(fixture)
|
|
driver = fixture.driver
|
|
|
|
|
|
def no_execution(mode="rejected_before_execution", *, admitted=True):
|
|
s = fixture.lifecycle_summary(captured=False)
|
|
g = s["generations"][0]
|
|
g.update(original_task_captured=False, submitted=admitted, not_submitted=not admitted,
|
|
no_execution_confirmed=True, lifecycle=None)
|
|
g["submission"] = dict(submitted_at=8.0, transfer_closed=True, transfer_closed_at=11.0, disposition=mode) if admitted else None
|
|
s.update(original_task_count=0, submitted_attempt_count=int(admitted), submitted_transfer_closed_count=int(admitted),
|
|
submitted_pending_attempt_count=0, submitted_without_identity_count=0, pre_execution_closed_attempt_count=int(admitted),
|
|
not_submitted_attempt_count=int(admitted and mode == "not_submitted"),
|
|
rejected_before_execution_attempt_count=int(admitted and mode == "rejected_before_execution"),
|
|
never_submitted_generation_count=int(not admitted), no_execution_generation_count=1, pending_generation_count=0)
|
|
return s
|
|
|
|
|
|
def row_for(audit):
|
|
return fixture.with_counts(dict(preparation_isolation_verified=True, private_wire_audit=audit,
|
|
state=dict(transport_closed=True, upstream_terminal_confirmed=True, upstream_pending_task_count=audit["pending_generation_count"])))
|
|
|
|
|
|
class SubmissionSummary(unittest.TestCase):
|
|
def test_no_execution_is_resolved_separately_from_independent_proof(self):
|
|
for value in [no_execution(), no_execution("not_submitted"), no_execution("not_submitted", admitted=False)]:
|
|
self.assertEqual(driver.safe_private_wire_summary(value), value)
|
|
self.assertFalse(driver.independent_cleanup_confirmed(row_for(value)))
|
|
self.assertEqual(value["independently_terminal_generation_count"], 0)
|
|
|
|
def test_explicit_rejection_plus_actual_terminal_fallback_is_not_unknown(self):
|
|
rejected = no_execution()
|
|
terminal = fixture.lifecycle_summary()
|
|
terminal["generations"][0]["generation"] = 2
|
|
value = copy.deepcopy(terminal)
|
|
for key in ("generation_count", "serialized_request_count", "submitted_attempt_count", "submitted_transfer_closed_count", "closed_generation_count",
|
|
"unidentified_generation_count", "pre_execution_closed_attempt_count", "rejected_before_execution_attempt_count", "no_execution_generation_count"):
|
|
value[key] += rejected[key]
|
|
value["generations"] = rejected["generations"] + terminal["generations"]
|
|
self.assertEqual(driver.safe_private_wire_summary(value), value)
|
|
self.assertTrue(driver.independent_cleanup_confirmed(row_for(value)))
|
|
self.assertEqual((value["pending_generation_count"], value["independently_terminal_generation_count"], value["no_execution_generation_count"]), (0, 1, 1))
|
|
|
|
def test_unknown_and_task_only_are_pending_even_when_transfer_closed(self):
|
|
unknown = fixture.lifecycle_summary(captured=False)
|
|
self.assertEqual(driver.safe_private_wire_summary(unknown), unknown)
|
|
self.assertFalse(driver.independent_cleanup_confirmed(row_for(unknown)))
|
|
bad = copy.deepcopy(unknown)
|
|
bad["generations"][0]["no_execution_confirmed"] = True
|
|
self.assertEqual(driver.safe_private_wire_summary(bad), {"persisted": False})
|
|
|
|
def test_every_control_counter_must_match_before_handoff(self):
|
|
value = fixture.lifecycle_summary()
|
|
row = row_for(value)
|
|
self.assertTrue(driver.independent_cleanup_confirmed(row))
|
|
for key in ("submitted_attempt_count", "submitted_transfer_closed_count", "submitted_pending_attempt_count", "submitted_without_identity_count",
|
|
"pre_execution_closed_attempt_count", "not_submitted_attempt_count", "rejected_before_execution_attempt_count", "upstream_pending_task_count"):
|
|
for mutation in (None, True, -1, 99, "0"):
|
|
invalid = copy.deepcopy(row)
|
|
invalid["state"][key] = mutation
|
|
self.assertFalse(driver.independent_cleanup_confirmed(invalid), (key, mutation))
|
|
|
|
def test_all_derived_counts_are_strict_and_non_forgeable(self):
|
|
valid = fixture.lifecycle_summary()
|
|
for key in [k for k in valid if k.endswith("_count")]:
|
|
for mutation in (True, -1, "1", 100):
|
|
value = copy.deepcopy(valid)
|
|
value[key] = mutation
|
|
self.assertEqual(driver.safe_private_wire_summary(value), {"persisted": False}, (key, mutation))
|
|
|
|
def test_clock_domains_and_late_terminal_proof_cannot_be_forged(self):
|
|
for kind in ("domain", "late", "preclosure", "false_budget", "too_many", "raw_token"):
|
|
value = fixture.lifecycle_summary()
|
|
state = value["generations"][0]["lifecycle"]
|
|
if kind == "domain": state["clock_domains"]["verification_started_at"] = "unix_seconds_microtime"
|
|
if kind == "late": state["verification_elapsed_ms"] = 2001
|
|
if kind == "preclosure": state["observations"][0]["observed_at"] = 0
|
|
if kind == "false_budget": state["observations"][0]["within_budget"] = False
|
|
if kind == "too_many": state["verification_attempts"] = 3
|
|
if kind == "raw_token": value["generations"][0]["submission"]["submission_id"] = "private-opaque-token"
|
|
self.assertEqual(driver.safe_private_wire_summary(value), {"persisted": False}, kind)
|
|
|
|
if __name__ == "__main__": unittest.main()
|