c9d02147d3
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02e56-e4ad-71a2-b3e2-b6193e0d8093
134 lines
5 KiB
Python
134 lines
5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPTS = ROOT / "scripts"
|
|
TESTS = ROOT / "tests"
|
|
for path in (SCRIPTS, TESTS):
|
|
if str(path) not in sys.path:
|
|
sys.path.insert(0, str(path))
|
|
|
|
from test_custody_contract import projection_contract
|
|
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"wp0025_broker_readiness", SCRIPTS / "wp0025-broker-readiness.py"
|
|
)
|
|
assert SPEC and SPEC.loader
|
|
module = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(module)
|
|
|
|
|
|
class BrokerReadinessTests(unittest.TestCase):
|
|
@staticmethod
|
|
def engagement_time(contract: dict) -> datetime:
|
|
return datetime.fromisoformat(
|
|
contract["window"]["starts_at"].replace("Z", "+00:00")
|
|
)
|
|
|
|
def test_show_is_direct_and_authorizes_no_live_mutation(self) -> None:
|
|
result = module.show(projection_contract())
|
|
self.assertEqual("whitehat-security", result["owner"])
|
|
self.assertIn("approve", result["approve_command"])
|
|
self.assertFalse(result["live_mutation_authorized"])
|
|
self.assertFalse(result["secret_values_observed"])
|
|
|
|
def test_adapter_verification_pins_revision_file_and_focused_test(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
adapter = root / module.DEFAULT_ADAPTER
|
|
test = root / module.DEFAULT_TEST
|
|
adapter.parent.mkdir(parents=True)
|
|
test.parent.mkdir(parents=True)
|
|
adapter.write_text("class PlatformCustodyBroker: pass\n", encoding="utf-8")
|
|
test.write_text("def test_adapter(): assert True\n", encoding="utf-8")
|
|
|
|
def runner(command, **kwargs):
|
|
if command[:2] == ["git", "rev-parse"]:
|
|
return subprocess.CompletedProcess(command, 0, stdout="a" * 40 + "\n", stderr="")
|
|
return subprocess.CompletedProcess(command, 0, stdout="1 passed\n", stderr="")
|
|
|
|
result = module.verify_adapter(root, runner=runner)
|
|
self.assertTrue(result["passed"])
|
|
self.assertEqual("a" * 40, result["adapter"]["revision"])
|
|
self.assertEqual(64, len(result["adapter"]["sha256"]))
|
|
self.assertTrue(result["adapter"]["tests_passed"])
|
|
|
|
def test_missing_adapter_fails_without_running_commands(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
result = module.verify_adapter(
|
|
Path(directory),
|
|
runner=lambda *args, **kwargs: self.fail("runner must not execute"),
|
|
)
|
|
self.assertFalse(result["passed"])
|
|
self.assertFalse(result["secret_values_observed"])
|
|
|
|
def test_approval_round_trip_and_change_request(self) -> None:
|
|
contract = projection_contract()
|
|
verification = {
|
|
"passed": True,
|
|
"adapter": {
|
|
"repo": "whitehat-security",
|
|
"revision": "a" * 40,
|
|
"path": str(module.DEFAULT_ADAPTER),
|
|
"sha256": "b" * 64,
|
|
"tests_passed": True,
|
|
},
|
|
"secret_values_observed": False,
|
|
}
|
|
now = self.engagement_time(contract)
|
|
receipt = module.build_approval(
|
|
contract, "whitehat-owner", verification, now=now
|
|
)
|
|
self.assertEqual("approve", receipt["decision"])
|
|
self.assertTrue(receipt["cleanup_request_supported"])
|
|
change = module.build_change_request(
|
|
contract,
|
|
"whitehat-owner",
|
|
"Bind cleanup acknowledgement.",
|
|
now=now,
|
|
)
|
|
self.assertEqual("request-changes", change["decision"])
|
|
with self.assertRaises(module.ReadinessError):
|
|
module.build_change_request(contract, "whitehat-owner", "", now=now)
|
|
|
|
def test_receipt_export_is_raw_canonical_document_with_private_mode(self) -> None:
|
|
contract = projection_contract()
|
|
verification = {
|
|
"passed": True,
|
|
"adapter": {
|
|
"repo": "whitehat-security",
|
|
"revision": "a" * 40,
|
|
"path": str(module.DEFAULT_ADAPTER),
|
|
"sha256": "b" * 64,
|
|
"tests_passed": True,
|
|
},
|
|
"secret_values_observed": False,
|
|
}
|
|
receipt = module.build_approval(
|
|
contract,
|
|
"whitehat-owner",
|
|
verification,
|
|
now=self.engagement_time(contract),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "nested" / "broker.json"
|
|
module.write_receipt(path, receipt)
|
|
observed = json.loads(path.read_text(encoding="utf-8"))
|
|
mode = stat.S_IMODE(path.stat().st_mode)
|
|
self.assertEqual(receipt, observed)
|
|
self.assertNotIn("submitted", observed)
|
|
self.assertEqual(0o600, mode)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|