Kamal: Created page with "Unit tests run a state machine against a simulated trading environment, exercising the paths of its specification. Tests feed the machine its..."

2026-07-13T19:00:55Z

Created page with "Unit tests run a state machine against a simulated trading environment, exercising the paths of its specification. Tests feed the machine its..."

New page

Unit tests run a state machine against a simulated trading environment,
exercising the paths of its [[State Machine Specifications|specification]].
Tests feed the machine its events one at a time and assert the machine's
reactions: what it submits, what it cancels, and how it terminates, etc... This
article covers the test environment, how to drive it, and the synchronization
rules that keep tests deterministic. It continues the <code>PeggedOrder</code>
example from [[State Machine Implementation]].

A module's tests live beside it in <code><module>_tester.py</code>, written
with <code>unittest</code>:

<pre>
import unittest

import beam
import nexus

import pegged_order
</pre>

Imports are grouped: standard library, then platform modules, then local
modules.

== The test environment ==

<code>nexus.TestEnvironment</code> simulates the entire trading platform, market
data, order execution, and time. <code>nexus.TestClients</code> gives
the machine under test the same <code>clients</code> interface it uses in
production. The machine can't tell the difference as the test plays the role
of both the market and the venue.

<pre>
class TestPeggedOrder(unittest.TestCase):
def setUp(self):
self.environment = nexus.TestEnvironment()
self.clients = nexus.TestClients(self.environment)
self.ticker = nexus.parse_ticker('ABX.TSX')
self.submissions = beam.Queue()
self.environment.monitor_order_submissions(self.submissions)
</pre>

<code>monitor_order_submissions</code> publishes every order any machine
submits into a queue for review. It is backed by a snapshot publisher, so
monitoring before or after a submission both work.

Machines that load reference data need it populated before they start. For
example, a TWAP order might need to know a ticker symbol's board lot, in that
case the <code>setUp</code> method should populate the market data environment's
test data store with a <code>TickerInfo</code> object that specifies the board
lot and any other information that is anticipated to be loaded:

<pre>
self.environment.get_market_data_environment().data_store.store(
nexus.TickerInfo(self.ticker, 'Barrick Gold Corp', '', 100))
</pre>

== Anatomy of a scenario ==

Each test is one scenario: a comment block stating the steps imperatively, then
a body that performs exactly those steps. The comments are the scenario's
specification, a reviewer should be able to check the body against them line by
line.

<pre>
# Start a PeggedOrder on the bid for 1000 shares with no limit price and a
# peg difference of 0.01.
# Set the BBO to 1.00 / 1.01.
# Expect a limit bid order submission for 0.99 for 1000 shares, published
# on the orders publisher.
# Accept and fill the order submission.
# Expect the PeggedOrder to terminate and close its orders publisher.
def test_fill(self):
order_fields = nexus.make_market_order_fields(
self.ticker, nexus.Side.BID, 1000)
order = pegged_order.PeggedOrder(
self.clients, order_fields, nexus.Money.CENT)
orders = beam.Queue()
order.orders.monitor(orders)
self.environment.update_bbo_price(
self.ticker, nexus.Money.parse('1.00'), nexus.Money.parse('1.01'))
expected_order = self.submissions.pop()
self.assertEqual(expected_order.info.fields.type, nexus.OrderType.LIMIT)
self.assertEqual(
expected_order.info.fields.price, nexus.Money.parse('0.99'))
self.assertEqual(expected_order.info.fields.quantity, 1000)
self.assertEqual(expected_order.info.fields.side, nexus.Side.BID)
published_order = orders.pop()
self.assertEqual(
published_order.info.id, expected_order.info.id)
self.environment.accept(expected_order)
self.environment.fill(expected_order, 1000)
order.wait()
self.assertRaises(beam.PipeBrokenException, orders.pop)
self.assertIsNone(self.submissions.try_pop())
</pre>

The structure is drive, observe, assert. Calls to the environment play the
market data, a blocking <code>pop</code> waits for the machine's reaction,
and assertions check the reaction's fields. Rinse and repeat.

== Driving the machine ==

* <code>update_bbo_price(ticker, bid, ask)</code> publishes a quote. The machine's subscription uses a current query, so a quote set before the machine subscribed is still delivered from the snapshot.
* <code>accept(order)</code>, <code>reject(order)</code> resolve a pending submission.
* <code>fill(order, quantity)</code> executes shares, a quantity below the order's fills partially and leaves it live, so multi-fill scenarios are just repeated calls.
* <code>cancel(order)</code> confirms a cancel the machine requested — drive it only once the request is observed (below).
* <code>advance_time(duration)</code> moves the simulated clock and triggers any timers whose expiry falls within it, for schedule-driven machines.

== Synchronization ==

Everything the machine does happens asynchronously in its own routine, so
tests synchronize by '''blocking on observations''', never by sleeping.

'''A blocking <code>pop</code> is the primary mechanism.''' It waits until
the machine acts, and a popped value proves everything causally before it
already happened. If the machine never acts, the pop blocks forever — which
is why suites run under a timeout as the backstop
(<code>timeout 60 python3 pegged_order_tester.py</code>): a hang is a failure.

'''Await order state through its reports.''' Asserting a status
point-in-time races the machine's handler. To wait for a cancel request,
monitor the order's reports and pop until the status arrives:

<pre>
reports = beam.Queue()
expected_order.publisher.monitor(reports)
order.cancel()
while reports.pop().status != nexus.OrderStatus.PENDING_CANCEL:
pass
self.environment.cancel(expected_order)
</pre>

'''The one legitimate flush.''' A timer-driven machine registers its first
timer asynchronously; <code>advance_time</code> immediately after
construction can win that race, and a test timer counts from its
<code>start()</code> so the missed window is unrecoverable. Settle the
routines once, after construction, before the first advance:

<pre>
order = market_twap_order.MarketTwapOrder(
self.clients, nexus.make_market_order_fields(
self.ticker, nexus.Side.ASK, 100), datetime.timedelta(minutes=1))
beam.flush_pending_routines()
self.environment.advance_time(datetime.timedelta(seconds=30))
</pre>

Avoid using <code>flush_pending_routines</code> as much as possible in favor of
other synchronization options.

== Testing composition ==

Tests should focus on the side effects of a machine, for example by monitoring
the <code>orders</code> publisher and matching published orders against the
environment's submissions by <code>info.id</code>, and asserting the close. A
normal termination breaks the monitor's queue with
<code>beam.PipeBrokenException</code> while a failure delivers the original
exception:

<pre>
self.environment.reject(expected_order)
order.wait()
orders.pop()
self.assertRaises(RuntimeError, orders.pop)
</pre>

== What to cover ==

Cover every edge of the specification's graph, and beyond the graph, the
policies:

* The happy path, including the surface (publication and normal close).
* Rejection, including the failure close.
* Every condition in both directions ie. the peg's away-move repricing ''and'' the toward-move hold.
* Partial fills, and terminal states that leave shares remaining to be filled.
* Every cancellation lane: while resting, while pending acceptance (resolved by acceptance ''and'' by rejection), while idle, and cancelling an already-completed order as a no-op.
* For schedule-driven machines: pacing between intervals, catch-up after a slow fill, and the overdue remainder when the schedule is exhausted.

A scenario that only exists to make a policy visible is worth having, the
toward-move test asserts that ''nothing'' happens, which is the policy.

== Debugging ==

* '''A silent hang — the suite terminates with no verbose output — usually is not a deadlock.''' It means an assertion raised mid-scenario and teardown wedged on the still-live machine. Diagnose by running tests individually (<code>python3 pegged_order_tester.py TestPeggedOrder.test_fill</code>) before suspecting the machine.
* A probe script that constructs the machine, drives one step, sleeps briefly, and prints <code>order._state</code> localizes where a machine actually stops — cheap and effective when a scenario disagrees with your reading of the specification.

State Machine Unit Testing - Revision history

Kamal: Created page with "Unit tests run a state machine against a simulated trading environment, exercising the paths of its specification. Tests feed the machine its..."