test(release): exercise installed Python distribution behavior

tests/test_installed_python_controls.py

@@ -0,0 +1,67 @@
+import importlib.util
+import json
+import unittest
+from pathlib import Path
+from unittest.mock import Mock
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location("installed_python_controls", ROOT / "tools/installed_python_controls.py")
+controls_module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(controls_module)
+
+
+class InstalledControlsTests(unittest.TestCase):
+    def setUp(self):
+        from agentrust_telemetry import (
+            EventFactory,
+            EventValidationError,
+            SchemaValidator,
+        )
+        self.error = EventValidationError
+        self.validator = SchemaValidator.bundled()
+        self.controls = {
+            "fixtures": {kind: [{"name": path.name, "event": json.loads(path.read_text())}
+                                for path in sorted((ROOT / "conformance/fixtures" / kind).glob("*.json"))]
+                         for kind in ("valid", "invalid")},
+            "golden": json.loads((ROOT / "compatibility/golden/event-factory.json").read_text()),
+        }
+        self.factory = EventFactory(self.validator, producer_name="parity-test", producer_version="1.0.0",
+                                   clock_ns=lambda: 1787079000000000000,
+                                   event_id_factory=lambda: "018f0f7d-7a13-7cc2-8000-000000000042")
+
+    def test_real_controls(self):
+        counts = controls_module.check_controls(self.controls, self.validator, self.factory, self.error)
+        self.assertEqual(counts, {"valid": 6, "invalid": 7})
+
+    def test_accept_all_is_caught(self):
+        with self.assertRaisesRegex(AssertionError, "invalid control admitted"):
+            controls_module.check_controls(self.controls, Mock(), self.factory, self.error)
+
+    def test_reject_all_is_caught(self):
+        validator = Mock()
+        validator.validate.side_effect = self.error("synthetic rejection")
+        with self.assertRaisesRegex(AssertionError, "valid control refused"):
+            controls_module.check_controls(self.controls, validator, self.factory, self.error)
+
+    def test_changed_factory_is_caught(self):
+        factory = Mock()
+        factory.build.return_value = {**self.controls["golden"], "input_tokens": 8}
+        with self.assertRaisesRegex(AssertionError, "golden"):
+            controls_module.check_controls(self.controls, self.validator, factory, self.error)
+
+    def test_empty_controls_are_caught(self):
+        for kind in ("valid", "invalid"):
+            with self.subTest(kind=kind):
+                controls = {**self.controls, "fixtures": {**self.controls["fixtures"], kind: []}}
+                with self.assertRaisesRegex(AssertionError, "missing"):
+                    controls_module.check_controls(controls, self.validator, self.factory, self.error)
+
+    def test_partial_and_renamed_controls_are_caught(self):
+        for kind in ("valid", "invalid"):
+            cases = self.controls["fixtures"][kind]
+            for changed in (cases[1:], [{**cases[0], "name": "renamed.json"}, *cases[1:]],
+                            [cases[0], *cases]):
+                with self.subTest(kind=kind, names=[c["name"] for c in changed]):
+                    controls = {**self.controls, "fixtures": {**self.controls["fixtures"], kind: changed}}
+                    with self.assertRaisesRegex(AssertionError, "missing or unexpected"):
+                        controls_module.check_controls(controls, self.validator, self.factory, self.error)

tools/installed_python_controls.py

@@ -0,0 +1,76 @@
+"""Behavioral controls executed by the installed interpreter outside the checkout."""
+from __future__ import annotations
+
+import json
+import sys
+from importlib.metadata import version
+from pathlib import Path
+
+
+EXPECTED_FIXTURES = {
+    "valid": [
+        "action.json",
+        "approval.json",
+        "data-flow.json",
+        "evidence.json",
+        "policy-decision.json",
+        "usage.json"
+    ],
+    "invalid": [
+        "action-without-digest.json",
+        "aggregate-cost-without-rollup.json",
+        "model-call-with-rollup.json",
+        "numeric-timestamp.json",
+        "raw-content.json",
+        "usage-without-measurement.json",
+        "zero-trace-id.json"
+    ]
+}
+
+
+def check_controls(controls, validator, factory, validation_error):
+    counts = {}
+    for kind in ("valid", "invalid"):
+        cases = controls["fixtures"][kind]
+        if sorted(case["name"] for case in cases) != EXPECTED_FIXTURES[kind]:
+            raise AssertionError(f"missing or unexpected {kind} controls")
+        for case in cases:
+            try:
+                validator.validate(case["event"])
+            except validation_error:
+                if kind == "valid":
+                    raise AssertionError(f"valid control refused: {case['name']}") from None
+            else:
+                if kind == "invalid":
+                    raise AssertionError(f"invalid control admitted: {case['name']}")
+        counts[kind] = len(cases)
+    actual = factory.build(
+        "usage.recorded", run_id="run-1", agent_id="agent-1", workflow_id="workflow-1",
+        scope="model_call", operation="chat", input_tokens=7,
+    )
+    if actual != controls["golden"]:
+        raise AssertionError("installed event construction differs from golden control")
+    return counts
+
+
+def main():
+    import agentrust_telemetry as package
+    from agentrust_telemetry import EventFactory, EventValidationError, SchemaValidator
+
+    controls = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
+    origin = Path(package.__file__).resolve()
+    if not origin.is_relative_to(Path(sys.prefix).resolve()):
+        raise AssertionError(f"import escaped installed environment: {origin}")
+    if version("agentrust-telemetry") != controls["version"]:
+        raise AssertionError("installed version differs from source package metadata")
+    validator = SchemaValidator.bundled()
+    factory = EventFactory(
+        validator, producer_name="parity-test", producer_version="1.0.0",
+        clock_ns=lambda: 1787079000000000000,
+        event_id_factory=lambda: "018f0f7d-7a13-7cc2-8000-000000000042",
+    )
+    print(json.dumps(check_controls(controls, validator, factory, EventValidationError)))
+
+
+if __name__ == "__main__":
+    main()

tools/smoke_wheel.py

@@ -1,31 +1,50 @@
 """Install a distribution into a fresh venv and verify bundled schemas load."""
 
-from pathlib import Path
 import argparse
+import json
+import os
+import shutil
 import subprocess
 import sys
 import tempfile
 import venv
+from pathlib import Path
 
 
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("distribution", type=Path)
     args = parser.parse_args()
     distribution = args.distribution.resolve()
+    root = Path(__file__).resolve().parents[1]
+    fixtures = {}
+    for kind in ("valid", "invalid"):
+        fixtures[kind] = [
+            {"name": path.name, "event": json.loads(path.read_text(encoding="utf-8"))}
+            for path in sorted((root / "conformance" / "fixtures" / kind).glob("*.json"))
+        ]
+    # Keep Python 3.10 support without introducing another TOML dependency.
+    import re
+    package_version = re.search(
+        r'^version = "([^"]+)"', (root / "pyproject.toml").read_text(), re.MULTILINE
+    ).group(1)
+    controls = {"fixtures": fixtures, "version": package_version,
+                "golden": json.loads((root / "compatibility/golden/event-factory.json").read_text())}
     with tempfile.TemporaryDirectory(prefix="agentrust-wheel-") as directory:
         env = Path(directory)
         venv.EnvBuilder(with_pip=True).create(env)
         python = env / ("Scripts/python.exe" if sys.platform == "win32" else "bin/python")
         subprocess.run([python, "-m", "pip", "install", str(distribution)], check=True)
+        (env / "controls.json").write_text(json.dumps(controls), encoding="utf-8")
+        shutil.copyfile(Path(__file__).with_name("installed_python_controls.py"), env / "controls.py")
+        clean_env = {key: value for key, value in os.environ.items()
+                     if key not in {"PYTHONPATH", "PYTHONHOME"}}
         subprocess.run(
             [
                 python,
-                "-c",
-                "from agentrust_telemetry import SchemaValidator; "
-                "v=SchemaValidator.bundled(); print(v.schema_directory)",
+                "-I", "controls.py", "controls.json",
             ],
-            check=True,
+            check=True, cwd=env, env=clean_env,
         )
     print(f"PASS installed distribution smoke test: {distribution.name}")
     return 0