diff --git a/tests/test_installed_python_controls.py b/tests/test_installed_python_controls.py new file mode 100644 index 0000000..710369b --- /dev/null +++ b/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) diff --git a/tools/installed_python_controls.py b/tools/installed_python_controls.py new file mode 100644 index 0000000..d197700 --- /dev/null +++ b/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() 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