371 lines
12 KiB
Python
371 lines
12 KiB
Python
|
'''
|
||
|
|
Real service-tree coverage for the offline replay provider.
|
||
|
|
|
||
|
|
'''
|
||
|
|
from contextlib import AbstractAsyncContextManager
|
||
|
|
from multiprocessing.shared_memory import SharedMemory
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import (
|
||
|
|
Any,
|
||
|
|
Callable,
|
||
|
|
)
|
||
|
|
|
||
|
|
import msgspec
|
||
|
|
import numpy as np
|
||
|
|
import pytest
|
||
|
|
import tractor
|
||
|
|
import trio
|
||
|
|
|
||
|
|
from piker.brokers import replay
|
||
|
|
from piker.data import open_feed
|
||
|
|
from piker.service import Services
|
||
|
|
|
||
|
|
|
||
|
|
FQME: str = 'btcusd.test.replay'
|
||
|
|
|
||
|
|
|
||
|
|
async def _request(
|
||
|
|
stream: tractor.MsgStream,
|
||
|
|
command: replay.ReplayCommand,
|
||
|
|
) -> replay.ReplayAck:
|
||
|
|
'''
|
||
|
|
Send one replay command and receive its correlated reply.
|
||
|
|
|
||
|
|
'''
|
||
|
|
await stream.send(command)
|
||
|
|
ack: replay.ReplayAck = await stream.receive()
|
||
|
|
assert ack.command_id == command.command_id
|
||
|
|
return ack
|
||
|
|
|
||
|
|
|
||
|
|
async def _search_replay_symbols(
|
||
|
|
portal: tractor.Portal,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
'''
|
||
|
|
Exercise the backend's normal symbol-search context.
|
||
|
|
|
||
|
|
'''
|
||
|
|
ctx: tractor.Context
|
||
|
|
async with portal.open_context(
|
||
|
|
replay.open_symbol_search,
|
||
|
|
) as (ctx, _):
|
||
|
|
async with ctx.open_stream() as stream:
|
||
|
|
await stream.send('btc')
|
||
|
|
return await stream.receive()
|
||
|
|
|
||
|
|
|
||
|
|
async def _run_basic_generation(
|
||
|
|
open_test_pikerd: Callable[
|
||
|
|
...,
|
||
|
|
AbstractAsyncContextManager,
|
||
|
|
],
|
||
|
|
loglevel: str,
|
||
|
|
) -> tuple[bytes, list[dict[str, Any]], list[float]]:
|
||
|
|
'''
|
||
|
|
Run one complete replay through real data services.
|
||
|
|
|
||
|
|
'''
|
||
|
|
quote_transcript: list[dict[str, Any]] = []
|
||
|
|
shm_names: list[str] = []
|
||
|
|
services: Services
|
||
|
|
async with (
|
||
|
|
open_test_pikerd() as (_, _, _, services),
|
||
|
|
open_feed(
|
||
|
|
[FQME],
|
||
|
|
loglevel=loglevel,
|
||
|
|
) as feed,
|
||
|
|
):
|
||
|
|
assert 'datad.replay' in services.service_tasks
|
||
|
|
assert 'samplerd' in services.service_tasks
|
||
|
|
|
||
|
|
stream: tractor.MsgStream = feed.streams['replay']
|
||
|
|
first_msg: dict[str, Any] = await stream.receive()
|
||
|
|
quote_transcript.append(first_msg)
|
||
|
|
assert first_msg[FQME]['replay_seq'] == 1
|
||
|
|
|
||
|
|
flume = feed.flumes[FQME]
|
||
|
|
history_closes: list[float] = (
|
||
|
|
flume.hist_shm.array['close'].tolist()
|
||
|
|
)
|
||
|
|
shm_names.extend([
|
||
|
|
flume.hist_shm._shm.name,
|
||
|
|
flume.rt_shm._shm.name,
|
||
|
|
])
|
||
|
|
assert history_closes == [
|
||
|
|
99.2,
|
||
|
|
99.4,
|
||
|
|
99.6,
|
||
|
|
99.8,
|
||
|
|
100.2,
|
||
|
|
100.6,
|
||
|
|
]
|
||
|
|
|
||
|
|
portal: tractor.Portal = feed.portals[replay]
|
||
|
|
matches: dict[str, Any] = await _search_replay_symbols(
|
||
|
|
portal
|
||
|
|
)
|
||
|
|
assert list(matches) == [FQME]
|
||
|
|
|
||
|
|
control_ctx: tractor.Context
|
||
|
|
started: replay.ReplaySnapshot
|
||
|
|
async with portal.open_context(
|
||
|
|
replay.open_replay_control,
|
||
|
|
) as (control_ctx, started):
|
||
|
|
assert started.state == 'ready'
|
||
|
|
assert started.event_sequence == 1
|
||
|
|
|
||
|
|
async with control_ctx.open_stream() as control:
|
||
|
|
with control_ctx.pld_rx.limit_plds(
|
||
|
|
spec=replay.ReplayAck,
|
||
|
|
):
|
||
|
|
await feed.pause()
|
||
|
|
paused: replay.ReplayAck = await _request(
|
||
|
|
control,
|
||
|
|
replay.AwaitState(
|
||
|
|
command_id=1,
|
||
|
|
state='paused',
|
||
|
|
),
|
||
|
|
)
|
||
|
|
assert paused.ok
|
||
|
|
assert not paused.snapshot.subscriber_active
|
||
|
|
|
||
|
|
rejected: replay.ReplayAck = await _request(
|
||
|
|
control,
|
||
|
|
replay.Advance(
|
||
|
|
command_id=2,
|
||
|
|
event_sequence=2,
|
||
|
|
),
|
||
|
|
)
|
||
|
|
assert not rejected.ok
|
||
|
|
assert rejected.snapshot.event_sequence == 1
|
||
|
|
|
||
|
|
await feed.resume()
|
||
|
|
ready: replay.ReplayAck = await _request(
|
||
|
|
control,
|
||
|
|
replay.AwaitState(
|
||
|
|
command_id=3,
|
||
|
|
state='ready',
|
||
|
|
),
|
||
|
|
)
|
||
|
|
assert ready.ok
|
||
|
|
assert ready.snapshot.subscriber_active
|
||
|
|
|
||
|
|
producer_paused = await _request(
|
||
|
|
control,
|
||
|
|
replay.Pause(command_id=4),
|
||
|
|
)
|
||
|
|
assert producer_paused.snapshot.state == 'paused'
|
||
|
|
gate_rejected = await _request(
|
||
|
|
control,
|
||
|
|
replay.Advance(
|
||
|
|
command_id=5,
|
||
|
|
event_sequence=2,
|
||
|
|
),
|
||
|
|
)
|
||
|
|
assert not gate_rejected.ok
|
||
|
|
producer_ready = await _request(
|
||
|
|
control,
|
||
|
|
replay.Resume(command_id=6),
|
||
|
|
)
|
||
|
|
assert producer_ready.snapshot.state == 'ready'
|
||
|
|
|
||
|
|
for command_id, event_sequence in (
|
||
|
|
(7, 2),
|
||
|
|
(8, 3),
|
||
|
|
(9, 4),
|
||
|
|
):
|
||
|
|
advanced: replay.ReplayAck = (
|
||
|
|
await _request(
|
||
|
|
control,
|
||
|
|
replay.Advance(
|
||
|
|
command_id=command_id,
|
||
|
|
event_sequence=(
|
||
|
|
event_sequence
|
||
|
|
),
|
||
|
|
),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
assert advanced.ok
|
||
|
|
quote: dict[str, Any] = (
|
||
|
|
await stream.receive()
|
||
|
|
)
|
||
|
|
quote_transcript.append(quote)
|
||
|
|
assert (
|
||
|
|
quote[FQME]['replay_seq']
|
||
|
|
== event_sequence
|
||
|
|
)
|
||
|
|
|
||
|
|
snap_ack: replay.ReplayAck = await _request(
|
||
|
|
control,
|
||
|
|
replay.Snapshot(command_id=10),
|
||
|
|
)
|
||
|
|
assert snap_ack.snapshot.state == 'exhausted'
|
||
|
|
transcript: bytes = msgspec.json.encode(
|
||
|
|
snap_ack.snapshot
|
||
|
|
)
|
||
|
|
|
||
|
|
np.testing.assert_allclose(
|
||
|
|
flume.rt_shm.array['close'][-1],
|
||
|
|
101.0,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert not services.service_tasks
|
||
|
|
for shm_name in shm_names:
|
||
|
|
with pytest.raises(FileNotFoundError):
|
||
|
|
SharedMemory(name=shm_name)
|
||
|
|
|
||
|
|
return transcript, quote_transcript, history_closes
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
'replay_scenario',
|
||
|
|
['basic-v1'],
|
||
|
|
indirect=True,
|
||
|
|
)
|
||
|
|
def test_real_feed_replay_is_repeatable(
|
||
|
|
replay_scenario: Path,
|
||
|
|
open_test_pikerd: Callable[
|
||
|
|
...,
|
||
|
|
AbstractAsyncContextManager,
|
||
|
|
],
|
||
|
|
loglevel: str,
|
||
|
|
) -> None:
|
||
|
|
'''
|
||
|
|
Replay identical typed transcripts through fresh actor trees.
|
||
|
|
|
||
|
|
Mocked feed calls can hide backend discovery, datad and samplerd
|
||
|
|
startup, history publication, SHM writes, stream subscription
|
||
|
|
control, and context teardown failures. Run the same tracked
|
||
|
|
scenario through two fresh `open_test_pikerd()` generations.
|
||
|
|
Each generation receives the initial quote, pauses the production
|
||
|
|
feed, awaits the typed subscriber-state barrier, proves
|
||
|
|
advancement is rejected, resumes, repeats that gate check
|
||
|
|
through typed producer controls, and explicitly advances every
|
||
|
|
remaining event. Exact control bytes, quote dictionaries, and
|
||
|
|
history values must match, while service and named-SHM checks
|
||
|
|
prove each generation fully tears down before the next one
|
||
|
|
starts.
|
||
|
|
|
||
|
|
'''
|
||
|
|
assert replay_scenario.name == 'basic-v1.json'
|
||
|
|
|
||
|
|
async def main() -> None:
|
||
|
|
first = await _run_basic_generation(
|
||
|
|
open_test_pikerd,
|
||
|
|
loglevel,
|
||
|
|
)
|
||
|
|
second = await _run_basic_generation(
|
||
|
|
open_test_pikerd,
|
||
|
|
loglevel,
|
||
|
|
)
|
||
|
|
assert first == second
|
||
|
|
|
||
|
|
trio.run(main)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
'replay_scenario',
|
||
|
|
['failure-v1'],
|
||
|
|
indirect=True,
|
||
|
|
)
|
||
|
|
def test_failure_injection_is_acknowledged(
|
||
|
|
replay_scenario: Path,
|
||
|
|
open_test_pikerd: Callable[
|
||
|
|
...,
|
||
|
|
AbstractAsyncContextManager,
|
||
|
|
],
|
||
|
|
loglevel: str,
|
||
|
|
) -> None:
|
||
|
|
'''
|
||
|
|
Stop at a fixture failure without publishing its quote.
|
||
|
|
|
||
|
|
An injected provider disconnect used to require timing a task
|
||
|
|
crash and inferring its position from logs. Start the real feed,
|
||
|
|
arm the fixture's third event through typed control, publish
|
||
|
|
event two, then request event three. Its correlated negative
|
||
|
|
acknowledgement must identify the fixture error, retain event
|
||
|
|
sequence two, and expose `failed` state. Reading SHM after the
|
||
|
|
acknowledgement proves the rejected event was not hidden in the
|
||
|
|
sampler before normal actor and SHM teardown.
|
||
|
|
|
||
|
|
'''
|
||
|
|
assert replay_scenario.name == 'failure-v1.json'
|
||
|
|
|
||
|
|
async def main() -> None:
|
||
|
|
services: Services
|
||
|
|
shm_names: list[str] = []
|
||
|
|
async with (
|
||
|
|
open_test_pikerd() as (_, _, _, services),
|
||
|
|
open_feed(
|
||
|
|
[FQME],
|
||
|
|
loglevel=loglevel,
|
||
|
|
) as feed,
|
||
|
|
):
|
||
|
|
quotes: tractor.MsgStream = feed.streams['replay']
|
||
|
|
first: dict[str, Any] = await quotes.receive()
|
||
|
|
assert first[FQME]['replay_seq'] == 1
|
||
|
|
|
||
|
|
flume = feed.flumes[FQME]
|
||
|
|
shm_names.extend([
|
||
|
|
flume.hist_shm._shm.name,
|
||
|
|
flume.rt_shm._shm.name,
|
||
|
|
])
|
||
|
|
portal: tractor.Portal = feed.portals[replay]
|
||
|
|
ctx: tractor.Context
|
||
|
|
async with portal.open_context(
|
||
|
|
replay.open_replay_control,
|
||
|
|
) as (ctx, _):
|
||
|
|
async with ctx.open_stream() as control:
|
||
|
|
with ctx.pld_rx.limit_plds(
|
||
|
|
spec=replay.ReplayAck,
|
||
|
|
):
|
||
|
|
armed: replay.ReplayAck = await _request(
|
||
|
|
control,
|
||
|
|
replay.FailAt(
|
||
|
|
command_id=1,
|
||
|
|
event_sequence=3,
|
||
|
|
),
|
||
|
|
)
|
||
|
|
assert armed.ok
|
||
|
|
|
||
|
|
advanced: replay.ReplayAck = (
|
||
|
|
await _request(
|
||
|
|
control,
|
||
|
|
replay.Advance(
|
||
|
|
command_id=2,
|
||
|
|
event_sequence=2,
|
||
|
|
),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
assert advanced.ok
|
||
|
|
second: dict[str, Any] = (
|
||
|
|
await quotes.receive()
|
||
|
|
)
|
||
|
|
assert second[FQME]['replay_seq'] == 2
|
||
|
|
|
||
|
|
failed: replay.ReplayAck = await _request(
|
||
|
|
control,
|
||
|
|
replay.Advance(
|
||
|
|
command_id=3,
|
||
|
|
event_sequence=3,
|
||
|
|
),
|
||
|
|
)
|
||
|
|
assert not failed.ok
|
||
|
|
assert failed.snapshot.state == 'failed'
|
||
|
|
assert failed.snapshot.event_sequence == 2
|
||
|
|
assert failed.error == (
|
||
|
|
'fixture_disconnect: offline provider '
|
||
|
|
'disconnected'
|
||
|
|
)
|
||
|
|
assert (
|
||
|
|
flume.rt_shm.array['close'][-1]
|
||
|
|
== 100.5
|
||
|
|
)
|
||
|
|
|
||
|
|
assert not services.service_tasks
|
||
|
|
for shm_name in shm_names:
|
||
|
|
with pytest.raises(FileNotFoundError):
|
||
|
|
SharedMemory(name=shm_name)
|
||
|
|
|
||
|
|
trio.run(main)
|