State Machine Unit Testing

Unit tests run a state machine against a simulated trading environment, exercising the paths of its 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 PeggedOrder example from State Machine Implementation.

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

import unittest

import beam
import nexus

import pegged_order

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

The test environment

nexus.TestEnvironment simulates the entire trading platform, market data, order execution, and time. nexus.TestClients gives the machine under test the same clients 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.

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)

monitor_order_submissions 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 setUp method should populate the market data environment's test data store with a TickerInfo object that specifies the board lot and any other information that is anticipated to be loaded:

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

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.

  # 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())

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

Driving the machine

update_bbo_price(ticker, bid, ask) 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.
accept(order), reject(order) resolve a pending submission.
fill(order, quantity) executes shares, a quantity below the order's fills partially and leaves it live, so multi-fill scenarios are just repeated calls.
cancel(order) confirms a cancel the machine requested — drive it only once the request is observed (below).
advance_time(duration) 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 pop 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 (timeout 60 python3 pegged_order_tester.py): 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:

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

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

    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))

Avoid using flush_pending_routines 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 orders publisher and matching published orders against the environment's submissions by info.id, and asserting the close. A normal termination breaks the monitor's queue with beam.PipeBrokenException while a failure delivers the original exception:

    self.environment.reject(expected_order)
    order.wait()
    orders.pop()
    self.assertRaises(RuntimeError, orders.pop)

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 (python3 pegged_order_tester.py TestPeggedOrder.test_fill) before suspecting the machine.
A probe script that constructs the machine, drives one step, sleeps briefly, and prints order._state localizes where a machine actually stops — cheap and effective when a scenario disagrees with your reading of the specification.

Navigation menu