| # Copyright 2026 The Fuchsia Authors. All rights reserved. | |
| # Use of this source code is governed by a BSD-style license that can be | |
| # found in the LICENSE file. | |
| """End-to-end tests for the plan-driven API over the fake proxy target.""" | |
| import json | |
| import tempfile | |
| import unittest | |
| from pathlib import Path | |
| from typing import Any | |
| from driver_lab.api import ( | |
| EXIT_ACTIVATION, | |
| EXIT_OPERATION, | |
| EXIT_PERMISSION, | |
| EXIT_STALE, | |
| EXIT_TRANSPORT, | |
| EXIT_UNSUPPORTED, | |
| DriverLab, | |
| DriverLabError, | |
| ) | |
| from driver_lab.consent import WRITE_WARNING, ConsentDecision | |
| from driver_lab.discovery import ( | |
| DEFAULT_PROXY_DRIVER_URL, | |
| FakeNodeDiscovery, | |
| FakeProxyActivator, | |
| NodeDescription, | |
| ) | |
| from driver_lab.evidence import EvidenceError | |
| from driver_lab.models import AccessClass, Decision, ReadGrant | |
| from driver_lab.permissions import save_grants | |
| from driver_lab.plans import PlanError, validate_plan | |
| from driver_lab.transport import ( | |
| FakeDirectTarget, | |
| FakeProxyTarget, | |
| OpenRejection, | |
| ProxyDescription, | |
| ResourceInfo, | |
| ) | |
| CTRL_DIGEST = "sha256:" + "ab" * 32 | |
| DESCRIPTION_DIGEST = "sha256:" + "ee" * 32 | |
| POLICY_DIGEST = "sha256:" + "ff" * 32 | |
| TARGET_SCOPE = "example-engineering-target" | |
| NODE_ID = "example-device" | |
| def make_description() -> ProxyDescription: | |
| return ProxyDescription( | |
| protocol_major=1, | |
| protocol_minor=0, | |
| proxy_generation=7, | |
| boot_id="boot-1", | |
| resource_digest=DESCRIPTION_DIGEST, | |
| policy_digest=POLICY_DIGEST, | |
| resources=( | |
| ResourceInfo( | |
| id=1, name="control", logical_size=0x100, digest=CTRL_DIGEST | |
| ), | |
| ), | |
| max_snapshot_items=64, | |
| audit_capacity=1024, | |
| ) | |
| def make_grant(**overrides: object) -> ReadGrant: | |
| fields: dict[str, Any] = dict( | |
| schema_version=1, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| resource_digest=CTRL_DIGEST, | |
| resource="control", | |
| offset=0x3C, | |
| width=4, | |
| access=AccessClass.READ_ONCE, | |
| decision=Decision.ALLOW, | |
| approved_at="2026-07-30T00:00:00Z", | |
| ) | |
| fields.update(overrides) | |
| return ReadGrant(**fields) | |
| def make_plan(**overrides: object) -> dict[str, Any]: | |
| plan: dict[str, Any] = { | |
| "schema_version": 1, | |
| "run_id": "run-1", | |
| "case_id": "case-1", | |
| "target": {"selector": "lab-target"}, | |
| "node": { | |
| "id": NODE_ID, | |
| "expected_resource_digest": DESCRIPTION_DIGEST, | |
| }, | |
| "access": {"mode": "proxy", "activation": "bind-unclaimed"}, | |
| "operations": [ | |
| {"kind": "mmio_read32", "resource": "control", "offset": "0x3c"}, | |
| ], | |
| } | |
| plan.update(overrides) | |
| return plan | |
| class RunPlanTest(unittest.IsolatedAsyncioTestCase): | |
| def setUp(self) -> None: | |
| self._dir = tempfile.TemporaryDirectory() | |
| self.addCleanup(self._dir.cleanup) | |
| base = Path(self._dir.name) | |
| self.grants_path = base / "grants.toml" | |
| self.evidence_root = base / "evidence" | |
| self.fake = FakeProxyTarget(make_description()) | |
| self.fake.set_value(1, 0x3C, 0xDEAD_BEEF) | |
| self.fake.set_value(1, 0x40, 0x1234_5678) | |
| self.lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| ) | |
| def read_manifest(self, evidence_dir: Path) -> dict[str, Any]: | |
| manifest = json.loads((evidence_dir / "manifest.json").read_text()) | |
| assert isinstance(manifest, dict) | |
| return manifest | |
| async def test_happy_read_produces_complete_evidence(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| result = await self.lab.run_plan(make_plan()) | |
| self.assertTrue(result.ok, result.failure) | |
| self.assertEqual(result.reads[0].value, 0xDEAD_BEEF) | |
| self.assertEqual(self.fake.sessions_opened, 1) | |
| manifest = self.read_manifest(result.evidence_dir) | |
| for name in ( | |
| "plan.requested.json", | |
| "plan.canonical.json", | |
| "access.capabilities.json", | |
| "target.description.json", | |
| "permission-resolution.json", | |
| "operations.jsonl", | |
| "target-audit.jsonl", | |
| ): | |
| self.assertIsInstance(manifest["files"][name], dict, name) | |
| # Artifacts a phase 1 read-only run cannot produce are marked | |
| # not applicable, never silently omitted. | |
| for name in ( | |
| "approval.json", | |
| "restoration.json", | |
| "proxy-access.jsonl", | |
| "serial.log", | |
| ): | |
| self.assertEqual(manifest["files"][name], "not_applicable", name) | |
| self.assertEqual(manifest["plan_digest"], result.plan_digest) | |
| audit_ops = [ | |
| json.loads(line)["operation"] | |
| for line in (result.evidence_dir / "target-audit.jsonl") | |
| .read_text() | |
| .splitlines() | |
| ] | |
| self.assertIn("open_session", audit_ops) | |
| self.assertIn("read32", audit_ops) | |
| # The session-closed entry is appended after the final drain (audit | |
| # is drained through the session, then the session closes), so it | |
| # is not part of this run's evidence. | |
| self.assertNotIn("session_closed", audit_ops) | |
| async def test_missing_grant_fails_closed_before_any_target_access( | |
| self, | |
| ) -> None: | |
| result = await self.lab.run_plan(make_plan()) | |
| self.assertEqual(result.exit_category, EXIT_PERMISSION) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| manifest = self.read_manifest(result.evidence_dir) | |
| self.assertEqual( | |
| manifest["files"]["operations.jsonl"], "not_applicable" | |
| ) | |
| self.assertIsInstance( | |
| manifest["files"]["permission-resolution.json"], dict | |
| ) | |
| async def test_persistent_deny_wins(self) -> None: | |
| save_grants(self.grants_path, [make_grant(decision=Decision.DENY)]) | |
| result = await self.lab.run_plan(make_plan()) | |
| self.assertEqual(result.exit_category, EXIT_PERMISSION) | |
| self.assertEqual(result.failure, "denied by persistent grant") | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_read_once_grant_does_not_cover_snapshot(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| plan = make_plan( | |
| operations=[ | |
| { | |
| "kind": "mmio_snapshot32", | |
| "items": [{"resource": "control", "offset": "0x3c"}], | |
| } | |
| ] | |
| ) | |
| result = await self.lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_PERMISSION) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_stale_resource_digest_fails_before_session(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| plan = make_plan() | |
| plan["node"] = dict( | |
| plan["node"], expected_resource_digest="sha256:" + "cd" * 32 | |
| ) | |
| result = await self.lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_STALE) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_unknown_resource_is_stale(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| plan = make_plan( | |
| operations=[ | |
| {"kind": "mmio_read32", "resource": "missing", "offset": "0x0"} | |
| ] | |
| ) | |
| result = await self.lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_STALE) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_session_rejection_maps_stale_and_activation(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| self.fake.reject_open = OpenRejection.STALE_PROXY_GENERATION | |
| result = await self.lab.run_plan(make_plan()) | |
| self.assertEqual(result.exit_category, EXIT_STALE) | |
| self.assertEqual(self.fake.sessions_opened, 0) | |
| self.fake.reject_open = OpenRejection.MUTATION_LEASE_HELD | |
| result = await self.lab.run_plan(make_plan(run_id="run-2")) | |
| self.assertEqual(result.exit_category, EXIT_ACTIVATION) | |
| async def test_backend_fault_still_drains_audit(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| self.fake.fail_at(1, 0x3C) | |
| result = await self.lab.run_plan(make_plan()) | |
| self.assertEqual(result.exit_category, EXIT_OPERATION) | |
| self.assertEqual(result.failure, "backend_fault") | |
| operations = [ | |
| json.loads(line) | |
| for line in (result.evidence_dir / "operations.jsonl") | |
| .read_text() | |
| .splitlines() | |
| ] | |
| self.assertEqual(operations[0]["error"], "backend_fault") | |
| audit_statuses = [ | |
| json.loads(line)["status"] | |
| for line in (result.evidence_dir / "target-audit.jsonl") | |
| .read_text() | |
| .splitlines() | |
| ] | |
| self.assertIn("backend_fault", audit_statuses) | |
| async def test_snapshot_happy_path(self) -> None: | |
| save_grants( | |
| self.grants_path, | |
| [ | |
| make_grant(access=AccessClass.SNAPSHOT), | |
| make_grant(access=AccessClass.SNAPSHOT, offset=0x40), | |
| ], | |
| ) | |
| plan = make_plan( | |
| operations=[ | |
| { | |
| "kind": "mmio_snapshot32", | |
| "items": [ | |
| {"resource": "control", "offset": "0x3c"}, | |
| {"resource": "control", "offset": "0x40"}, | |
| ], | |
| } | |
| ] | |
| ) | |
| result = await self.lab.run_plan(plan) | |
| self.assertTrue(result.ok, result.failure) | |
| [row] = [ | |
| json.loads(line) | |
| for line in (result.evidence_dir / "operations.jsonl") | |
| .read_text() | |
| .splitlines() | |
| ] | |
| self.assertTrue(row["complete"]) | |
| self.assertEqual( | |
| [item["value"] for item in row["results"]], | |
| [0xDEAD_BEEF, 0x1234_5678], | |
| ) | |
| async def test_evidence_directory_is_never_reused(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| await self.lab.run_plan(make_plan()) | |
| with self.assertRaises(EvidenceError): | |
| await self.lab.run_plan(make_plan()) | |
| async def test_direct_mode_is_unsupported_not_substituted(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| result = await self.lab.run_plan(make_plan(access={"mode": "direct"})) | |
| self.assertEqual(result.exit_category, EXIT_UNSUPPORTED) | |
| # A direct-mode plan must never silently execute over the proxy. | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_direct_mode_happy_path(self) -> None: | |
| fake_direct = FakeDirectTarget( | |
| node_id=NODE_ID, | |
| protocol_name="fuchsia.hardware.example/Device", | |
| boot_id="boot-1", | |
| ) | |
| fake_direct.register_method( | |
| "GetStatus", lambda args: {"status": "ok", "code": 0} | |
| ) | |
| direct_lab = DriverLab( | |
| fake_direct, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| ) | |
| plan = { | |
| "schema_version": 1, | |
| "run_id": "direct-run-1", | |
| "case_id": "direct-status-case", | |
| "target": {"selector": "lab-target", "expected_boot_id": "boot-1"}, | |
| "node": {"id": NODE_ID}, | |
| "access": {"mode": "direct"}, | |
| "operations": [ | |
| {"kind": "fidl_call", "method": "GetStatus", "args": {}} | |
| ], | |
| } | |
| result = await direct_lab.run_plan(plan) | |
| self.assertTrue(result.ok, result.failure) | |
| self.assertEqual(len(result.calls), 1) | |
| self.assertEqual(result.calls[0]["method"], "GetStatus") | |
| self.assertEqual( | |
| result.calls[0]["response"], {"status": "ok", "code": 0} | |
| ) | |
| self.assertEqual(fake_direct.sessions_opened, 1) | |
| manifest = self.read_manifest(result.evidence_dir) | |
| self.assertEqual( | |
| manifest["files"]["target-audit.jsonl"], "not_applicable" | |
| ) | |
| self.assertEqual( | |
| manifest["files"]["permission-resolution.json"], "not_applicable" | |
| ) | |
| capabilities = json.loads( | |
| (result.evidence_dir / "access.capabilities.json").read_text() | |
| ) | |
| self.assertEqual(capabilities["mode"], "direct") | |
| self.assertFalse(capabilities["target_policy"]) | |
| self.assertFalse(capabilities["target_audit"]) | |
| target_desc = json.loads( | |
| (result.evidence_dir / "target.description.json").read_text() | |
| ) | |
| self.assertEqual(target_desc["node_id"], NODE_ID) | |
| self.assertEqual( | |
| target_desc["protocol_name"], "fuchsia.hardware.example/Device" | |
| ) | |
| ops_lines = ( | |
| (result.evidence_dir / "operations.jsonl").read_text().splitlines() | |
| ) | |
| self.assertEqual(len(ops_lines), 1) | |
| row = json.loads(ops_lines[0]) | |
| self.assertEqual(row["method"], "GetStatus") | |
| self.assertEqual(row["response"], {"status": "ok", "code": 0}) | |
| async def test_direct_mode_rejects_private_mmio(self) -> None: | |
| fake_direct = FakeDirectTarget() | |
| direct_lab = DriverLab( | |
| fake_direct, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| ) | |
| plan = { | |
| "schema_version": 1, | |
| "run_id": "direct-run-mmio", | |
| "case_id": "case-mmio", | |
| "target": {"selector": "lab-target"}, | |
| "node": {"id": NODE_ID}, | |
| "access": {"mode": "direct"}, | |
| "operations": [ | |
| {"kind": "mmio_read32", "resource": "control", "offset": 0x3C} | |
| ], | |
| } | |
| result = await direct_lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_UNSUPPORTED) | |
| self.assertEqual( | |
| result.failure, "direct mode does not provide private MMIO access" | |
| ) | |
| self.assertEqual(fake_direct.sessions_opened, 0) | |
| async def test_direct_mode_rejects_target_audit_requirement(self) -> None: | |
| fake_direct = FakeDirectTarget() | |
| direct_lab = DriverLab( | |
| fake_direct, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| ) | |
| plan = { | |
| "schema_version": 1, | |
| "run_id": "direct-run-audit", | |
| "case_id": "case-audit", | |
| "target": {"selector": "lab-target"}, | |
| "node": {"id": NODE_ID}, | |
| "access": {"mode": "direct", "requires_target_audit": True}, | |
| "operations": [{"kind": "fidl_call", "method": "GetStatus"}], | |
| } | |
| result = await direct_lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_UNSUPPORTED) | |
| self.assertIn("target_audit", str(result.failure)) | |
| self.assertEqual(fake_direct.sessions_opened, 0) | |
| async def test_proxy_mode_rejects_direct_transport(self) -> None: | |
| fake_direct = FakeDirectTarget() | |
| direct_lab = DriverLab( | |
| fake_direct, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| ) | |
| plan = make_plan() | |
| result = await direct_lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_UNSUPPORTED) | |
| self.assertEqual( | |
| result.failure, "transport does not support proxy mode" | |
| ) | |
| async def test_direct_mode_no_fallback_on_connection_failure(self) -> None: | |
| fake_direct = FakeDirectTarget() | |
| fake_direct.fail_open = True | |
| direct_lab = DriverLab( | |
| fake_direct, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| ) | |
| plan = { | |
| "schema_version": 1, | |
| "run_id": "direct-fail-open", | |
| "case_id": "case-fail", | |
| "target": {"selector": "lab-target"}, | |
| "node": {"id": NODE_ID}, | |
| "access": {"mode": "direct"}, | |
| "operations": [{"kind": "fidl_call", "method": "GetStatus"}], | |
| } | |
| result = await direct_lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_TRANSPORT) | |
| self.assertIn("direct connection failed", str(result.failure)) | |
| self.assertEqual(fake_direct.sessions_opened, 0) | |
| def test_direct_mode_refuses_takeover_activation(self) -> None: | |
| plan = { | |
| "schema_version": 1, | |
| "run_id": "direct-takeover", | |
| "case_id": "case-takeover", | |
| "target": {"selector": "lab-target"}, | |
| "node": {"id": NODE_ID}, | |
| "access": {"mode": "direct", "activation": "takeover"}, | |
| "operations": [{"kind": "fidl_call", "method": "GetStatus"}], | |
| } | |
| with self.assertRaises(PlanError): | |
| validate_plan(plan) | |
| async def test_typed_protocol_discovery(self) -> None: | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker="node-1", | |
| offers=("fuchsia.hardware.gpio/Device",), | |
| ), | |
| NodeDescription( | |
| moniker="node-2", | |
| offers=("fuchsia.hardware.i2c/Device",), | |
| ), | |
| ] | |
| ) | |
| gpio_nodes = await discovery.find_nodes_offering( | |
| "fuchsia.hardware.gpio/Device" | |
| ) | |
| self.assertEqual(len(gpio_nodes), 1) | |
| self.assertEqual(gpio_nodes[0].moniker, "node-1") | |
| self.assertEqual( | |
| gpio_nodes[0].offers, ("fuchsia.hardware.gpio/Device",) | |
| ) | |
| async def test_required_guarantee_fails_before_connection(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| plan = make_plan( | |
| access={ | |
| "mode": "proxy", | |
| "activation": "bind-unclaimed", | |
| "requires_target_local_timing": True, | |
| } | |
| ) | |
| result = await self.lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_UNSUPPORTED) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_expected_unclaimed_is_unverifiable_and_fails_closed( | |
| self, | |
| ) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| plan = make_plan() | |
| plan["node"] = dict(plan["node"], expected_unclaimed=True) | |
| result = await self.lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_UNSUPPORTED) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_expected_unclaimed_verified_with_discovery(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker=NODE_ID, | |
| bound_driver_url=None, | |
| ) | |
| ] | |
| ) | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| discovery=discovery, | |
| ) | |
| plan = make_plan() | |
| plan["node"] = dict(plan["node"], expected_unclaimed=True) | |
| result = await lab.run_plan(plan) | |
| self.assertTrue(result.ok, result.failure) | |
| self.assertEqual(self.fake.sessions_opened, 1) | |
| async def test_expected_unclaimed_fails_when_bound(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker=NODE_ID, | |
| bound_driver_url="fuchsia-boot:///driver#meta/driver.cm", | |
| ) | |
| ] | |
| ) | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| discovery=discovery, | |
| ) | |
| plan = make_plan() | |
| plan["node"] = dict(plan["node"], expected_unclaimed=True) | |
| result = await lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_ACTIVATION) | |
| self.assertIsNotNone(result.failure) | |
| assert result.failure is not None | |
| self.assertIn("managed takeover is phase 2", result.failure) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_expected_bound_fails_when_unclaimed(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker=NODE_ID, | |
| bound_driver_url=None, | |
| ) | |
| ] | |
| ) | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| discovery=discovery, | |
| ) | |
| plan = make_plan() | |
| plan["node"] = dict(plan["node"], expected_unclaimed=False) | |
| result = await lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_ACTIVATION) | |
| self.assertIsNotNone(result.failure) | |
| assert result.failure is not None | |
| self.assertIn("is unclaimed, expected bound driver", result.failure) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_expected_unclaimed_fails_when_node_not_found(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| discovery = FakeNodeDiscovery([]) | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| discovery=discovery, | |
| ) | |
| plan = make_plan() | |
| plan["node"] = dict(plan["node"], expected_unclaimed=True) | |
| result = await lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_ACTIVATION) | |
| self.assertIsNotNone(result.failure) | |
| assert result.failure is not None | |
| self.assertIn("not found during discovery", result.failure) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_list_nodes_and_describe_node(self) -> None: | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker=NODE_ID, | |
| bound_driver_url=None, | |
| ) | |
| ] | |
| ) | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| discovery=discovery, | |
| ) | |
| nodes = await lab.list_nodes() | |
| self.assertEqual(len(nodes), 1) | |
| self.assertEqual(nodes[0].moniker, NODE_ID) | |
| desc = await lab.describe_node(NODE_ID) | |
| self.assertIsNotNone(desc) | |
| assert desc is not None | |
| self.assertEqual(desc.moniker, NODE_ID) | |
| async def test_expected_boot_id_is_enforced(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| plan = make_plan() | |
| plan["target"] = dict(plan["target"], expected_boot_id="boot-2") | |
| result = await self.lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_STALE) | |
| self.assertEqual( | |
| result.failure, "expected_boot_id does not match target" | |
| ) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| plan = make_plan(run_id="run-2") | |
| plan["target"] = dict(plan["target"], expected_boot_id="boot-1") | |
| result = await self.lab.run_plan(plan) | |
| self.assertTrue(result.ok, result.failure) | |
| async def test_audit_entries_carry_run_and_instance_identity(self) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| result = await self.lab.run_plan(make_plan()) | |
| self.assertTrue(result.ok, result.failure) | |
| rows = [ | |
| json.loads(line) | |
| for line in (result.evidence_dir / "target-audit.jsonl") | |
| .read_text() | |
| .splitlines() | |
| ] | |
| opens = [row for row in rows if row["operation"] == "open_session"] | |
| self.assertEqual(opens[0]["run_id"], "run-1") | |
| for row in rows: | |
| self.assertEqual(row["boot_id"], "boot-1") | |
| self.assertEqual(row["proxy_generation"], 7) | |
| async def test_write32_execution_with_readback(self) -> None: | |
| class Prompt: | |
| def __init__(self) -> None: | |
| self.warnings: list[str] = [] | |
| async def request_consent( | |
| self, req: Any, warning: str | |
| ) -> ConsentDecision: | |
| self.warnings.append(warning) | |
| return ConsentDecision.ALLOW_ONCE | |
| prompt = Prompt() | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| consent=prompt, | |
| ) | |
| plan = make_plan( | |
| operations=[ | |
| { | |
| "kind": "mmio_write32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "value": 0xCAFE_BABE, | |
| "readback": True, | |
| } | |
| ] | |
| ) | |
| result = await lab.run_plan(plan) | |
| self.assertTrue(result.ok, result.failure) | |
| self.assertEqual(len(result.writes), 1) | |
| self.assertEqual(result.writes[0].value, 0xCAFE_BABE) | |
| self.assertEqual(result.writes[0].readback_value, 0xCAFE_BABE) | |
| self.assertEqual(self.fake.get_value(1, 0x3C), 0xCAFE_BABE) | |
| self.assertEqual(prompt.warnings, [WRITE_WARNING]) | |
| self.assertIsNone(self.fake.active_mutating_session) | |
| # Confirm audit draining captured the write. | |
| audit_ops = [ | |
| json.loads(line)["operation"] | |
| for line in (result.evidence_dir / "target-audit.jsonl") | |
| .read_text() | |
| .splitlines() | |
| ] | |
| self.assertIn("write32", audit_ops) | |
| async def test_write32_precondition_success_and_failure(self) -> None: | |
| class Prompt: | |
| async def request_consent( | |
| self, req: Any, warning: str | |
| ) -> ConsentDecision: | |
| return ConsentDecision.ALLOW_ONCE | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| consent=Prompt(), | |
| ) | |
| # 1. Matching precondition succeeds | |
| plan_good = make_plan( | |
| run_id="run-precond-good", | |
| operations=[ | |
| { | |
| "kind": "mmio_write32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "value": 0x1111_2222, | |
| "precondition": { | |
| "expected": 0xDEAD_BEEF, | |
| "mask": 0xFFFF_FFFF, | |
| }, | |
| } | |
| ], | |
| ) | |
| result_good = await lab.run_plan(plan_good) | |
| self.assertTrue(result_good.ok, result_good.failure) | |
| self.assertEqual(self.fake.get_value(1, 0x3C), 0x1111_2222) | |
| # 2. Mismatched precondition fails and does not mutate | |
| plan_bad = make_plan( | |
| run_id="run-precond-bad", | |
| operations=[ | |
| { | |
| "kind": "mmio_write32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "value": 0x9999_9999, | |
| "precondition": { | |
| "expected": 0x0000_0000, | |
| "mask": 0xFFFF_FFFF, | |
| }, | |
| } | |
| ], | |
| ) | |
| result_bad = await lab.run_plan(plan_bad) | |
| self.assertEqual(result_bad.exit_category, EXIT_OPERATION) | |
| self.assertEqual(result_bad.failure, "precondition_failed") | |
| self.assertEqual(self.fake.get_value(1, 0x3C), 0x1111_2222) | |
| async def test_poll32_execution_success_and_timeout(self) -> None: | |
| class Prompt: | |
| async def request_consent( | |
| self, req: Any, warning: str | |
| ) -> ConsentDecision: | |
| return ConsentDecision.ALLOW_ONCE | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| consent=Prompt(), | |
| ) | |
| # Success | |
| plan_good = make_plan( | |
| run_id="run-poll-good", | |
| operations=[ | |
| { | |
| "kind": "mmio_poll32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "expected": 0xDEAD_BEEF, | |
| "mask": 0xFFFF_FFFF, | |
| "interval_ns": 1000, | |
| "timeout_ns": 100_000, | |
| } | |
| ], | |
| ) | |
| result_good = await lab.run_plan(plan_good) | |
| self.assertTrue(result_good.ok, result_good.failure) | |
| self.assertEqual(result_good.polls[0].value, 0xDEAD_BEEF) | |
| # Timeout | |
| plan_timeout = make_plan( | |
| run_id="run-poll-timeout", | |
| operations=[ | |
| { | |
| "kind": "mmio_poll32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "expected": 0x1234_0000, | |
| "mask": 0xFFFF_FFFF, | |
| "interval_ns": 1000, | |
| "timeout_ns": 10_000, | |
| } | |
| ], | |
| ) | |
| result_timeout = await lab.run_plan(plan_timeout) | |
| self.assertEqual(result_timeout.exit_category, EXIT_OPERATION) | |
| self.assertEqual(result_timeout.failure, "timeout") | |
| async def test_sequence_execution(self) -> None: | |
| class Prompt: | |
| async def request_consent( | |
| self, req: Any, warning: str | |
| ) -> ConsentDecision: | |
| return ConsentDecision.ALLOW_ONCE | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| consent=Prompt(), | |
| ) | |
| plan = make_plan( | |
| operations=[ | |
| { | |
| "kind": "sequence", | |
| "items": [ | |
| { | |
| "kind": "mmio_read32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| }, | |
| {"kind": "delay_ns", "duration_ns": 1000}, | |
| {"kind": "barrier", "variant": "memory"}, | |
| { | |
| "kind": "mmio_write32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "value": 0x5555_AAAA, | |
| "readback": True, | |
| }, | |
| { | |
| "kind": "mmio_poll32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "expected": 0x5555_AAAA, | |
| "interval_ns": 1000, | |
| "timeout_ns": 100_000, | |
| }, | |
| ], | |
| } | |
| ] | |
| ) | |
| result = await lab.run_plan(plan) | |
| self.assertTrue(result.ok, result.failure) | |
| self.assertEqual(len(result.sequences), 1) | |
| seq = result.sequences[0] | |
| self.assertTrue(seq.complete) | |
| self.assertEqual(len(seq.results), 5) | |
| self.assertEqual(seq.results[0].value, 0xDEAD_BEEF) | |
| self.assertEqual(seq.results[3].readback_value, 0x5555_AAAA) | |
| self.assertEqual(seq.results[4].value, 0x5555_AAAA) | |
| self.assertEqual(self.fake.get_value(1, 0x3C), 0x5555_AAAA) | |
| async def test_mutation_lease_contention_rejected(self) -> None: | |
| class Prompt: | |
| async def request_consent( | |
| self, req: Any, warning: str | |
| ) -> ConsentDecision: | |
| return ConsentDecision.ALLOW_ONCE | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| consent=Prompt(), | |
| ) | |
| self.fake.active_mutating_session = 999 | |
| plan = make_plan( | |
| operations=[ | |
| { | |
| "kind": "mmio_write32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "value": 0x1234, | |
| } | |
| ] | |
| ) | |
| result = await lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_ACTIVATION) | |
| self.assertEqual( | |
| result.failure, "session rejected: mutation_lease_held" | |
| ) | |
| async def test_unattended_mutation_fails_closed(self) -> None: | |
| plan = make_plan( | |
| operations=[ | |
| { | |
| "kind": "mmio_write32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "value": 0x1234, | |
| } | |
| ] | |
| ) | |
| result = await self.lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_PERMISSION) | |
| self.assertEqual( | |
| result.failure, | |
| "consent required; unattended operation fails closed", | |
| ) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_interactive_write_always_allow_rejected(self) -> None: | |
| class Prompt: | |
| async def request_consent( | |
| self, req: Any, warning: str | |
| ) -> ConsentDecision: | |
| return ConsentDecision.ALWAYS_ALLOW | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| consent=Prompt(), | |
| ) | |
| plan = make_plan( | |
| operations=[ | |
| { | |
| "kind": "mmio_write32", | |
| "resource": "control", | |
| "offset": "0x3c", | |
| "value": 0x1234, | |
| } | |
| ] | |
| ) | |
| result = await lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_PERMISSION) | |
| self.assertEqual( | |
| result.failure, "persistent write grants are not supported" | |
| ) | |
| self.assertFalse(self.grants_path.exists()) | |
| self.assertEqual(self.fake.open_attempts, 0) | |
| async def test_direct_mode_rejects_mutation_and_sequence(self) -> None: | |
| fake_direct = FakeDirectTarget() | |
| direct_lab = DriverLab( | |
| fake_direct, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| ) | |
| plan_write = { | |
| "schema_version": 1, | |
| "run_id": "direct-write", | |
| "case_id": "case-write", | |
| "target": {"selector": "lab-target"}, | |
| "node": {"id": NODE_ID}, | |
| "access": {"mode": "direct"}, | |
| "operations": [ | |
| { | |
| "kind": "mmio_write32", | |
| "resource": "control", | |
| "offset": 0x3C, | |
| "value": 0x1234, | |
| } | |
| ], | |
| } | |
| result_write = await direct_lab.run_plan(plan_write) | |
| self.assertEqual(result_write.exit_category, EXIT_UNSUPPORTED) | |
| self.assertEqual( | |
| result_write.failure, | |
| "direct mode does not provide private MMIO access", | |
| ) | |
| plan_seq = { | |
| "schema_version": 1, | |
| "run_id": "direct-seq", | |
| "case_id": "case-seq", | |
| "target": {"selector": "lab-target"}, | |
| "node": {"id": NODE_ID}, | |
| "access": {"mode": "direct"}, | |
| "operations": [ | |
| { | |
| "kind": "sequence", | |
| "items": [{"kind": "delay_ns", "duration_ns": 100}], | |
| } | |
| ], | |
| } | |
| result_seq = await direct_lab.run_plan(plan_seq) | |
| self.assertEqual(result_seq.exit_category, EXIT_UNSUPPORTED) | |
| self.assertEqual( | |
| result_seq.failure, | |
| "direct mode does not provide private MMIO access", | |
| ) | |
| async def test_bind_unclaimed_activates_and_verifies_teardown( | |
| self, | |
| ) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker=NODE_ID, | |
| bound_driver_url=None, | |
| ) | |
| ] | |
| ) | |
| activator = FakeProxyActivator(discovery) | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| discovery=discovery, | |
| activator=activator, | |
| ) | |
| plan = make_plan() | |
| result = await lab.run_plan(plan) | |
| self.assertTrue(result.ok, result.failure) | |
| self.assertEqual( | |
| activator.bind_calls, [(NODE_ID, DEFAULT_PROXY_DRIVER_URL)] | |
| ) | |
| self.assertEqual( | |
| activator.end_calls, [(NODE_ID, DEFAULT_PROXY_DRIVER_URL)] | |
| ) | |
| # Verify node is again unclaimed in discovery | |
| node = await discovery.describe_node(NODE_ID) | |
| self.assertIsNotNone(node) | |
| assert node is not None | |
| self.assertTrue(node.is_unclaimed) | |
| # Verify node.before.json and node.after.json evidence files | |
| before_file = result.evidence_dir / "node.before.json" | |
| after_file = result.evidence_dir / "node.after.json" | |
| self.assertTrue(before_file.exists()) | |
| self.assertTrue(after_file.exists()) | |
| before_data = json.loads(before_file.read_text()) | |
| after_data = json.loads(after_file.read_text()) | |
| self.assertIsNone(before_data["bound_driver_url"]) | |
| self.assertIsNone(after_data["bound_driver_url"]) | |
| async def test_bind_unclaimed_verified_teardown_failure_when_still_bound( | |
| self, | |
| ) -> None: | |
| save_grants(self.grants_path, [make_grant()]) | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker=NODE_ID, | |
| bound_driver_url=None, | |
| ) | |
| ] | |
| ) | |
| class BrokenActivator(FakeProxyActivator): | |
| async def end_proxy( | |
| self, | |
| node_id: str, | |
| driver_url: str = DEFAULT_PROXY_DRIVER_URL, | |
| ) -> None: | |
| # Intentionally leave node bound! | |
| self.end_calls.append((node_id, driver_url)) | |
| activator = BrokenActivator(discovery) | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| discovery=discovery, | |
| activator=activator, | |
| ) | |
| plan = make_plan() | |
| result = await lab.run_plan(plan) | |
| self.assertEqual(result.exit_category, EXIT_ACTIVATION) | |
| self.assertIsNotNone(result.failure) | |
| assert result.failure is not None | |
| self.assertIn("verified teardown failed", result.failure) | |
| async def test_bind_proxy_and_end_proxy_explicit_methods(self) -> None: | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker=NODE_ID, | |
| bound_driver_url=None, | |
| ) | |
| ] | |
| ) | |
| activator = FakeProxyActivator(discovery) | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| discovery=discovery, | |
| activator=activator, | |
| ) | |
| await lab.bind_proxy(NODE_ID) | |
| node = await discovery.describe_node(NODE_ID) | |
| self.assertIsNotNone(node) | |
| assert node is not None | |
| self.assertFalse(node.is_unclaimed) | |
| self.assertEqual(node.bound_driver_url, DEFAULT_PROXY_DRIVER_URL) | |
| await lab.end_proxy(NODE_ID) | |
| node_after = await discovery.describe_node(NODE_ID) | |
| self.assertIsNotNone(node_after) | |
| assert node_after is not None | |
| self.assertTrue(node_after.is_unclaimed) | |
| async def test_bind_proxy_raises_when_no_activator(self) -> None: | |
| lab = DriverLab( | |
| self.fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope=TARGET_SCOPE, | |
| node_id=NODE_ID, | |
| ) | |
| with self.assertRaises(DriverLabError): | |
| await lab.bind_proxy(NODE_ID) | |
| with self.assertRaises(DriverLabError): | |
| await lab.end_proxy(NODE_ID) | |
| if __name__ == "__main__": | |
| unittest.main() |