dau_driver.testing¶

Shared test doubles for DAU driver consumers.

Used by the dau-driver, dau, and dau-polars test suites so the fake device’s semantics cannot drift between packages. Not imported by production code.

Classes¶

FakeTransport

In-memory DeviceTransport: records DMA traffic and register writes.

Functions¶

`default_int32_capabilities`([operators, max_transfer_bytes]) → dau_core.capabilities.DeviceCapabilities	The standard fake INT32 aggregation device capability set.
`fake_device`(tmp_path, transport, *[, capabilities]) → dau_driver.device.DauDevice	A DauDevice backed by a FakeTransport (paths are never opened).
`golden_aggregation_trigger`(transport, output_address)	A StreamTrigger that emulates the aggregation hardware: decode the

Module Contents¶

class dau_driver.testing.FakeTransport[source]¶

In-memory DeviceTransport: records DMA traffic and register writes.

Advertises a current (register map 0.3) shell by default, including the job capacity a host batches against — an all-zero register file models a shell nobody builds any more, and every job test would be exercising the not-advertised refusal instead of the path it means to test. Tests that want an older shell clear or lower these.

DEFAULT_INPUT_ROW_CAPACITY = 16384¶

DEFAULT_OUTPUT_RECORD_CAPACITY = 819¶

h2c_writes: list[tuple[int, bytes]] = []¶

c2h_reads: list[tuple[int, int]] = []¶

c2h_payloads: dict[int, bytes]¶

registers: dict[int, int]¶

register_reads: list[int] = []¶

read_u32(offset: int) → int[source]¶

write_u32(offset: int, value: int) → None[source]¶

read_dma(device_address: int, byte_count: int) → bytes[source]¶

write_dma(device_address: int, data: bytes | bytearray | memoryview) → int[source]¶

dau_driver.testing.default_int32_capabilities(operators: tuple[dau_core.capabilities.OperatorCapability, ...] | None = None, *, max_transfer_bytes: int = 4096) → dau_core.capabilities.DeviceCapabilities[source]¶: The standard fake INT32 aggregation device capability set.

dau_driver.testing.fake_device(tmp_path: pathlib.Path, transport: FakeTransport, *, capabilities: dau_core.capabilities.DeviceCapabilities | None = None) → dau_driver.device.DauDevice[source]¶: A DauDevice backed by a FakeTransport (paths are never opened).

dau_driver.testing.golden_aggregation_trigger(transport: FakeTransport, output_address: int)[source]¶: A StreamTrigger that emulates the aggregation hardware: decode the encoded input batch, compute the golden scalar, stage the result stream at output_address for the subsequent C2H read.