| # 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. | |
| """Tests for the driver-lab CLI.""" | |
| import asyncio | |
| import contextlib | |
| import io | |
| import json | |
| import tempfile | |
| import unittest | |
| from pathlib import Path | |
| from typing import Any | |
| from unittest import mock | |
| from driver_lab import cli | |
| from driver_lab.api import EXIT_ACTIVATION | |
| from driver_lab.discovery import ( | |
| FakeNodeDiscovery, | |
| FakeProxyActivator, | |
| NodeDescription, | |
| ) | |
| from driver_lab.models import AccessClass, Decision, ReadGrant | |
| from driver_lab.permissions import load_grants, save_grants | |
| from driver_lab.plans import plan_digest | |
| from driver_lab.transport import FakeProxyTarget, ProxyDescription, ResourceInfo | |
| DIGEST = "sha256:" + "ab" * 32 | |
| def make_grant(**overrides: object) -> ReadGrant: | |
| fields: dict[str, Any] = dict( | |
| schema_version=1, | |
| target_scope="scope", | |
| node_id="node", | |
| resource_digest=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() -> dict[str, Any]: | |
| return { | |
| "schema_version": 1, | |
| "run_id": "run-1", | |
| "case_id": "case-1", | |
| "target": {"selector": "lab-target"}, | |
| "node": {"id": "node"}, | |
| "access": {"mode": "proxy", "activation": "bind-unclaimed"}, | |
| "operations": [ | |
| {"kind": "mmio_read32", "resource": "control", "offset": "0x3c"}, | |
| ], | |
| } | |
| class CliTest(unittest.TestCase): | |
| def setUp(self) -> None: | |
| self._dir = tempfile.TemporaryDirectory() | |
| self.addCleanup(self._dir.cleanup) | |
| self.base = Path(self._dir.name) | |
| self.grants = self.base / "grants.toml" | |
| def run_cli(self, *argv: str) -> tuple[int, dict[str, Any], str]: | |
| stdout = io.StringIO() | |
| stderr = io.StringIO() | |
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr( | |
| stderr | |
| ): | |
| code = cli.main(list(argv)) | |
| payload = json.loads(stdout.getvalue()) if stdout.getvalue() else {} | |
| return code, payload, stderr.getvalue() | |
| def test_permissions_list(self) -> None: | |
| save_grants(self.grants, [make_grant()]) | |
| code, payload, _ = self.run_cli( | |
| "permissions", "list", "--grants", str(self.grants) | |
| ) | |
| self.assertEqual(code, 0) | |
| [row] = payload["grants"] | |
| self.assertEqual(row["offset"], 0x3C) | |
| self.assertTrue(row["grant_id"].startswith("grant-")) | |
| def test_permissions_revoke(self) -> None: | |
| save_grants(self.grants, [make_grant()]) | |
| grant_id = make_grant().grant_id | |
| code, payload, _ = self.run_cli( | |
| "permissions", | |
| "revoke", | |
| "--grants", | |
| str(self.grants), | |
| "--grant-id", | |
| grant_id, | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["revoked"], grant_id) | |
| self.assertEqual(load_grants(self.grants), []) | |
| code, _, stderr = self.run_cli( | |
| "permissions", | |
| "revoke", | |
| "--grants", | |
| str(self.grants), | |
| "--grant-id", | |
| grant_id, | |
| ) | |
| self.assertEqual(code, 2) | |
| self.assertIn("no grant", stderr) | |
| def test_permissions_explain(self) -> None: | |
| save_grants(self.grants, [make_grant()]) | |
| code, payload, _ = self.run_cli( | |
| "permissions", | |
| "explain", | |
| "--grants", | |
| str(self.grants), | |
| "--target-scope", | |
| "scope", | |
| "--node-id", | |
| "node", | |
| "--resource-digest", | |
| DIGEST, | |
| "--resource", | |
| "control", | |
| "--offset", | |
| "0x3c", | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["outcome"], "allowed") | |
| code, payload, _ = self.run_cli( | |
| "permissions", | |
| "explain", | |
| "--grants", | |
| str(self.grants), | |
| "--target-scope", | |
| "scope", | |
| "--node-id", | |
| "node", | |
| "--resource-digest", | |
| DIGEST, | |
| "--resource", | |
| "control", | |
| "--offset", | |
| "0x40", | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["outcome"], "undecided") | |
| def test_plan_digest_matches_library(self) -> None: | |
| plan_path = self.base / "plan.json" | |
| plan_path.write_text(json.dumps(make_plan())) | |
| code, payload, _ = self.run_cli( | |
| "plan", "digest", "--plan", str(plan_path) | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["digest"], plan_digest(make_plan())) | |
| def test_plan_validate_rejects_takeover(self) -> None: | |
| plan = make_plan() | |
| plan["access"] = {"mode": "proxy", "activation": "takeover"} | |
| plan_path = self.base / "plan.json" | |
| plan_path.write_text(json.dumps(plan)) | |
| code, _, stderr = self.run_cli( | |
| "plan", "validate", "--plan", str(plan_path) | |
| ) | |
| self.assertEqual(code, 2) | |
| self.assertIn("phase 2", stderr) | |
| def test_malformed_grants_file_is_an_error(self) -> None: | |
| self.grants.write_text("not toml [") | |
| code, _, stderr = self.run_cli( | |
| "permissions", "list", "--grants", str(self.grants) | |
| ) | |
| self.assertEqual(code, 2) | |
| self.assertIn("error", stderr) | |
| def test_permissions_add_persists_and_prints_exact_rule(self) -> None: | |
| code, payload, _ = self.run_cli( | |
| "permissions", | |
| "add", | |
| "--grants", | |
| str(self.grants), | |
| "--target-scope", | |
| "scope", | |
| "--node-id", | |
| "node", | |
| "--resource-digest", | |
| DIGEST, | |
| "--resource", | |
| "control", | |
| "--offset", | |
| "0x3c", | |
| "--decision", | |
| "allow", | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["added"]["offset"], 0x3C) | |
| self.assertEqual(payload["added"]["decision"], "allow") | |
| self.assertEqual(payload["added"]["approval_source"], "cli") | |
| [stored] = load_grants(self.grants) | |
| self.assertEqual(stored.grant_id, payload["added"]["grant_id"]) | |
| # A duplicate exact rule is rejected, not silently merged. | |
| code, _, stderr = self.run_cli( | |
| "permissions", | |
| "add", | |
| "--grants", | |
| str(self.grants), | |
| "--target-scope", | |
| "scope", | |
| "--node-id", | |
| "node", | |
| "--resource-digest", | |
| DIGEST, | |
| "--resource", | |
| "control", | |
| "--offset", | |
| "0x3c", | |
| "--decision", | |
| "deny", | |
| ) | |
| self.assertEqual(code, 2) | |
| self.assertIn("error", stderr) | |
| def _run_fixture(self) -> FakeProxyTarget: | |
| description = ProxyDescription( | |
| protocol_major=1, | |
| protocol_minor=0, | |
| proxy_generation=7, | |
| boot_id="boot-1", | |
| resource_digest="sha256:" + "ee" * 32, | |
| policy_digest="sha256:" + "ff" * 32, | |
| resources=( | |
| ResourceInfo( | |
| id=1, name="control", logical_size=0x100, digest=DIGEST | |
| ), | |
| ), | |
| max_snapshot_items=64, | |
| audit_capacity=1024, | |
| ) | |
| fake = FakeProxyTarget(description) | |
| fake.set_value(1, 0x3C, 0xDEAD_BEEF) | |
| return fake | |
| def run_cli_run( | |
| self, fake: FakeProxyTarget, run_id: str | |
| ) -> tuple[int, dict[str, Any], str]: | |
| plan = make_plan() | |
| plan["run_id"] = run_id | |
| plan_path = self.base / f"{run_id}.json" | |
| plan_path.write_text(json.dumps(plan)) | |
| with ( | |
| mock.patch.object(cli, "_connect_transport", return_value=fake), | |
| mock.patch.object(cli, "_connect_discovery", return_value=None), | |
| mock.patch.object(cli, "_connect_activator", return_value=None), | |
| ): | |
| return self.run_cli( | |
| "run", | |
| "--plan", | |
| str(plan_path), | |
| "--evidence-dir", | |
| str(self.base / "evidence"), | |
| "--grants", | |
| str(self.grants), | |
| "--target-scope", | |
| "scope", | |
| "--node-id", | |
| "node", | |
| "--moniker", | |
| "bootstrap/driver-lab", | |
| ) | |
| def test_run_executes_plan_to_finalized_evidence(self) -> None: | |
| save_grants(self.grants, [make_grant()]) | |
| fake = self._run_fixture() | |
| code, payload, _ = self.run_cli_run(fake, "run-1") | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["reads"][0]["value"], 0xDEAD_BEEF) | |
| manifest_path = Path(payload["evidence_dir"]) / "manifest.json" | |
| self.assertTrue(manifest_path.exists()) | |
| def test_run_unattended_fails_closed(self) -> None: | |
| fake = self._run_fixture() | |
| code, payload, _ = self.run_cli_run(fake, "run-1") | |
| self.assertEqual(code, 2) | |
| self.assertIn("fails closed", payload["failure"]) | |
| self.assertEqual(fake.open_attempts, 0) | |
| def test_run_evidence_reuse_maps_to_evidence_category(self) -> None: | |
| save_grants(self.grants, [make_grant()]) | |
| fake = self._run_fixture() | |
| code, _, _ = self.run_cli_run(fake, "run-1") | |
| self.assertEqual(code, 0) | |
| code, _, stderr = self.run_cli_run(fake, "run-1") | |
| self.assertEqual(code, 7) | |
| self.assertIn("no reuse", stderr) | |
| def test_permissions_explain_plan(self) -> None: | |
| save_grants(self.grants, [make_grant()]) | |
| plan = make_plan() | |
| plan["target"]["selector"] = "scope" | |
| plan["node"]["expected_resource_digest"] = DIGEST | |
| plan_path = self.base / "explain_plan.json" | |
| plan_path.write_text(json.dumps(plan)) | |
| # With grant present: | |
| code, payload, _ = self.run_cli( | |
| "permissions", | |
| "explain", | |
| "--grants", | |
| str(self.grants), | |
| "--plan", | |
| str(plan_path), | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertTrue(payload["all_allowed"]) | |
| self.assertEqual(len(payload["resolutions"]), 1) | |
| self.assertEqual(payload["resolutions"][0]["outcome"], "allowed") | |
| # With no grants: | |
| save_grants(self.grants, []) | |
| code, payload, _ = self.run_cli( | |
| "permissions", | |
| "explain", | |
| "--grants", | |
| str(self.grants), | |
| "--plan", | |
| str(plan_path), | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertFalse(payload["all_allowed"]) | |
| self.assertEqual(payload["resolutions"][0]["outcome"], "undecided") | |
| def test_list_nodes(self) -> None: | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription(moniker="node-1", bound_driver_url=None), | |
| NodeDescription( | |
| moniker="node-2", bound_driver_url="fuchsia-boot:///driver" | |
| ), | |
| ] | |
| ) | |
| with mock.patch.object( | |
| cli, "_connect_discovery", return_value=discovery | |
| ): | |
| code, payload, _ = self.run_cli("list") | |
| self.assertEqual(code, 0) | |
| self.assertEqual(len(payload["nodes"]), 2) | |
| code, payload_unclaimed, _ = self.run_cli("list", "--unclaimed") | |
| self.assertEqual(code, 0) | |
| self.assertEqual(len(payload_unclaimed["nodes"]), 1) | |
| self.assertEqual(payload_unclaimed["nodes"][0]["moniker"], "node-1") | |
| def test_describe_node(self) -> None: | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker="node-1", | |
| bound_driver_url=None, | |
| offers=("fuchsia.examples.Echo",), | |
| ) | |
| ] | |
| ) | |
| with mock.patch.object( | |
| cli, "_connect_discovery", return_value=discovery | |
| ): | |
| code, payload, _ = self.run_cli("describe", "--node", "node-1") | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["node"]["moniker"], "node-1") | |
| self.assertEqual( | |
| payload["node"]["offers"], ["fuchsia.examples.Echo"] | |
| ) | |
| code, _, stderr = self.run_cli("describe", "--node", "nonexistent") | |
| self.assertEqual(code, 2) | |
| self.assertIn("not found", stderr) | |
| def test_direct_mode_cli(self) -> None: | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker="node-1", | |
| bound_driver_url="fuchsia-boot:///driver", | |
| offers=("fuchsia.examples.Echo",), | |
| ) | |
| ] | |
| ) | |
| with mock.patch.object( | |
| cli, "_connect_discovery", return_value=discovery | |
| ): | |
| code, payload, _ = self.run_cli( | |
| "direct", | |
| "--node", | |
| "node-1", | |
| "--protocol", | |
| "fuchsia.examples.Echo", | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["status"], "verified") | |
| code, _, stderr = self.run_cli( | |
| "direct", | |
| "--node", | |
| "node-1", | |
| "--protocol", | |
| "fuchsia.hardware.other", | |
| ) | |
| self.assertEqual(code, 2) | |
| self.assertIn("does not offer protocol", stderr) | |
| def test_bind_proxy_and_end_proxy_cli(self) -> None: | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription(moniker="node-1", bound_driver_url=None), | |
| NodeDescription( | |
| moniker="node-2", bound_driver_url="fuchsia-boot:///driver" | |
| ), | |
| ] | |
| ) | |
| activator = FakeProxyActivator(discovery) | |
| with mock.patch.object( | |
| cli, "_connect_discovery", return_value=discovery | |
| ), mock.patch.object(cli, "_connect_activator", return_value=activator): | |
| # Cannot bind already bound node | |
| code, _, stderr = self.run_cli("bind-proxy", "--node", "node-2") | |
| self.assertEqual(code, EXIT_ACTIVATION) | |
| self.assertIn("is not unclaimed", stderr) | |
| # Bind unclaimed node | |
| code, payload, _ = self.run_cli("bind-proxy", "--node", "node-1") | |
| self.assertEqual(code, 0) | |
| self.assertTrue(payload["bound"]) | |
| node = asyncio.run(discovery.describe_node("node-1")) | |
| self.assertIsNotNone(node) | |
| assert node is not None | |
| self.assertFalse(node.is_unclaimed) | |
| # End proxy access | |
| code, payload, _ = self.run_cli("end-proxy", "--node", "node-1") | |
| self.assertEqual(code, 0) | |
| self.assertTrue(payload["unclaimed"]) | |
| node = asyncio.run(discovery.describe_node("node-1")) | |
| self.assertIsNotNone(node) | |
| assert node is not None | |
| self.assertTrue(node.is_unclaimed) | |
| def test_run_with_automated_bind_and_verified_teardown(self) -> None: | |
| save_grants(self.grants, [make_grant()]) | |
| fake = self._run_fixture() | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker="node", | |
| bound_driver_url=None, | |
| ) | |
| ] | |
| ) | |
| activator = FakeProxyActivator(discovery) | |
| plan = make_plan() | |
| plan_path = self.base / "run_bind.json" | |
| plan_path.write_text(json.dumps(plan)) | |
| with mock.patch.object( | |
| cli, "_connect_transport", return_value=fake | |
| ), mock.patch.object( | |
| cli, "_connect_discovery", return_value=discovery | |
| ), mock.patch.object( | |
| cli, "_connect_activator", return_value=activator | |
| ): | |
| code, payload, _ = self.run_cli( | |
| "run", | |
| "--plan", | |
| str(plan_path), | |
| "--evidence-dir", | |
| str(self.base / "evidence_bind"), | |
| "--grants", | |
| str(self.grants), | |
| "--target-scope", | |
| "scope", | |
| "--node-id", | |
| "node", | |
| "--moniker", | |
| "bootstrap/driver-lab", | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["reads"][0]["value"], 0xDEAD_BEEF) | |
| self.assertEqual(len(activator.bind_calls), 1) | |
| self.assertEqual(len(activator.end_calls), 1) | |
| # Verified teardown left node unclaimed | |
| node = asyncio.run(discovery.describe_node("node")) | |
| self.assertIsNotNone(node) | |
| assert node is not None | |
| self.assertTrue(node.is_unclaimed) | |
| def test_list_debug_capable_filters_active_drivers_exposing_service( | |
| self, | |
| ) -> None: | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker="dev.sys.platform.unclaimed", | |
| bound_driver_url=None, | |
| offers=("fuchsia.driver.lab.Service",), | |
| ), | |
| NodeDescription( | |
| moniker="dev.sys.platform.normal-driver", | |
| bound_driver_url="fuchsia-pkg://fuchsia.com/normal#meta/normal.cm", | |
| offers=("fuchsia.hardware.gpio.Service",), | |
| ), | |
| NodeDescription( | |
| moniker="dev.sys.platform.sample-device", | |
| bound_driver_url="fuchsia-pkg://fuchsia.com/lab_root#meta/lab_sample_driver.cm", | |
| offers=("fuchsia.driver.lab.Service",), | |
| ), | |
| ] | |
| ) | |
| with mock.patch.object( | |
| cli, "_connect_discovery", return_value=discovery | |
| ): | |
| code, payload, _ = self.run_cli("list", "--debug-capable") | |
| self.assertEqual(code, 0) | |
| self.assertEqual(len(payload["nodes"]), 1) | |
| self.assertEqual( | |
| payload["nodes"][0]["moniker"], | |
| "dev.sys.platform.sample-device", | |
| ) | |
| self.assertTrue(payload["nodes"][0]["debug_capable"]) | |
| def test_describe_with_moniker_fetches_embedded_proxy_description( | |
| self, | |
| ) -> None: | |
| fake = self._run_fixture() | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker="dev.sys.platform.sample-device", | |
| bound_driver_url="fuchsia-pkg://fuchsia.com/lab_root#meta/lab_sample_driver.cm", | |
| offers=("fuchsia.driver.lab.Service",), | |
| ) | |
| ] | |
| ) | |
| with mock.patch.object( | |
| cli, "_connect_transport", return_value=fake | |
| ), mock.patch.object(cli, "_connect_discovery", return_value=discovery): | |
| code, payload, _ = self.run_cli( | |
| "describe", | |
| "--moniker", | |
| "dev.sys.platform.sample-device", | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual( | |
| payload["moniker"], "dev.sys.platform.sample-device" | |
| ) | |
| self.assertEqual(payload["description"]["boot_id"], "boot-1") | |
| self.assertEqual( | |
| payload["node"]["bound_driver_url"], | |
| "fuchsia-pkg://fuchsia.com/lab_root#meta/lab_sample_driver.cm", | |
| ) | |
| def test_inspect_moniker_reads_register_and_drains_audit(self) -> None: | |
| fake = self._run_fixture() | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker="dev.sys.platform.sample-device", | |
| bound_driver_url="fuchsia-pkg://fuchsia.com/lab_root#meta/lab_sample_driver.cm", | |
| offers=("fuchsia.driver.lab.Service",), | |
| ) | |
| ] | |
| ) | |
| with mock.patch.object( | |
| cli, "_connect_transport", return_value=fake | |
| ), mock.patch.object(cli, "_connect_discovery", return_value=discovery): | |
| code, payload, _ = self.run_cli( | |
| "inspect", | |
| "--moniker", | |
| "dev.sys.platform.sample-device", | |
| "--resource", | |
| "control", | |
| "--offset", | |
| "0x3c", | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual( | |
| payload["driver_url"], | |
| "fuchsia-pkg://fuchsia.com/lab_root#meta/lab_sample_driver.cm", | |
| ) | |
| self.assertEqual(payload["read"]["value"], 0xDEAD_BEEF) | |
| self.assertTrue(len(payload["audit"]) >= 1) | |
| def test_in_situ_run_preserves_bound_driver_and_records_manifest_metadata( | |
| self, | |
| ) -> None: | |
| save_grants(self.grants, [make_grant()]) | |
| fake = self._run_fixture() | |
| driver_url = ( | |
| "fuchsia-pkg://fuchsia.com/lab_root#meta/lab_sample_driver.cm" | |
| ) | |
| discovery = FakeNodeDiscovery( | |
| [ | |
| NodeDescription( | |
| moniker="node", | |
| bound_driver_url=driver_url, | |
| offers=("fuchsia.driver.lab.Service",), | |
| ) | |
| ] | |
| ) | |
| activator = FakeProxyActivator(discovery) | |
| plan = make_plan() | |
| plan["access"] = {"mode": "in-situ"} | |
| plan["node"] = { | |
| "id": "node", | |
| "driver_moniker": "/bootstrap/boot-drivers:dev.sys.platform.sample-device", | |
| "expected_unclaimed": False, | |
| } | |
| plan_path = self.base / "run_in_situ.json" | |
| plan_path.write_text(json.dumps(plan)) | |
| with mock.patch.object( | |
| cli, "_connect_transport", return_value=fake | |
| ), mock.patch.object( | |
| cli, "_connect_discovery", return_value=discovery | |
| ), mock.patch.object( | |
| cli, "_connect_activator", return_value=activator | |
| ): | |
| code, payload, _ = self.run_cli( | |
| "run", | |
| "--plan", | |
| str(plan_path), | |
| "--evidence-dir", | |
| str(self.base / "evidence_in_situ"), | |
| "--grants", | |
| str(self.grants), | |
| "--target-scope", | |
| "scope", | |
| "--node-id", | |
| "node", | |
| "--moniker", | |
| "/bootstrap/boot-drivers:dev.sys.platform.sample-device", | |
| ) | |
| self.assertEqual(code, 0) | |
| self.assertEqual(payload["reads"][0]["value"], 0xDEAD_BEEF) | |
| # In-situ mode never calls bind_proxy or end_proxy | |
| self.assertEqual(len(activator.bind_calls), 0) | |
| self.assertEqual(len(activator.end_calls), 0) | |
| # Driver remains actively bound after in-situ session | |
| node = asyncio.run(discovery.describe_node("node")) | |
| self.assertIsNotNone(node) | |
| assert node is not None | |
| self.assertFalse(node.is_unclaimed) | |
| self.assertEqual(node.bound_driver_url, driver_url) | |
| # Evidence manifest records driver_moniker and driver_url | |
| manifest = json.loads( | |
| (Path(payload["evidence_dir"]) / "manifest.json").read_text() | |
| ) | |
| self.assertEqual( | |
| manifest["driver_moniker"], | |
| "/bootstrap/boot-drivers:dev.sys.platform.sample-device", | |
| ) | |
| self.assertEqual(manifest["driver_url"], driver_url) | |
| if __name__ == "__main__": | |
| unittest.main() |