Skip to content

Testing Helpers

chumicro_runner.testing ships three host-side helpers: CallRecorder records handler invocations, validate_service asserts an object has the shape Runner.add expects, and FakePoller stands in for select.poll().ipoll so tests can drive Runner.wait() without real file descriptors. The module declares itself test support, so the deploy walker and the bundle builder drop it and it never lands on a microcontroller.

CallRecorder

CallRecorder is a callable, so it registers anywhere a handler does:

from chumicro_runner import Runner
from chumicro_runner.testing import CallRecorder
from chumicro_timing.testing import FakeTicks

fake = FakeTicks()
recorder = CallRecorder()
runner = Runner(ticks=fake)
runner.add_periodic(recorder, period_ms=100)

# Not due yet, so nothing fires.
runner.tick()
assert len(recorder) == 0

# Advance past the period.
fake.advance(100)
runner.tick()
assert recorder.calls == [100]

recorder.calls is a plain list of the now_ms value from each invocation, and clear() resets it between phases of a test:

assert recorder.calls[0] == 100
assert len(recorder) == 1

recorder.clear()
assert len(recorder) == 0

A handler registered without a period fires on every tick:

recorder = CallRecorder()
runner.add(handler=recorder)
runner.tick()
assert len(recorder) == 1

validate_service

validate_service(service) reads which contract members your service exposes and raises ValueError naming the offending one when the set is incoherent. It checks shape only: it never calls check, handle, or any hook, so it is safe to run against a service that would talk to hardware.

from chumicro_runner.testing import validate_service

class Blinker:
    def check(self, now_ms):
        return True

    def handle(self, now_ms):
        pass

validate_service(Blinker())        # passes, nothing raised

The rules it enforces are the ones Runner dispatch relies on: check and handle are both required; io_socket and io_interest come as a pair; io_error needs an io_socket to report errors on. next_deadline is optional and stands alone.

class HalfWired:
    io_socket = None                # no io_interest to go with it

    def check(self, now_ms):
        return False

    def handle(self, now_ms):
        pass

validate_service(HalfWired())
# ValueError: a service with io_socket must also define io_interest;
# the runner polls the socket only through io_interest

FakePoller

Runner.wait() hands its poll set to a poller object. CPython's real select.poll needs live file descriptors, which in-memory fake sockets do not have, so pass poller=FakePoller() and assert on what the runner did with the poll set. register_calls, modify_calls, unregister_calls, and ipoll_calls record every call; set_ready(obj, eventmask) queues a pair for the next ipoll() return.

import select

from chumicro_runner import IO_READ, Runner
from chumicro_runner.testing import FakePoller
from chumicro_timing.testing import FakeTicks

class ReadService:
    def __init__(self, sock):
        self.io_socket = sock

    def io_interest(self, now_ms):
        return IO_READ

    def check(self, now_ms):
        return False

    def handle(self, now_ms):
        pass

poller = FakePoller()
runner = Runner(ticks=FakeTicks(), poller=poller)
sock = object()
runner.add(ReadService(sock), period_ms=100)

runner.wait(0)

assert (sock, select.POLLIN) in poller.register_calls
assert poller.ipoll_calls == [100]      # idled until the next period

Using these fakes in your own tests

Your test suite imports them straight from the installed package, the same way this library's own tests do:

from chumicro_runner.testing import CallRecorder, FakePoller, validate_service

Project convention: libraries that expose injectable services ship their own test fakes alongside the production code.

API Reference

chumicro_runner.testing

Test helpers for libraries that use chumicro-runner.

Provides validate_service (asserts a service's runner-contract shape), CallRecorder (records handler invocations), and FakePoller (host-test stand-in for select.poll().ipoll).

FakePoller

Host-test fake for select.poll().ipoll.

ipoll(timeout_ms)

Record the call; return whatever set_ready queued.

set_ready(obj, eventmask)

Queue obj / eventmask for the next ipoll return.

CallRecorder

Callable that records each invocation for test assertions.

clear()

Discard all recorded calls.

validate_service(service)

Assert service has a coherent runner-service shape.

Checks shape only, never behavior: it reads which contract members the service exposes and enforces the coherence rules the Runner dispatch relies on. It never calls check / handle / any hook.

The rules the Runner dispatch relies on. Only the first raises on its own; the other two silently no-op when violated, which is why this checker exists:

  • check and handle are both required. Runner.add reads task.check and task.handle unconditionally, so a service missing either cannot register.
  • io_socket and io_interest come as a pair. The poll sync reads interest through io_interest and the socket through the io_socket attribute; one without the other never reaches the poller.
  • io_error requires io_socket. It is dispatched only when the service's io_socket reports a poll error.

next_deadline is optional and stands alone. io_socket is a data attribute (the socket itself, or None before connect); the other members are callables.

Parameters:

Name Type Description Default
service object

The object a consumer would pass to Runner.add.

 required

Raises:

Type Description
ValueError

A required member is missing or a coherence rule is broken; the message names the offending member.

← Home

Source · \ PyPI · \ Bundle · \ Experimental Bundle