Testing With Boc

Testing with Behavior-Oriented Concurrency (BOC)

This skill describes how to write tests for code that uses the bocpy library for behavior-oriented concurrency. BOC schedules work as behaviors — decorated functions that run once all required cowns (concurrently-owned data) are available. Testing BOC programs requires specific patterns because behaviors execute asynchronously on worker interpreters.

Design guidance: if you find yourself reaching for time.sleep, threading.Event, polling loops, or wait_for_* helpers inside a behavior or in test code that drives one, stop and read thinking-in-boc first. The right answer is almost always to express the dependency through the cown graph, not through a classical synchronization primitive.

Key Concepts

Cown count, parameter count, and captures

The first N parameters of the decorated function bind positionally to the N arguments of @when. Every additional trailing parameter must carry a default value; that default is evaluated when the def runs and snapshotted as the capture at schedule time:

• def b(c, factor=factor) — captures factor (read from the enclosing scope at schedule time).
• def b(c, i=i) — captures i (the canonical loop-snapshot idiom).
• def b(c, x=y) — captures y and binds it into param x.

The runtime enforces this at decoration time:

• A bare extra parameter (def b(c, factor) — no default) raises TypeError: the capture count must equal the trailing-parameter count.
• A closure over an enclosing-function local raises SyntaxError — a behavior runs in another interpreter and cannot capture by closure. The error names the variable and suggests the x=x fix. (A plain module-level global is not a closure: the bindings reducer keeps module-level assignments, so a behavior may read them directly. The closure rule only bites names bound in an enclosing function — the common case when a behavior is written inside a test method.)
• async def / generator behaviors raise SyntaxError.

Computed defaults (def b(c, k=expensive())) are allowed — the value is evaluated once and snapshotted like any other capture. A default on a cown position (def b(c=c) under @when(x)) is just a capture-count mismatch and raises TypeError.

# CORRECT — 1 @when arg, 1 function param
@when(x)
def good(x):
    return x.value * 2

# CORRECT — extra params beyond the cown count are captures and must
# carry a default. The default is snapshotted at schedule time.
def schedule(x):
    factor = 2
    @when(x)
    def with_extra(x, factor=factor):   # capture ``factor`` via default
        return x.value * factor

# WRONG — bare extra parameter (no default) raises TypeError
@when(x)
def missing_default(x, factor):         # TypeError at decoration
    return x.value * factor

# WRONG — closing over an enclosing-function local raises SyntaxError
def schedule_bad(x):
    factor = 2
    @when(x)
    def via_closure(x):                  # SyntaxError: closes over 'factor'
        return x.value * factor          # pass it as ``factor=factor`` instead

The def _(c, i=i) loop-capture idiom is supported

The canonical Python idiom for snapshotting a loop variable as a default argument works transparently:

for i, c in enumerate(cowns):
    @when(c)
    def _(c, i=i):                      # ``i`` captured at schedule time
        send("done", i)

Parameters beyond the cown count are captures, snapshotted at schedule time. A capture is a trailing parameter carrying a default whose value is read when the behavior is scheduled: def b(c, i=i) captures i (the rename form def b(c, x=y) captures y). A bare extra parameter (def b(c, factor)) raises TypeError at decoration time, and a closure over a free variable raises SyntaxError — a behavior runs in another interpreter and cannot capture by closure.

A behavior cannot reference a loop variable by closure; you must capture it explicitly:

for i, c in enumerate(cowns):
    @when(c)
    def _(c, i=i):                      # capture i; bare `i` reference would fail
        send("done", i)

See the "Inspecting the Worker Bindings Module" section of .github/copilot-instructions.md for how @when captures work and how to inspect the bindings module workers import.

If you want a fresh scope per iteration (e.g. to avoid sharing mutable state between iterations), use a helper function:

def _schedule(c, i):                    # fresh scope per iteration
    @when(c)
    def _(c):
        send("done", i)

for i, c in enumerate(cowns):
    _schedule(c, i)

Critical rule: classes and functions must be declared at module level

Behaviors run in separate sub-interpreters. Workers import the module's bindings, which means any class or function referenced inside a @when behavior must be defined at module level. A class defined inside a test method or local function cannot be resolved by the worker and will crash.

# CORRECT — class at module level
class Accumulator:
    def __init__(self):
        self.items = []
    def add(self, item):
        self.items.append(item)

def test_accumulator(self):
    acc = Cown(Accumulator())

    @when(acc)
    def _(a):
        a.value.add(42)       # Accumulator is importable ✓

# WRONG — class inside test method
def test_accumulator_bad(self):
    class Accumulator:        # local class — worker can't import it
        ...
    acc = Cown(Accumulator())

    @when(acc)
    def _(a):
        a.value.add(42)       # will crash

Project Test Setup

• Tests use pytest and live in the test/ directory.
• Install test dependencies: pip install -e .[test]
• Run: pytest -vv

Pattern 1 — quiesce() + unwrap() (default)

This is the preferred pattern for result-shipping assertions: the behavior computes a value (or raises), and the test thread verifies it. A behavior returns its result, so @when hands back a Cown holding that result. The test blocks once on quiesce() (which lets every in-flight behavior finish without tearing down the workers), then reads the result with Cown.unwrap().

unwrap() returns the stored value on success, and re-raises any exception the behavior captured — verbatim, on the test thread — so a failing assertion inside a behavior surfaces as a real AssertionError in the test. It also guards against misuse: calling it while behaviors are still in flight raises RuntimeError, so always quiesce() first.

Always pass a timeout to quiesce(). If the barrier is not reached in time (e.g. a behavior never fires because of a @when arg-count mismatch) it raises TimeoutError, so the test fails fast instead of hanging forever. Use a module-level constant such as QUIESCE_TIMEOUT = 5.

from bocpy import Cown, when, quiesce, wait

QUIESCE_TIMEOUT = 5  # seconds; quiesce() raises TimeoutError if exceeded


class TestExample:
    @classmethod
    def teardown_class(cls):
        wait()  # tear the runtime down after all tests

    def test_double(self):
        x = Cown(3)

        @when(x)
        def result(x):
            return x.value * 2

        quiesce(QUIESCE_TIMEOUT)
        assert result.unwrap() == 6

Tuple results read just as naturally:

    def test_pair(self):
        v = Cown(Matrix(1, 2, [3.0, 4.0]))

        @when(v)
        def result(v):
            n = v.value.normalize()
            return (n[0, 0], n[0, 1])

        quiesce(QUIESCE_TIMEOUT)
        n0, n1 = result.unwrap()
        assert n0 == pytest.approx(0.6)
        assert n1 == pytest.approx(0.8)

Asserting a behavior raised

Because unwrap() re-raises the captured exception, use pytest.raises to assert that a behavior failed:

    def test_raises(self):
        x = Cown(1)

        @when(x)
        def boom(x):
            raise ValueError("bad input")

        quiesce(QUIESCE_TIMEOUT)
        with pytest.raises(ValueError, match="bad input"):
            boom.unwrap()

Once unwrap() consumes an exception it clears the cown's exception flag, so the runtime will not later report it as unhandled, and a second unwrap() returns the (now None) value rather than re-raising.

Why this pattern?

@when returns immediately — the behavior has not executed yet. quiesce() is the test-thread barrier that lets it (and any behaviors it chains) finish. Unlike wait(), it leaves the workers alive, so the result cowns remain readable and further behaviors can still be scheduled. Reading results directly avoids the message-queue round-trip and per-assert timeout polling that a send/receive-based assertion would require.

Pattern 2 — Testing Nested / Chained Behaviors

Behaviors can schedule further behaviors. A behavior that chains more work can return the inner result cown; after quiesce() the whole chain has run, so you unwrap the final result on the test thread.

def test_nested(self):
    x = Cown(1)

    @when(x)
    def step1(x):
        x.value *= 2          # x is now 2

        @when(x)
        def step2(x):
            x.value *= 3      # x is now 6
            return x.value

        return step2

    quiesce(QUIESCE_TIMEOUT)
    # step1 returns the inner step2 cown; unwrap it to read the final value.
    assert step1.unwrap().unwrap() == 6

Pattern 3 — Multi-Cown Coordination

Pass multiple cowns to @when to atomically operate on several pieces of data. The scheduler guarantees deadlock-free acquisition.

def test_transfer(self):
    x = Cown(100)
    y = Cown(0)

    @when(x, y)
    def _(x, y):
        y.value += 50
        x.value -= 50

    @when(x)
    def read_x(x):
        return x.value

    @when(y)
    def read_y(y):
        return y.value

    quiesce(QUIESCE_TIMEOUT)
    assert read_x.unwrap() == 50
    assert read_y.unwrap() == 50

Pattern 4 — Cown Grouping

When you have a dynamic number of cowns (e.g., a list), you can pass them to @when as a list (or slice) rather than individual arguments. Inside the behavior, that parameter is delivered as a list[Cown] — each element is an acquired cown whose .value you can read or write.

You can mix single cowns and groups freely in any order. Each distinct argument to @when becomes its own parameter in the decorated function:

Full group example

from bocpy import Cown, when, quiesce

cowns = [Cown(i) for i in range(10)]  # values 0..9, sum = 45

# All cowns as a single group
@when(cowns)
def group_sum(group: list[Cown[int]]):
    return sum(c.value for c in group)

# Group + single cown
@when(cowns[:9], cowns[9])
def group_then_single(group: list[Cown[int]], single: Cown[int]):
    return sum(c.value for c in group) + single.value

# Single cown + group
@when(cowns[0], cowns[1:])
def single_then_group(single: Cown[int], group: list[Cown[int]]):
    return single.value + sum(c.value for c in group)

# Group + single + group
@when(cowns[:4], cowns[4], cowns[5:])
def group_single_group(g0: list[Cown[int]], single: Cown[int], g1: list[Cown[int]]):
    return sum(c.value for c in g0) + single.value + sum(c.value for c in g1)

Testing grouped results

The results are all cowns, so collect them after quiesce() with unwrap(). You can also pass a list of result cowns as a group to a follow-up @when:

def test_cown_grouping(self):
    expected = 45
    results = [group_sum, group_then_single, single_then_group, group_single_group]

    quiesce(QUIESCE_TIMEOUT)
    for r in results:
        assert r.unwrap() == expected

Key rules for grouping

• Pass a list (or slice) of cowns to @when — the behavior receives the corresponding parameter as list[Cown].
• Pass a single cown — the parameter receives that Cown directly.
• You can interleave singles and groups in any order. The positional mapping between @when(...) arguments and the decorated function's parameters is 1:1.
• Type-annotate grouped parameters as list[Cown[T]] for clarity.