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 | Concept | Description | |---------|-------------| | `Cown(value)` | A concurrently-owned wrapper. Behaviors receive exclusive temporal access to the cown's `.value`. | | `@when(*cowns)` | Decorator that schedules the function as a behavior. The decorator replaces the function with a `Cown` holding the return value. The first N parameters bind to the N cowns; every trailing parameter **must carry a default**, whose value is snapshotted as a capture at schedule time (see below). | | `quiesce(timeout=None)` | Blocks until all in-flight behaviors complete **without** tearing down the workers. The preferred test-thread barrier before reading results. | | `Cown.unwrap()` | After quiescence, returns the behavior's result value, or **re-raises** the exception the behavior captured (verbatim, on the test thread). Raises `RuntimeError` if called while behaviors are still in flight. | | `send(tag, contents)` | Sends a cross-interpreter message with the given tag. | | `receive(tags, timeout)` | Blocks until a message with a matching tag arrives (or times out). Returns `(TIMEOUT, None)` on timeout. | | `TIMEOUT` | Sentinel string returned as the tag by `receive` when a timeout elapses. | | `wait(timeout)` | Blocks until all scheduled behaviors have completed, then tears the runtime down. | ### 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`. ```python # 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: ```python 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: ```python 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: ```python 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. ```python # 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`. ```python 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: ```python 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: ```python 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. ```python 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. ```python 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: | `@when(...)` arguments | Behavior parameters | |------------------------|---------------------| | `@when(list_of_cowns)` | `(group: list[Cown])` | | `@when(cowns[:9], cowns[9])` | `(group: list[Cown], single: Cown)` | | `@when(cowns[0], cowns[1:])` | `(single: Cown, group: list[Cown])` | | `@when(cowns[:4], cowns[4], cowns[5:])` | `(g0: list[Cown], single: Cown, g1: list[Cown])` | | `@when(cowns[0], cowns[1:9], cowns[9])` | `(s0: Cown, group: list[Cown], s1: Cown)` | ### Full group example ```python 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`: ```python 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. ## Pattern 5 — Exception Propagation If a behavior raises, the exception is captured in the returned cown's `.value`