Testing without hardware

pycosmicsdk.testing lets you write tests for code that drives a Binho host adapter without owning one. Open a fake adapter, say what its commands should do, and drive it with the ordinary SDK API:

from pycosmicsdk import I2cPullUp
from pycosmicsdk.testing import FakeAdapter


def test_reads_the_id_register() -> None:
    adapter = FakeAdapter()
    adapter.returns("i2c_read", b"\x2a")

    with adapter.open() as dev:
        i2c = dev.i2c().initialize(frequency_hz=400_000, pull_up=I2cPullUp.OHM_330)
        assert i2c.read(address=0x50, subaddress=b"\x10", length=1) == b"\x2a"

    assert adapter.calls_to("i2c_read")[0].args[1] == 0x50

Install the package as usual; there is nothing extra to add to your test dependencies.

What is real and what is faked

open returns a real Device and open a real AsyncDevice. Only the compiled facade underneath is substituted, so everything above it is the shipping code:

Real	Faked
The device lifecycle and its state machine	The USB HID transport
All seven interface classes and their async twins	The firmware’s command handling
Argument validation and `ValidationError`	The bytes on the wire
The notification dispatcher and `Subscription` filters	Notification arrival
The asyncio request/response correlation engine
The status-to-exception translation

A test that passes against the fake exercises the same objects a test against an adapter on the bench does.

Programming commands

Commands are named by facade method — "i2c_read", "spi_transfer", "i3c_ccc_getpid" — not by public method name, because one public call may issue more than one command. Programming a name the facade does not carry raises FakeAdapterError on the spot, so a typo cannot quietly do nothing.

returns: Succeed with a value. The value is the logical result — the bytes a read yields, the int a voltage reader reports — and the fake re-encodes it for whichever surface asked, so the same call programs the blocking and the asyncio adapter identically.
fails: Report a firmware status. The status travels the real translator, so the exception is the same class with the same status_code and operation that hardware would have produced.
responds: Compute the outcome from the call — a register map, a counter, a value derived from the address the caller passed.

adapter.returns("i2c_read", b"\x2a")
adapter.returns("i2c_scan_bus", [0x50, 0x68])          # a scan that finds two devices
adapter.fails("i2c_write", StatusCode.FW_I2C_NACK_ADDRESS)

registers = {0x0F: b"\x6b", 0x20: b"\x01"}
adapter.responds("i2c_read", lambda call: registers[call.args[2][0]])

All three take times=. Without it they install a standing default; with it they queue that many one-shot outcomes, taken in order ahead of the default — which is how you write the transient-failure case:

adapter.fails("i2c_write", StatusCode.FW_I2C_NACK_ADDRESS, times=1)  # NACK once, then succeed

A responds handler reports a failure by returning a StatusCode other than SUCCESS; anything else it returns is a result. A command nobody programmed succeeds with a plausible default: zero bytes of the requested length for a read, an empty scan, the firmware default for a length limit.

Asserting on the conversation

calls is every facade call the SDK made, oldest first, and calls_to filters it by method. Each entry is a RecordedCall whose args are plain Python values — enums unwrapped to their int, payloads as bytes, and on the asyncio surface with the transaction id already stripped, so the two surfaces record identically.

write = adapter.calls_to("i2c_write")[0]
assert write.args == (I2cBus.A.value, 0x50, b"\x10", b"\x01", False)

Notifications

Real notifications arrive on the dispatcher thread. Build one with the helpers in this package and hand it to publish, which returns only once the dispatcher has finished fanning it out — so no test sleeps and no test polls:

from pycosmicsdk.testing import gpio_interrupt, i3c_ibi, i3c_target_bus_event, uart_rx

with adapter.open() as dev:
    seen = []
    dev.subscribe(I3cIbiNotification, lambda n: seen.append(n.address))
    adapter.publish(i3c_ibi(address=0x08, payload=b"\x01"))
    assert seen == [0x08]        # already delivered

The builders cover every event type the SDK reports: uart_rx, i3c_ibi, i3c_hot_join, i3c_target_bus_event, gpio_interrupt, i3c_connector_event and unknown_notification. Each produces a real notification object, so isinstance filtering, where= predicates, receive and wait_for_event all see it as they see a firmware event.

Publishing is rejected with FakeAdapterError when nothing is listening yet — subscribe, or bring up the interface that arms reception, first. A silently dropped event would only show up later as a confusing assertion failure.

The asyncio surface

FakeAsyncAdapter is the twin: the same programming calls, the same values, an opener that works as async with adapter.open() as dev: or dev = await adapter.open(), and await adapter.publish(...) as the notification barrier.

async def test_scan_finds_the_bench_targets() -> None:
    adapter = FakeAsyncAdapter()
    adapter.returns("i2c_scan_bus", [0x50, 0x68])

    async with adapter.open() as dev:
        i2c = await dev.i2c().initialize(frequency_hz=400_000, pull_up=I2cPullUp.OHM_330)
        found = await i2c.scan()

    assert sorted(found.addresses_7bit) == [0x50, 0x68]

Responses reach the correlation engine inside the submit call, so an awaited command resolves on the next event-loop pass and no wall-clock time passes. A timeout is programmed like any other status — adapter.fails("i2c_read", StatusCode.SDK_ERROR_TIMEOUT) raises CosmicTimeoutError immediately rather than after the per-request deadline.

The bulk plane, and why a fake adapter has none

The bulk transfer plane is the narrative account of what this section fakes; Target mode covers the two arenas that have no other transport.

Large SPI payloads, large I2C and I3C controller transfers, and the I2C and SPI target arenas do not travel over the same USB pipe as ordinary commands. They use a second, vendor-specific bulk transport that the host has to claim when it connects — bound to WinUSB on Windows, resting on udev permissions elsewhere — and a host that never claimed it has no bulk plane whatever the connected device supports. On top of that the device is asked, once per connection, which bulk commands it implements and how large a payload each protocol group accepts; firmware that predates the question does not answer, and silence is read as supports nothing.

A FakeAdapter reports no bulk plane, and that is the case worth testing. It is what most real hosts report, so the SPI fallback is the path your code will actually take:

adapter = FakeAdapter()
with adapter.open() as dev:
    spi = dev.spi().initialize(...)
    spi.write(b"\x01\x02\x03")               # HID transport, as on a host with no bulk
    assert not dev.capabilities.supports_spi_bulk()

Two behaviours of the real negotiation carry over exactly:

It is asked once and never retried. The SDK records the query as done before issuing it, so even a device that never answers costs one query per connection and every later call is free. A test that needs the negotiation to run again must close and reopen the device.
Demanding the plane fails before anything is sent. spi.write(..., transport=SpiTransport.BULK) raises CapabilityError on a host with no bulk plane, and no command reaches the adapter.

To exercise the other branch — a host that did claim the interface talking to a device that advertises bulk — describe the pair with BulkProfile and hand it to bulk_facade:

from pycosmicsdk._capabilities import probe_bulk
from pycosmicsdk.testing import (
    BulkGroup,
    BulkProfile,
    bulk_capabilities,
    bulk_facade,
)

profile = BulkProfile(
    transport_open=True,   # this host claimed the vendor interface
    capabilities=bulk_capabilities(groups={BulkGroup.SPI: 32_768}),
)
assert probe_bulk(bulk_facade(profile)).max_spi_payload_bytes == 32_768

transport_open defaults to False, because that is what almost every real host reports and it is the branch most tests should exercise. Set it only when the has-bulk path is the thing under test.

bulk_capabilities takes a payload ceiling per protocol group and works out the command mask; pass commands= to fake a device that implements only part of a group’s bulk surface, and supported=False for one that answered but said no. no_bulk_capabilities is the other reply — what a device with no bulk plane at all yields. To model what the device advertises rather than what this SDK binds — which is not the same list, since firmware carries bulk commands the SDK has no wrapper for — use firmware_bulk_capabilities.

BulkProfile exists because “has a bulk plane” is two independent facts, and the interesting case is when they disagree. transport_open is whether this host claimed the vendor interface; webusb_enabled is whether the interface is in the device’s configuration descriptor at all. A device with webusb_enabled=False still answers the capability query saying it supports bulk — firmware reports support unconditionally — so that combination is the one a real host actually meets, and it is worth a test:

absent = BulkProfile(
    webusb_enabled=False,
    capabilities=bulk_capabilities(groups={BulkGroup.SPI: 32_768}),
)
negotiated = probe_bulk(bulk_facade(absent))
assert negotiated.device_supported          # the device said yes
assert not negotiated.host_transport_open   # the endpoints are not there
assert not negotiated.available

Keeping those two flags apart is the point: it is what lets a caller tell “this adapter cannot” from “this machine cannot”, which need different fixes.

The ceiling is per call, and per protocol group. It is not the same number as a target’s arena capacity — though on these products it happens to be, since firmware derives both from one constant per model: SUPERNOVA_BULK_ARENA_BYTES is 32768 and PULSAR_BULK_ARENA_BYTES is 16384. Prefer those names to literals; the Pulsar figure in particular moved several times during development, and tests written against an older number pass for the wrong reason.

Target mode and the memory arena

In target mode the adapter exposes a block of memory that a controller on the bus reads and writes, and that your code preloads and reads back with a pair of commands. Those two commands are the host’s own window onto memory the firmware holds — not bus traffic — so a read after a write returns what was written even with nothing attached. That makes it the one piece of target-mode behaviour a test can rely on without hardware, and memory_arena provides it:

from pycosmicsdk.testing import FakeAdapter, memory_arena

adapter = FakeAdapter()
arena = memory_arena(1024)
adapter.responds("i3c_target_write_memory", arena.on_write)
adapter.responds("i3c_target_read_memory", arena.on_read)

with adapter.open() as dev:
    target = dev.i3c_target().initialize(...)
    target.write_memory(address=0x10, data=b"\xde\xad")
    assert target.read_memory(address=0x10, length=2) == b"\xde\xad"

arena.preload(address, data) seeds the starting state without it showing up in adapter.calls, and arena.contents is the whole block for asserting on what the code under test stored. The same arena serves every protocol, because each takes the same two positional arguments. Pass register_bytes= when the target’s addresses are register indices rather than byte offsets — the I3C target’s memory layout selects 1, 2 or 4 — and fill=0xFF for an arena that starts out erased.

It is a flat block of bytes and nothing more: no emulation models, no page wrapping, no busy windows, no write-protect windows, no per-region access policies. A transfer running off the end raises FakeAdapterError rather than guessing a status, because which one firmware reports differs per protocol — pass overflow=StatusCode.... when that case is what the test is about.

Re-arming a target is not the same call on every protocol, and this is the one to get right:

Target	A second `initialize()`	The idempotent way to reconfigure
I3C	Succeeds. Re-initialises the peripheral unconditionally, and takes the role from the controller when it has to.	Either — `initialize()` or `set_parameters()`.
I2C	Reports `FW_INTERFACE_ALREADY_INITIALIZED`.	`set_parameters()`, which keeps the arena contents.
SPI	Reports `FW_INTERFACE_ALREADY_INITIALIZED`.	`set_parameters()`, which keeps the arena contents.

The fake enforces exactly that split. Reading the I3C rule across to the other two — or the reverse — is how a test comes to assert the opposite of what hardware does.

What the fake rejects

A fake that accepts everything is worse than useless, so these fail here for the same reason and with the same exception they would on hardware:

A call before initialize() reports FW_INTERFACE_NOT_INITIALIZED as a FirmwareError. Bring the interface up in the test, as you would against hardware. GPIO is the exception, as it is on the device: it has no interface-level bring-up, and the SDK already rejects an unconfigured pin in Python.
A second initialize() reports FW_INTERFACE_ALREADY_INITIALIZED — for I2C, UART, SPI and the I3C controller, which are the four bring-ups firmware refuses a repeat of today. The I2C and SPI target bring-ups arriving with the target-mode surface refuse a repeat too, and offer a separate set_parameters command as the idempotent path; the I3C target bring-up is the odd one out and does not. See Target mode.
A call after close(), or after reboot() raises DisconnectError — that is the real Device state machine, not something the fake adds.
A protocol the model does not have raises CapabilityError. FakeAdapter(model=DeviceModel.PULSAR) refuses i3c and offers two I2C buses, which come up independently.
Bad arguments raise ValidationError before the fake is reached at all, because that check is the SDK’s own.

What it deliberately accepts

The mirror image of the list above, and the more important half: a fake that invents a refusal is worse than one that permits too much, because it has you writing code around a case that cannot happen. Each of these succeeds here because it succeeds on the device, and each is sourced to upstream’s @retval documentation and to the firmware handler that answers the command:

A second initialize. Target bring-up is idempotent. Upstream states that i3cTargetInit re-initialises the I3C peripheral unconditionally and never reports FW_INTERFACE_ALREADY_INITIALIZED, and the firmware handler inspects no state before doing it. Re-arming a target mid-test is the ordinary case, not an error.
Bringing the I3C controller up after target mode. The peripheral holds one role, and the target bring-up took it, so the next initialize is not a repeat.
get_device_table and set_target_configuration before bring-up. Both are answered from the controller’s internal target table rather than from the bus, and neither is gated on the interface state. A table read before bring-up reports an empty table.
Every voltage-rail command before bring-up. They are system commands, and servicing them first is what makes power-then-initialise work.

What it deliberately does not fake

The fake is a programmable stand-in, not a device simulator. It has no model of the bus, so nothing below is reproduced:

Enumeration and the connect handshake. adapter.open() does no USB enumeration, so it never raises DeviceNotFoundError, and list_devices still reports the real machine.
Bus semantics. A write is not remembered, so a following read does not return it unless you programmed it that way (responds over a dict is the usual answer). Addresses, voltages and clock rates are recorded, never simulated.
Firmware parameter validation. A frequency or payload size the firmware would reject is accepted; program the rejection with fails when that is what the test is about.
I3C mode exclusivity. One adapter has one I3C peripheral in one role, and firmware answers a controller command taken in target mode — or a target command taken in controller mode — with FW_INVALID_COMMAND. Here the controller and target families are tracked independently, so both can be up at once and either can be driven. Only bring-up is role-accurate: an I3C controller initialize after a target initialize succeeds, as it does on the device. A test that must pin the cross-role refusal should program it with adapter.fails(..., StatusCode.FW_INVALID_COMMAND).
I2C and SPI role exclusivity, which is settled and still not modelled. FW_I2C_ROLE_CONFLICT (0x0216) and FW_SPI_ROLE_CONFLICT (0x0316) are real and are returned from six firmware guard sites, so the fake is no longer declining to choose between two authorities. It permits both orders, for two better reasons: on the device the exclusivity is recoverable — I2C_DEINIT and SPI_DEINIT release the role and the shared target arena, and I2cTarget.release / SpiTarget.release send them — so a refusal here would model a terminal state the hardware does not have; and a refusal would break the controller-then-target pattern with nothing to offer in its place. The third reason this entry used to give is superseded, and is named because a stale reading would keep it: “this SDK cannot issue that recovery, because no CosmicSDK revision binds an i2cDeinit or spiDeinit, so from Python a role taken is held for the session”. It can, and it is not. Programme the conflict when it is what the test is about — adapter.fails("spi_controller_init", 0x0316) reaches it, and pycosmicsdk.testing.SPI_ROLE_CONFLICT names the number. Which way round matters: the I2C guards are order-sensitive, because a target bring-up is refused by any role on either bus while a controller bring-up looks only at its own. Still do not read a passing test here as evidence that a controller-then-target sequence works on hardware — the fake permits an order the device refuses, and the device’s refusal is what a release() clears.
Firmware’s GPIO pin state. FW_GPIO_NOT_CONFIGURED and FW_GPIO_WRONG_CONFIGURATION are never reached, because Gpio refuses an unconfigured or wrong-direction pin in Python first — which is the SDK’s own behaviour, not the fake’s, so the two agree.
Firmware’s target-table lookup. set_target_configuration answers FW_TARGET_ADDRESS_NOT_FOUND_IN_THE_TABLE on the device for an address the controller does not know. The fake keeps no table, so it accepts any address.
The UART’s VCOM mode. On a device whose CDC interface is enabled in UART passthrough and has been claimed by the host’s serial port, firmware answers uart_init with FW_UART_IN_VCOM_PORT_MODE. There is no host serial port here, so UART bring-up always sees an idle peripheral. Program the refusal with fails if that is the case under test.
Timing and concurrency. Commands complete instantly, notifications arrive exactly when published, and nothing is reordered. That is the point, but it means the fake cannot surface a race that only real latency would.

Nothing here needs hardware, so these tests belong in the ordinary unit-test job. The hardware-only behaviours above stay the province of tests run against an attached adapter.

Reference

class pycosmicsdk.testing.FakeAdapter(*, model: DeviceModel = DeviceModel.SUPERNOVA, serial_number: str = _DEFAULT_SERIAL, hw_version: str = _DEFAULT_HW_VERSION, fw_version: str = _DEFAULT_FW_VERSION, bulk: BulkProfile | None = None)

Bases: _FakeAdapterBase

A blocking Binho host adapter that exists only in this process.

Open it to get a real pycosmicsdk.Device, program what its commands should do, then drive it with the ordinary SDK API:

adapter = FakeAdapter()
adapter.returns("i2c_read", b"\\x2a")
with adapter.open() as dev:
    i2c = dev.i2c().initialize(frequency_hz=400_000, pull_up=I2cPullUp.OHM_330)
    assert i2c.read(address=0x50, subaddress=b"\\x10", length=1) == b"\\x2a"

The device enforces its real state machine, and the fake enforces the interface bring-up firmware enforces, so a transfer before initialize(), after close(), or on an adapter left TERMINAL by reboot() fails here for the same reason and with the same exception it would on hardware.

open() → Device

Return a real pycosmicsdk.Device bound to this fake adapter.

Unlike pycosmicsdk.Device.open this issues no enumeration and no connect handshake – there is no bus to enumerate – so it cannot raise DeviceNotFoundError. The device it returns is otherwise ordinary, close() included; use it as a context manager.

Returns:: An open Device in DeviceState.OPEN.
Raises:: FakeAdapterError – If this adapter was already opened.

property device: Device

The pycosmicsdk.Device this adapter opened.

Raises:: FakeAdapterError – If the adapter has not been opened.

publish(notification: Notification) → None

Deliver one notification and return once every callback has run.

Notifications reach subscribers on the dispatcher thread, exactly as firmware events do. This posts a barrier behind the notification and waits for it, so when the call returns the fan-out is complete: no sleeps, no polling, no flaky assertions.

Parameters:: notification – Build it with one of this package’s notification helpers, e.g. uart_rx(b"ok") or i3c_ibi(address=0x08).
Raises:: FakeAdapterError – If the adapter is not open, nothing has armed notification delivery yet, or the dispatcher failed to drain.

property calls: tuple[RecordedCall, ...]

Every facade call the SDK made, oldest first.

Includes the notification-bridge installs and the teardown call, so a test can assert on the whole conversation, not just the transfers.

calls_to(method: str) → tuple[RecordedCall, ...]

Return only the recorded calls to one facade method.

Parameters:: method – Facade method name, e.g. "i2c_write".
Returns:: The matching calls, oldest first.

fails(method: str, status: StatusCode | int, *, times: int | None = None) → None

Make one facade command report a firmware failure.

The status travels the real translation path, so the exception a test catches is the same class, with the same status_code and operation, that hardware would have produced.

Parameters:

method – Facade method name, e.g. "i2c_write".
status – The status code to report, e.g. StatusCode.FW_I2C_NACK_ADDRESS. A raw int stands in for a code this SDK release does not name.
times – When None (the default), applies to every call. An integer applies to that many calls – times=1 is the transient-failure-then-recover case.

Raises:

FakeAdapterError – If method is not a facade command, or times is not positive.

property info: DeviceInfo: The identity this adapter reports, as dev.info will show it.

responds(method: str, handler: Callable[[RecordedCall], Any], *, times: int | None = None) → None

Compute one facade command’s outcome from the call itself.

The escape hatch for anything the flat forms cannot express: a register map, a counter, a value derived from the address the caller passed.

Parameters:

method – Facade method name, e.g. "i2c_read".
handler – Called with the RecordedCall. Return a logical result to succeed, or a pycosmicsdk.StatusCode other than SUCCESS to report a firmware failure.
times – When None (the default), applies to every call. An integer applies to that many calls.

Raises:

FakeAdapterError – If method is not a facade command, or times is not positive.

returns(method: str, value: Any, *, times: int | None = None) → None

Make one facade command succeed with value.

Parameters:

method – Facade method name, e.g. "i2c_read".
value – The logical result – the bytes a read yields, the addresses a scan finds, the int a voltage reader reports. It is re-encoded per surface, so the same value serves the blocking and the asyncio device.
times – When None (the default), applies to every call. An integer applies to that many calls, consumed before whatever default is in place.

Raises:

FakeAdapterError – If method is not a facade command, or times is not positive.

class pycosmicsdk.testing.FakeAsyncAdapter(*, model: DeviceModel = DeviceModel.SUPERNOVA, serial_number: str = _DEFAULT_SERIAL, hw_version: str = _DEFAULT_HW_VERSION, fw_version: str = _DEFAULT_FW_VERSION, bulk: BulkProfile | None = None)

Bases: _FakeAdapterBase

An asyncio Binho host adapter that exists only in this process.

The FakeAdapter twin: it hands back a real pycosmicsdk.AsyncDevice, and commands are programmed with the same returns / fails / responds calls and the same logical values:

adapter = FakeAsyncAdapter()
adapter.returns("i2c_scan_bus", [0x50, 0x51])
async with adapter.open() as dev:
    i2c = await dev.i2c().initialize(frequency_hz=400_000, pull_up=I2cPullUp.OHM_330)
    found = await i2c.scan()

Responses are handed to the correlation engine inside the submit call, so an awaited command resolves on the next event-loop pass and no wall-clock time passes. A timeout is programmed like any other status – adapter.fails("i2c_read", StatusCode.SDK_ERROR_TIMEOUT) raises pycosmicsdk.CosmicTimeoutError immediately rather than after the per-request deadline.

open(*, timeout_ms: int = 2000) → _FakeAsyncOpener

Return an opener for a real pycosmicsdk.AsyncDevice on this adapter.

Use it as async with adapter.open() as dev: for automatic cleanup, or dev = await adapter.open() to own close() yourself. As on the blocking side there is no enumeration and no connect handshake.

Parameters:: timeout_ms – Per-request timeout applied to awaited commands. Only reachable by a command programmed never to answer.
Returns:: An awaitable async context manager yielding the open device.

property device: AsyncDevice

The pycosmicsdk.AsyncDevice this adapter opened.

Raises:: FakeAdapterError – If the adapter has not been opened.

async publish(notification: Notification) → None

Deliver one notification and return once every callback has run.

The asyncio twin of FakeAdapter.publish. Awaiting it drains the dispatcher thread and then the event loop, so by the time it returns an async subscription callback has run and an iter_notifications stream already holds the item.

Parameters:: notification – Build it with one of this package’s notification helpers, e.g. uart_rx(b"ok") or gpio_interrupt(pin=GpioPinNumber.PIN_3, edge=GpioTriggerType.RISING_EDGE).
Raises:: FakeAdapterError – If the adapter is not open, nothing has armed notification delivery yet, or the dispatcher failed to drain.

property calls: tuple[RecordedCall, ...]

Every facade call the SDK made, oldest first.

Includes the notification-bridge installs and the teardown call, so a test can assert on the whole conversation, not just the transfers.

calls_to(method: str) → tuple[RecordedCall, ...]

Return only the recorded calls to one facade method.

Parameters:: method – Facade method name, e.g. "i2c_write".
Returns:: The matching calls, oldest first.

fails(method: str, status: StatusCode | int, *, times: int | None = None) → None

Make one facade command report a firmware failure.

The status travels the real translation path, so the exception a test catches is the same class, with the same status_code and operation, that hardware would have produced.

Parameters:

method – Facade method name, e.g. "i2c_write".
status – The status code to report, e.g. StatusCode.FW_I2C_NACK_ADDRESS. A raw int stands in for a code this SDK release does not name.
times – When None (the default), applies to every call. An integer applies to that many calls – times=1 is the transient-failure-then-recover case.

Raises:

FakeAdapterError – If method is not a facade command, or times is not positive.

property info: DeviceInfo: The identity this adapter reports, as dev.info will show it.

responds(method: str, handler: Callable[[RecordedCall], Any], *, times: int | None = None) → None

Compute one facade command’s outcome from the call itself.

The escape hatch for anything the flat forms cannot express: a register map, a counter, a value derived from the address the caller passed.

Parameters:

method – Facade method name, e.g. "i2c_read".
handler – Called with the RecordedCall. Return a logical result to succeed, or a pycosmicsdk.StatusCode other than SUCCESS to report a firmware failure.
times – When None (the default), applies to every call. An integer applies to that many calls.

Raises:

FakeAdapterError – If method is not a facade command, or times is not positive.

returns(method: str, value: Any, *, times: int | None = None) → None

Make one facade command succeed with value.

Parameters:

method – Facade method name, e.g. "i2c_read".
value – The logical result – the bytes a read yields, the addresses a scan finds, the int a voltage reader reports. It is re-encoded per surface, so the same value serves the blocking and the asyncio device.
times – When None (the default), applies to every call. An integer applies to that many calls, consumed before whatever default is in place.

Raises:

FakeAdapterError – If method is not a facade command, or times is not positive.

class pycosmicsdk.testing.RecordedCall(method: str, args: tuple[Any, ...] = ())

Bases: object

One facade call the fake adapter serviced.

method

The facade method name, e.g. "i2c_write". These are the low-level names of the device facade, not the public interface method names – one public call may issue more than one of them.

Type:: str

args

Positional arguments the SDK passed, already unwrapped to plain Python values (enums arrive as their int values, payloads as bytes). On the asyncio surface the leading transaction id is stripped, so args is the same tuple on either surface.

Type:: tuple[Any, …]

class pycosmicsdk.testing.I2cScanFound(addresses_7bit: Sequence[int] = (), addresses_10bit: Sequence[int] = ())

Bases: object

The addresses a faked I2C bus scan reports.

Program it with adapter.returns("i2c_scan_bus", I2cScanFound(...)). A plain sequence of ints is accepted in its place and read as 7-bit addresses, which is the common case:

adapter.returns("i2c_scan_bus", [0x50, 0x51])

addresses_7bit

7-bit addresses that acknowledged.

Type:: collections.abc.Sequence[int]

addresses_10bit

10-bit addresses that acknowledged. Only reported when the caller asked for them.

Type:: collections.abc.Sequence[int]

class pycosmicsdk.testing.MemoryArena(size: int, register_bytes: int = 1, overflow: ~pycosmicsdk._cpp._pycosmicsdk_cpp.StatusCode | int | None = None, _store: bytearray = <factory>)

Bases: object

A block of target memory a fake adapter serves reads and writes from.

Build one with memory_arena rather than constructing it directly. Install on_write and on_read on the write and read commands of the target under test with adapter.responds, then assert against contents or against what the read command returned.

size

Capacity in bytes.

Type:: int

register_bytes

Bytes one addressable unit spans. Addresses are multiplied by this to reach a byte offset, as firmware does for the I3C target’s memory layout. 1 for a byte-addressed arena.

Type:: int

overflow

Status reported when a transfer runs past the end of the arena, or None to raise FakeAdapterError instead of guessing a status. See memory_arena for why there is no default.

Type:: pycosmicsdk._cpp._pycosmicsdk_cpp.StatusCode | int | None

property contents: bytes: The whole arena as it stands, for asserting against.

preload(address: int, data: bytes) → None

Seed part of the arena from the test, without going through a command.

Use it for the starting state a test needs before the code under test runs – the register defaults a driver expects to read back. A preload is not a facade call and does not appear in adapter.calls.

Parameters:

address – Address to start at, in the arena’s addressing units.
data – Bytes to store.

Raises:

FakeAdapterError – If the write would run past the end of the arena. A preload is the test’s own doing, so this is always an error rather than a programmed status.

on_write(call: RecordedCall) → object

Serve one target-memory write command. Pass to adapter.responds.

Parameters:: call – The recorded call, whose arguments are (address, data).
Returns:: None for a write that fits, or the overflow status for one that does not.
Raises:: FakeAdapterError – If the write runs past the end of the arena and no overflow status was chosen.

on_read(call: RecordedCall) → object

Serve one target-memory read command. Pass to adapter.responds.

Parameters:: call – The recorded call, whose arguments are (address, length).
Returns:: The bytes held at that address, or the overflow status when the read runs past the end of the arena.
Raises:: FakeAdapterError – If the read runs past the end of the arena and no overflow status was chosen.

pycosmicsdk.testing.memory_arena(size: int, *, register_bytes: int = 1, fill: int = 0x00, overflow: StatusCode | int | None = None) → MemoryArena

Build a target-memory arena a fake adapter can serve reads and writes from.

Parameters:

size – Capacity in bytes. Use the capacity of the target under test – the I3C target exposes 1024 bytes, and the I2C and SPI targets report their arena capacity from the bring-up call.
register_bytes – Bytes one addressable unit spans, for a target whose addresses are register indices rather than byte offsets. 1 (the default) is a byte-addressed arena; the I3C target’s memory layout selects 1, 2 or 4.
fill – Byte value the arena starts out holding. 0xFF matches an erased flash or EEPROM part.
overflow – Status a transfer running past the end of the arena reports. None (the default) raises FakeAdapterError instead, because which status firmware reports differs per protocol – and for the I3C target the SDK’s own validation refuses an out-of-range address before any command is sent, so the fake never sees one. Pass the code the protocol under test reports if that case is what the test is about.

Returns:

An empty MemoryArena, ready to install with adapter.responds.

Raises:

FakeAdapterError – If size is not positive, register_bytes is not positive, or fill is not a byte value.

pycosmicsdk.testing.bulk_capabilities(*, groups: Mapping[int, int], commands: Iterable[int] | None = None, supported: bool = True, struct_version: int = 1) → tuple[int, bool, tuple[tuple[int, int, int], ...]]

Build the capability reply a device with a bulk plane answers with.

The reply is the plain (struct_version, supported, groups) tuple the negotiation consumes, so it can be returned straight from a duck-typed facade’s get_bulk_capabilities and will be programmable with adapter.returns("get_bulk_capabilities", ...) once the compiled facade carries the command.

Parameters:

groups – Protocol group id to that group’s payload ceiling in bytes – {BulkGroup.SPI: SUPERNOVA_BULK_ARENA_BYTES} for a Supernova, or PULSAR_BULK_ARENA_BYTES for a Pulsar. Prefer those constants to a literal: the 2048 that appears in several of our documents was Pulsar’s figure for exactly one commit, and the ceiling is equal to the arena by construction on both models. The ceiling is per group and per call; it is not a bound the host can apply uniformly to every command in the group – see BULK_TRANSMIT_CEILINGS. It is masked to the 16 bits the wire field carries, so 65_536 becomes 0 exactly as the device would report it.
commands – Which bulk commands the device advertises. None (the default) advertises every bulk command upstream defines for each named group, which is what a device implementing the group’s bulk surface reports. Pass an explicit list to fake a device that implements only part of it.
supported – Whether the device implements the capability query at all. False reproduces a device that answered, but said no – which no build carrying the handler can do. Firmware computes it as (groupCount > 0) ? 1u : 0u and increments groupCount unconditionally for two groups, so on the wire it is a constant 1 and proves nothing about whether the endpoints exist. It stays settable because the SDK has to read it: firmware that predates the query does not answer at all, and the SDK records silence, an error and an explicit no identically – which is no_bulk_capabilities. For the device a real host meets, use webusb_enabled=False on the BulkProfile instead.
struct_version – Wire format version the device reports. Carried through the negotiation untouched; nothing compares it yet.

Returns:

(struct_version, supported, groups), groups being a tuple of (group, max_payload_length, command_mask) triples in the order groups supplied them.

Raises:

ValueError – If a ceiling is negative.

pycosmicsdk.testing.no_bulk_capabilities() → tuple[int, bool, tuple[tuple[int, int, int], ...]]

Build the reply the SDK records for a device with no bulk plane.

This is the majority case and the one a FakeAdapter already produces unaided. A device whose firmware predates the capability command does not answer it at all, and the SDK reads silence, an error, and an explicit “no” identically: supports nothing. It records that the query ran before issuing it, so an unanswerable device is asked exactly once per connection and every later call is free – and there is no retry. A test that needs the negotiation to run again must close and reopen the device.

Returns:: (0, False, ()) – struct version zero, unsupported, no groups.

class pycosmicsdk.testing.BulkGroup(*values)

Bases: IntEnum

Protocol group a bulk capability entry describes.

The GROUP_* macros of definitions.h, restricted to the three groups a bulk entry can currently describe. A capability reply carries one entry per group the device supports, and the payload ceiling is a property of the group rather than of an individual command.

I2C: GROUP_I2C. Advertised for the I2C target arena commands.

SPI: GROUP_SPI. Advertised for the SPI controller bulk pair and the SPI target arena commands.

I3C: GROUP_I3C. Advertised by a Supernova for I3C_CONTROLLER_PRIVATE_TRANSFER_BULK, which the device implements and no CosmicSDK branch wraps – so the group is real on the wire and unconsumable here. See UnboundBulkCommand. A Pulsar has no I3C peripheral and advertises no I3C group at all.

class pycosmicsdk.testing.BulkCommand(*values)

Bases: IntEnum

One bulk command, by its full 16-bit command code.

A device advertises support per command, not per group: each group entry carries a mask whose bit code & 0x7F is set for every bulk command the device implements. Pass these to bulk_capabilities rather than computing the bits by hand.

SPI_CONTROLLER_WRITE_BULK: CMD_SPI_CONTROLLER_WRITE_BULK, the large SPI write whose MISO bytes are discarded.

SPI_CONTROLLER_READ_BULK: CMD_SPI_CONTROLLER_READ_BULK, a short command phase followed by a large read.

I2C_TARGET_WRITE_MEMORY: CMD_I2C_TARGET_WRITE_MEMORY, preloading the I2C target arena.

I2C_TARGET_READ_MEMORY: CMD_I2C_TARGET_READ_MEMORY, reading the I2C target arena back.

SPI_TARGET_WRITE_MEMORY: CMD_SPI_TARGET_WRITE_MEMORY, preloading the SPI target arena.

SPI_TARGET_READ_MEMORY: CMD_SPI_TARGET_READ_MEMORY, reading the SPI target arena back.

exception pycosmicsdk.testing.FakeAdapterError

Bases: RuntimeError

Raised when a fake adapter is driven in a way no device could be.

Subclasses RuntimeError, so except RuntimeError catches it. It is never raised to represent something a real host adapter would report — those always surface as the real pycosmicsdk.errors.CosmicError subclasses.

pycosmicsdk.testing.uart_rx(data: bytes) → UartRxNotification

Build a UART receive notification.

data is the received message and is what a subscriber should read. The inherited payload is the raw frame that message arrived in, which is UART_RX_FRAME_HEADER_BYTES longer – see uart_rx_frame, and note that this is the shape hardware produces, not a convenience of the fake.

Parameters:: data – The message the peer sent.
Returns:: A UartRxNotification whose data is data and whose payload is uart_rx_frame(data).

pycosmicsdk.testing.uart_rx_frame(data: bytes) → bytes

Build the raw frame a UART receive notification arrives in.

This is what the notification’s inherited ``payload`` really holds, and the reason uart_rx reports two different byte strings: data is the received message, payload is this frame – the same message with UART_RX_FRAME_HEADER_BYTES of header in front of it. The extension fills the base class’s member from upstream’s Notification::payload / payloadLength, which is the entire raw frame (native/facade/facade_util.cpp:49-:54), while data is a pointer offset past the header onto the flexible array member (CosmicSDK.cpp:223:uartRxCache). So a handler that reads payload for the received bytes gets eight bytes of header ahead of them, and len(payload) is eight more than the message length.

Every field of the header is established rather than assumed:

id is 0x00, NOTIFICATION_ID (definitions.h:72), which is what marks a frame unsolicited rather than a reply echoing a request id.
code is 0x1081, CMD_UART_NOTIFICATION resolved through MAKE_COMMAND_CODE – not the 0x1401 upstream’s doc comment on definitions.h:276:CMD_UART_NOTIFICATION claims, which is copy-pasted onto all six UART codes and lands in the GPIO group.
result is 0x0000, SUCCESS. Firmware loads it once when it arms the RX buffer and never recomputes it: uart_mgr.c:627-:629 hold the only writes to that header anywhere in the UART manager (fw-hostadapters @ origin/develop-4.5.0). There is no error result on this path – a UART RX notification is always SUCCESS.
payloadLength is len(data): firmware counts it up one per received byte (uart_mgr.c:320) and sends exactly sizeof(CommandResponseHeader_t) + sizeof(UartTransferParameters_t) + payloadLength bytes (uart_mgr.c:770), which is this frame.

Frames shorter than the header degrade silently. Notification::uartRx() returns nullptr when payloadLength is under sizeof(UartNotification) (CosmicSDK.cpp:211:UartNotification) and build_typed_notification then falls through to pycosmicsdk.UnknownNotification rather than raising. UART_RX_FRAME_HEADER_BYTES is that floor.

Parameters:: data – The received message the frame carries.
Returns:: UART_RX_FRAME_HEADER_BYTES of header, then data.
Return type:: The packed frame

pycosmicsdk.testing.i3c_ibi(*, address: int, payload: bytes = b'') → I3cIbiNotification

Build an I3C in-band interrupt notification.

Parameters:

address – Dynamic address of the target raising the IBI.
payload – Mandatory data bytes the target sent with it, if any.

Returns:

An I3cIbiNotification reporting that request.

pycosmicsdk.testing.i3c_hot_join(*, dynamic_address: int, pid: bytes = bytes(6), bcr: int = 0, dcr: int = 0) → I3cHotJoinNotification

Build an I3C hot-join notification.

Parameters:

dynamic_address – Address the controller assigned to the joining target.
pid – The target’s 48-bit provisional id, six bytes.
bcr – The target’s bus characteristics register.
dcr – The target’s device characteristics register.

Returns:

An I3cHotJoinNotification reporting that join.

pycosmicsdk.testing.i3c_target_bus_event(*, kind: I3cTargetBusEventKind | int, target_address: int = 0, memory_address: int = 0, payload: bytes = b'') → I3cTargetBusEventNotification

Build an I3C target-mode bus-event notification.

This is what an adapter running in target mode reports for every transfer, CCC and address change an external controller performs on it.

Parameters:

kind – What the controller did, e.g. I3cTargetBusEventKind.WRITE_TRANSFER. A raw int is accepted so a test can inject a kind this SDK release does not name.
target_address – Dynamic address this target answered on.
memory_address – Register index the transfer touched.
payload – Bytes the transfer carried.

Returns:

An I3cTargetBusEventNotification reporting that event.

pycosmicsdk.testing.gpio_interrupt(*, pin: GpioPinNumber | int, edge: GpioTriggerType | int, level: GpioLogicLevel | int = GpioLogicLevel.HIGH) → GpioInterruptNotification

Build a GPIO interrupt notification.

Parameters:

pin – The pin that fired. Pass the GpioPinNumber member, not the printed pin label – firmware reports the zero-based index, so GpioPinNumber.PIN_3 is 2.
edge – The edge that fired, e.g. GpioTriggerType.RISING_EDGE.
level – Level read on the pin after the edge.

Returns:

A GpioInterruptNotification reporting that edge.

pycosmicsdk.testing.i3c_connector_event(*, port: I3cConnectorPort | int, event: I3cConnectorEvent | int, connector_type: I3cConnectorType | int = 0) → I3cConnectorEventNotification

Build an I3C connector plug or unplug notification.

Parameters:

port – Which I3C port the accessory was attached to or removed from.
event – Whether it was plugged in or unplugged.
connector_type – What kind of accessory the firmware detected.

Returns:

An I3cConnectorEventNotification reporting that change.

pycosmicsdk.testing.unknown_notification(payload: bytes = b'') → UnknownNotification

Build the notification the SDK reports for a packet it cannot type.

Use it to check that application code tolerates an event from a firmware newer than the SDK.

Parameters:: payload – The raw packet bytes.
Returns:: An UnknownNotification carrying payload.