From 7d81258a3052f7c8c50e8b2eb446f82a5e7e080f Mon Sep 17 00:00:00 2001 From: Louielunz <48041247+lywinged@users.noreply.github.com> Date: Mon, 24 Aug 2026 22:01:30 +0000 Subject: [PATCH] test(trace-adapter): exercise every refusal finalize_trace can reach `trace_adapter.py` has sixteen `raise TraceFinalizationError` sites. Replacing each one with `pass` on its own and re-running the suite leaves ten of them green, so ten correct refusals had nothing exercising them. With this file the same sweep turns all sixteen red. The sites are located by AST rather than by grep, so the two compound conditions and the two `except` handlers are counted alongside the twelve plain guards. The eight cases that a bad event stream can cause are built through `EvidenceAccumulator` rather than by handing the adapter a dict, so a passing case is also evidence the refusal is reachable from a schema-valid event stream rather than defensive. That distinction matters for at least one of them: `policy.required` is `["engine", "engine_version"]`, so a conforming producer can emit a decision without a `bundle_digest`, and the guard is the only thing between that and a signed record. Two of the sixteen are different in kind and the file says so where they sit. No event stream can produce either. The missing optional dependency is reached by blocking the import, which is the only way in, so that case does not go through `finalize_trace` at all. The signing wrapper does go through `finalize_trace`, on an accumulator-built snapshot, but it needs a key that satisfies the shape check and then fails when used, because `object()` is refused earlier for having no `sign` and a case built that way would pass on the wrong refusal. Behaviour is unchanged. Probing `_policy_binding` directly with six malformed event streams produced the correct refusal every time before this was written; the gap was coverage, not conduct. CI as configured, in order: check_versions, check_schemas, check_otel_compatibility, check_typescript_schemas, validate.py, unittest discover, compileall, and on the 3.12 leg build and smoke_wheel on both artefacts. All exit 0. sync_schemas is not a CI step, it is on the pull request template's evidence list; it also exits 0 and leaves the tree unchanged. On 3.11, 3.12 and 3.13 the suite goes from `Ran 98 tests` to `Ran 108 tests`. On 3.10 it stays at `Ran 89 tests` and gains one skip: `agentrust-trace` is pinned to `python_version >= '3.11'`, so this class skips whole on the same guard and the same message as `tests/test_trace_adapter.py`. Signed-off-by: Louielunz <48041247+lywinged@users.noreply.github.com> --- tests/test_trace_adapter_refusals.py | 204 +++++++++++++++++++++++++++ 1 file changed, 204 insertions(+) create mode 100644 tests/test_trace_adapter_refusals.py diff --git a/tests/test_trace_adapter_refusals.py b/tests/test_trace_adapter_refusals.py new file mode 100644 index 0000000..956e6d0 --- /dev/null +++ b/tests/test_trace_adapter_refusals.py @@ -0,0 +1,204 @@ +"""Every refusal in `trace_adapter.py`, exercised. + +`trace_adapter.py` has sixteen `raise TraceFinalizationError` sites. Five test +methods in `tests/test_trace_adapter.py` hold six of them up. Replacing any of +the other ten with `pass`, one at a time, leaves `python -m unittest discover +-s tests` at exit 0, so those ten are correct and unexercised. This file covers +those ten. + +Eight of them are refused because of the events. Each is built with +`EvidenceAccumulator` and driven through `finalize_trace` rather than by +handing the adapter a dict, so a case that passes is also evidence the refusal +is reachable from a schema-valid event stream rather than defensive. + +Two are different in kind, and the file says so where they sit. No event +stream can produce either. The missing optional dependency is reached by +blocking the import, which is the only way in, so that case does not go +through `finalize_trace` at all. The failure inside official signing does go +through `finalize_trace`, on an accumulator-built snapshot, with a signing key +that passes the shape check and then fails when used. +""" + +import copy +import json +import sys +import unittest +from dataclasses import replace +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) + +from agentrust_telemetry import ( # noqa: E402 + EvidenceAccumulator, + SchemaValidator, + TraceConfiguration, + TraceFinalizationError, + finalize_trace, +) +from agentrust_telemetry import trace_adapter # noqa: E402 + + +def fixture(name): + return json.loads((ROOT / "conformance" / "fixtures" / "valid" / name).read_text()) + + +class TraceAdapterRefusalTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + if sys.version_info < (3, 11): + raise unittest.SkipTest("agentrust-trace requires Python 3.11+") + from agentrust_trace import generate_key + + cls.key = generate_key() + cls.validator = SchemaValidator(ROOT / "spec" / "schema") + cls.config = TraceConfiguration( + subject="spiffe://example.test/agent/workflow", + model_provider="example", + model_id="example-model", + model_version="2026-08", + build_digest="sha256:" + "b" * 64, + build_slsa_level=1, + origin_kind="self", + origin_producer="example-runtime", + appraisal_verifier="https://example.test/verifier", + classification_taxonomy="example.enterprise.v1", + classification_order=("public", "internal", "confidential", "restricted"), + ) + + def snapshot(self, events, *, completeness="complete"): + accumulator = EvidenceAccumulator("run-governed-sdlc-001", self.validator) + for event in events: + accumulator.append(event) + return accumulator.seal(completeness=completeness) + + def finalize(self, events, *, config=None, completeness="complete"): + return finalize_trace( + self.snapshot(events, completeness=completeness), + config or self.config, + signing_key=self.key, + ) + + def assertRefuses(self, message, events, *, config=None, completeness="complete"): + with self.assertRaises(TraceFinalizationError) as caught: + self.finalize(events, config=config, completeness=completeness) + self.assertIn(message, str(caught.exception)) + + # -- the evidence itself ------------------------------------------------- + + def test_an_empty_run_carries_no_chained_evidence(self): + self.assertRefuses("non-empty chained evidence", []) + + # -- the trusted configuration ------------------------------------------- + + def test_every_required_configuration_field_is_named_when_absent(self): + for field in ("subject", "model_provider", "model_id", "build_digest", + "origin_producer", "appraisal_verifier"): + with self.subTest(field=field): + config = replace(self.config, **{field: ""}) + with self.assertRaises(TraceFinalizationError) as caught: + self.finalize([fixture("policy-decision.json"), + fixture("data-flow.json")], config=config) + message = str(caught.exception) + self.assertIn("trusted TRACE configuration is missing", message) + self.assertIn(field, message) + + def test_a_repeated_classification_cannot_rank_anything(self): + order = ("public", "internal", "public") + config = replace(self.config, classification_order=order) + self.assertRefuses( + "classification_order contains duplicates", + [fixture("policy-decision.json"), fixture("data-flow.json")], + config=config, + ) + + # -- the policy binding --------------------------------------------------- + + def test_a_run_with_no_policy_decision_has_no_policy_to_bind(self): + self.assertRefuses( + "no policy decision evidence is present", [fixture("data-flow.json")] + ) + + def test_a_policy_decision_without_a_bundle_digest_is_refused(self): + event = copy.deepcopy(fixture("policy-decision.json")) + del event["policy"]["bundle_digest"] + self.validator.validate(event) # the schema permits it; the adapter must not + self.assertRefuses( + "every policy decision must carry bundle_digest", + [event, fixture("data-flow.json")], + ) + + def test_two_decisions_disagreeing_on_enforcement_cannot_produce_one_mode(self): + first = fixture("policy-decision.json") + second = copy.deepcopy(first) + second["event_id"] = "018f0f7d-7a13-7cc2-8000-0000000000f1" + second["enforcement_mode"] = "monitor" + self.assertRefuses( + "conflicting policy enforcement modes are present", + [first, second, fixture("data-flow.json")], + ) + + # -- the data classification ---------------------------------------------- + + def test_a_run_with_no_data_flow_has_nothing_to_classify(self): + self.assertRefuses( + "no classified data-flow evidence is present", + [fixture("policy-decision.json")], + ) + + def test_a_taxonomy_the_configuration_does_not_name_is_refused(self): + config = replace(self.config, classification_taxonomy="example.other.v1") + self.assertRefuses( + "data-flow taxonomy conflicts with TRACE configuration", + [fixture("policy-decision.json"), fixture("data-flow.json")], + config=config, + ) + + # -- the two that no event stream can produce ------------------------------ + + def test_the_optional_dependency_is_named_when_it_is_absent(self): + import builtins + + original = builtins.__import__ + + def refuse(name, *args, **kwargs): + if name == "agentrust_trace": + raise ImportError("not installed") + return original(name, *args, **kwargs) + + builtins.__import__ = refuse + try: + with self.assertRaises(TraceFinalizationError) as caught: + trace_adapter._trace_package() + self.assertIn("optional dependency", str(caught.exception)) + finally: + builtins.__import__ = original + + def test_a_failure_inside_official_signing_is_reported_as_one(self): + """A key that satisfies the shape check and then fails when used. + + `object()` would not reach here: it is refused earlier for having no + `sign`, so a case built that way passes on the wrong refusal. + """ + key = self.key + + class FailsWhenUsed: + def sign(self, *args, **kwargs): + raise RuntimeError("hardware signer unavailable") + + def public_key(self): + return key.public_key() + + snapshot = self.snapshot( + [fixture("policy-decision.json"), fixture("data-flow.json")] + ) + with self.assertRaises(TraceFinalizationError) as caught: + finalize_trace(snapshot, self.config, signing_key=FailsWhenUsed()) + message = str(caught.exception) + self.assertIn("official TRACE signing or validation failed", message) + self.assertIn("hardware signer unavailable", message) + + +if __name__ == "__main__": + unittest.main()