| # 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. | |
| import asyncio | |
| import tempfile | |
| import unittest | |
| from pathlib import Path | |
| from typing import Any | |
| from driver_lab.api import DriverLab, DriverLabError | |
| from driver_lab.consent import ConsentDecision | |
| from driver_lab.models import AccessRequest, ResourceKind | |
| from driver_lab.session import ( | |
| AccessRequirements, | |
| Gpio, | |
| HardwareSession, | |
| I2c, | |
| Interrupt, | |
| MmioRegion, | |
| SessionCapabilities, | |
| Spi, | |
| StateHandle, | |
| TranslationMetadata, | |
| UnsupportedCapabilityError, | |
| ) | |
| from driver_lab.transport import ( | |
| Denial, | |
| DirectDescription, | |
| FakeDirectTarget, | |
| FakeProxyTarget, | |
| OpenRejection, | |
| OpenSessionRejected, | |
| OperationDenied, | |
| ProxyDescription, | |
| ResourceInfo, | |
| SequenceItem, | |
| SessionMode, | |
| ) | |
| CTRL_DIGEST = "sha256:" + "11" * 32 | |
| POLICY_DIGEST = "sha256:" + "22" * 32 | |
| class ScriptedPrompt: | |
| def __init__(self, decision: ConsentDecision) -> None: | |
| self.decision = decision | |
| async def request_consent( | |
| self, request: AccessRequest, warning: str | |
| ) -> ConsentDecision: | |
| return self.decision | |
| def make_proxy_description() -> ProxyDescription: | |
| return ProxyDescription( | |
| protocol_major=1, | |
| protocol_minor=0, | |
| proxy_generation=1, | |
| boot_id="boot-test-1", | |
| resource_digest="sha256:" + "33" * 32, | |
| policy_digest=POLICY_DIGEST, | |
| resources=( | |
| ResourceInfo( | |
| id=1, | |
| name="control", | |
| logical_size=0x100, | |
| digest=CTRL_DIGEST, | |
| ), | |
| ResourceInfo( | |
| id=2, | |
| name="status", | |
| logical_size=0x80, | |
| digest="sha256:" + "44" * 32, | |
| ), | |
| ), | |
| max_snapshot_items=64, | |
| audit_capacity=256, | |
| node_moniker="sample/hardware/node", | |
| ) | |
| def make_extended_proxy_description() -> ProxyDescription: | |
| return ProxyDescription( | |
| protocol_major=1, | |
| protocol_minor=0, | |
| proxy_generation=1, | |
| boot_id="boot-test-1", | |
| resource_digest="sha256:" + "33" * 32, | |
| policy_digest=POLICY_DIGEST, | |
| resources=( | |
| ResourceInfo( | |
| id=1, | |
| name="control", | |
| logical_size=0x100, | |
| digest=CTRL_DIGEST, | |
| ), | |
| ResourceInfo( | |
| id=2, | |
| name="status", | |
| logical_size=0x80, | |
| digest="sha256:" + "44" * 32, | |
| ), | |
| ResourceInfo( | |
| id=3, | |
| name="pin0", | |
| kind=ResourceKind.GPIO, | |
| logical_size=1, | |
| digest="sha256:" + "55" * 32, | |
| ), | |
| ResourceInfo( | |
| id=4, | |
| name="i2c-bus", | |
| kind=ResourceKind.I2C, | |
| logical_size=32, | |
| digest="sha256:" + "66" * 32, | |
| ), | |
| ResourceInfo( | |
| id=5, | |
| name="spi-bus", | |
| kind=ResourceKind.SPI, | |
| logical_size=32, | |
| digest="sha256:" + "77" * 32, | |
| ), | |
| ResourceInfo( | |
| id=6, | |
| name="irq0", | |
| kind=ResourceKind.INTERRUPT, | |
| logical_size=0, | |
| digest="sha256:" + "88" * 32, | |
| ), | |
| ResourceInfo( | |
| id=7, | |
| name="state0", | |
| kind=ResourceKind.MMIO, | |
| logical_size=0x20, | |
| digest="sha256:" + "99" * 32, | |
| ), | |
| ), | |
| max_snapshot_items=64, | |
| audit_capacity=256, | |
| node_moniker="sample/hardware/node", | |
| ) | |
| class SessionTest(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_proxy = FakeProxyTarget(make_proxy_description()) | |
| self.fake_direct = FakeDirectTarget( | |
| node_id="sample-node", | |
| protocol_name="fuchsia.hardware.gpio/Device", | |
| bound_driver_url="fuchsia-boot:///sample#meta/sample.cm", | |
| ) | |
| async def test_proxy_mmio_read_and_write(self) -> None: | |
| self.fake_proxy.set_value(1, 0x10, 0xCAFE_BABE) | |
| lab = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| prompt = ScriptedPrompt(ConsentDecision.ALLOW_ONCE) | |
| async with await lab.attach( | |
| "node-1", | |
| mode="proxy", | |
| session_mode=SessionMode.MUTATING, | |
| consent=prompt, | |
| ) as session: | |
| self.assertEqual(session.mode, "proxy") | |
| self.assertTrue(session.capabilities.target_policy) | |
| self.assertTrue(session.capabilities.target_audit) | |
| self.assertTrue(session.capabilities.target_local_timing) | |
| self.assertFalse(session.capabilities.production_driver_active) | |
| mmio = await session.mmio("control") | |
| self.assertEqual(mmio.name, "control") | |
| self.assertEqual(mmio.id, 1) | |
| self.assertEqual(mmio.logical_size, 0x100) | |
| # Read32 | |
| val = await mmio.read32(0x10) | |
| self.assertEqual(val, 0xCAFE_BABE) | |
| # Write32 with readback | |
| write_res = await mmio.write32(0x14, 0x1234_5678) | |
| self.assertEqual(write_res.readback_value, 0x1234_5678) | |
| self.assertEqual(self.fake_proxy.get_value(1, 0x14), 0x1234_5678) | |
| # Write32 with precondition | |
| write_res2 = await mmio.write32( | |
| 0x14, 0x8765_4321, expected_before=0x1234_5678 | |
| ) | |
| self.assertEqual(write_res2.readback_value, 0x8765_4321) | |
| self.assertEqual(self.fake_proxy.get_value(1, 0x14), 0x8765_4321) | |
| # Precondition failure | |
| with self.assertRaises(OperationDenied) as ctx: | |
| await mmio.write32( | |
| 0x14, 0xAAAA_BBBB, expected_before=0xDEAD_BEEF | |
| ) | |
| self.assertEqual(ctx.exception.denial, Denial.PRECONDITION_FAILED) | |
| async def test_proxy_mmio_poll_and_snapshot(self) -> None: | |
| self.fake_proxy.set_value(1, 0x20, 0x0) | |
| lab = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| async with await lab.attach("node-1", mode="proxy") as session: | |
| mmio = await session.mmio("control") | |
| # Poll matching value | |
| self.fake_proxy.set_value(1, 0x20, 0x1) | |
| poll_res = await mmio.poll32(0x20, expected=0x1, mask=0x1) | |
| self.assertEqual(poll_res.value, 0x1) | |
| # Poll timeout | |
| self.fake_proxy.set_value(1, 0x24, 0x0) | |
| with self.assertRaises(OperationDenied) as ctx: | |
| await mmio.poll32(0x24, expected=0x1, mask=0x1, timeout_s=0.001) | |
| self.assertEqual(ctx.exception.denial, Denial.TIMEOUT) | |
| # Snapshot32 | |
| self.fake_proxy.set_value(1, 0x0, 0x1111) | |
| self.fake_proxy.set_value(1, 0x4, 0x2222) | |
| values = await mmio.snapshot32([0x0, 0x4]) | |
| self.assertEqual(values, [0x1111, 0x2222]) | |
| async def test_mmio_bounds_and_alignment_validation(self) -> None: | |
| lab = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| async with await lab.attach("node-1", mode="proxy") as session: | |
| mmio = await session.mmio("control") | |
| with self.assertRaises(ValueError): | |
| await mmio.read32(-4) | |
| with self.assertRaises(ValueError): | |
| await mmio.read32( | |
| 0x100 | |
| ) # Out of bounds (logical_size is 0x100) | |
| with self.assertRaises(ValueError): | |
| await mmio.read32(0x2) # Misaligned | |
| with self.assertRaises(ValueError): | |
| await mmio.write32(0x100, 0x1) | |
| with self.assertRaises(ValueError): | |
| await mmio.poll32(0x1, expected=0) | |
| with self.assertRaises(ValueError): | |
| await session.mmio("nonexistent_resource") | |
| async def test_translation_metadata_present(self) -> None: | |
| self.assertIsInstance(MmioRegion.read32_metadata, TranslationMetadata) | |
| self.assertTrue(MmioRegion.read32_metadata.directly_translatable) | |
| self.assertIn("Read32", MmioRegion.read32_metadata.cpp_analogue) | |
| self.assertIsInstance(MmioRegion.write32_metadata, TranslationMetadata) | |
| self.assertTrue(MmioRegion.write32_metadata.directly_translatable) | |
| self.assertIsInstance(MmioRegion.poll32_metadata, TranslationMetadata) | |
| self.assertTrue(MmioRegion.poll32_metadata.target_local_timing) | |
| self.assertIsInstance( | |
| MmioRegion.snapshot32_metadata, TranslationMetadata | |
| ) | |
| self.assertTrue(MmioRegion.snapshot32_metadata.experiment_only) | |
| # Protocol adapter translation metadata | |
| self.assertIsInstance(Gpio.read_metadata, TranslationMetadata) | |
| self.assertTrue(Gpio.read_metadata.directly_translatable) | |
| self.assertIsInstance(Gpio.write_metadata, TranslationMetadata) | |
| self.assertTrue(Gpio.write_metadata.directly_translatable) | |
| self.assertIsInstance(I2c.transfer_metadata, TranslationMetadata) | |
| self.assertTrue(I2c.transfer_metadata.directly_translatable) | |
| self.assertIsInstance(Spi.transmit_metadata, TranslationMetadata) | |
| self.assertTrue(Spi.transmit_metadata.directly_translatable) | |
| self.assertIsInstance(Interrupt.wait_metadata, TranslationMetadata) | |
| self.assertTrue(Interrupt.wait_metadata.directly_translatable) | |
| async def test_direct_mode_capabilities_and_protocols(self) -> None: | |
| # Register handlers on fake direct target | |
| gpio_state = {"value": True, "direction": "output"} | |
| def handle_read(args: Any) -> dict[str, Any]: | |
| return {"value": gpio_state["value"]} | |
| def handle_write(args: Any) -> dict[str, Any]: | |
| gpio_state["value"] = args.get("value", False) | |
| return {} | |
| self.fake_direct.register_method("read", handle_read) | |
| self.fake_direct.register_method("write", handle_write) | |
| lab = DriverLab( | |
| self.fake_direct, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| async with await lab.attach("node-1", mode="direct") as session: | |
| self.assertEqual(session.mode, "direct") | |
| self.assertFalse(session.capabilities.target_policy) | |
| self.assertFalse(session.capabilities.target_audit) | |
| self.assertFalse(session.capabilities.target_local_timing) | |
| self.assertTrue(session.capabilities.production_driver_active) | |
| # Direct published protocol | |
| proto = await session.protocol("fuchsia.hardware.gpio/Device") | |
| res = await proto.read(resource="gpio") | |
| self.assertTrue(res.response["value"]) | |
| # Gpio adapter | |
| gpio = await session.gpio("gpio") | |
| self.assertTrue(await gpio.read()) | |
| await gpio.write(False) | |
| self.assertFalse(await gpio.read()) | |
| # Attempting MMIO or target sequence in direct mode fails closed | |
| with self.assertRaises(UnsupportedCapabilityError): | |
| await session.mmio("control") | |
| with self.assertRaises(UnsupportedCapabilityError): | |
| await session.sequence([]) | |
| async def test_proxy_mode_rejects_direct_protocols(self) -> None: | |
| lab = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| async with await lab.attach("node-1", mode="proxy") as session: | |
| with self.assertRaises(UnsupportedCapabilityError): | |
| await session.protocol("fuchsia.hardware.gpio/Device") | |
| with self.assertRaises(UnsupportedCapabilityError): | |
| await session.serial("serial") | |
| with self.assertRaises(UnsupportedCapabilityError): | |
| await session.clock("clock") | |
| with self.assertRaises(UnsupportedCapabilityError): | |
| await session.reset("reset") | |
| # Gpio request on proxy that has no GPIO resource raises ValueError | |
| with self.assertRaises(ValueError): | |
| await session.gpio("gpio") | |
| # Interrupt request on proxy that has no interrupt resource raises ValueError | |
| with self.assertRaises(ValueError): | |
| await session.interrupt("interrupt") | |
| async def test_proxy_mode_protocol_adapters(self) -> None: | |
| fake = FakeProxyTarget(make_extended_proxy_description()) | |
| fake.set_i2c_response(4, b"\xde\xad") | |
| fake.set_spi_response(5, b"\xbe\xef") | |
| lab = DriverLab( | |
| fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| prompt = ScriptedPrompt(ConsentDecision.ALLOW_ONCE) | |
| async with await lab.attach( | |
| "node-1", | |
| mode="proxy", | |
| session_mode=SessionMode.MUTATING, | |
| consent=prompt, | |
| ) as session: | |
| # GPIO adapter in proxy mode | |
| gpio = await session.gpio("pin0") | |
| self.assertEqual(gpio.name, "pin0") | |
| self.assertEqual(gpio.id, 3) | |
| await gpio.write(True) | |
| self.assertTrue(await gpio.read()) | |
| self.assertTrue(fake.get_gpio(3)) | |
| await gpio.write(False) | |
| self.assertFalse(await gpio.read()) | |
| self.assertFalse(fake.get_gpio(3)) | |
| # I2C adapter in proxy mode | |
| i2c = await session.i2c("i2c-bus") | |
| self.assertEqual(i2c.name, "i2c-bus") | |
| self.assertEqual(i2c.id, 4) | |
| rx_i2c = await i2c.transfer(b"\x01\x02", read_length=2) | |
| self.assertEqual(rx_i2c, b"\xde\xad") | |
| # SPI adapter in proxy mode | |
| spi = await session.spi("spi-bus") | |
| self.assertEqual(spi.name, "spi-bus") | |
| self.assertEqual(spi.id, 5) | |
| rx_spi = await spi.transmit(b"\x03\x04") | |
| self.assertEqual(rx_spi, b"\xbe\xef") | |
| # Interrupt adapter in proxy mode | |
| irq = await session.interrupt("irq0") | |
| self.assertEqual(irq.name, "irq0") | |
| self.assertEqual(irq.id, 6) | |
| # Immediate wait when after_sequence < sequence | |
| fake.trigger_interrupt(6, timestamp_ns=500_000) | |
| irq_res = await irq.wait(after_sequence=0) | |
| self.assertEqual(irq_res.resource, 6) | |
| self.assertEqual(irq_res.sequence, 1) | |
| self.assertEqual(irq_res.count, 1) | |
| self.assertEqual(irq_res.timestamp_ns, 500_000) | |
| self.assertEqual(irq_res.coalesced_count, 1) | |
| # Asynchronous wait satisfied by trigger_interrupt | |
| wait_task = asyncio.create_task( | |
| irq.wait(after_sequence=1, timeout_s=1.0) | |
| ) | |
| await asyncio.sleep(0) | |
| fake.trigger_interrupt(6, timestamp_ns=600_000) | |
| irq_res2 = await wait_task | |
| self.assertEqual(irq_res2.resource, 6) | |
| self.assertEqual(irq_res2.sequence, 2) | |
| self.assertEqual(irq_res2.count, 2) | |
| self.assertEqual(irq_res2.timestamp_ns, 600_000) | |
| self.assertEqual(irq_res2.coalesced_count, 0) | |
| async def test_heterogeneous_sequence_proxy_mode(self) -> None: | |
| fake = FakeProxyTarget(make_extended_proxy_description()) | |
| fake.set_value(1, 0x10, 0x1111_2222) | |
| fake.set_i2c_response(4, b"\xca\xfe") | |
| fake.set_spi_response(5, b"\xba\xbe") | |
| lab = DriverLab( | |
| fake, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| prompt = ScriptedPrompt(ConsentDecision.ALLOW_ONCE) | |
| async with await lab.attach( | |
| "node-1", | |
| mode="proxy", | |
| session_mode=SessionMode.MUTATING, | |
| consent=prompt, | |
| ) as session: | |
| seq_items = [ | |
| SequenceItem(kind="read32", resource=1, offset=0x10), | |
| SequenceItem( | |
| kind="write32", | |
| resource=1, | |
| offset=0x14, | |
| value=0x9999, | |
| readback=True, | |
| ), | |
| SequenceItem(kind="delay_ns", delay_ns=1000), | |
| SequenceItem(kind="barrier", barrier="memory"), | |
| SequenceItem(kind="gpio_write", resource=3, value=True), | |
| SequenceItem(kind="gpio_read", resource=3), | |
| SequenceItem( | |
| kind="i2c_transfer", | |
| resource=4, | |
| write_data=b"\x01", | |
| read_length=2, | |
| ), | |
| SequenceItem(kind="spi_transmit", resource=5, tx_data=b"\x02"), | |
| ] | |
| outcome = await session.sequence(seq_items) | |
| self.assertTrue(outcome.complete) | |
| self.assertEqual(len(outcome.results), 8) | |
| self.assertEqual(outcome.results[0].kind, "read32") | |
| self.assertEqual(outcome.results[0].value, 0x1111_2222) | |
| self.assertEqual(outcome.results[1].kind, "write32") | |
| self.assertEqual(outcome.results[1].readback_value, 0x9999) | |
| self.assertEqual(outcome.results[2].kind, "delay_ns") | |
| self.assertEqual(outcome.results[3].kind, "barrier") | |
| self.assertEqual(outcome.results[4].kind, "gpio_write") | |
| self.assertEqual(outcome.results[5].kind, "gpio_read") | |
| self.assertEqual(outcome.results[5].value, 1) | |
| self.assertEqual(outcome.results[6].kind, "i2c_transfer") | |
| self.assertEqual(outcome.results[6].data, b"\xca\xfe") | |
| self.assertEqual(outcome.results[7].kind, "spi_transmit") | |
| self.assertEqual(outcome.results[7].data, b"\xba\xbe") | |
| async def test_selection_rules_in_attach(self) -> None: | |
| lab_proxy = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| # auto with needs_mmio selects proxy | |
| session = await lab_proxy.attach( | |
| "node-1", | |
| mode="auto", | |
| requirements=AccessRequirements(needs_mmio=True), | |
| ) | |
| self.assertEqual(session.mode, "proxy") | |
| await session.close() | |
| # direct mode with needs_mmio raises DriverLabError | |
| with self.assertRaises(DriverLabError): | |
| await lab_proxy.attach( | |
| "node-1", | |
| mode="direct", | |
| requirements=AccessRequirements(needs_mmio=True), | |
| ) | |
| # Invalid mode string | |
| with self.assertRaises(ValueError): | |
| await lab_proxy.attach("node-1", mode="invalid") # type: ignore[arg-type] | |
| async def test_mutating_session_consent_enforcement(self) -> None: | |
| # Consent denied | |
| prompt_denied = ScriptedPrompt(ConsentDecision.DENY_ONCE) | |
| lab_denied = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| consent=prompt_denied, | |
| ) | |
| with self.assertRaises(DriverLabError): | |
| await lab_denied.attach( | |
| "node-1", mode="proxy", session_mode=SessionMode.MUTATING | |
| ) | |
| # No consent prompt configured for mutating session | |
| lab_unattended = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| consent=None, | |
| ) | |
| with self.assertRaises(DriverLabError): | |
| await lab_unattended.attach( | |
| "node-1", mode="proxy", session_mode=SessionMode.MUTATING | |
| ) | |
| async def test_async_cancellation_cleans_up_session(self) -> None: | |
| prompt = ScriptedPrompt(ConsentDecision.ALLOW_ONCE) | |
| lab = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| consent=prompt, | |
| ) | |
| session = await lab.attach( | |
| "node-1", mode="proxy", session_mode=SessionMode.MUTATING | |
| ) | |
| self.assertEqual(self.fake_proxy.active_mutating_session, 1) | |
| async def worker() -> None: | |
| async with session: | |
| mmio = await session.mmio("control") | |
| await mmio.read32(0x10) | |
| await asyncio.sleep(10.0) | |
| task = asyncio.create_task(worker()) | |
| await asyncio.sleep(0.01) | |
| task.cancel() | |
| with self.assertRaises(asyncio.CancelledError): | |
| await task | |
| self.assertTrue(session.is_closed) | |
| self.assertIsNone(self.fake_proxy.active_mutating_session) | |
| # Calling on closed session fails | |
| with self.assertRaises(UnsupportedCapabilityError): | |
| await session.mmio("control") | |
| async def test_representative_protocols_direct_mode(self) -> None: | |
| target = FakeDirectTarget(node_id="dev", protocol_name="multi") | |
| target.register_method("transfer", lambda a: {"read_data": [1, 2, 3]}) | |
| target.register_method("transmit", lambda a: {"rx_data": [4, 5, 6]}) | |
| target.register_method("write", lambda a: {"bytes_written": 2}) | |
| target.register_method("read", lambda a: {"data": [7, 8]}) | |
| target.register_method("enable", lambda a: {}) | |
| target.register_method("disable", lambda a: {}) | |
| target.register_method("assert_reset", lambda a: {}) | |
| target.register_method("deassert_reset", lambda a: {}) | |
| target.register_method("wait", lambda a: {"timestamp_ns": 12345}) | |
| caps = SessionCapabilities( | |
| mode="direct", | |
| target_policy=False, | |
| target_audit=False, | |
| target_local_timing=False, | |
| fault_isolation="driver_host", | |
| production_driver_active=True, | |
| ) | |
| desc = DirectDescription( | |
| node_id="dev", protocol_name="multi", boot_id="b1" | |
| ) | |
| fake_session = await target.open_direct_session(None) # type: ignore[arg-type] | |
| session = HardwareSession( | |
| capabilities=caps, | |
| direct_session=fake_session, | |
| direct_description=desc, | |
| ) | |
| i2c = await session.i2c("i2c") | |
| rx_i2c = await i2c.transfer(b"abc", read_length=3) | |
| self.assertEqual(rx_i2c, bytes([1, 2, 3])) | |
| spi = await session.spi("spi") | |
| rx_spi = await spi.transmit(b"xyz") | |
| self.assertEqual(rx_spi, bytes([4, 5, 6])) | |
| serial = await session.serial("serial") | |
| written = await serial.write(b"hi") | |
| self.assertEqual(written, 2) | |
| rx_serial = await serial.read(2) | |
| self.assertEqual(rx_serial, bytes([7, 8])) | |
| clock = await session.clock("clock") | |
| await clock.enable() | |
| await clock.disable() | |
| reset = await session.reset("reset") | |
| await reset.assert_reset() | |
| await reset.deassert_reset() | |
| irq = await session.interrupt("irq") | |
| ts = await irq.wait(1.0) | |
| self.assertEqual(ts, 12345) | |
| await session.close() | |
| async def test_disabled_target_rejected(self) -> None: | |
| self.fake_proxy.enabled = False | |
| lab = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| with self.assertRaises(OpenSessionRejected) as cm: | |
| await lab.attach("node-1", mode="proxy") | |
| self.assertEqual(cm.exception.reason, OpenRejection.NOT_ACCEPTING) | |
| async def test_rate_limit_exceeded(self) -> None: | |
| self.fake_proxy.max_ops_per_second = 2 | |
| self.fake_proxy.set_value(1, 0x10, 42) | |
| lab = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| async with await lab.attach("node-1", mode="proxy") as session: | |
| mmio = await session.mmio("control") | |
| val1 = await mmio.read32(0x10) | |
| self.assertEqual(val1, 42) | |
| val2 = await mmio.read32(0x10) | |
| self.assertEqual(val2, 42) | |
| with self.assertRaises(OperationDenied) as cm: | |
| await mmio.read32(0x10) | |
| self.assertEqual(cm.exception.denial, Denial.LIMIT_EXCEEDED) | |
| async def test_deadline_exceeded(self) -> None: | |
| self.fake_proxy.max_deadline_ns = 50_000_000 | |
| self.fake_proxy.set_value(1, 0x10, 42) | |
| lab = DriverLab( | |
| self.fake_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| ) | |
| async with await lab.attach("node-1", mode="proxy") as session: | |
| mmio = await session.mmio("control") | |
| with self.assertRaises(OperationDenied) as cm: | |
| await mmio.poll32(0x10, expected=42, timeout_s=0.1) | |
| self.assertEqual(cm.exception.denial, Denial.LIMIT_EXCEEDED) | |
| async def test_proxy_state_handle_read_knob_trigger_poll_snapshot( | |
| self, | |
| ) -> None: | |
| ext_proxy = FakeProxyTarget(make_extended_proxy_description()) | |
| ext_proxy.set_value(7, 0x00, 0) # STATE_BUSY_FLAG | |
| ext_proxy.set_value(7, 0x04, 0) # STATE_INVARIANT_VIOLATIONS | |
| ext_proxy.set_value(7, 0x08, 0) # KNOB_RACE_DELAY_US | |
| ext_proxy.set_value(7, 0x0C, 0) # KNOB_FIX_ENABLED | |
| ext_proxy.set_value(7, 0x10, 0) # TRIGGER_CONCURRENT_OP | |
| def on_trigger(arg: int) -> None: | |
| delay_us = ext_proxy.get_value(7, 0x08) | |
| fix_enabled = ext_proxy.get_value(7, 0x0C) | |
| if delay_us > 0 and fix_enabled == 0: | |
| violations = ext_proxy.get_value(7, 0x04) | |
| ext_proxy.set_value(7, 0x04, violations + max(arg, 1)) | |
| # Trigger slot always reads as 0. | |
| ext_proxy.set_value(7, 0x10, 0) | |
| ext_proxy.set_write_hook(7, 0x10, on_trigger) | |
| prompt = ScriptedPrompt(ConsentDecision.ALLOW_ONCE) | |
| lab = DriverLab( | |
| ext_proxy, | |
| grants_path=self.grants_path, | |
| evidence_root=self.evidence_root, | |
| target_scope="target-1", | |
| node_id="node-1", | |
| consent=prompt, | |
| ) | |
| async with await lab.attach( | |
| "node-1", | |
| mode="proxy", | |
| session_mode=SessionMode.MUTATING, | |
| ) as session: | |
| state = await session.state("state0") | |
| self.assertIsInstance(state, StateHandle) | |
| self.assertEqual(state.name, "state0") | |
| self.assertEqual(state.id, 7) | |
| # 1. Repro bug: widen race window via knob and fire trigger. | |
| knob_res = await state.write_knob(0x08, 5000) | |
| self.assertEqual(knob_res.readback_value, 5000) | |
| trig_res = await state.trigger(0x10, arg=1) | |
| self.assertIsNone(trig_res.readback_value) | |
| self.assertEqual(await state.read32(0x04), 1) | |
| # 2. Prove fix in-situ: enable fix knob, fire trigger, verify counter stays 1. | |
| await state.write_knob(0x0C, 1) | |
| await state.trigger(0x10, arg=2) | |
| poll_res = await state.poll32(0x04, expected=1) | |
| self.assertEqual(poll_res.value, 1) | |
| snap = await state.snapshot32([0x04, 0x08, 0x0C]) | |
| self.assertEqual(snap, [1, 5000, 1]) | |
| if __name__ == "__main__": | |
| unittest.main() |