indikit.testing

Test a driver by driving it the way a client would - no indiserver, no sockets, no hardware. See Testing without hardware for the guide, and tests/test_weather_example.py for a worked set.

DeviceHarness

indikit.testing

DeviceHarness: drive a driver in a test without indiserver.

A driver's value is in what it emits - the def it announces, the set it publishes from a poll, the message it logs when hardware refuses a command. This module gives that a first-class seam::

async def test_shutter_opens():
    harness = DeviceHarness(MyDome())
    await harness.setup()

    await harness.write("DOME_SHUTTER", SHUTTER_OPEN=True)

    assert harness.latest("DOME_SHUTTER").state is IPState.BUSY
    assert "Shutter" in harness.messages[-1]

No sockets, no subprocess, no XML. Writes go through the device's real dispatch path - the @on_new map, the device-name guard, the serialisation lock - so a handler that works here works under indiserver.

Failures are not isolated here, and that is the one place the harness departs from the runtime on purpose. Under indiserver a raising handler or tick is caught, reported to the client as an ERROR message and stepped over; in a test that would turn a bug into a missing emission and an assertion failure ten lines away from its cause. So :meth:DeviceHarness.write and :meth:DeviceHarness.tick let the exception out with its traceback, and the runtime's isolation is covered where it lives, in tests/test_driver.py.

For coverage of the wire itself (framing, chunk boundaries, the codec), drive a :class:~indikit.driver.runtime.DriverRuntime over byte streams instead; see tests/test_driver.py.

DeviceHarness

DeviceHarness(device: Device, *, config_dir: Path | None = None)

A device plus every message it has emitted, driven like a client would.

Parameters:

Name	Type	Description	Default
device	Device	The device under test. It is bound to this harness on construction, so define_* and set work immediately.	required
config_dir	Path or None	Where the device saves and loads its configuration, standing in for what the runtime would inject. Pass pytest's tmp_path to exercise CONFIG_PROCESS; leave it out and the persistence methods raise :class:~indikit.ConfigError, which is a real driver on a machine with nowhere to save.	None

Attach to device and start recording what it emits.

Source code in src/indikit/testing.py

def __init__(self, device: Device, *, config_dir: Path | None = None) -> None:
    """Attach to ``device`` and start recording what it emits."""
    self._device = device
    self._emitted: list[IndiMessage] = []
    device._bind(self._emitted.append, config_dir=config_dir)

device property

device: Device

The device under test.

config_path property

config_path: Path

The file this device's configuration is saved to.

It need not exist: before a save there is nothing there, and asserting that is most of what a persistence test does.

Returns:

Name	Type	Description
path	Path	<config_dir>/<device>.json.

Raises:

Type	Description
ConfigError	Raised if the harness was built without a config_dir. Also an OSError.

emitted property

emitted: list[IndiMessage]

Every message the device has emitted, in order.

messages property

messages: list[str]

The text of every message the device has sent.

setup async

setup() -> None

Send a getProperties, running the device's :meth:Device.setup.

The same trigger indiserver provides at startup. Calling it again re-announces the already-defined properties, as a late-joining client would see.

Source code in src/indikit/testing.py

async def setup(self) -> None:
    """Send a ``getProperties``, running the device's :meth:`Device.setup`.

    The same trigger ``indiserver`` provides at startup. Calling it again
    re-announces the already-defined properties, as a late-joining client
    would see.
    """
    await self._device._dispatch_get_properties(GetProperties(device=self._device.device))

write async

write(name: str, values: dict[str, Any] | None = None, **kwargs: Any) -> None

Send a client write to property name, as a real client would.

Only the named elements are sent - a partial write, which is what INDI clients actually do - and switch values accept bool or the wire strings as well as :class:~indikit.protocol.ISState.

Parameters:

Name	Type	Description	Default
name	str	The property to write to. It must already be defined.	required
values	dict	Element values keyed by name, for names that are not valid Python identifiers.	None
**kwargs	object	Element values by name (the common case).	{}

Raises:

Type	Description
KeyError	Raised if no property with that name has been defined.

Source code in src/indikit/testing.py

async def write(self, name: str, values: dict[str, Any] | None = None, **kwargs: Any) -> None:
    """Send a client write to property ``name``, as a real client would.

    Only the named elements are sent - a partial write, which is what INDI
    clients actually do - and switch values accept `bool` or the wire
    strings as well as :class:`~indikit.protocol.ISState`.

    Parameters
    ----------
    name : str
        The property to write to. It must already be defined.
    values : dict, optional
        Element values keyed by name, for names that are not valid Python
        identifiers.
    **kwargs : object
        Element values by name (the common case).

    Raises
    ------
    KeyError
        Raised if no property with that name has been defined.
    """
    merged = {**(values or {}), **kwargs}
    defined = self._device[name].vector
    await self._device._dispatch_new(_client_write(defined, merged))

tick async

tick(job: str) -> None

Run one iteration of an @every job, by method name.

Runs the job body directly under the device guard - the schedule is the runtime's business, and a test should not have to wait out an interval to see one tick. A job that raises raises here too, rather than being swallowed the way the runtime swallows it; see the module docstring.

Parameters:

Name	Type	Description	Default
job	str	The name of the @every-decorated method.	required

Raises:

Type	Description
KeyError	Raised if the device has no @every job by that name.

Source code in src/indikit/testing.py

async def tick(self, job: str) -> None:
    """Run one iteration of an ``@every`` job, by method name.

    Runs the job body directly under the device guard - the schedule is the
    runtime's business, and a test should not have to wait out an interval
    to see one tick. A job that raises raises here too, rather than being
    swallowed the way the runtime swallows it; see the module docstring.

    Parameters
    ----------
    job : str
        The name of the ``@every``-decorated method.

    Raises
    ------
    KeyError
        Raised if the device has no ``@every`` job by that name.
    """
    jobs = {method.__name__: method for _, method in iter_periodic(self._device)}
    if job not in jobs:
        raise KeyError(f"{self._device.device} has no @every job named {job!r}")
    async with self._device._guard():
        result = jobs[job]()
        if inspect.isawaitable(result):
            await result

defs

defs(name: str | None = None) -> list[Vector]

Return the vectors the device has defined.

Parameters:

Name	Type	Description	Default
name	str	Restrict to one property name; all of them when omitted.	None

Returns:

Name	Type	Description
vectors	list of Vector	The defined vectors, in emission order.

Source code in src/indikit/testing.py

def defs(self, name: str | None = None) -> list[Vector]:
    """Return the vectors the device has defined.

    Parameters
    ----------
    name : str, optional
        Restrict to one property name; all of them when omitted.

    Returns
    -------
    vectors : list of Vector
        The defined vectors, in emission order.
    """
    return self._vectors(DefVector, name)

sets

sets(name: str | None = None) -> list[Vector]

Return the value updates the device has published.

Parameters:

Name	Type	Description	Default
name	str	Restrict to one property name; all of them when omitted.	None

Returns:

Name	Type	Description
vectors	list of Vector	The published vectors, in emission order.

Source code in src/indikit/testing.py

def sets(self, name: str | None = None) -> list[Vector]:
    """Return the value updates the device has published.

    Parameters
    ----------
    name : str, optional
        Restrict to one property name; all of them when omitted.

    Returns
    -------
    vectors : list of Vector
        The published vectors, in emission order.
    """
    return self._vectors(SetVector, name)

deletes

deletes() -> list[DelProperty]

Return every delProperty the device has sent.

Returns:

Name	Type	Description
deletions	list of DelProperty	The deletions, in emission order.

Source code in src/indikit/testing.py

def deletes(self) -> list[DelProperty]:
    """Return every ``delProperty`` the device has sent.

    Returns
    -------
    deletions : list of DelProperty
        The deletions, in emission order.
    """
    return [msg for msg in self._emitted if isinstance(msg, DelProperty)]

latest

latest(name: str) -> Vector

Return what a client would currently hold for one property.

The last def or set recorded for it, falling back to the device's live vector when nothing has been emitted - after :meth:clear, say, or for a property whose "on_change" policy has had nothing to report. The two agree, because every change a client would care about is exactly what gets published.

Parameters:

Name	Type	Description	Default
name	str	The property name.	required

Returns:

Name	Type	Description
vector	Vector	The latest state of that property.

Raises:

Type	Description
KeyError	Raised if no property with that name has been defined.

Source code in src/indikit/testing.py

def latest(self, name: str) -> Vector:
    """Return what a client would currently hold for one property.

    The last ``def`` or ``set`` recorded for it, falling back to the
    device's live vector when nothing has been emitted - after
    :meth:`clear`, say, or for a property whose ``"on_change"`` policy has
    had nothing to report. The two agree, because every change a client
    would care about is exactly what gets published.

    Parameters
    ----------
    name : str
        The property name.

    Returns
    -------
    vector : Vector
        The latest state of that property.

    Raises
    ------
    KeyError
        Raised if no property with that name has been defined.
    """
    for msg in reversed(self._emitted):
        if isinstance(msg, DefVector | SetVector) and msg.vector.name == name:
            return msg.vector
    return cast("Vector", self._device[name].vector)

clear

clear() -> None

Drop the recorded messages, keeping the device as it is.

Useful between the arrange and act halves of a test, so assertions run against what one action produced rather than the whole history.

Source code in src/indikit/testing.py

def clear(self) -> None:
    """Drop the recorded messages, keeping the device as it is.

    Useful between the arrange and act halves of a test, so assertions run
    against what one action produced rather than the whole history.
    """
    self._emitted.clear()