80 lines
2.1 KiB
Python
80 lines
2.1 KiB
Python
|
'''
|
||
|
|
EMS broadcast ownership regressions.
|
||
|
|
|
||
|
|
'''
|
||
|
|
from contextlib import asynccontextmanager as acm
|
||
|
|
|
||
|
|
import trio
|
||
|
|
|
||
|
|
from piker.clearing import _ems
|
||
|
|
|
||
|
|
|
||
|
|
def test_dark_clearing_owns_non_strict_subscription(
|
||
|
|
monkeypatch,
|
||
|
|
) -> None:
|
||
|
|
'''
|
||
|
|
Dark clearing must not mutate another feed consumer's lag policy.
|
||
|
|
|
||
|
|
`clear_dark_triggers()` previously assigned `_raise_on_lag` on
|
||
|
|
whichever root or temporary receiver was stored in `Flume.stream`.
|
||
|
|
That changed private shared state and made behavior depend on feed
|
||
|
|
cache ownership. Supply a stream whose subscription records the
|
||
|
|
public lag-policy argument and yields a distinct child. Replace
|
||
|
|
the hot trigger loop with a checkpointing probe, then prove the
|
||
|
|
non-strict child is entered, passed to the core, and closed only
|
||
|
|
after processing completes while the source remains untouched.
|
||
|
|
|
||
|
|
'''
|
||
|
|
events: list[tuple] = []
|
||
|
|
child = object()
|
||
|
|
|
||
|
|
class QuoteStream:
|
||
|
|
'''
|
||
|
|
Record public subscription ownership without quote traffic.
|
||
|
|
|
||
|
|
'''
|
||
|
|
@acm
|
||
|
|
async def subscribe(self, raise_on_lag: bool = True):
|
||
|
|
'''
|
||
|
|
Yield one dedicated child for the wrapper lifetime.
|
||
|
|
|
||
|
|
'''
|
||
|
|
events.append(('enter', raise_on_lag))
|
||
|
|
try:
|
||
|
|
yield child
|
||
|
|
finally:
|
||
|
|
events.append(('exit', raise_on_lag))
|
||
|
|
|
||
|
|
async def clear_core(*args) -> None:
|
||
|
|
'''
|
||
|
|
Prove the child remains owned while processing runs.
|
||
|
|
|
||
|
|
'''
|
||
|
|
events.append(('core', args[2] is child))
|
||
|
|
await trio.lowlevel.checkpoint()
|
||
|
|
|
||
|
|
monkeypatch.setattr(
|
||
|
|
_ems,
|
||
|
|
'_clear_dark_triggers',
|
||
|
|
clear_core,
|
||
|
|
)
|
||
|
|
|
||
|
|
async def main() -> None:
|
||
|
|
stream = QuoteStream()
|
||
|
|
await _ems.clear_dark_triggers(
|
||
|
|
object(),
|
||
|
|
object(),
|
||
|
|
stream,
|
||
|
|
'ib',
|
||
|
|
'nvda.nasdaq.ib',
|
||
|
|
object(),
|
||
|
|
)
|
||
|
|
assert not hasattr(stream, '_raise_on_lag')
|
||
|
|
|
||
|
|
trio.run(main)
|
||
|
|
assert events == [
|
||
|
|
('enter', False),
|
||
|
|
('core', True),
|
||
|
|
('exit', False),
|
||
|
|
]
|