Rewrite order ctl tests as a parametrization

More or less a complete rework which allows passing a detailed
clearing/fills input and allows for *not* rebooting the runtime / ems
between each position check.

Some further enhancements:
- use (unit) fractional sizes to simulate both the more realistic and
  more "complex position calculation" case; since this is crypto.
- add a no-fqme-found test.
- factor cross-session/offline pos storage (pps.toml) checks into
  a `load_and_check_pos()` helper which does all entry loading directly
  from a provided `BrokerdPosition` msg.
- use the new `OrderClient.send()` async api.

tests/test_paper.py

@@ -18,7 +18,6 @@ from exceptiongroup import BaseExceptionGroup
 import pytest
 import tractor
 from uuid import uuid4
-from functools import partial
 
 from piker.service import Services
 from piker.log import get_logger
@@ -53,14 +52,15 @@ async def open_pikerd(
         yield services
 
 
-async def submit_order(
+async def order_and_and_wait_for_ppmsg(
     client: OrderClient,
     trades_stream: tractor.MsgStream,
     fqme: str,
+
     action: Literal['buy', 'sell'],
-    price: float = 30000.,
+    price: float = 100e3,  # just a super high price.
-    executions: int = 1,
     size: float = 0.01,
+
     exec_mode: str = 'live',
     account: str = 'paper',
 
@@ -73,11 +73,9 @@ async def submit_order(
     sent: list[Order] = []
     broker, key, suffix = unpack_fqme(fqme)
 
-    for _ in range(executions):
-
     order = Order(
         exec_mode=exec_mode,
-            action=action,
+        action=action,  # TODO: remove this from our schema?
         oid=str(uuid4()),
         account=account,
         size=size,
@@ -107,17 +105,17 @@ async def submit_order(
     return sent, msgs
 
 
-def run_and_catch(
+def run_and_tollerate_cancels(
     fn: Callable[..., Awaitable],
 
-    expect_errs: tuple[Exception] = (
+    expect_errs: tuple[Exception] | None = None,
-        KeyboardInterrupt,
+    tollerate_errs: tuple[Exception] = (tractor.ContextCancelled,),
-        tractor.ContextCancelled,
-    )
 
 ):
     '''
-    Close position and assert empty position in pps
+    Run ``trio``-``piker`` runtime with potential tolerance for
+    inter-actor cancellation during teardown (normally just
+    `tractor.ContextCancelled`s).
 
     '''
     if expect_errs:
@@ -127,7 +125,10 @@ def run_and_catch(
         for err in exc_info.value.exceptions:
             assert type(err) in expect_errs
     else:
+        try:
             trio.run(fn)
+        except tollerate_errs:
+            pass
 
 
 @cm
@@ -139,6 +140,12 @@ def load_and_check_pos(
 
     with open_pps(ppmsg.broker, ppmsg.account) as table:
 
+        if ppmsg.size == 0:
+            assert ppmsg.symbol not in table.pps
+            yield None
+            return
+
+        else:
             # NOTE: a special case is here since the `PpTable.pps` are
             # normally indexed by the particular broker's
             # `Position.bs_mktid: str` (a unique market / symbol id provided
@@ -154,7 +161,7 @@ def load_and_check_pos(
 
 @pytest.mark.trio
 async def test_ems_err_on_bad_broker(
-    open_pikerd: Services,
+    open_test_pikerd: Services,
     loglevel: str,
 ):
     try:
@@ -168,9 +175,60 @@ async def test_ems_err_on_bad_broker(
         pass
 
 
-async def atest_buy(
+async def match_ppmsgs_on_ems_boot(
-    loglevel: str,
+    ppmsgs: list[BrokerdPosition],
+
+) -> None:
+    '''
+    Given a list of input position msgs, verify they match
+    what is loaded from the EMS on connect.
+
+    '''
+    by_acct: dict[tuple, list[BrokerdPosition]] = {}
+    for msg in ppmsgs:
+        by_acct.setdefault(
+            (msg.broker, msg.account),
+            [],
+        ).append(msg)
+
+    # TODO: actually support multi-mkts to `open_ems()`
+    # but for now just pass the first fqme.
+    fqme = msg.symbol
+
+    # disconnect from EMS, reconnect and ensure we get our same
+    # position relayed to us again in the startup msg.
+    async with (
+        open_ems(
+            fqme,
+            mode='paper',
+            loglevel='info',
+        ) as (
+            _,  # OrderClient
+            _,  # tractor.MsgStream
+            startup_pps,
+            accounts,
+            _,  # dialogs,
+        )
     ):
+        for (broker, account), ppmsgs in by_acct.items():
+            assert account in accounts
+
+            # lookup all msgs rx-ed for this account
+            rx_msgs = startup_pps[(broker, account)]
+
+            for expect_ppmsg in ppmsgs:
+                rx_msg = BrokerdPosition(**rx_msgs[expect_ppmsg.symbol])
+                assert rx_msg == expect_ppmsg
+
+
+async def submit_and_check(
+    fills: tuple[dict],
+    loglevel: str,
+
+) -> tuple[
+    BrokerdPosition,
+    Position,
+]:
     '''
     Enter a trade and assert entries are made in pps and ledger files.
 
@@ -203,125 +261,108 @@ async def atest_buy(
         assert not startup_pps
         assert 'paper' in accounts
 
-        sent, msgs = await submit_order(
+        od: dict
+        for od in fills:
+            print(f'Sending order {od} for fill')
+            sent, msgs = await order_and_and_wait_for_ppmsg(
                 client,
                 trades_stream,
                 fqme,
                 action='buy',
-            size=0.01,
+                size=od['size'],
             )
 
-        last_order = sent[-1]
+        last_order: Order = sent[-1]
-
         last_resp = msgs[-1]
         assert isinstance(last_resp, BrokerdPosition)
+        ppmsg = last_resp
 
         # check that pps.toml for account has been updated
+        # and all ems position msgs match that state.
         with load_and_check_pos(
             last_order,
-            last_resp,
+            ppmsg,
         ) as pos:
-            return pos
+            pass
 
-    # disconnect from EMS, then reconnect and ensure we get our same
+    return ppmsg, pos
-    # position relayed to us again.
-
-    # _run_test_and_check(
-    #     partial(
-    #         _async_main,
-    #         open_test_pikerd_and_ems=open_test_pikerd_and_ems,
-    #         action='buy',
-    #         assert_entries=True,
-    #     ),
-    # )
-
-    # await _async_main(
-    #     open_test_pikerd_and_ems=open_test_pikerd_and_ems,
-    #     assert_pps=True,
-    # )
-    # _run_test_and_check(
-    #     partial(
-    #         _async_main,
-    #         open_test_pikerd_and_ems=open_test_pikerd_and_ems,
-    #         assert_pps=True,
-    #     ),
-    # )
 
 
-def test_open_long(
+@pytest.mark.parametrize(
+    'fills',
+    [
+        # buy and leave
+        ({'size': 0.001},),
+
+        # sell short, then buy back to net-zero in dst
+        (
+            {'size': -0.001},
+            {'size': 0.001},
+        ),
+
+        # multi-partial entry and exits.
+        (
+            # enters
+            {'size': 0.001},
+            {'size': 0.002},
+
+            # partial exit
+            {'size': -0.001},
+
+            # partial enter
+            {'size': 0.0015},
+            {'size': 0.001},
+            {'size': 0.002},
+
+            # exits to get back to zero.
+            {'size': -0.001},
+            {'size': -0.025},
+            {'size': -0.0195},
+        ),
+    ],
+    ids='fills={}'.format,
+)
+def test_multi_fill_positions(
     open_test_pikerd: AsyncContextManager,
     loglevel: str,
 
+    fills: tuple[dict],
+
+    check_cross_session: bool = True,
+
 ) -> None:
 
+    ppmsg: BrokerdPosition
+    pos: Position
+
+    accum_size: float = 0
+    for fill in fills:
+        accum_size += fill['size']
+
     async def atest():
+
+        # export to outer scope for audit on second runtime-boot.
+        nonlocal ppmsg, pos
+
         async with (
             open_test_pikerd() as (_, _, _, services),
         ):
-            assert await atest_buy(loglevel)
+            ppmsg, pos = await submit_and_check(
-
+                fills=fills,
-            # Teardown piker like a user would from cli
+                loglevel=loglevel,
-            # raise KeyboardInterrupt
-
-    run_and_catch(
-        atest,
-        expect_errs=None,
             )
-    # Open ems another time and assert existence of prior
+            assert ppmsg.size == accum_size
-    # pps entries confirming they persisted
 
+    run_and_tollerate_cancels(atest)
 
+    if check_cross_session or accum_size != 0:
+        # rerun just to check that position info is persistent for the paper
+        # account (i.e. a user can expect to see paper pps persist across
+        # runtime sessions.
+        async def just_check_pp():
+            async with (
+                open_test_pikerd() as (_, _, _, services),
+            ):
+                await match_ppmsgs_on_ems_boot([ppmsg])
 
-# def test_sell(
+        run_and_tollerate_cancels(just_check_pp)
-#     open_test_pikerd_and_ems: AsyncContextManager,
-# ):
-#     '''
-#     Sell position and ensure pps are zeroed.
-
-#     '''
-#     _run_test_and_check(
-#         partial(
-#             _async_main,
-#             open_test_pikerd_and_ems=open_test_pikerd_and_ems,
-#             action='sell',
-#             price=1,
-#         ),
-#     )
-
-#     _run_test_and_check(
-#         partial(
-#             _async_main,
-#             open_test_pikerd_and_ems=open_test_pikerd_and_ems,
-#             assert_zeroed_pps=True,
-#         ),
-#     )
-
-
-# def test_multi_sell(
-#     open_test_pikerd_and_ems: AsyncContextManager,
-# ):
-#     '''
-#     Make 5 market limit buy orders and
-#     then sell 5 slots at the same price.
-#     Finally, assert cleared positions.
-
-#     '''
-#     _run_test_and_check(
-#         partial(
-#             _async_main,
-#             open_test_pikerd_and_ems=open_test_pikerd_and_ems,
-#             action='buy',
-#             executions=5,
-#         ),
-#     )
-
-#     _run_test_and_check(
-#         partial(
-#             _async_main,
-#             open_test_pikerd_and_ems=open_test_pikerd_and_ems,
-#             action='sell',
-#             executions=5,
-#             price=1,
-#             assert_zeroed_pps=True,
-#         ),
-#     )