diff --git a/python/tests/test_control_contract_composition.py b/python/tests/test_control_contract_composition.py new file mode 100644 index 0000000..e9fcbd8 --- /dev/null +++ b/python/tests/test_control_contract_composition.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +import importlib.util +import json +import math +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" +AS_OF = "2026-08-05T12:00:00Z" + + +def _load_module(name: str): + spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +deployment_bundle_contract = _load_module("deployment_bundle_contract") +activation_contract = _load_module("activation_contract") +reconciliation_record_contract = _load_module("reconciliation_record_contract") + + +class ControlContractCompositionTest(unittest.TestCase): + @staticmethod + def _sha(character: str) -> str: + return character * 64 + + @staticmethod + def _revision(character: str) -> str: + return character * 40 + + def _bundle(self, *, platform: str = "interactive-brokers") -> dict[str, object]: + bundle: dict[str, object] = { + "schema": "qsl.deployment_bundle.v1", + "bundle_id": f"bundle.synthetic.{platform}.20260805", + "created_at": "2026-08-05T08:00:00Z", + "digest_algorithm": "sha256", + "strategy": { + "id": "synthetic-strategy", + "source_id": "synthetic-strategy-source", + "revision": self._revision("a"), + "artifact_sha256": self._sha("b"), + }, + "profile": { + "id": "synthetic-profile", + "revision": self._revision("c"), + "artifact_sha256": self._sha("d"), + }, + "config": { + "id": "synthetic-config", + "revision": self._revision("e"), + "artifact_sha256": self._sha("f"), + }, + "evidence": { + "id": "synthetic-evidence", + "revision": self._revision("1"), + "artifact_sha256": self._sha("2"), + }, + "target": {"id": f"synthetic-{platform}", "platform_id": platform}, + "dependencies": { + "qpk": {"id": "quant-platform-kit", "revision": self._revision("3"), "artifact_sha256": self._sha("4")}, + "strategy": {"id": "synthetic-strategy-source", "revision": self._revision("a"), "artifact_sha256": self._sha("5")}, + "pipeline": {"id": "synthetic-pipeline", "revision": self._revision("6"), "artifact_sha256": self._sha("7")}, + "platform": {"id": platform, "revision": self._revision("8"), "artifact_sha256": self._sha("9")}, + }, + } + bundle["bundle_sha256"] = deployment_bundle_contract.calculate_bundle_sha256(bundle) + return bundle + + def _activation(self, bundle: dict[str, object]) -> dict[str, object]: + activation: dict[str, object] = { + "schema": "qsl.activation.v1", + "activation_id": "activation.synthetic.disabled.20260805", + "created_at": "2026-08-05T09:00:00Z", + "digest_algorithm": "sha256", + "contract_only": True, + "deployment_bundle": { + "schema": bundle["schema"], + "bundle_id": bundle["bundle_id"], + "bundle_sha256": bundle["bundle_sha256"], + }, + "stage": "DISABLED", + "effective_at": "2026-08-05T10:00:00Z", + "expires_at": "2026-08-05T18:00:00Z", + "human_authority": { + "stage": "DISABLED", + "authority_id": "synthetic-human-authority.disabled.20260805", + "authority_version": "v1", + "authority_receipt_sha256": self._sha("c"), + }, + "target": { + "platform": bundle["target"]["platform_id"], + "repository": "QuantStrategyLab/InteractiveBrokersPlatform", + "revision": bundle["dependencies"]["platform"]["revision"], + "environment": "synthetic-disabled", + "account_alias": "synthetic-research", + "account_digest_sha256": self._sha("d"), + }, + } + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + return activation + + def _composition(self) -> tuple[dict[str, object], dict[str, object], dict[str, object]]: + bundle = self._bundle() + activation = self._activation(bundle) + record = reconciliation_record_contract.build_missing_record( + record_id="reconciliation.synthetic.disabled.20260805", + produced_at="2026-08-05T11:00:00Z", + expires_at="2026-08-05T17:00:00Z", + expected_bundle=bundle, + expected_activation=activation, + as_of=AS_OF, + ) + return bundle, activation, record + + def _validate(self, bundle, activation, record): + deployment_bundle_contract.validate_bundle(bundle) + activation_contract.validate_activation(activation, expected_bundle=bundle, as_of=AS_OF) + return reconciliation_record_contract.validate_reconciliation_record( + record, + expected_bundle=bundle, + expected_activation=activation, + as_of=AS_OF, + ) + + def test_synthetic_disabled_composition_is_exact_and_deterministic(self): + bundle, activation, record = self._composition() + validated = self._validate(bundle, activation, record) + self.assertEqual(activation["stage"], "DISABLED") + self.assertTrue(activation["contract_only"]) + self.assertEqual(record["status"], "MISSING") + self.assertTrue(record["contract_only"]) + self.assertEqual( + record["assertions"], + { + "apply_performed": False, + "config_sync_performed": False, + "runtime_mutation_performed": False, + "runtime_active_verified": False, + "fills_verified": False, + "capital_use_verified": False, + }, + ) + self.assertEqual(validated["deployment_bundle"]["bundle_sha256"], bundle["bundle_sha256"]) + self.assertEqual(validated["activation"]["activation_sha256"], activation["activation_sha256"]) + self.assertEqual(validated["expected_identity"]["deployment_bundle_sha256"], bundle["bundle_sha256"]) + self.assertEqual(validated["expected_identity"]["activation_sha256"], activation["activation_sha256"]) + self.assertEqual( + deployment_bundle_contract.calculate_bundle_sha256(bundle), + deployment_bundle_contract.calculate_bundle_sha256(dict(reversed(bundle.items()))), + ) + self.assertEqual( + activation_contract.calculate_activation_sha256(activation), + activation_contract.calculate_activation_sha256(dict(reversed(activation.items()))), + ) + self.assertEqual( + reconciliation_record_contract.calculate_reconciliation_sha256(record), + reconciliation_record_contract.calculate_reconciliation_sha256(dict(reversed(record.items()))), + ) + rebuilt = self._composition() + self.assertEqual(json.dumps((bundle, activation, record), sort_keys=True), json.dumps(rebuilt, sort_keys=True)) + + def test_bundle_mutations_fail_closed_through_activation_and_reconciliation_chain(self): + for mutate in ( + lambda value: value["strategy"].update({"id": "mutated-strategy"}), + lambda value: value["dependencies"]["strategy"].update({"artifact_sha256": self._sha("0")}), + lambda value: value["evidence"].update({"artifact_sha256": self._sha("0")}), + ): + with self.subTest(mutate=mutate): + bundle, activation, record = self._composition() + mutate(bundle) + with self.assertRaises((activation_contract.ActivationValidationError, reconciliation_record_contract.ReconciliationValidationError)): + activation_contract.validate_activation(activation, expected_bundle=bundle, as_of=AS_OF) + with self.assertRaises(reconciliation_record_contract.ReconciliationValidationError): + reconciliation_record_contract.validate_reconciliation_record( + record, expected_bundle=bundle, expected_activation=activation, as_of=AS_OF + ) + + def test_activation_mutations_fail_closed_through_reconciliation_chain(self): + mutations = ( + lambda value: value.update({"activation_id": "activation.synthetic.other.20260805"}), + lambda value: value.update({"activation_sha256": self._sha("0")}), + lambda value: value["target"].update({"environment": "synthetic-other"}), + lambda value: value.update({"stage": "PAPER_DRY_RUN"}), + lambda value: value.update({"effective_at": "2026-08-05T11:30:00Z"}), + ) + for mutate in mutations: + with self.subTest(mutate=mutate): + bundle, activation, record = self._composition() + mutate(activation) + if activation["stage"] != activation["human_authority"]["stage"]: + activation["human_authority"]["stage"] = activation["stage"] + if activation["activation_sha256"] != self._sha("0"): + activation["activation_sha256"] = activation_contract.calculate_activation_sha256(activation) + with self.assertRaises(reconciliation_record_contract.ReconciliationValidationError): + reconciliation_record_contract.validate_reconciliation_record( + record, expected_bundle=bundle, expected_activation=activation, as_of=AS_OF + ) + + def test_non_missing_observer_and_comparison_material_is_rejected(self): + for field, value in ( + ("status", "MATCHED"), + ("status", "MISMATCHED"), + ("observer", {"id": "synthetic-observer"}), + ("observation", {"observed_at": "2026-08-05T11:00:00Z"}), + ("comparison", {"complete": False}), + ): + with self.subTest(field=field): + bundle, activation, record = self._composition() + record[field] = value + record["reconciliation_sha256"] = reconciliation_record_contract.calculate_reconciliation_sha256(record) + with self.assertRaises(reconciliation_record_contract.ReconciliationValidationError): + self._validate(bundle, activation, record) + + def test_unsafe_material_and_invalid_synthetic_inputs_fail_closed(self): + for key, value in ( + ("unknown", {"nested": True}), + ("credential", "not-a-credential"), + ("raw_payload", {"synthetic": True}), + ("account", {"number": "00000000"}), + ("order", {"id": "synthetic-order"}), + ("fill", {"id": "synthetic-fill"}), + ("capital", {"amount": 1}), + ): + with self.subTest(key=key): + bundle, activation, record = self._composition() + record["expected_identity"][key] = value + record["reconciliation_sha256"] = reconciliation_record_contract.calculate_reconciliation_sha256(record) + with self.assertRaises(reconciliation_record_contract.ReconciliationValidationError): + self._validate(bundle, activation, record) + + bundle, activation, record = self._composition() + bundle["created_at"] = "2026-02-30T08:00:00Z" + bundle["bundle_sha256"] = deployment_bundle_contract.calculate_bundle_sha256(bundle) + with self.assertRaises(deployment_bundle_contract.BundleValidationError): + deployment_bundle_contract.validate_bundle(bundle) + + bundle, activation, record = self._composition() + record["produced_at"] = "2026-02-30T11:00:00Z" + record["reconciliation_sha256"] = reconciliation_record_contract.calculate_reconciliation_sha256(record) + with self.assertRaises(reconciliation_record_contract.ReconciliationValidationError): + self._validate(bundle, activation, record) + + bundle, activation, record = self._composition() + record["assertions"]["runtime_active_verified"] = math.nan + with self.assertRaises(reconciliation_record_contract.ReconciliationValidationError): + self._validate(bundle, activation, record) + + bundle, activation, record = self._composition() + other_bundle = self._bundle(platform="binance-platform") + with self.assertRaises(reconciliation_record_contract.ReconciliationValidationError): + reconciliation_record_contract.validate_reconciliation_record( + record, expected_bundle=other_bundle, expected_activation=activation, as_of=AS_OF + ) + + def test_serialized_synthetic_composition_has_no_credentials_or_business_payload(self): + composition = json.dumps(self._composition(), sort_keys=True) + for forbidden in ("credential", "secret", "token", "password", "raw_payload", "orders", "capital_value"): + self.assertNotIn(forbidden, composition.lower()) + + +if __name__ == "__main__": + unittest.main()