From a3ce03824ef991d7ad8ebb5c2e85623e60969661 Mon Sep 17 00:00:00 2001 From: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com> Date: Sun, 25 Oct 2026 20:03:51 +0530 Subject: [PATCH 1/2] test(release): exercise installed Python distribution behavior Signed-off-by: Srinivasa Dinakar Jainti --- tests/test_installed_python_controls.py | 58 +++++++++++++++++++++++++ tools/installed_python_controls.py | 55 +++++++++++++++++++++++ tools/smoke_wheel.py | 29 ++++++++++--- 3 files changed, 137 insertions(+), 5 deletions(-) create mode 100644 tests/test_installed_python_controls.py create mode 100644 tools/installed_python_controls.py diff --git a/tests/test_installed_python_controls.py b/tests/test_installed_python_controls.py new file mode 100644 index 0000000..ad837ec --- /dev/null +++ b/tests/test_installed_python_controls.py @@ -0,0 +1,58 @@ +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.assertGreater(counts["valid"], 0) + self.assertGreater(counts["invalid"], 0) + + 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) diff --git a/tools/installed_python_controls.py b/tools/installed_python_controls.py new file mode 100644 index 0000000..d7dcf0b --- /dev/null +++ b/tools/installed_python_controls.py @@ -0,0 +1,55 @@ +"""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 + + +def check_controls(controls, validator, factory, validation_error): + counts = {} + for kind in ("valid", "invalid"): + cases = controls["fixtures"][kind] + if not cases: + raise AssertionError(f"missing {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() diff --git a/tools/smoke_wheel.py b/tools/smoke_wheel.py index d8de74d..24d633a 100644 --- a/tools/smoke_wheel.py +++ b/tools/smoke_wheel.py @@ -1,11 +1,14 @@ """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: @@ -13,19 +16,35 @@ def main() -> int: 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 From cb6c0d76bfdb4ac1b64152fe4b426ade54b1f19e Mon Sep 17 00:00:00 2001 From: Srinivasa Consultants <78673153+dinakarjs@users.noreply.github.com> Date: Sat, 31 Oct 2026 09:25:43 +0530 Subject: [PATCH 2/2] test(release): enforce the complete installed control inventory Prepared with OpenAI Codex. Signed-off-by: Srinivasa Dinakar Jainti --- tests/test_installed_python_controls.py | 13 +++++++++++-- tools/installed_python_controls.py | 25 +++++++++++++++++++++++-- 2 files changed, 34 insertions(+), 4 deletions(-) diff --git a/tests/test_installed_python_controls.py b/tests/test_installed_python_controls.py index ad837ec..710369b 100644 --- a/tests/test_installed_python_controls.py +++ b/tests/test_installed_python_controls.py @@ -31,8 +31,7 @@ def setUp(self): def test_real_controls(self): counts = controls_module.check_controls(self.controls, self.validator, self.factory, self.error) - self.assertGreater(counts["valid"], 0) - self.assertGreater(counts["invalid"], 0) + self.assertEqual(counts, {"valid": 6, "invalid": 7}) def test_accept_all_is_caught(self): with self.assertRaisesRegex(AssertionError, "invalid control admitted"): @@ -56,3 +55,13 @@ def test_empty_controls_are_caught(self): 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) diff --git a/tools/installed_python_controls.py b/tools/installed_python_controls.py index d7dcf0b..d197700 100644 --- a/tools/installed_python_controls.py +++ b/tools/installed_python_controls.py @@ -7,12 +7,33 @@ 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 not cases: - raise AssertionError(f"missing {kind} controls") + 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"])