gitadmin/dzentra_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Build 060.26: complete Integration and Regression

Browse Source
This commit is contained in:
Sergey
2026-07-31 14:14:30 +03:00
parent 60bec1eaf9
commit cb8acfe5fe
24 changed files with 4732 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
app/pytest.ini
View File

@@ -15,4 +15,11 @@ python_functions =
addopts =
-ra
--strict-markers
--strict-markers
-m
"not integration and not stress and not live"
markers =
integration: deterministic integration tests using local network endpoints
stress: explicit local stress and soak verification
live: opt-in verification against explicitly configured live endpoints

41
app/src/market_data/acquisition/consistency/trade_stream_state.py
View File

@@ -10,6 +10,10 @@ from src.market_data.acquisition.consistency.trade_stream_exceptions import (
TradeOrderingError,
)
from src.market_data.acquisition.models.trade import Trade
from src.market_data.acquisition.trade_id_sequence import (
is_trade_id_newer,
validate_signed_trade_id,
)
DEFAULT_DEDUPLICATION_WINDOW_SIZE = 10_000
@@ -65,15 +69,22 @@ class TradeStreamState:
)
trade_id = trade.trade_id
validate_signed_trade_id(trade_id)
if self.last_trade_id is not None:
if trade_id < self.last_trade_id:
if not is_trade_id_newer(
trade_id,
self.last_trade_id,
) and trade_id != self.last_trade_id:
previous = self._trades.get(trade_id)
if previous is None:
raise TradeOrderingError()
if previous == trade:
if self._is_same_market_trade(
previous,
trade,
):
return None
raise TradeConsistencyError()
@@ -81,7 +92,10 @@ class TradeStreamState:
previous = self._trades.get(trade_id)
if previous is not None:
if previous == trade:
if self._is_same_market_trade(
previous,
trade,
):
return None
raise TradeConsistencyError()
@@ -93,6 +107,27 @@ class TradeStreamState:
return trade
@staticmethod
def _is_same_market_trade(
first: Trade,
second: Trade,
) -> bool:
"""
Сравнить биржевой факт независимо от transport source.
Одна сделка закономерно приходит через WebSocket и повторяется
через REST recovery. Поле ``source`` описывает путь доставки и
поэтому не является частью identity биржевой сделки.
"""
return (
first.symbol == second.symbol
and first.trade_id == second.trade_id
and first.price == second.price
and first.quantity == second.quantity
and first.executed_at == second.executed_at
and first.aggressor_side is second.aggressor_side
)
def _append(
self,
trade: Trade,

28
app/src/market_data/acquisition/recovery/trade_recovery_normalizer.py
View File

@@ -5,6 +5,9 @@ from __future__ import annotations
from collections.abc import Iterable
from src.market_data.acquisition.models.trade import Trade
from src.market_data.acquisition.trade_id_sequence import (
trade_id_relative_offset,
)
def normalize_recovered_trades(
@@ -13,13 +16,18 @@ def normalize_recovered_trades(
"""
Нормализовать порядок восстановленных сделок.
Сделки возвращаются в возрастающем порядке по ``trade_id``.
Сделки возвращаются в хронологическом порядке signed 32-bit
``trade_id`` с учётом rollover.
Исходная последовательность не изменяется.
Нормализатор намеренно не выполняет дедупликацию и не проверяет
согласованность сделок. Идентичные и конфликтующие дубликаты должны
обрабатываться экземпляром ``TradeStreamConsistencyController``.
Входной набор должен покрывать меньше половины 32-битного цикла.
Это ограничение делает modular-порядок однозначным и существенно
шире максимального часового Recovery window Runtime.
Parameters
----------
trades
@@ -29,12 +37,22 @@ def normalize_recovered_trades(
-------
tuple[Trade, ...]
Неизменяемая последовательность сделок, отсортированная
по возрастанию ``trade_id``.
по rollover-aware порядку ``trade_id``.
"""
source = tuple(trades)
if not source:
return ()
reference_trade_id = source[0].trade_id
return tuple(
sorted(
trades,
key=lambda trade: trade.trade_id,
source,
key=lambda trade: trade_id_relative_offset(
trade.trade_id,
reference_trade_id,
),
)
)
)

75
app/src/market_data/acquisition/trade_id_sequence.py Normal file
View File

@@ -0,0 +1,75 @@
from __future__ import annotations
SIGNED_TRADE_ID_BITS = 32
SIGNED_TRADE_ID_HALF_RANGE = 1 << (SIGNED_TRADE_ID_BITS - 1)
SIGNED_TRADE_ID_MODULUS = 1 << SIGNED_TRADE_ID_BITS
SIGNED_TRADE_ID_MIN = -SIGNED_TRADE_ID_HALF_RANGE
SIGNED_TRADE_ID_MAX = SIGNED_TRADE_ID_HALF_RANGE - 1
def validate_signed_trade_id(trade_id: int) -> None:
"""Проверить raw Trade ID провайдера как signed 32-bit integer."""
if isinstance(trade_id, bool) or not isinstance(trade_id, int):
raise TypeError("trade_id must be an integer")
if not SIGNED_TRADE_ID_MIN <= trade_id <= SIGNED_TRADE_ID_MAX:
raise ValueError("trade_id must fit signed 32-bit range")
def trade_id_relative_offset(
candidate_trade_id: int,
reference_trade_id: int,
) -> int:
"""
Вернуть signed modular-смещение candidate относительно reference.
Положительное значение означает более новый ID, отрицательное —
более старый. Контракт однозначен для расстояний меньше половины
32-битного цикла. Ровно половина цикла не имеет определимого
направления и отклоняется явно.
"""
validate_signed_trade_id(candidate_trade_id)
validate_signed_trade_id(reference_trade_id)
forward_distance = (
candidate_trade_id - reference_trade_id
) % SIGNED_TRADE_ID_MODULUS
if forward_distance == SIGNED_TRADE_ID_HALF_RANGE:
raise ValueError(
"Trade ID distance is exactly half of the 32-bit cycle"
)
if forward_distance > SIGNED_TRADE_ID_HALF_RANGE:
return forward_distance - SIGNED_TRADE_ID_MODULUS
return forward_distance
def is_trade_id_newer(
candidate_trade_id: int,
reference_trade_id: int,
) -> bool:
"""Проверить, следует ли candidate после reference по 32-bit циклу."""
return (
trade_id_relative_offset(
candidate_trade_id,
reference_trade_id,
)
> 0
)
def is_trade_id_same_or_newer(
candidate_trade_id: int,
reference_trade_id: int,
) -> bool:
"""Проверить равенство либо продвижение по 32-bit циклу."""
return (
trade_id_relative_offset(
candidate_trade_id,
reference_trade_id,
)
>= 0
)

21
app/src/market_data/acquisition/validation/values.py
View File

@@ -27,6 +27,9 @@ from src.market_data.acquisition.exceptions import (
QuoteValueError,
TradeValueError,
)
from src.market_data.acquisition.trade_id_sequence import (
validate_signed_trade_id,
)
_SUPPORTED_WEBSOCKET_OHLC_INTERVALS = frozenset(
@@ -831,7 +834,7 @@ def validate_dzengi_websocket_trade_values(
значения в Decimal и не выполняет mapping во внутреннюю модель Trade.
"""
_trade_positive_int(
_trade_signed_int(
event.trade_id,
path="$.payload.id",
)
@@ -881,7 +884,7 @@ def _validate_rest_agg_trade(
*,
path: str,
) -> None:
_trade_positive_int(
_trade_signed_int(
trade.aggregate_trade_id,
path=f"{path}.aggregateTradeId",
)
@@ -910,6 +913,20 @@ def _trade_positive_int(
)
def _trade_signed_int(
value: int,
*,
path: str,
) -> None:
try:
validate_signed_trade_id(value)
except (TypeError, ValueError) as error:
raise TradeValueError(
f"{path} должно быть целым числом "
"в диапазоне signed 32-bit."
) from error
def _trade_positive_decimal(
value: DzengiRawNumeric,
*,

904
app/tests/integration/market_data/acquisition/runtime/loopback_trade_exchange.py Normal file
View File

@@ -0,0 +1,904 @@
from __future__ import annotations
import asyncio
import json
import threading
from collections import deque
from collections.abc import Callable
from contextlib import AsyncExitStack
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from types import TracebackType
from typing import Any
from urllib.parse import parse_qs, urlsplit
from websockets.asyncio.server import (
Server,
ServerConnection,
serve,
)
from websockets.exceptions import ConnectionClosed
from websockets.protocol import State
from websockets.typing import Subprotocol
from tests.support.async_wait import wait_until
RESOURCE_CLEANUP_TIMEOUT_SECONDS = 5.0
ENVIRONMENT_CLEANUP_TIMEOUT_SECONDS = 20.0
@dataclass(frozen=True, slots=True)
class WebSocketSubscriptionRecord:
connection_index: int
correlation_id: str
symbols: tuple[str, ...]
class LoopbackTradeWebSocketServer:
"""Управляемый локальный Dzengi-подобный WebSocket endpoint."""
def __init__(
self,
*,
events: list[str] | None = None,
auto_ack: bool = True,
) -> None:
self._events = events if events is not None else []
self._auto_ack = auto_ack
self._server: Server | None = None
self._host = "127.0.0.1"
self._port: int | None = None
self._connections: list[ServerConnection] = []
self._handler_tasks: set[asyncio.Task[Any]] = set()
self._subscriptions: list[WebSocketSubscriptionRecord] = []
self._received_documents: list[dict[str, Any]] = []
@property
def url(self) -> str:
if self._port is None:
raise RuntimeError("WebSocket server is not started.")
return f"ws://{self._host}:{self._port}"
@property
def port(self) -> int:
if self._port is None:
raise RuntimeError("WebSocket server is not started.")
return self._port
@property
def connection_count(self) -> int:
return len(self._connections)
@property
def active_handler_count(self) -> int:
return sum(
not task.done()
for task in self._handler_tasks
)
@property
def active_connection_count(self) -> int:
return sum(
connection.state is State.OPEN
for connection in self._connections
)
@property
def subscriptions(
self,
) -> tuple[WebSocketSubscriptionRecord, ...]:
return tuple(self._subscriptions)
@property
def received_documents(
self,
) -> tuple[dict[str, Any], ...]:
return tuple(self._received_documents)
async def start(self) -> None:
if self._server is not None:
raise RuntimeError("WebSocket server is already started.")
server = await serve(
self._handle_connection,
self._host,
0,
subprotocols=(Subprotocol("json"),),
ping_interval=None,
close_timeout=0.2,
)
self._server = server
sockets = tuple(server.sockets)
if not sockets:
server.close()
await server.wait_closed()
self._server = None
raise RuntimeError(
"WebSocket server did not expose a listening socket."
)
socket = sockets[0]
address = socket.getsockname()
self._port = int(address[1])
async def stop(self) -> None:
server = self._server
if server is None:
return
self._server = None
self._port = None
server.close()
try:
await asyncio.wait_for(
server.wait_closed(),
timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS,
)
except TimeoutError:
for connection in tuple(self._connections):
if connection.state is State.OPEN:
connection.transport.abort()
await asyncio.wait_for(
server.wait_closed(),
timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS,
)
if self.active_handler_count:
raise AssertionError(
"WebSocket server handlers did not stop."
)
async def wait_for_connections(
self,
count: int,
*,
timeout_seconds: float = 3.0,
) -> None:
await wait_until(
lambda: self.connection_count >= count,
timeout_seconds=timeout_seconds,
)
async def wait_for_subscriptions(
self,
count: int,
*,
timeout_seconds: float = 3.0,
) -> None:
await wait_until(
lambda: len(self._subscriptions) >= count,
timeout_seconds=timeout_seconds,
)
async def send_raw(
self,
connection_index: int,
message: str | bytes,
) -> None:
await self._connections[connection_index].send(message)
async def send_trade(
self,
connection_index: int,
*,
symbol: str,
trade_id: int,
timestamp_ms: int,
price: str = "64555.55",
quantity: str = "0.002",
buyer: bool = True,
) -> None:
await self.send_raw(
connection_index,
json.dumps(
{
"status": "OK",
"destination": "internal.trade",
"payload": {
"id": trade_id,
"price": price,
"size": quantity,
"ts": timestamp_ms,
"symbol": symbol,
"buyer": buyer,
"orderId": f"order-{trade_id}",
},
}
),
)
async def abort_connection(
self,
connection_index: int,
) -> None:
connection = self._connections[connection_index]
connection.transport.abort()
await asyncio.sleep(0)
async def close_connection(
self,
connection_index: int,
*,
code: int = 1012,
reason: str = "integration test reconnect",
) -> None:
await self._connections[connection_index].close(
code=code,
reason=reason,
)
async def _handle_connection(
self,
connection: ServerConnection,
) -> None:
task = asyncio.current_task()
if task is None:
raise RuntimeError("WebSocket handler has no asyncio task.")
self._handler_tasks.add(task)
connection_index = len(self._connections)
self._connections.append(connection)
self._events.append(f"ws.connect:{connection_index}")
try:
async for raw_message in connection:
if not isinstance(raw_message, str):
continue
try:
document = json.loads(raw_message)
except json.JSONDecodeError:
continue
if not isinstance(document, dict):
continue
self._received_documents.append(document)
if document.get("destination") != "trades.subscribe":
continue
correlation_id = document.get("correlationId")
payload = document.get("payload")
if (
not isinstance(correlation_id, str)
or not isinstance(payload, dict)
):
continue
raw_symbols = payload.get("symbols")
if not isinstance(raw_symbols, list) or not all(
isinstance(symbol, str)
for symbol in raw_symbols
):
continue
record = WebSocketSubscriptionRecord(
connection_index=connection_index,
correlation_id=correlation_id,
symbols=tuple(raw_symbols),
)
self._subscriptions.append(record)
self._events.append(
f"ws.subscribe:{connection_index}"
)
if self._auto_ack:
await connection.send(
json.dumps(
{
"correlationId": correlation_id,
"destination": "trades.subscribe",
"status": "OK",
}
)
)
except ConnectionClosed:
pass
finally:
self._events.append(
f"ws.disconnect:{connection_index}"
)
self._handler_tasks.discard(task)
@dataclass(frozen=True, slots=True)
class LoopbackHttpResponse:
body: object
status: int = 200
release: threading.Event | None = None
@dataclass(frozen=True, slots=True)
class LoopbackHttpRequest:
path: str
query: dict[str, tuple[str, ...]]
class _LoopbackThreadingHttpServer(ThreadingHTTPServer):
daemon_threads = False
block_on_close = True
class LoopbackTradeRestServer:
"""Локальный HTTP endpoint для настоящего ExchangeRestClient."""
def __init__(
self,
*,
responses: tuple[LoopbackHttpResponse, ...] = (),
events: list[str] | None = None,
) -> None:
self._events = events if events is not None else []
self._responses = deque(responses)
self._requests: list[LoopbackHttpRequest] = []
self._lock = threading.Lock()
self._release_events = {
response.release
for response in responses
if response.release is not None
}
self._server: _LoopbackThreadingHttpServer | None = None
self._thread: threading.Thread | None = None
self._host = "127.0.0.1"
self._port: int | None = None
@property
def base_url(self) -> str:
if self._port is None:
raise RuntimeError("HTTP server is not started.")
return f"http://{self._host}:{self._port}"
@property
def request_count(self) -> int:
with self._lock:
return len(self._requests)
@property
def requests(self) -> tuple[LoopbackHttpRequest, ...]:
with self._lock:
return tuple(self._requests)
@property
def thread_is_alive(self) -> bool:
thread = self._thread
return thread is not None and thread.is_alive()
def start(self) -> None:
if self._server is not None:
raise RuntimeError("HTTP server is already started.")
controller = self
class RequestHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def do_GET(self) -> None: # noqa: N802
controller._handle_get(self)
def log_message(
self,
format: str,
*args: object,
) -> None:
del format, args
server = _LoopbackThreadingHttpServer(
(self._host, 0),
RequestHandler,
)
self._server = server
self._port = int(server.server_address[1])
self._thread = threading.Thread(
target=lambda: server.serve_forever(
poll_interval=0.05,
),
name="loopback-trade-rest-server",
)
try:
self._thread.start()
except BaseException:
server.server_close()
self._server = None
self._thread = None
self._port = None
raise
def stop(self) -> None:
server = self._server
thread = self._thread
if server is None:
return
self._server = None
self._thread = None
self._port = None
for release_event in self._release_events:
release_event.set()
shutdown_error: BaseException | None = None
try:
server.shutdown()
except BaseException as error:
shutdown_error = error
finally:
try:
server.server_close()
except BaseException as error:
if shutdown_error is None:
shutdown_error = error
else:
shutdown_error.add_note(
"Loopback HTTP server_close also failed: "
f"{type(error).__name__}."
)
if thread is not None:
thread.join(timeout=3)
if thread.is_alive():
thread_error = AssertionError(
"Loopback HTTP server thread did not stop."
)
if shutdown_error is None:
shutdown_error = thread_error
else:
shutdown_error.add_note(str(thread_error))
if shutdown_error is not None:
raise shutdown_error
async def wait_for_requests(
self,
count: int,
*,
timeout_seconds: float = 3.0,
) -> None:
await wait_until(
lambda: self.request_count >= count,
timeout_seconds=timeout_seconds,
)
def _handle_get(
self,
handler: BaseHTTPRequestHandler,
) -> None:
parsed = urlsplit(handler.path)
query = {
key: tuple(values)
for key, values in parse_qs(
parsed.query,
keep_blank_values=True,
).items()
}
request = LoopbackHttpRequest(
path=parsed.path,
query=query,
)
self._events.append("rest.request")
with self._lock:
self._requests.append(request)
response = (
self._responses.popleft()
if self._responses
else LoopbackHttpResponse(body=[])
)
if response.release is not None:
response.release.wait()
if isinstance(response.body, bytes):
body = response.body
elif isinstance(response.body, str):
body = response.body.encode("utf-8")
else:
body = json.dumps(response.body).encode("utf-8")
try:
handler.send_response(response.status)
handler.send_header(
"Content-Type",
"application/json",
)
handler.send_header(
"Content-Length",
str(len(body)),
)
handler.send_header(
"Connection",
"close",
)
handler.end_headers()
handler.wfile.write(body)
except (BrokenPipeError, ConnectionResetError):
pass
class LoopbackTcpFaultProxy:
"""TCP relay с управляемой потерей client → server traffic."""
def __init__(
self,
*,
target_host: str,
target_port: int,
) -> None:
self._target_host = target_host
self._target_port = target_port
self._server: asyncio.Server | None = None
self._host = "127.0.0.1"
self._port: int | None = None
self._connection_count = 0
self._blackholed_connections: set[int] = set()
self._handler_tasks: set[asyncio.Task[Any]] = set()
self._relay_tasks: set[asyncio.Task[Any]] = set()
self._writers: set[asyncio.StreamWriter] = set()
@property
def url(self) -> str:
if self._port is None:
raise RuntimeError("TCP fault proxy is not started.")
return f"ws://{self._host}:{self._port}"
@property
def connection_count(self) -> int:
return self._connection_count
@property
def active_handler_count(self) -> int:
return sum(
not task.done()
for task in self._handler_tasks
)
@property
def active_relay_count(self) -> int:
return sum(
not task.done()
for task in self._relay_tasks
)
@property
def tracked_writer_count(self) -> int:
return len(self._writers)
async def start(self) -> None:
if self._server is not None:
raise RuntimeError("TCP fault proxy is already started.")
server = await asyncio.start_server(
self._handle_client,
self._host,
0,
)
self._server = server
sockets = server.sockets
if not sockets:
server.close()
await server.wait_closed()
self._server = None
raise RuntimeError(
"TCP fault proxy did not expose a listening socket."
)
socket = sockets[0]
address = socket.getsockname()
self._port = int(address[1])
def blackhole_client_to_server(
self,
connection_index: int,
) -> None:
self._blackholed_connections.add(connection_index)
async def stop(self) -> None:
server = self._server
if server is None:
return
self._server = None
self._port = None
server.close()
tasks = tuple(
task
for task in (
*self._relay_tasks,
*self._handler_tasks,
)
if not task.done()
)
writers = tuple(self._writers)
for task in tasks:
task.cancel()
for writer in writers:
writer.close()
async def wait_for_writer(
writer: asyncio.StreamWriter,
) -> None:
try:
await writer.wait_closed()
except (BrokenPipeError, ConnectionResetError):
pass
await asyncio.wait_for(
server.wait_closed(),
timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS,
)
if writers:
try:
await asyncio.wait_for(
asyncio.gather(
*(
wait_for_writer(writer)
for writer in writers
),
),
timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS,
)
except TimeoutError:
for writer in writers:
writer.transport.abort()
await asyncio.wait_for(
asyncio.gather(
*(
wait_for_writer(writer)
for writer in writers
),
),
timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS,
)
for writer in writers:
self._writers.discard(writer)
if tasks:
await asyncio.wait_for(
asyncio.gather(
*tasks,
return_exceptions=True,
),
timeout=RESOURCE_CLEANUP_TIMEOUT_SECONDS,
)
if (
self.active_handler_count
or self.active_relay_count
or self.tracked_writer_count
):
raise AssertionError(
"TCP fault proxy resources did not stop."
)
async def _handle_client(
self,
client_reader: asyncio.StreamReader,
client_writer: asyncio.StreamWriter,
) -> None:
task = asyncio.current_task()
if task is None:
raise RuntimeError("TCP proxy handler has no asyncio task.")
self._handler_tasks.add(task)
connection_index = self._connection_count
self._connection_count += 1
server_writer: asyncio.StreamWriter | None = None
relay_tasks: set[asyncio.Task[None]] = set()
try:
server_reader, server_writer = await asyncio.open_connection(
self._target_host,
self._target_port,
)
self._writers.update(
{
client_writer,
server_writer,
}
)
upstream = asyncio.create_task(
self._relay(
client_reader,
server_writer,
should_drop=lambda: (
connection_index
in self._blackholed_connections
),
),
name="loopback-proxy-upstream",
)
downstream = asyncio.create_task(
self._relay(
server_reader,
client_writer,
should_drop=lambda: False,
),
name="loopback-proxy-downstream",
)
relay_tasks = {
upstream,
downstream,
}
self._relay_tasks.update(relay_tasks)
done, pending = await asyncio.wait(
relay_tasks,
return_when=asyncio.FIRST_COMPLETED,
)
for relay_task in pending:
relay_task.cancel()
await asyncio.gather(
*done,
*pending,
return_exceptions=True,
)
finally:
self._relay_tasks.difference_update(relay_tasks)
for writer in (
client_writer,
server_writer,
):
if writer is None:
continue
self._writers.discard(writer)
writer.close()
try:
await writer.wait_closed()
except (BrokenPipeError, ConnectionResetError):
pass
self._handler_tasks.discard(task)
@staticmethod
async def _relay(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
*,
should_drop: Callable[[], bool],
) -> None:
while True:
data = await reader.read(65_536)
if not data:
return
if should_drop():
continue
writer.write(data)
await writer.drain()
class LoopbackTradeEnvironment:
"""Exception-safe владелец локальных сетевых ресурсов теста."""
def __init__(
self,
*,
websocket: LoopbackTradeWebSocketServer,
rest: LoopbackTradeRestServer,
use_fault_proxy: bool = False,
) -> None:
self.websocket = websocket
self.rest = rest
self.use_fault_proxy = use_fault_proxy
self.proxy: LoopbackTcpFaultProxy | None = None
self._exit_stack: AsyncExitStack | None = None
@property
def websocket_url(self) -> str:
proxy = self.proxy
if proxy is not None:
return proxy.url
return self.websocket.url
async def __aenter__(self) -> LoopbackTradeEnvironment:
if self._exit_stack is not None:
raise RuntimeError(
"Loopback environment is already active."
)
stack = AsyncExitStack()
await stack.__aenter__()
self._exit_stack = stack
try:
self.rest.start()
stack.push_async_callback(self._stop_rest)
await self.websocket.start()
stack.push_async_callback(self._stop_websocket)
if self.use_fault_proxy:
proxy = LoopbackTcpFaultProxy(
target_host="127.0.0.1",
target_port=self.websocket.port,
)
self.proxy = proxy
await proxy.start()
stack.push_async_callback(self._stop_proxy)
except BaseException:
self._exit_stack = None
await stack.aclose()
raise
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: TracebackType | None,
) -> bool | None:
stack = self._exit_stack
self._exit_stack = None
if stack is None:
return None
return await stack.__aexit__(
exc_type,
exc_value,
traceback,
)
async def _stop_proxy(self) -> None:
proxy = self.proxy
if proxy is None:
return
await asyncio.wait_for(
proxy.stop(),
timeout=ENVIRONMENT_CLEANUP_TIMEOUT_SECONDS,
)
async def _stop_websocket(self) -> None:
await asyncio.wait_for(
self.websocket.stop(),
timeout=ENVIRONMENT_CLEANUP_TIMEOUT_SECONDS,
)
async def _stop_rest(self) -> None:
await asyncio.wait_for(
asyncio.to_thread(
self.rest.stop,
),
timeout=ENVIRONMENT_CLEANUP_TIMEOUT_SECONDS,
)

782
app/tests/integration/market_data/acquisition/runtime/test_trade_stream_loopback_integration.py Normal file
View File

@@ -0,0 +1,782 @@
from __future__ import annotations
import asyncio
import threading
import time
from typing import Any
import pytest
from tests.integration.market_data.acquisition.runtime import (
loopback_trade_exchange,
)
from tests.integration.market_data.acquisition.runtime.loopback_trade_exchange import (
LoopbackHttpResponse,
LoopbackTcpFaultProxy,
LoopbackTradeEnvironment,
LoopbackTradeRestServer,
LoopbackTradeWebSocketServer,
wait_until,
)
from src.market_data.acquisition.exceptions import (
TradeTransportError,
WebSocketMessageDecodeError,
)
from src.market_data.acquisition.runtime.runtime_reconnect_recovery_coordinator import (
RuntimeReconnectRecoveryCoordinator,
)
from src.market_data.acquisition.runtime.trade_stream_production_runtime import (
TradeStreamProductionRuntime,
TradeStreamProductionRuntimeState,
)
from tests.support.trade_stream_runtime import (
SYMBOL,
assert_no_owned_tasks,
build_runtime,
reconnect_coordinator_from,
run_scenario,
start_runtime,
state_store_from,
stop_runtime,
)
pytestmark = pytest.mark.integration
def make_recovered_trade(
*,
trade_id: int,
timestamp_ms: int,
) -> dict[str, object]:
return {
"a": trade_id,
"p": "64555.56",
"q": "0.003",
"T": timestamp_ms,
"m": False,
}
class HangingStartupRuntime:
def __init__(self) -> None:
self._state = TradeStreamProductionRuntimeState.STARTING
self.stop_calls = 0
self.released = asyncio.Event()
self.finished = asyncio.Event()
@property
def state(self) -> TradeStreamProductionRuntimeState:
return self._state
async def run(self) -> None:
try:
await self.released.wait()
finally:
self.finished.set()
async def stop(self) -> None:
self.stop_calls += 1
self.released.set()
self._state = TradeStreamProductionRuntimeState.STOPPED
def test_start_runtime_timeout_cleans_task_it_created() -> None:
async def scenario() -> None:
runtime = HangingStartupRuntime()
with pytest.raises(TimeoutError):
await start_runtime(
runtime,
timeout_seconds=0.01,
)
assert runtime.stop_calls == 1
assert runtime.finished.is_set()
await assert_no_owned_tasks()
run_scenario(scenario())
def test_environment_cleans_rest_after_websocket_setup_failure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
async def scenario() -> None:
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
async def broken_start() -> None:
raise RuntimeError("websocket setup failed")
monkeypatch.setattr(
websocket,
"start",
broken_start,
)
with pytest.raises(
RuntimeError,
match="websocket setup failed",
):
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
):
raise AssertionError(
"environment entered after setup failure"
)
assert rest.thread_is_alive is False
run_scenario(scenario())
def test_environment_continues_after_websocket_cleanup_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
async def scenario() -> None:
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
original_stop = websocket.stop
async def broken_stop() -> None:
await original_stop()
raise RuntimeError("websocket cleanup failed")
monkeypatch.setattr(
websocket,
"stop",
broken_stop,
)
with pytest.raises(
RuntimeError,
match="websocket cleanup failed",
):
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
):
assert rest.thread_is_alive is True
assert websocket.active_handler_count == 0
assert rest.thread_is_alive is False
run_scenario(scenario())
def test_proxy_waits_for_aborted_writer_before_removing_tracking(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class RecordingServer:
def __init__(self) -> None:
self.close_calls = 0
self.wait_closed_calls = 0
def close(self) -> None:
self.close_calls += 1
async def wait_closed(self) -> None:
self.wait_closed_calls += 1
class HangingWriter:
def __init__(self) -> None:
self.close_calls = 0
self.wait_closed_calls = 0
self.aborted = False
self.transport = self
def close(self) -> None:
self.close_calls += 1
async def wait_closed(self) -> None:
self.wait_closed_calls += 1
if not self.aborted:
await asyncio.Event().wait()
def abort(self) -> None:
self.aborted = True
async def scenario() -> None:
proxy = LoopbackTcpFaultProxy(
target_host="127.0.0.1",
target_port=1,
)
proxy_graph: Any = proxy
server = RecordingServer()
writer = HangingWriter()
proxy_graph._server = server
proxy_graph._port = 1
proxy_graph._writers.add(writer)
monkeypatch.setattr(
loopback_trade_exchange,
"RESOURCE_CLEANUP_TIMEOUT_SECONDS",
0.01,
)
await proxy.stop()
assert server.close_calls == 1
assert server.wait_closed_calls == 1
assert writer.close_calls == 1
assert writer.wait_closed_calls == 2
assert writer.aborted is True
assert proxy.tracked_writer_count == 0
run_scenario(scenario())
def test_real_loopback_websocket_updates_shared_checkpoint() -> None:
async def scenario() -> None:
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
) as environment:
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(runtime)
await websocket.wait_for_subscriptions(1)
timestamp_ms = time.time_ns() // 1_000_000
await websocket.send_trade(
0,
symbol=SYMBOL,
trade_id=100,
timestamp_ms=timestamp_ms,
)
state_store = state_store_from(runtime)
await wait_until(
lambda: (
state_store.contains(SYMBOL)
and state_store.get(SYMBOL).last_trade_id == 100
)
)
assert websocket.subscriptions[0].symbols == (SYMBOL,)
assert rest.request_count == 0
assert runtime_task.done() is False
finally:
if runtime_task is not None:
await stop_runtime(runtime, runtime_task)
assert websocket.active_handler_count == 0
assert rest.thread_is_alive is False
await assert_no_owned_tasks()
run_scenario(scenario())
def test_reconnect_restores_then_recovers_before_buffered_live() -> None:
async def scenario() -> None:
events: list[str] = []
recovery_release = threading.Event()
base_time_ms = time.time_ns() // 1_000_000 - 1_000
websocket = LoopbackTradeWebSocketServer(
events=events,
)
rest = LoopbackTradeRestServer(
responses=(
LoopbackHttpResponse(
body=[
make_recovered_trade(
trade_id=201,
timestamp_ms=base_time_ms + 100,
),
],
release=recovery_release,
),
),
events=events,
)
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
) as environment:
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(runtime)
await websocket.wait_for_subscriptions(1)
await websocket.send_trade(
0,
symbol=SYMBOL,
trade_id=200,
timestamp_ms=base_time_ms,
)
state_store = state_store_from(runtime)
await wait_until(
lambda: (
state_store.contains(SYMBOL)
and state_store.get(SYMBOL).last_trade_id == 200
)
)
await websocket.abort_connection(0)
await websocket.wait_for_subscriptions(2)
await rest.wait_for_requests(1)
await websocket.send_trade(
1,
symbol=SYMBOL,
trade_id=202,
timestamp_ms=base_time_ms + 200,
)
await asyncio.sleep(0)
assert state_store.get(SYMBOL).last_trade_id == 200
assert events.index("ws.subscribe:1") < events.index(
"rest.request"
)
recovery_release.set()
await wait_until(
lambda: state_store.get(SYMBOL).last_trade_id == 202,
)
assert runtime_task.done() is False
assert rest.requests[0].path == "/api/v1/aggTrades"
assert rest.requests[0].query["symbol"] == (SYMBOL,)
assert (
reconnect_coordinator_from(runtime).generation
== 1
)
finally:
recovery_release.set()
if runtime_task is not None:
await stop_runtime(runtime, runtime_task)
await assert_no_owned_tasks()
run_scenario(scenario())
def test_repeated_network_disconnects_advance_one_generation_each() -> None:
async def scenario() -> None:
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
base_time_ms = time.time_ns() // 1_000_000 - 1_000
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
) as environment:
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(runtime)
await websocket.wait_for_subscriptions(1)
await websocket.send_trade(
0,
symbol=SYMBOL,
trade_id=300,
timestamp_ms=base_time_ms,
)
state_store = state_store_from(runtime)
await wait_until(
lambda: (
state_store.contains(SYMBOL)
and state_store.get(SYMBOL).last_trade_id == 300
)
)
for generation in range(1, 4):
previous_connection = generation - 1
if generation == 1:
await websocket.close_connection(
previous_connection,
)
else:
await websocket.abort_connection(
previous_connection,
)
await websocket.wait_for_subscriptions(
generation + 1,
)
await rest.wait_for_requests(generation)
trade_id = 300 + generation
await websocket.send_trade(
generation,
symbol=SYMBOL,
trade_id=trade_id,
timestamp_ms=(
base_time_ms + generation * 100
),
)
await wait_until(
lambda trade_id=trade_id: (
state_store.get(SYMBOL).last_trade_id
== trade_id
)
)
assert websocket.connection_count == 4
assert len(websocket.subscriptions) == 4
assert rest.request_count == 3
assert (
reconnect_coordinator_from(runtime).generation
== 3
)
assert runtime_task.done() is False
finally:
if runtime_task is not None:
await stop_runtime(runtime, runtime_task)
await assert_no_owned_tasks()
run_scenario(scenario())
def test_http_recovery_error_rejects_buffered_live_and_cleans_up() -> None:
async def scenario() -> None:
recovery_release = threading.Event()
base_time_ms = time.time_ns() // 1_000_000 - 1_000
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer(
responses=(
LoopbackHttpResponse(
body={
"error": "recovery unavailable",
},
status=503,
release=recovery_release,
),
),
)
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
) as environment:
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(runtime)
await websocket.wait_for_subscriptions(1)
await websocket.send_trade(
0,
symbol=SYMBOL,
trade_id=400,
timestamp_ms=base_time_ms,
)
state_store = state_store_from(runtime)
await wait_until(
lambda: (
state_store.contains(SYMBOL)
and state_store.get(SYMBOL).last_trade_id == 400
)
)
await websocket.abort_connection(0)
await websocket.wait_for_subscriptions(2)
await rest.wait_for_requests(1)
await websocket.send_trade(
1,
symbol=SYMBOL,
trade_id=402,
timestamp_ms=base_time_ms + 200,
)
recovery_release.set()
with pytest.raises(
TradeTransportError,
match="Не удалось получить агрегированные сделки",
):
await asyncio.wait_for(
runtime_task,
timeout=3,
)
assert state_store.get(SYMBOL).last_trade_id == 400
assert (
reconnect_coordinator_from(runtime)
.live_processing_gate
.failed
)
assert (
runtime.state
is TradeStreamProductionRuntimeState.FAILED
)
finally:
recovery_release.set()
if (
runtime_task is not None
and not runtime_task.done()
):
await stop_runtime(runtime, runtime_task)
await assert_no_owned_tasks()
run_scenario(scenario())
def test_invalid_websocket_json_is_terminal_and_releases_resources() -> None:
async def scenario() -> None:
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
) as environment:
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(runtime)
await websocket.wait_for_subscriptions(1)
await websocket.send_raw(0, "{not-json")
with pytest.raises(
WebSocketMessageDecodeError,
):
await asyncio.wait_for(
runtime_task,
timeout=3,
)
assert (
runtime.state
is TradeStreamProductionRuntimeState.FAILED
)
assert rest.request_count == 0
finally:
if (
runtime_task is not None
and not runtime_task.done()
):
await stop_runtime(runtime, runtime_task)
await assert_no_owned_tasks()
run_scenario(scenario())
def test_stop_waits_for_real_http_recovery_worker() -> None:
async def scenario() -> None:
recovery_release = threading.Event()
base_time_ms = time.time_ns() // 1_000_000 - 1_000
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer(
responses=(
LoopbackHttpResponse(
body=[
make_recovered_trade(
trade_id=501,
timestamp_ms=base_time_ms + 100,
),
],
release=recovery_release,
),
),
)
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
) as environment:
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(runtime)
await websocket.wait_for_subscriptions(1)
await websocket.send_trade(
0,
symbol=SYMBOL,
trade_id=500,
timestamp_ms=base_time_ms,
)
state_store = state_store_from(runtime)
await wait_until(
lambda: (
state_store.contains(SYMBOL)
and state_store.get(SYMBOL).last_trade_id == 500
)
)
await websocket.abort_connection(0)
await websocket.wait_for_subscriptions(2)
await rest.wait_for_requests(1)
stop_task = asyncio.create_task(runtime.stop())
await asyncio.sleep(0)
await asyncio.sleep(0)
assert stop_task.done() is False
recovery_release.set()
await asyncio.wait_for(
stop_task,
timeout=3,
)
await asyncio.wait_for(
runtime_task,
timeout=3,
)
assert state_store.get(SYMBOL).last_trade_id == 501
assert (
runtime.state
is TradeStreamProductionRuntimeState.STOPPED
)
finally:
recovery_release.set()
if (
runtime_task is not None
and not runtime_task.done()
):
await stop_runtime(runtime, runtime_task)
await assert_no_owned_tasks()
run_scenario(scenario())
def test_ping_timeout_and_receive_failure_share_one_reconnect(
monkeypatch: pytest.MonkeyPatch,
) -> None:
caller_task_names: list[str] = []
original_reconnect = (
RuntimeReconnectRecoveryCoordinator
.reconnect_after_transport_failure
)
async def recording_reconnect(
self: RuntimeReconnectRecoveryCoordinator,
*,
observed_generation: int,
) -> None:
task = asyncio.current_task()
caller_task_names.append(
task.get_name()
if task is not None
else "<no-task>"
)
await original_reconnect(
self,
observed_generation=observed_generation,
)
monkeypatch.setattr(
RuntimeReconnectRecoveryCoordinator,
"reconnect_after_transport_failure",
recording_reconnect,
)
async def scenario() -> None:
base_time_ms = time.time_ns() // 1_000_000 - 1_000
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
use_fault_proxy=True,
) as environment:
proxy = environment.proxy
assert proxy is not None
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
probe_timeout_seconds=0.05,
close_timeout_seconds=0.05,
heartbeat_timeout_seconds=0.15,
scheduler_interval_seconds=0.05,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(runtime)
await websocket.wait_for_subscriptions(1)
await websocket.send_trade(
0,
symbol=SYMBOL,
trade_id=600,
timestamp_ms=base_time_ms,
)
state_store = state_store_from(runtime)
await wait_until(
lambda: (
state_store.contains(SYMBOL)
and state_store.get(SYMBOL).last_trade_id == 600
)
)
proxy.blackhole_client_to_server(0)
await websocket.wait_for_subscriptions(2)
await rest.wait_for_requests(1)
await websocket.send_trade(
1,
symbol=SYMBOL,
trade_id=601,
timestamp_ms=base_time_ms + 100,
)
await wait_until(
lambda: state_store.get(SYMBOL).last_trade_id == 601,
)
assert proxy.connection_count == 2
assert websocket.connection_count == 2
assert len(websocket.subscriptions) == 2
assert rest.request_count == 1
assert (
reconnect_coordinator_from(runtime).generation
== 1
)
assert sorted(caller_task_names) == [
"trade-stream-receive",
"trade-stream-scheduler",
]
assert runtime_task.done() is False
finally:
if runtime_task is not None:
await stop_runtime(runtime, runtime_task)
assert proxy.active_handler_count == 0
assert proxy.active_relay_count == 0
await assert_no_owned_tasks()
run_scenario(scenario())

176
app/tests/live/market_data/acquisition/runtime/test_trade_stream_runtime_live.py Normal file
View File

@@ -0,0 +1,176 @@
from __future__ import annotations
import asyncio
from typing import Any
import pytest
from tests.support.live_trade_stream import (
LIVE_RUNTIME_CLEANUP_TIMEOUT_SECONDS,
LiveTestConfigurationError,
LiveTradeStreamTestConfig,
build_live_trade_stream_settings,
load_live_trade_stream_test_config,
)
pytestmark = pytest.mark.live
RUNNING_TASK_NAMES = (
"trade-stream-receive",
"trade-stream-runtime",
"trade-stream-scheduler",
)
def require_live_config() -> LiveTradeStreamTestConfig:
try:
config = load_live_trade_stream_test_config()
except LiveTestConfigurationError as error:
pytest.fail(
f"Invalid opt-in live configuration: {error}",
pytrace=False,
)
if config is None:
pytest.skip(
"Live verification requires DZENTRA_RUN_LIVE_TESTS=1."
)
return config
def test_live_runtime_reconnects_recovers_and_resumes_trades() -> None:
config = require_live_config()
from src.bootstrap.trade_stream_runtime import (
build_trade_stream_production_runtime,
)
from src.market_data.acquisition.runtime.trade_stream_production_runtime import (
TradeStreamProductionRuntimeState,
)
from src.market_data.acquisition.trade_id_sequence import (
is_trade_id_newer,
is_trade_id_same_or_newer,
)
from tests.support.trade_stream_runtime import (
active_owned_task_names,
assert_no_owned_tasks,
reconnect_coordinator_from,
run_scenario,
start_runtime,
state_store_from,
stop_runtime,
wait_until_or_runtime_exit,
)
settings = build_live_trade_stream_settings(config)
runtime = build_trade_stream_production_runtime(settings)
assert runtime is not None
async def scenario() -> None:
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(
runtime,
timeout_seconds=config.trade_timeout_seconds,
)
state_store = state_store_from(runtime)
await wait_until_or_runtime_exit(
lambda: (
state_store.contains(config.symbol)
and state_store.get(
config.symbol
).last_trade_id
is not None
),
runtime_task=runtime_task,
timeout_seconds=config.trade_timeout_seconds,
)
first_state = state_store.get(config.symbol)
first_trade_id = first_state.last_trade_id
assert first_trade_id is not None
assert first_state.last_trade is not None
assert first_state.last_trade.symbol == config.symbol
assert runtime.state is (
TradeStreamProductionRuntimeState.RUNNING
)
assert runtime_task.done() is False
coordinator: Any = reconnect_coordinator_from(runtime)
source_generation = coordinator.generation
await asyncio.wait_for(
coordinator.reconnect(),
timeout=config.trade_timeout_seconds,
)
assert coordinator.generation == source_generation + 1
assert coordinator.live_processing_gate.locked is False
assert coordinator.live_processing_gate.failed is False
assert runtime.state is (
TradeStreamProductionRuntimeState.RUNNING
)
assert runtime_task.done() is False
checkpoint_after_recovery = state_store.get(
config.symbol
).last_trade_id
assert checkpoint_after_recovery is not None
assert is_trade_id_same_or_newer(
checkpoint_after_recovery,
first_trade_id,
)
def checkpoint_advanced_after_recovery() -> bool:
last_trade_id = state_store.get(
config.symbol
).last_trade_id
return (
last_trade_id is not None
and is_trade_id_newer(
last_trade_id,
checkpoint_after_recovery,
)
)
await wait_until_or_runtime_exit(
checkpoint_advanced_after_recovery,
runtime_task=runtime_task,
timeout_seconds=config.trade_timeout_seconds,
)
final_state = state_store.get(config.symbol)
assert final_state.last_trade_id is not None
assert is_trade_id_newer(
final_state.last_trade_id,
checkpoint_after_recovery,
)
assert final_state.last_trade is not None
assert final_state.last_trade.symbol == config.symbol
assert active_owned_task_names() == RUNNING_TASK_NAMES
assert runtime_task.done() is False
finally:
if runtime_task is not None:
await stop_runtime(
runtime,
runtime_task,
timeout_seconds=(
LIVE_RUNTIME_CLEANUP_TIMEOUT_SECONDS
),
)
await assert_no_owned_tasks()
run_scenario(
scenario(),
timeout_seconds=config.scenario_timeout_seconds,
)

589
app/tests/stress/market_data/acquisition/runtime/test_trade_stream_runtime_stress.py Normal file
View File

@@ -0,0 +1,589 @@
from __future__ import annotations
import asyncio
import gc
import os
import time
import tracemalloc
from typing import Any
import pytest
from src.market_data.acquisition.consistency.trade_stream_state import (
DEFAULT_DEDUPLICATION_WINDOW_SIZE,
TradeStreamState,
)
from src.market_data.acquisition.runtime.trade_stream_production_runtime import (
TradeStreamProductionRuntime,
TradeStreamProductionRuntimeState,
)
from tests.integration.market_data.acquisition.runtime.loopback_trade_exchange import (
LoopbackTcpFaultProxy,
LoopbackTradeEnvironment,
LoopbackTradeRestServer,
LoopbackTradeWebSocketServer,
wait_until,
)
from tests.support.trade_stream_runtime import (
SYMBOL,
active_owned_task_names,
assert_no_owned_tasks,
build_runtime,
reconnect_coordinator_from,
run_scenario,
start_runtime,
state_store_from,
stop_runtime,
)
pytestmark = pytest.mark.stress
TRADE_BURST_COUNT = 20_000
TRADE_BURST_SIZE = 250
TRADE_BURST_TIMEOUT_SECONDS = 90.0
MEMORY_WARMUP_TRADE_COUNT = 12_000
MEMORY_SAMPLE_TRADE_INTERVAL = 2_000
MAX_RETAINED_MEMORY_GROWTH_BYTES = 4 * 1024 * 1024
RECONNECT_GENERATION_COUNT = 30
RECONNECT_STORM_TIMEOUT_SECONDS = 90.0
DEFAULT_SOAK_SECONDS = 120
SOAK_TRADES_PER_SECOND = 50
SOAK_RECONNECT_INTERVAL_SECONDS = 5
SOAK_CLEANUP_ALLOWANCE_SECONDS = 60.0
STRESS_CONDITION_TIMEOUT_SECONDS = 10.0
RUNNING_TASK_NAMES = (
"trade-stream-receive",
"trade-stream-runtime",
"trade-stream-scheduler",
)
def read_soak_seconds() -> int:
raw_value = os.environ.get(
"DZENTRA_SOAK_SECONDS",
str(DEFAULT_SOAK_SECONDS),
)
try:
duration = int(raw_value)
except ValueError as error:
raise ValueError(
"DZENTRA_SOAK_SECONDS must be an integer."
) from error
if duration <= 0:
raise ValueError(
"DZENTRA_SOAK_SECONDS must be positive."
)
return duration
def acquisition_memory_bytes() -> int:
snapshot = tracemalloc.take_snapshot().filter_traces(
(
tracemalloc.Filter(
True,
"*/src/market_data/acquisition/*",
),
)
)
return sum(
statistic.size
for statistic in snapshot.statistics("filename")
)
def assert_runtime_is_settled(
runtime: TradeStreamProductionRuntime,
*,
expected_generation: int,
) -> None:
coordinator: Any = reconnect_coordinator_from(runtime)
assert runtime.state is TradeStreamProductionRuntimeState.RUNNING
assert coordinator.generation == expected_generation
assert coordinator.live_processing_gate.locked is False
assert coordinator.live_processing_gate.failed is False
assert coordinator._recovery_task is None
assert active_owned_task_names() == RUNNING_TASK_NAMES
async def send_trade_batch(
websocket: LoopbackTradeWebSocketServer,
*,
connection_index: int,
first_trade_id: int,
count: int,
first_timestamp_ms: int,
) -> int:
for offset in range(count):
await websocket.send_trade(
connection_index,
symbol=SYMBOL,
trade_id=first_trade_id + offset,
timestamp_ms=first_timestamp_ms + offset,
)
return first_trade_id + count - 1
def test_memory_filter_observes_acquisition_allocations() -> None:
tracing_was_active = tracemalloc.is_tracing()
if not tracing_was_active:
tracemalloc.start()
states: tuple[TradeStreamState, ...] = ()
try:
states = tuple(
TradeStreamState(
symbol=f"MEMORY_TEST_{index}",
)
for index in range(100)
)
gc.collect()
baseline = acquisition_memory_bytes()
assert baseline > 0, (
"Acquisition tracemalloc filter returned an empty baseline."
)
finally:
del states
gc.collect()
if not tracing_was_active:
tracemalloc.stop()
def test_trade_burst_remains_ordered_and_memory_bounded() -> None:
async def scenario() -> None:
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
memory_samples: list[int] = []
first_trade_id = 1_000_000
first_timestamp_ms = time.time_ns() // 1_000_000
tracing_was_active = tracemalloc.is_tracing()
if not tracing_was_active:
tracemalloc.start()
try:
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
) as environment:
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(
runtime,
timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS,
)
await websocket.wait_for_subscriptions(
1,
timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS,
)
state_store = state_store_from(runtime)
for sent_count in range(
TRADE_BURST_SIZE,
TRADE_BURST_COUNT + 1,
TRADE_BURST_SIZE,
):
batch_first_id = (
first_trade_id
+ sent_count
- TRADE_BURST_SIZE
)
last_trade_id = await send_trade_batch(
websocket,
connection_index=0,
first_trade_id=batch_first_id,
count=TRADE_BURST_SIZE,
first_timestamp_ms=(
first_timestamp_ms
+ sent_count
- TRADE_BURST_SIZE
),
)
await wait_until(
lambda last_trade_id=last_trade_id: (
state_store.contains(SYMBOL)
and state_store.get(
SYMBOL
).last_trade_id
== last_trade_id
),
timeout_seconds=(
STRESS_CONDITION_TIMEOUT_SECONDS
),
)
if (
sent_count >= MEMORY_WARMUP_TRADE_COUNT
and (
sent_count
- MEMORY_WARMUP_TRADE_COUNT
)
% MEMORY_SAMPLE_TRADE_INTERVAL
== 0
):
gc.collect()
memory_samples.append(
acquisition_memory_bytes()
)
state: Any = state_store.get(SYMBOL)
expected_last_trade_id = (
first_trade_id + TRADE_BURST_COUNT - 1
)
assert state.last_trade_id == expected_last_trade_id
assert len(state._trade_window) == (
DEFAULT_DEDUPLICATION_WINDOW_SIZE
)
assert len(state._trades) == (
DEFAULT_DEDUPLICATION_WINDOW_SIZE
)
assert len(memory_samples) == 5
assert memory_samples[0] > 0, (
"Acquisition tracemalloc filter returned an "
f"empty baseline: samples={memory_samples!r}"
)
retained_memory_growth = (
memory_samples[-1] - memory_samples[0]
)
assert (
retained_memory_growth
<= MAX_RETAINED_MEMORY_GROWTH_BYTES
), (
"Acquisition retained memory exceeded the "
"post-warmup limit: "
f"growth={retained_memory_growth} "
f"samples={memory_samples!r}"
)
assert websocket.connection_count == 1
assert websocket.active_connection_count == 1
assert len(websocket.subscriptions) == 1
assert rest.request_count == 0
assert_runtime_is_settled(
runtime,
expected_generation=0,
)
assert runtime_task.done() is False
finally:
if runtime_task is not None:
await stop_runtime(runtime, runtime_task)
assert websocket.active_connection_count == 0
assert websocket.active_handler_count == 0
assert rest.thread_is_alive is False
await assert_no_owned_tasks()
finally:
if not tracing_was_active:
tracemalloc.stop()
run_scenario(
scenario(),
timeout_seconds=TRADE_BURST_TIMEOUT_SECONDS,
)
def test_reconnect_storm_advances_exactly_one_generation_per_fault() -> None:
async def scenario() -> None:
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
first_trade_id = 2_000_000
first_timestamp_ms = time.time_ns() // 1_000_000
proxy: LoopbackTcpFaultProxy | None = None
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
use_fault_proxy=True,
) as environment:
proxy = environment.proxy
assert proxy is not None
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
probe_timeout_seconds=0.05,
close_timeout_seconds=0.05,
heartbeat_timeout_seconds=0.15,
scheduler_interval_seconds=0.05,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(
runtime,
timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS,
)
await websocket.wait_for_subscriptions(
1,
timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS,
)
state_store = state_store_from(runtime)
await websocket.send_trade(
0,
symbol=SYMBOL,
trade_id=first_trade_id,
timestamp_ms=first_timestamp_ms,
)
await wait_until(
lambda: (
state_store.contains(SYMBOL)
and state_store.get(SYMBOL).last_trade_id
== first_trade_id
),
timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS,
)
for generation in range(
1,
RECONNECT_GENERATION_COUNT + 1,
):
previous_connection = generation - 1
fault_kind = generation % 3
if fault_kind == 1:
await websocket.close_connection(
previous_connection,
)
elif fault_kind == 2:
await websocket.abort_connection(
previous_connection,
)
else:
proxy.blackhole_client_to_server(
previous_connection,
)
await websocket.wait_for_subscriptions(
generation + 1,
timeout_seconds=(
STRESS_CONDITION_TIMEOUT_SECONDS
),
)
await rest.wait_for_requests(
generation,
timeout_seconds=(
STRESS_CONDITION_TIMEOUT_SECONDS
),
)
trade_id = first_trade_id + generation
await websocket.send_trade(
generation,
symbol=SYMBOL,
trade_id=trade_id,
timestamp_ms=(
first_timestamp_ms + generation
),
)
await wait_until(
lambda trade_id=trade_id: (
state_store.get(SYMBOL).last_trade_id
== trade_id
),
timeout_seconds=(
STRESS_CONDITION_TIMEOUT_SECONDS
),
)
assert_runtime_is_settled(
runtime,
expected_generation=generation,
)
assert proxy.connection_count == (
RECONNECT_GENERATION_COUNT + 1
)
assert websocket.connection_count == (
RECONNECT_GENERATION_COUNT + 1
)
assert websocket.active_connection_count == 1
assert len(websocket.subscriptions) == (
RECONNECT_GENERATION_COUNT + 1
)
assert rest.request_count == RECONNECT_GENERATION_COUNT
assert runtime_task.done() is False
finally:
if runtime_task is not None:
await stop_runtime(runtime, runtime_task)
assert proxy is not None
assert proxy.active_handler_count == 0
assert proxy.active_relay_count == 0
assert proxy.tracked_writer_count == 0
assert websocket.active_connection_count == 0
assert websocket.active_handler_count == 0
assert rest.thread_is_alive is False
await assert_no_owned_tasks()
run_scenario(
scenario(),
timeout_seconds=RECONNECT_STORM_TIMEOUT_SECONDS,
)
def test_runtime_remains_stable_during_local_soak() -> None:
soak_seconds = read_soak_seconds()
expected_reconnects = (
soak_seconds // SOAK_RECONNECT_INTERVAL_SECONDS
)
expected_trade_count = soak_seconds * SOAK_TRADES_PER_SECOND
async def scenario() -> None:
websocket = LoopbackTradeWebSocketServer()
rest = LoopbackTradeRestServer()
first_trade_id = 3_000_000
first_timestamp_ms = time.time_ns() // 1_000_000
async with LoopbackTradeEnvironment(
websocket=websocket,
rest=rest,
) as environment:
runtime = build_runtime(
websocket_url=environment.websocket_url,
rest_base_url=rest.base_url,
)
runtime_task: asyncio.Task[None] | None = None
try:
runtime_task = await start_runtime(
runtime,
timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS,
)
await websocket.wait_for_subscriptions(
1,
timeout_seconds=STRESS_CONDITION_TIMEOUT_SECONDS,
)
state_store = state_store_from(runtime)
loop = asyncio.get_running_loop()
started_at = loop.time()
connection_index = 0
next_trade_id = first_trade_id
for second in range(1, soak_seconds + 1):
if (
second
% SOAK_RECONNECT_INTERVAL_SECONDS
== 0
):
if connection_index % 2 == 0:
await websocket.close_connection(
connection_index,
)
else:
await websocket.abort_connection(
connection_index,
)
connection_index += 1
await websocket.wait_for_subscriptions(
connection_index + 1,
timeout_seconds=(
STRESS_CONDITION_TIMEOUT_SECONDS
),
)
await rest.wait_for_requests(
connection_index,
timeout_seconds=(
STRESS_CONDITION_TIMEOUT_SECONDS
),
)
last_trade_id = await send_trade_batch(
websocket,
connection_index=connection_index,
first_trade_id=next_trade_id,
count=SOAK_TRADES_PER_SECOND,
first_timestamp_ms=(
first_timestamp_ms
+ next_trade_id
- first_trade_id
),
)
next_trade_id = last_trade_id + 1
await wait_until(
lambda last_trade_id=last_trade_id: (
state_store.contains(SYMBOL)
and state_store.get(
SYMBOL
).last_trade_id
== last_trade_id
),
timeout_seconds=(
STRESS_CONDITION_TIMEOUT_SECONDS
),
)
if (
second
% SOAK_RECONNECT_INTERVAL_SECONDS
== 0
):
assert_runtime_is_settled(
runtime,
expected_generation=connection_index,
)
remaining = started_at + second - loop.time()
if remaining > 0:
await asyncio.sleep(remaining)
expected_last_trade_id = (
first_trade_id + expected_trade_count - 1
)
state: Any = state_store.get(SYMBOL)
assert state.last_trade_id == expected_last_trade_id
assert len(state._trade_window) <= (
DEFAULT_DEDUPLICATION_WINDOW_SIZE
)
assert len(state._trades) <= (
DEFAULT_DEDUPLICATION_WINDOW_SIZE
)
assert websocket.connection_count == (
expected_reconnects + 1
)
assert websocket.active_connection_count == 1
assert len(websocket.subscriptions) == (
expected_reconnects + 1
)
assert rest.request_count == expected_reconnects
assert_runtime_is_settled(
runtime,
expected_generation=expected_reconnects,
)
assert runtime_task.done() is False
finally:
if runtime_task is not None:
await stop_runtime(runtime, runtime_task)
assert websocket.active_connection_count == 0
assert websocket.active_handler_count == 0
assert rest.thread_is_alive is False
await assert_no_owned_tasks()
run_scenario(
scenario(),
timeout_seconds=(
soak_seconds + SOAK_CLEANUP_ALLOWANCE_SECONDS
),
)

1
app/tests/support/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Общие вспомогательные компоненты тестов Dzentra."""

18
app/tests/support/async_wait.py Normal file
View File

@@ -0,0 +1,18 @@
from __future__ import annotations
import asyncio
from collections.abc import Callable
Predicate = Callable[[], bool]
async def wait_until(
predicate: Predicate,
*,
timeout_seconds: float = 3.0,
) -> None:
"""Дождаться наблюдаемого условия без фиксированной длинной паузы."""
async with asyncio.timeout(timeout_seconds):
while not predicate():
await asyncio.sleep(0.005)

222
app/tests/support/live_trade_stream.py Normal file
View File

@@ -0,0 +1,222 @@
from __future__ import annotations
import math
import os
from collections.abc import Mapping
from dataclasses import dataclass
from typing import TYPE_CHECKING
from urllib.parse import urlsplit
if TYPE_CHECKING:
from src.core.config import Settings
RUN_LIVE_TESTS_ENV = "DZENTRA_RUN_LIVE_TESTS"
LIVE_REST_URL_ENV = "DZENTRA_LIVE_REST_URL"
LIVE_WEBSOCKET_URL_ENV = "DZENTRA_LIVE_WS_URL"
LIVE_SYMBOLS_ENV = "DZENTRA_LIVE_SYMBOLS"
LIVE_TRADE_TIMEOUT_ENV = "DZENTRA_LIVE_TRADE_TIMEOUT_SECONDS"
DEFAULT_LIVE_TRADE_TIMEOUT_SECONDS = 600.0
LIVE_RUNTIME_CLEANUP_TIMEOUT_SECONDS = 30.0
class LiveTestConfigurationError(ValueError):
"""Ошибка явной конфигурации opt-in live verification."""
@dataclass(frozen=True, slots=True)
class LiveTradeStreamTestConfig:
rest_url: str
websocket_url: str
symbol: str
trade_timeout_seconds: float
@property
def scenario_timeout_seconds(self) -> float:
return (
self.trade_timeout_seconds * 2
+ LIVE_RUNTIME_CLEANUP_TIMEOUT_SECONDS * 2
)
def load_live_trade_stream_test_config(
environment: Mapping[str, str] | None = None,
) -> LiveTradeStreamTestConfig | None:
values = environment if environment is not None else os.environ
opt_in = values.get(RUN_LIVE_TESTS_ENV, "").strip()
if opt_in in {"", "0"}:
return None
if opt_in != "1":
raise LiveTestConfigurationError(
f"{RUN_LIVE_TESTS_ENV} must be exactly 1 when enabled."
)
rest_url = _require_environment_value(
values,
LIVE_REST_URL_ENV,
).rstrip("/")
websocket_url = _require_environment_value(
values,
LIVE_WEBSOCKET_URL_ENV,
)
symbols_value = _require_environment_value(
values,
LIVE_SYMBOLS_ENV,
)
symbols = tuple(
symbol.strip()
for symbol in symbols_value.split(",")
)
if any(not symbol for symbol in symbols):
raise LiveTestConfigurationError(
f"{LIVE_SYMBOLS_ENV} must not contain empty symbols."
)
if len(symbols) != 1:
raise LiveTestConfigurationError(
f"{LIVE_SYMBOLS_ENV} must contain exactly one symbol "
"for bounded live verification."
)
_validate_secure_url(
rest_url,
name=LIVE_REST_URL_ENV,
expected_scheme="https",
required_path=None,
)
_validate_secure_url(
websocket_url,
name=LIVE_WEBSOCKET_URL_ENV,
expected_scheme="wss",
required_path="/connect",
)
timeout_seconds = _parse_positive_timeout(
values.get(
LIVE_TRADE_TIMEOUT_ENV,
str(DEFAULT_LIVE_TRADE_TIMEOUT_SECONDS),
),
)
return LiveTradeStreamTestConfig(
rest_url=rest_url,
websocket_url=websocket_url,
symbol=symbols[0],
trade_timeout_seconds=timeout_seconds,
)
def build_live_trade_stream_settings(
config: LiveTradeStreamTestConfig,
) -> Settings:
from src.core.config import Settings, TradeStreamSettings
return Settings(
bot_token="live-verification-does-not-use-telegram",
bot_parse_mode="HTML",
app_env="live-verification",
log_level="INFO",
tz="UTC",
exchange_enabled=True,
exchange_name="dzengi",
exchange_base_url=config.rest_url,
exchange_ws_url="",
exchange_api_key="",
exchange_api_secret="",
exchange_timeout_sec=20,
exchange_testnet=False,
default_symbol=config.symbol,
trade_stream=TradeStreamSettings(
enabled=True,
websocket_url=config.websocket_url,
symbols=(config.symbol,),
open_timeout_seconds=10.0,
probe_timeout_seconds=20.0,
close_timeout_seconds=10.0,
heartbeat_timeout_seconds=20.0,
scheduler_interval_seconds=5.0,
recovery_window_ms=3_599_999,
),
db_host="localhost",
db_port=5432,
db_name="live-verification",
db_user="live-verification",
db_password="",
debug_enabled=False,
journal_debug_enabled=False,
)
def _require_environment_value(
environment: Mapping[str, str],
name: str,
) -> str:
value = environment.get(name, "").strip()
if not value:
raise LiveTestConfigurationError(
f"{name} is required when {RUN_LIVE_TESTS_ENV}=1."
)
return value
def _validate_secure_url(
raw_url: str,
*,
name: str,
expected_scheme: str,
required_path: str | None,
) -> None:
parsed = urlsplit(raw_url)
if parsed.scheme.lower() != expected_scheme or not parsed.netloc:
raise LiveTestConfigurationError(
f"{name} must be an absolute {expected_scheme} URL."
)
if parsed.username is not None or parsed.password is not None:
raise LiveTestConfigurationError(
f"{name} must not contain embedded credentials."
)
if parsed.query or parsed.fragment:
raise LiveTestConfigurationError(
f"{name} must not contain a query or fragment."
)
if required_path is None:
if parsed.path not in {"", "/"}:
raise LiveTestConfigurationError(
f"{name} must be a base URL without an endpoint path."
)
return
if parsed.path.rstrip("/") != required_path:
raise LiveTestConfigurationError(
f"{name} must end with {required_path}."
)
def _parse_positive_timeout(raw_value: str) -> float:
try:
timeout_seconds = float(raw_value.strip())
except ValueError as error:
raise LiveTestConfigurationError(
f"{LIVE_TRADE_TIMEOUT_ENV} must be a number."
) from error
if (
not math.isfinite(timeout_seconds)
or timeout_seconds <= 0
):
raise LiveTestConfigurationError(
f"{LIVE_TRADE_TIMEOUT_ENV} must be positive and finite."
)
return timeout_seconds

314
app/tests/support/trade_stream_runtime.py Normal file
View File

@@ -0,0 +1,314 @@
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from typing import Any, Protocol
from src.bootstrap.trade_stream_runtime import (
build_trade_stream_production_runtime,
)
from src.core.config import Settings, TradeStreamSettings
from src.market_data.acquisition.consistency.trade_stream_state_store import (
TradeStreamStateStore,
)
from src.market_data.acquisition.runtime.trade_stream_production_runtime import (
TradeStreamProductionRuntime,
TradeStreamProductionRuntimeState,
)
from tests.support.async_wait import wait_until
SYMBOL = "BTC/USD_LEVERAGE"
RUNTIME_CLEANUP_TIMEOUT_SECONDS = 5.0
OWNED_TASK_NAMES = frozenset(
{
"trade-stream-receive",
"trade-stream-runtime",
"trade-stream-runtime-recovery",
"trade-stream-scheduler",
"trade-stream-startup",
}
)
class ManagedRuntimeProtocol(Protocol):
@property
def state(self) -> TradeStreamProductionRuntimeState:
...
async def run(self) -> None:
...
async def stop(self) -> None:
...
def run_scenario(
scenario: Awaitable[None],
*,
timeout_seconds: float = 30.0,
) -> None:
asyncio.run(
asyncio.wait_for(
scenario,
timeout=timeout_seconds,
)
)
def make_settings(
*,
websocket_url: str,
rest_base_url: str,
open_timeout_seconds: float = 0.5,
probe_timeout_seconds: float = 0.2,
close_timeout_seconds: float = 0.2,
heartbeat_timeout_seconds: float = 60.0,
scheduler_interval_seconds: float = 60.0,
) -> Settings:
return Settings(
bot_token="integration-test-token",
bot_parse_mode="HTML",
app_env="test",
log_level="INFO",
tz="UTC",
exchange_enabled=True,
exchange_name="dzengi",
exchange_base_url=rest_base_url,
exchange_ws_url="",
exchange_api_key="",
exchange_api_secret="",
exchange_timeout_sec=2,
exchange_testnet=True,
default_symbol=SYMBOL,
trade_stream=TradeStreamSettings(
enabled=True,
websocket_url=websocket_url,
symbols=(SYMBOL,),
open_timeout_seconds=open_timeout_seconds,
probe_timeout_seconds=probe_timeout_seconds,
close_timeout_seconds=close_timeout_seconds,
heartbeat_timeout_seconds=heartbeat_timeout_seconds,
scheduler_interval_seconds=scheduler_interval_seconds,
recovery_window_ms=3_599_999,
),
db_host="localhost",
db_port=5432,
db_name="integration",
db_user="integration",
db_password="",
debug_enabled=False,
journal_debug_enabled=False,
)
def build_runtime(
*,
websocket_url: str,
rest_base_url: str,
probe_timeout_seconds: float = 0.2,
close_timeout_seconds: float = 0.2,
heartbeat_timeout_seconds: float = 60.0,
scheduler_interval_seconds: float = 60.0,
) -> TradeStreamProductionRuntime:
runtime = build_trade_stream_production_runtime(
make_settings(
websocket_url=websocket_url,
rest_base_url=rest_base_url,
probe_timeout_seconds=probe_timeout_seconds,
close_timeout_seconds=close_timeout_seconds,
heartbeat_timeout_seconds=heartbeat_timeout_seconds,
scheduler_interval_seconds=scheduler_interval_seconds,
)
)
assert runtime is not None
return runtime
def state_store_from(
runtime: TradeStreamProductionRuntime,
) -> TradeStreamStateStore:
runtime_graph: Any = runtime
return (
runtime_graph
._reconnect_recovery_coordinator
._recovery_coordinator
._state_store
)
def reconnect_coordinator_from(
runtime: TradeStreamProductionRuntime,
) -> Any:
runtime_graph: Any = runtime
return runtime_graph._reconnect_recovery_coordinator
def active_owned_task_names() -> tuple[str, ...]:
current_task = asyncio.current_task()
return tuple(
sorted(
task.get_name()
for task in asyncio.all_tasks()
if task is not current_task
and not task.done()
and task.get_name() in OWNED_TASK_NAMES
)
)
async def start_runtime(
runtime: ManagedRuntimeProtocol,
*,
timeout_seconds: float = 3.0,
) -> asyncio.Task[None]:
task = asyncio.create_task(
runtime.run(),
name="trade-stream-runtime",
)
try:
await wait_until(
lambda: runtime.state
in {
TradeStreamProductionRuntimeState.RUNNING,
TradeStreamProductionRuntimeState.FAILED,
},
timeout_seconds=timeout_seconds,
)
if (
runtime.state
is TradeStreamProductionRuntimeState.FAILED
):
await task
except BaseException as error:
try:
await stop_runtime(runtime, task)
except BaseException as cleanup_error:
error.add_note(
"Runtime startup cleanup also failed: "
f"{type(cleanup_error).__name__}."
)
raise
return task
async def wait_until_or_runtime_exit(
predicate: Callable[[], bool],
*,
runtime_task: asyncio.Task[None],
timeout_seconds: float,
) -> None:
"""
Дождаться условия, немедленно распространяя завершение Runtime.
Runtime task не отменяется: helper владеет только внутренней задачей
ожидания условия. Если Runtime завершился без ошибки до выполнения
условия, это считается ошибкой проверочного сценария.
"""
condition_task = asyncio.create_task(
wait_until(
predicate,
timeout_seconds=timeout_seconds,
)
)
try:
await asyncio.wait(
(
condition_task,
runtime_task,
),
return_when=asyncio.FIRST_COMPLETED,
)
if runtime_task.done():
condition_task.cancel()
try:
await condition_task
except asyncio.CancelledError:
pass
await runtime_task
raise RuntimeError(
"Trade Stream Runtime exited before the expected "
"live condition was reached."
)
await condition_task
finally:
if not condition_task.done():
condition_task.cancel()
try:
await condition_task
except asyncio.CancelledError:
pass
async def stop_runtime(
runtime: ManagedRuntimeProtocol,
task: asyncio.Task[None],
*,
timeout_seconds: float = RUNTIME_CLEANUP_TIMEOUT_SECONDS,
) -> None:
primary_error: BaseException | None = None
if not task.done():
try:
await asyncio.wait_for(
runtime.stop(),
timeout=timeout_seconds,
)
except BaseException as error:
primary_error = error
if not task.done() and primary_error is not None:
task.cancel()
try:
await asyncio.wait_for(
asyncio.shield(task),
timeout=timeout_seconds,
)
except BaseException as error:
if isinstance(error, TimeoutError):
task.cancel()
try:
await asyncio.wait_for(
task,
timeout=timeout_seconds,
)
except asyncio.CancelledError:
pass
except BaseException as cancellation_error:
error.add_note(
"Runtime task cancellation also failed: "
f"{type(cancellation_error).__name__}."
)
if primary_error is None:
primary_error = error
else:
primary_error.add_note(
"Runtime task cleanup also failed: "
f"{type(error).__name__}."
)
if primary_error is not None:
raise primary_error
async def assert_no_owned_tasks() -> None:
await asyncio.sleep(0)
assert active_owned_task_names() == ()

83
app/tests/unit/market_data/acquisition/consistency/test_trade_stream_state.py
View File

@@ -73,6 +73,25 @@ def test_accepts_first_trade() -> None:
assert state.last_trade is trade
@pytest.mark.parametrize(
"trade_id",
(
-(2**31) - 1,
2**31,
),
)
def test_rejects_out_of_range_first_trade_id(
trade_id: int,
) -> None:
state = TradeStreamState(symbol="BTCUSD")
with pytest.raises(ValueError, match="signed 32-bit"):
state.accept(_trade(trade_id=trade_id))
assert state.last_trade_id is None
assert state.last_trade is None
def test_first_accepted_trade_becomes_checkpoint() -> None:
state = TradeStreamState(symbol="BTCUSD")
trade = _trade()
@@ -121,6 +140,49 @@ def test_accepts_trade_with_gap() -> None:
assert state.last_trade is trade_after_gap
@pytest.mark.parametrize(
("first_trade_id", "next_trade_id"),
(
(2**31 - 1, -(2**31)),
(-1, 0),
),
)
def test_accepts_next_trade_across_signed_rollover(
first_trade_id: int,
next_trade_id: int,
) -> None:
state = TradeStreamState(symbol="BTCUSD")
first_trade = _trade(trade_id=first_trade_id)
next_trade = _trade(trade_id=next_trade_id)
state.accept(first_trade)
result = state.accept(next_trade)
assert result is next_trade
assert state.last_trade_id == next_trade_id
assert state.last_trade is next_trade
@pytest.mark.parametrize(
("current_trade_id", "older_trade_id"),
(
(-(2**31), 2**31 - 1),
(0, -1),
),
)
def test_rejects_reverse_direction_across_signed_rollover(
current_trade_id: int,
older_trade_id: int,
) -> None:
state = TradeStreamState(symbol="BTCUSD")
state.accept(_trade(trade_id=current_trade_id))
with pytest.raises(TradeOrderingError):
state.accept(_trade(trade_id=older_trade_id))
assert state.last_trade_id == current_trade_id
def test_checkpoint_preserves_trade_identity() -> None:
state = TradeStreamState(symbol="BTCUSD")
trade = _trade()
@@ -158,6 +220,26 @@ def test_identical_duplicate_does_not_change_checkpoint() -> None:
assert state.last_trade_id == original_trade.trade_id
def test_duplicate_from_rest_and_websocket_is_same_market_trade() -> None:
state = TradeStreamState(symbol="BTCUSD")
websocket_trade = _trade(
trade_id=-2_036_886_697,
source="dzengi_websocket_trade",
)
rest_trade = _trade(
trade_id=-2_036_886_697,
source="dzengi",
)
state.accept(websocket_trade)
result = state.accept(rest_trade)
assert result is None
assert state.last_trade is websocket_trade
assert state.last_trade_id == websocket_trade.trade_id
def test_raises_consistency_error_for_conflicting_duplicate() -> None:
state = TradeStreamState(symbol="BTCUSD")
original_trade = _trade(
@@ -167,6 +249,7 @@ def test_raises_consistency_error_for_conflicting_duplicate() -> None:
conflicting_trade = _trade(
trade_id=100,
price=Decimal("50001.00"),
source="dzengi_websocket_trade",
)
state.accept(original_trade)

88
app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_controller.py
View File

@@ -13,6 +13,12 @@ from src.market_data.acquisition.adapters.dzengi.rest import (
from src.market_data.acquisition.consistency.trade_stream_exceptions import (
TradeConsistencyError,
)
from src.market_data.acquisition.consistency.trade_stream_consistency_controller import (
TradeStreamConsistencyController,
)
from src.market_data.acquisition.consistency.trade_stream_state_store import (
TradeStreamStateStore,
)
from src.market_data.acquisition.models.trade import (
Trade,
TradeAggressorSide,
@@ -281,6 +287,47 @@ def test_returns_trades_in_normalized_order() -> None:
)
@pytest.mark.parametrize(
("before_rollover", "after_rollover"),
(
(2**31 - 1, -(2**31)),
(-1, 0),
),
)
def test_recovers_trades_in_rollover_aware_order(
before_rollover: int,
after_rollover: int,
) -> None:
source = StubTradesDocumentSource(
document=[
_raw_trade(trade_id=after_rollover),
_raw_trade(trade_id=before_rollover),
]
)
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
result = controller.recover(_request())
assert tuple(
trade.trade_id
for trade in result.recovered_trades
) == (
before_rollover,
after_rollover,
)
assert [
trade.trade_id
for trade in consistency_controller.received_trades
] == [
before_rollover,
after_rollover,
]
def test_excludes_duplicate_rejected_by_consistency_controller() -> None:
source = StubTradesDocumentSource(
document=[
@@ -375,6 +422,45 @@ def test_preserves_consistency_controller_returned_instance() -> None:
assert result.recovered_trades[0] is replacement_trade
def test_rest_recovery_deduplicates_same_websocket_market_trade() -> None:
trade_id = -2_036_886_697
timestamp = 1_700_000_000_000
state_store = TradeStreamStateStore()
consistency_controller = TradeStreamConsistencyController(
state_store,
)
websocket_trade = Trade(
symbol="BTCUSD",
trade_id=trade_id,
price=Decimal("50000.00"),
quantity=Decimal("0.25"),
executed_at=datetime.fromtimestamp(
timestamp / 1000,
tz=timezone.utc,
),
aggressor_side=TradeAggressorSide.BUY,
source="dzengi_websocket_trade",
)
consistency_controller.accept(websocket_trade)
controller = TradeRecoveryController(
document_source=StubTradesDocumentSource(
document=[
_raw_trade(
trade_id=trade_id,
timestamp=timestamp,
),
]
),
consistency_controller=consistency_controller,
)
result = controller.recover(_request())
assert result.is_empty is True
state = state_store.get("BTCUSD")
assert state.last_trade is websocket_trade
def test_propagates_source_error() -> None:
expected_error = RuntimeError("source error")
@@ -572,4 +658,4 @@ def test_controller_does_not_create_additional_consistency_state() -> None:
assert controller.__dict__ == {
"_document_source": controller._document_source,
"_consistency_controller": consistency_controller,
}
}

64
app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_normalizer.py
View File

@@ -107,6 +107,68 @@ def test_sorts_arbitrary_order_by_trade_id() -> None:
)
def test_sorts_across_int32_max_to_int32_min_rollover() -> None:
before_rollover = _trade(trade_id=2**31 - 1)
after_rollover = _trade(trade_id=-(2**31))
result = normalize_recovered_trades(
(
after_rollover,
before_rollover,
)
)
assert result == (
before_rollover,
after_rollover,
)
def test_sorts_sequence_spanning_int32_sign_boundary() -> None:
trades_by_id = {
trade_id: _trade(trade_id=trade_id)
for trade_id in (
2**31 - 2,
2**31 - 1,
-(2**31),
-(2**31) + 1,
)
}
result = normalize_recovered_trades(
(
trades_by_id[-(2**31) + 1],
trades_by_id[2**31 - 1],
trades_by_id[2**31 - 2],
trades_by_id[-(2**31)],
)
)
assert tuple(trade.trade_id for trade in result) == (
2**31 - 2,
2**31 - 1,
-(2**31),
-(2**31) + 1,
)
def test_sorts_across_minus_one_to_zero_rollover() -> None:
before_rollover = _trade(trade_id=-1)
after_rollover = _trade(trade_id=0)
result = normalize_recovered_trades(
(
after_rollover,
before_rollover,
)
)
assert result == (
before_rollover,
after_rollover,
)
def test_preserves_stable_order_for_equal_trade_ids() -> None:
first_duplicate = _trade(
trade_id=100,
@@ -221,4 +283,4 @@ def test_returns_new_tuple_for_tuple_input() -> None:
result = normalize_recovered_trades(source)
assert result == source
assert result is not source
assert result is not source

90
app/tests/unit/market_data/acquisition/test_trade_id_sequence.py Normal file
View File

@@ -0,0 +1,90 @@
from __future__ import annotations
import pytest
from src.market_data.acquisition.trade_id_sequence import (
SIGNED_TRADE_ID_MAX,
SIGNED_TRADE_ID_MIN,
is_trade_id_newer,
is_trade_id_same_or_newer,
trade_id_relative_offset,
validate_signed_trade_id,
)
@pytest.mark.parametrize(
"trade_id",
(
SIGNED_TRADE_ID_MIN,
-1,
0,
1,
SIGNED_TRADE_ID_MAX,
),
)
def test_validate_signed_trade_id_accepts_full_range(
trade_id: int,
) -> None:
assert validate_signed_trade_id(trade_id) is None
@pytest.mark.parametrize(
"trade_id",
(
SIGNED_TRADE_ID_MIN - 1,
SIGNED_TRADE_ID_MAX + 1,
),
)
def test_validate_signed_trade_id_rejects_out_of_range(
trade_id: int,
) -> None:
with pytest.raises(ValueError, match="signed 32-bit"):
validate_signed_trade_id(trade_id)
def test_validate_signed_trade_id_rejects_boolean() -> None:
with pytest.raises(TypeError, match="must be an integer"):
validate_signed_trade_id(True)
@pytest.mark.parametrize(
("reference_trade_id", "candidate_trade_id"),
(
(100, 101),
(-100, -99),
(SIGNED_TRADE_ID_MAX, SIGNED_TRADE_ID_MIN),
(-1, 0),
),
)
def test_rollover_aware_contract_recognizes_next_id(
reference_trade_id: int,
candidate_trade_id: int,
) -> None:
assert (
trade_id_relative_offset(
candidate_trade_id,
reference_trade_id,
)
== 1
)
assert is_trade_id_newer(
candidate_trade_id,
reference_trade_id,
) is True
assert is_trade_id_newer(
reference_trade_id,
candidate_trade_id,
) is False
def test_same_id_is_same_or_newer_but_not_newer() -> None:
assert is_trade_id_same_or_newer(-100, -100) is True
assert is_trade_id_newer(-100, -100) is False
def test_half_cycle_distance_is_rejected_as_ambiguous() -> None:
with pytest.raises(ValueError, match="exactly half"):
trade_id_relative_offset(
SIGNED_TRADE_ID_MIN,
0,
)

72
app/tests/unit/market_data/acquisition/validation/test_values.py
View File

@@ -535,21 +535,40 @@ def test_validate_empty_rest_agg_trade_values() -> None:
@pytest.mark.parametrize(
("aggregate_trade_id", "timestamp"),
"aggregate_trade_id",
[
(0, 1000),
(-1, 1000),
(1, 0),
(1, -1),
-(2**31),
-2_037_115_004,
-1,
0,
1,
2**31 - 1,
],
)
def test_reject_non_positive_rest_agg_trade_integer_value(
def test_accept_signed_rest_agg_trade_id(
aggregate_trade_id: int,
timestamp: int,
) -> None:
trades = (
_valid_rest_agg_trade(
aggregate_trade_id=aggregate_trade_id,
),
)
assert validate_rest_agg_trade_values(trades) is None
@pytest.mark.parametrize(
"timestamp",
[
0,
-1,
],
)
def test_reject_non_positive_rest_agg_trade_timestamp(
timestamp: int,
) -> None:
trades = (
_valid_rest_agg_trade(
timestamp=timestamp,
),
)
@@ -561,6 +580,43 @@ def test_reject_non_positive_rest_agg_trade_integer_value(
validate_rest_agg_trade_values(trades)
def test_reject_boolean_rest_agg_trade_id() -> None:
trades = (
_valid_rest_agg_trade(
aggregate_trade_id=True,
),
)
with pytest.raises(
TradeValueError,
match=r"\$\[0\]\.aggregateTradeId должно быть целым числом",
):
validate_rest_agg_trade_values(trades)
@pytest.mark.parametrize(
"aggregate_trade_id",
(
-(2**31) - 1,
2**31,
),
)
def test_reject_out_of_range_rest_agg_trade_id(
aggregate_trade_id: int,
) -> None:
trades = (
_valid_rest_agg_trade(
aggregate_trade_id=aggregate_trade_id,
),
)
with pytest.raises(
TradeValueError,
match="signed 32-bit",
):
validate_rest_agg_trade_values(trades)
@pytest.mark.parametrize(
("field", "value"),
[
@@ -684,4 +740,4 @@ def test_rest_agg_trade_value_error_reports_item_index() -> None:
TradeValueError,
match=r"\$\[1\]\.quantity должно быть больше нуля",
):
validate_rest_agg_trade_values(trades)
validate_rest_agg_trade_values(trades)

38
app/tests/unit/market_data/acquisition/validation/test_websocket_trade_values.py
View File

@@ -85,17 +85,45 @@ def test_validate_websocket_trade_values_does_not_modify_event() -> None:
@pytest.mark.parametrize(
"trade_id",
[
0,
-(2**31),
-2_037_129_153,
-1,
-123456,
0,
1,
2**31 - 1,
],
)
def test_validate_websocket_trade_values_rejects_non_positive_trade_id(
def test_validate_websocket_trade_values_accepts_signed_trade_id(
trade_id: int,
) -> None:
validate_dzengi_websocket_trade_values(
_event(trade_id=trade_id)
)
def test_validate_websocket_trade_values_rejects_boolean_trade_id() -> None:
with pytest.raises(
TradeValueError,
match=r"\$\.payload\.id должно быть целым числом",
):
validate_dzengi_websocket_trade_values(
_event(trade_id=True)
)
@pytest.mark.parametrize(
"trade_id",
(
-(2**31) - 1,
2**31,
),
)
def test_validate_websocket_trade_values_rejects_out_of_range_trade_id(
trade_id: int,
) -> None:
with pytest.raises(
TradeValueError,
match=r"\$\.payload\.id должно быть целым числом больше нуля",
match="signed 32-bit",
):
validate_dzengi_websocket_trade_values(
_event(trade_id=trade_id)
@@ -272,4 +300,4 @@ def test_validate_websocket_trade_values_rejects_empty_order_id(
):
validate_dzengi_websocket_trade_values(
_event(order_id=order_id)
)
)

173
app/tests/unit/test_live_trade_stream_support.py Normal file
View File

@@ -0,0 +1,173 @@
from __future__ import annotations
import pytest
from tests.support.live_trade_stream import (
DEFAULT_LIVE_TRADE_TIMEOUT_SECONDS,
LIVE_REST_URL_ENV,
LIVE_SYMBOLS_ENV,
LIVE_TRADE_TIMEOUT_ENV,
LIVE_WEBSOCKET_URL_ENV,
RUN_LIVE_TESTS_ENV,
LiveTestConfigurationError,
build_live_trade_stream_settings,
load_live_trade_stream_test_config,
)
REST_URL = "https://api-adapter.dzengi.com"
WEBSOCKET_URL = "wss://api-adapter.dzengi.com/connect"
SYMBOL = "BTC/USD_LEVERAGE"
def make_environment() -> dict[str, str]:
return {
RUN_LIVE_TESTS_ENV: "1",
LIVE_REST_URL_ENV: REST_URL,
LIVE_WEBSOCKET_URL_ENV: WEBSOCKET_URL,
LIVE_SYMBOLS_ENV: SYMBOL,
}
@pytest.mark.parametrize(
"opt_in",
(
"",
"0",
),
)
def test_live_config_is_disabled_without_exact_opt_in(
opt_in: str,
) -> None:
assert load_live_trade_stream_test_config(
{
RUN_LIVE_TESTS_ENV: opt_in,
}
) is None
def test_invalid_live_opt_in_is_rejected() -> None:
with pytest.raises(
LiveTestConfigurationError,
match="must be exactly 1",
):
load_live_trade_stream_test_config(
{
RUN_LIVE_TESTS_ENV: "true",
}
)
@pytest.mark.parametrize(
"missing_name",
(
LIVE_REST_URL_ENV,
LIVE_WEBSOCKET_URL_ENV,
LIVE_SYMBOLS_ENV,
),
)
def test_enabled_live_config_requires_every_explicit_value(
missing_name: str,
) -> None:
environment = make_environment()
del environment[missing_name]
with pytest.raises(
LiveTestConfigurationError,
match=missing_name,
):
load_live_trade_stream_test_config(environment)
@pytest.mark.parametrize(
("name", "value"),
(
(
LIVE_REST_URL_ENV,
"http://api-adapter.dzengi.com",
),
(
LIVE_REST_URL_ENV,
"https://key@example.com",
),
(
LIVE_WEBSOCKET_URL_ENV,
"ws://api-adapter.dzengi.com/connect",
),
(
LIVE_WEBSOCKET_URL_ENV,
"wss://api-adapter.dzengi.com/not-connect",
),
),
)
def test_live_config_rejects_unsafe_or_wrong_endpoints(
name: str,
value: str,
) -> None:
environment = make_environment()
environment[name] = value
with pytest.raises(LiveTestConfigurationError):
load_live_trade_stream_test_config(environment)
def test_live_config_requires_exactly_one_symbol() -> None:
environment = make_environment()
environment[LIVE_SYMBOLS_ENV] = (
"BTC/USD_LEVERAGE,ETH/USD_LEVERAGE"
)
with pytest.raises(
LiveTestConfigurationError,
match="exactly one symbol",
):
load_live_trade_stream_test_config(environment)
def test_live_config_uses_bounded_default_trade_timeout() -> None:
config = load_live_trade_stream_test_config(
make_environment()
)
assert config is not None
assert (
config.trade_timeout_seconds
== DEFAULT_LIVE_TRADE_TIMEOUT_SECONDS
== 600.0
)
@pytest.mark.parametrize(
"timeout_value",
(
"0",
"-1",
"nan",
"inf",
"not-a-number",
),
)
def test_live_config_rejects_invalid_trade_timeout(
timeout_value: str,
) -> None:
environment = make_environment()
environment[LIVE_TRADE_TIMEOUT_ENV] = timeout_value
with pytest.raises(LiveTestConfigurationError):
load_live_trade_stream_test_config(environment)
def test_live_settings_are_explicit_and_credential_free() -> None:
config = load_live_trade_stream_test_config(
make_environment()
)
assert config is not None
settings = build_live_trade_stream_settings(config)
assert settings.exchange_base_url == REST_URL
assert settings.exchange_api_key == ""
assert settings.exchange_api_secret == ""
assert settings.trade_stream.enabled is True
assert settings.trade_stream.websocket_url == WEBSOCKET_URL
assert settings.trade_stream.symbols == (SYMBOL,)

92
app/tests/unit/test_trade_stream_runtime_support.py Normal file
View File

@@ -0,0 +1,92 @@
from __future__ import annotations
import asyncio
import pytest
from tests.support.trade_stream_runtime import (
run_scenario,
wait_until_or_runtime_exit,
)
def test_wait_until_or_runtime_exit_returns_when_condition_is_reached() -> None:
async def scenario() -> None:
condition_reached = False
async def keep_runtime_alive() -> None:
await asyncio.Event().wait()
runtime_task = asyncio.create_task(
keep_runtime_alive(),
)
try:
condition_reached = True
await wait_until_or_runtime_exit(
lambda: condition_reached,
runtime_task=runtime_task,
timeout_seconds=0.5,
)
assert runtime_task.done() is False
finally:
runtime_task.cancel()
with pytest.raises(asyncio.CancelledError):
await runtime_task
run_scenario(scenario())
def test_wait_until_or_runtime_exit_propagates_runtime_error_immediately() -> None:
expected_error = RuntimeError("runtime failed")
async def scenario() -> None:
async def fail_runtime() -> None:
await asyncio.sleep(0)
raise expected_error
runtime_task = asyncio.create_task(
fail_runtime(),
)
with pytest.raises(RuntimeError) as exc_info:
await asyncio.wait_for(
wait_until_or_runtime_exit(
lambda: False,
runtime_task=runtime_task,
timeout_seconds=10.0,
),
timeout=0.5,
)
assert exc_info.value is expected_error
run_scenario(scenario())
def test_wait_until_or_runtime_exit_rejects_clean_early_runtime_exit() -> None:
async def scenario() -> None:
async def finish_runtime() -> None:
await asyncio.sleep(0)
runtime_task = asyncio.create_task(
finish_runtime(),
)
with pytest.raises(
RuntimeError,
match="exited before the expected live condition",
):
await asyncio.wait_for(
wait_until_or_runtime_exit(
lambda: False,
runtime_task=runtime_task,
timeout_seconds=10.0,
),
timeout=0.5,
)
run_scenario(scenario())