481 lines
19 KiB
Python
481 lines
19 KiB
Python
"""BAT-047 — Event publisher connection reliability tests (REQ-091/092/093).
|
|
|
|
All tests run without a live RabbitMQ broker using unittest.mock. The
|
|
RabbitMQTransport is exercised through its public interface; pika internals
|
|
are replaced by lightweight mocks so the retry / keepalive / DLQ logic can
|
|
be unit-tested deterministically.
|
|
|
|
AC coverage:
|
|
REQ-091 AC-1 : half-open → reconnect + retry → success (no error raised)
|
|
REQ-091 AC-2 : half-open + retry also fails → TransportUnavailableError
|
|
REQ-091 AC-3 : timeout budget — _force_reconnect uses connect_timeout=2.0
|
|
REQ-091 AC-4 : normal publish (conn alive) — no retry overhead
|
|
REQ-091 AC-5 : consume/RPC paths untouched by this change (interface check)
|
|
REQ-091 AC-6 : broker restart scenario — reconnect + retry succeeds
|
|
REQ-092 AC-1 : keepalive_tick calls process_data_events on live conn
|
|
REQ-092 AC-2 : keepalive_tick executes only on owning object (no lock escape)
|
|
REQ-092 AC-3 : simultaneous publish + keepalive — no crash (single-thread)
|
|
REQ-092 AC-4 : keepalive non-blocking (time_limit=0 passed to pika)
|
|
REQ-092 AC-5 : consume/RPC paths unaffected (keepalive not called there)
|
|
REQ-093 AC-1 : unroutable → _publish_dlq called + ERROR log emitted
|
|
REQ-093 AC-3 : emit ERROR on final publish failure (daemon layer)
|
|
REQ-093 AC-4 : _publish_dlq itself fails → ERROR log, no silent loss
|
|
REQ-093 AC-6 : consume/RPC paths unaffected (DLQ change local to publish)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pika.exceptions
|
|
import pytest
|
|
|
|
from trex_core.adapters.messaging.rabbitmq import RabbitMQTransport
|
|
from trex_core.application.messaging import TransportUnavailableError
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_transport() -> RabbitMQTransport:
|
|
"""Return a RabbitMQTransport instance without connecting to a real broker."""
|
|
return RabbitMQTransport("amqp://localhost/", timeout=5.0)
|
|
|
|
|
|
def _inject_connected(t: RabbitMQTransport, conn=None, channel=None):
|
|
"""Put the transport into CONNECTED state with mock pika objects."""
|
|
from trex_core.application.messaging import TransportHealth
|
|
if conn is None:
|
|
conn = MagicMock()
|
|
conn.is_open = True
|
|
if channel is None:
|
|
channel = MagicMock()
|
|
channel.is_open = True
|
|
t._conn = conn
|
|
t._channel = channel
|
|
t._health = TransportHealth.CONNECTED
|
|
return conn, channel
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-091 AC-1: half-open → reconnect + retry → success
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPublishRetryOnHalfOpen:
|
|
"""REQ-091 AC-1: publish succeeds after forced reconnect."""
|
|
|
|
def test_half_open_first_attempt_fails_retry_succeeds(self, caplog):
|
|
"""Simulates a half-open connection: is_open=True but basic_publish raises
|
|
AMQPConnectionError on first call. The transport must reconnect and retry,
|
|
and publish_event must NOT raise.
|
|
"""
|
|
t = _make_transport()
|
|
mock_conn, mock_ch = _inject_connected(t)
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def fake_basic_publish(**kwargs):
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
raise pika.exceptions.AMQPConnectionError("connection reset")
|
|
# Second call (after reconnect) succeeds silently.
|
|
|
|
mock_ch.basic_publish.side_effect = fake_basic_publish
|
|
|
|
reconnect_called = {"yes": False}
|
|
|
|
def fake_force_reconnect():
|
|
reconnect_called["yes"] = True
|
|
# After reconnect, the channel is fresh and functional.
|
|
new_ch = MagicMock()
|
|
new_ch.is_open = True
|
|
new_ch.basic_publish.return_value = None # success
|
|
t._channel = new_ch
|
|
from trex_core.application.messaging import TransportHealth
|
|
t._health = TransportHealth.CONNECTED
|
|
|
|
with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect):
|
|
# Must not raise.
|
|
t.publish_event("slot_filled_buy.A.005930", {"slot": "s1"})
|
|
|
|
assert reconnect_called["yes"], "_force_reconnect must be called on first failure"
|
|
|
|
def test_sell_retry_reuses_identical_payload_and_mandatory_flag(self):
|
|
t = _make_transport()
|
|
_inject_connected(t)
|
|
payload = {
|
|
"slot_id": "slot-95",
|
|
"event_type": "slot_filled_sell",
|
|
"sequence_at": 11,
|
|
}
|
|
routing_key = "slot_filled_sell.SELL-B.005930"
|
|
|
|
with (
|
|
patch.object(
|
|
t,
|
|
"_do_basic_publish",
|
|
side_effect=[OSError("first attempt"), None],
|
|
) as publish,
|
|
patch.object(t, "_force_reconnect"),
|
|
):
|
|
t.publish_event(routing_key, payload)
|
|
|
|
assert publish.call_count == 2
|
|
assert publish.call_args_list[0] == publish.call_args_list[1]
|
|
assert publish.call_args_list[0].args[0] == routing_key
|
|
assert publish.call_args_list[0].kwargs["mandatory"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-091 AC-2: retry also fails → TransportUnavailableError raised
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPublishRetryExhausted:
|
|
"""REQ-091 AC-2: when reconnect+retry also fails, error is propagated."""
|
|
|
|
def test_retry_exhausted_raises_transport_unavailable(self):
|
|
t = _make_transport()
|
|
mock_conn, mock_ch = _inject_connected(t)
|
|
|
|
mock_ch.basic_publish.side_effect = pika.exceptions.AMQPConnectionError("gone")
|
|
|
|
def fake_force_reconnect():
|
|
from trex_core.application.messaging import TransportHealth
|
|
new_ch = MagicMock()
|
|
new_ch.is_open = True
|
|
new_ch.basic_publish.side_effect = pika.exceptions.AMQPConnectionError(
|
|
"still gone"
|
|
)
|
|
t._channel = new_ch
|
|
t._health = TransportHealth.CONNECTED
|
|
|
|
with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect):
|
|
with pytest.raises(TransportUnavailableError) as exc_info:
|
|
t.publish_event("slot_filled_buy.A.005930", {"slot": "s1"})
|
|
|
|
assert "reconnect+retry" in str(exc_info.value)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-091 AC-3: timeout budget — _force_reconnect uses connect_timeout=2.0
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestForceReconnectTimeout:
|
|
"""REQ-091 AC-3 / NFR-2: _force_reconnect passes connect_timeout=2.0."""
|
|
|
|
def test_force_reconnect_uses_2s_timeout(self):
|
|
t = _make_transport()
|
|
_inject_connected(t)
|
|
|
|
connect_calls: list = []
|
|
|
|
def fake_connect(connect_timeout=None):
|
|
connect_calls.append(connect_timeout)
|
|
from trex_core.application.messaging import TransportHealth
|
|
new_conn = MagicMock()
|
|
new_conn.is_open = True
|
|
new_ch = MagicMock()
|
|
new_ch.is_open = True
|
|
t._conn = new_conn
|
|
t._channel = new_ch
|
|
t._health = TransportHealth.CONNECTED
|
|
|
|
with patch.object(t, "_connect", side_effect=fake_connect):
|
|
t._force_reconnect()
|
|
|
|
assert len(connect_calls) == 1
|
|
assert connect_calls[0] == 2.0, (
|
|
f"_force_reconnect must pass connect_timeout=2.0, got {connect_calls[0]}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-091 AC-4: normal publish (conn alive) — no retry path triggered
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNormalPublishNoRetry:
|
|
"""REQ-091 AC-4: successful first-attempt publish does not invoke retry."""
|
|
|
|
def test_normal_publish_does_not_call_force_reconnect(self):
|
|
t = _make_transport()
|
|
_inject_connected(t)
|
|
t._channel.basic_publish.return_value = None # success
|
|
|
|
with patch.object(t, "_force_reconnect") as mock_fr:
|
|
t.publish_event("slot_filled_buy.A.005930", {"ok": True})
|
|
|
|
mock_fr.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-091 AC-6: broker restart — force reconnect + retry succeeds
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBrokerRestartScenario:
|
|
"""REQ-091 AC-6: after broker restart the transport reconnects and retries."""
|
|
|
|
def test_broker_restart_publish_recovers(self):
|
|
t = _make_transport()
|
|
_inject_connected(t)
|
|
|
|
# Simulate broker restart: first publish raises OSError (TCP reset).
|
|
t._channel.basic_publish.side_effect = OSError("connection refused")
|
|
|
|
recovered = {"yes": False}
|
|
|
|
def fake_force_reconnect():
|
|
from trex_core.application.messaging import TransportHealth
|
|
new_ch = MagicMock()
|
|
new_ch.is_open = True
|
|
new_ch.basic_publish.return_value = None
|
|
t._channel = new_ch
|
|
t._health = TransportHealth.CONNECTED
|
|
recovered["yes"] = True
|
|
|
|
with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect):
|
|
# publish_event must succeed without raising.
|
|
t.publish_event("slot_filled_buy.A.005930", {"ev": "restart"})
|
|
|
|
assert recovered["yes"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-092 AC-1: keepalive_tick calls process_data_events on live conn
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestKeepaliveTick:
|
|
"""REQ-092 AC-1: keepalive_tick delegates to conn.process_data_events."""
|
|
|
|
def test_keepalive_tick_calls_process_data_events_on_live_connection(self):
|
|
t = _make_transport()
|
|
mock_conn, _ = _inject_connected(t)
|
|
mock_conn.is_open = True
|
|
|
|
t.keepalive_tick()
|
|
|
|
mock_conn.process_data_events.assert_called_once_with(time_limit=0)
|
|
|
|
def test_keepalive_tick_noop_when_disconnected(self):
|
|
"""REQ-092 AC-1 variant: no-op if connection is None (shutdown path)."""
|
|
t = _make_transport()
|
|
# _conn is None by default — keepalive_tick must not raise.
|
|
t.keepalive_tick() # should return silently
|
|
|
|
def test_keepalive_tick_noop_when_conn_closed(self):
|
|
"""keepalive_tick skips when is_open=False."""
|
|
t = _make_transport()
|
|
mock_conn = MagicMock()
|
|
mock_conn.is_open = False
|
|
t._conn = mock_conn
|
|
from trex_core.application.messaging import TransportHealth
|
|
t._health = TransportHealth.DISCONNECTED
|
|
|
|
t.keepalive_tick()
|
|
|
|
mock_conn.process_data_events.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-092 AC-4: keepalive non-blocking — time_limit=0
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestKeepaliveTimeLimitZero:
|
|
"""REQ-092 AC-4: process_data_events is called with time_limit=0."""
|
|
|
|
def test_keepalive_uses_time_limit_zero(self):
|
|
t = _make_transport()
|
|
mock_conn, _ = _inject_connected(t)
|
|
mock_conn.is_open = True
|
|
|
|
t.keepalive_tick()
|
|
|
|
args, kwargs = mock_conn.process_data_events.call_args
|
|
assert kwargs.get("time_limit", args[0] if args else None) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-092 AC-3: publish + keepalive in same thread — no crash
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestKeepaliveAndPublishSameThread:
|
|
"""REQ-092 AC-3: calling keepalive then publish in the same thread is safe."""
|
|
|
|
def test_keepalive_then_publish_sequential_no_error(self):
|
|
t = _make_transport()
|
|
mock_conn, _ = _inject_connected(t)
|
|
mock_conn.is_open = True
|
|
t._channel.basic_publish.return_value = None
|
|
|
|
# Sequential in same thread — simulates execution-loop tick.
|
|
t.keepalive_tick()
|
|
t.publish_event("slot_filled_buy.A.005930", {"seq": 1})
|
|
|
|
mock_conn.process_data_events.assert_called_once_with(time_limit=0)
|
|
t._channel.basic_publish.assert_called_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-092 AC-5: consume/RPC paths are unaffected by keepalive
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestKeepaliveDoesNotAffectConsumePaths:
|
|
"""REQ-092 AC-5: keepalive is specific to publish path; consume API unchanged."""
|
|
|
|
def test_keepalive_tick_not_present_on_consume_rpc_forever_signature(self):
|
|
"""The consume_rpc_forever signature is unchanged."""
|
|
import inspect
|
|
sig = inspect.signature(RabbitMQTransport.consume_rpc_forever)
|
|
params = list(sig.parameters)
|
|
assert "queue" in params
|
|
assert "handler" in params
|
|
assert "stop" in params
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-093 AC-1: unroutable → _publish_dlq called + ERROR log emitted
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDlqVisibilityOnUnroutable:
|
|
"""REQ-093 AC-1: unroutable event triggers _publish_dlq + ERROR log."""
|
|
|
|
def test_unroutable_calls_publish_dlq_and_logs_error(self, caplog):
|
|
t = _make_transport()
|
|
_inject_connected(t)
|
|
|
|
t._channel.basic_publish.side_effect = pika.exceptions.UnroutableError([])
|
|
|
|
dlq_calls: list = []
|
|
|
|
def fake_dlq(rk, payload, *, reason):
|
|
dlq_calls.append((rk, reason))
|
|
|
|
with patch.object(t, "_publish_dlq", side_effect=fake_dlq):
|
|
with caplog.at_level(logging.DEBUG):
|
|
# Must not raise — unroutable is handled via DLQ.
|
|
t.publish_event("slot_filled_buy.A.005930", {"x": 1})
|
|
|
|
assert len(dlq_calls) == 1
|
|
assert dlq_calls[0][1] == "unroutable"
|
|
|
|
def test_publish_dlq_logs_error_on_entry(self, caplog):
|
|
"""REQ-093 AC-1: _publish_dlq itself logs ERROR when DLQ entry is created."""
|
|
t = _make_transport()
|
|
_inject_connected(t)
|
|
# Make DLQ publish succeed (no-op).
|
|
t._channel.basic_publish.return_value = None
|
|
|
|
with caplog.at_level(logging.ERROR, logger="trex_core.adapters.messaging.rabbitmq"):
|
|
t._publish_dlq("slot_filled_buy.A.005930", {"x": 1}, reason="unroutable")
|
|
|
|
errors = [r for r in caplog.records if r.levelno >= logging.ERROR]
|
|
assert any("DLQ" in r.message for r in errors), (
|
|
"Expected ERROR log with 'DLQ' when _publish_dlq is called"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-093 AC-4: _publish_dlq itself fails → ERROR log, no silent loss
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDlqPublishFailureLogsError:
|
|
"""REQ-093 AC-4: when DLQ publish itself raises, ERROR is logged."""
|
|
|
|
def test_dlq_publish_failure_logs_error_not_warning(self, caplog):
|
|
t = _make_transport()
|
|
_inject_connected(t)
|
|
|
|
# First basic_publish for DLQ raises (after _ensure_connected succeeds).
|
|
t._channel.basic_publish.side_effect = Exception("DLQ broker down")
|
|
|
|
with caplog.at_level(logging.ERROR, logger="trex_core.adapters.messaging.rabbitmq"):
|
|
# _publish_dlq must swallow the exception but log ERROR.
|
|
t._publish_dlq("slot_filled_buy.A.005930", {"x": 1}, reason="unroutable")
|
|
|
|
error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
|
|
assert len(error_records) >= 1
|
|
# The DLQ publish failure must be ERROR, not WARNING.
|
|
dlq_fail_records = [r for r in error_records if "DLQ" in r.message]
|
|
assert dlq_fail_records, "Expected ERROR log mentioning 'DLQ' on publish failure"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-093 AC-3: emit ERROR on final publish failure (daemon layer)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestEmitErrorOnFinalPublishFailure:
|
|
"""REQ-093 AC-3: _make_event_emitter logs ERROR (not WARNING) on failure."""
|
|
|
|
def test_emit_logs_error_on_transport_unavailable(self, caplog):
|
|
"""The emitter created by _make_event_emitter must log ERROR when
|
|
publish_event raises TransportUnavailableError after retry.
|
|
|
|
Uses the production Daemon._make_event_emitter() directly (via a
|
|
lightweight stub instance) so that any change to the production error
|
|
path is caught as a regression.
|
|
"""
|
|
from trex_core.runtime.daemon import Daemon
|
|
from trex_core.application.messaging import TransportUnavailableError
|
|
|
|
# Construct a minimal Daemon stub without triggering __init__ (which
|
|
# requires DB / config / broker connections). Only event_publisher is
|
|
# accessed by _make_event_emitter, so that is the only attribute we set.
|
|
daemon_stub = object.__new__(Daemon)
|
|
fake_publisher = MagicMock()
|
|
fake_publisher.publish_event.side_effect = TransportUnavailableError("no broker")
|
|
daemon_stub.event_publisher = fake_publisher
|
|
|
|
# Obtain the emitter from the production code path.
|
|
emit = daemon_stub._make_event_emitter()
|
|
|
|
# Minimal slot stub.
|
|
slot = MagicMock()
|
|
slot.strategy_id = "A"
|
|
slot.symbol = "005930"
|
|
slot.slot_id = "slot-1"
|
|
slot.status = MagicMock()
|
|
slot.status.value = "holding"
|
|
slot.updated_at = None
|
|
slot.portfolio_id = "PF-1"
|
|
slot.entry_price = None
|
|
slot.qty = 10
|
|
slot.instance_key = "default"
|
|
slot.actor = None
|
|
slot.reason = None
|
|
|
|
with caplog.at_level(logging.ERROR, logger="trex_core.runtime.daemon"):
|
|
emit("slot_filled_buy", slot)
|
|
|
|
error_records = [
|
|
r for r in caplog.records
|
|
if r.levelno >= logging.ERROR and "event publish failed" in r.message
|
|
]
|
|
assert error_records, (
|
|
"Expected ERROR log 'event publish failed' from Daemon._make_event_emitter"
|
|
)
|
|
# Must NOT be mere WARNING.
|
|
warn_only = [
|
|
r for r in caplog.records
|
|
if r.levelno == logging.WARNING and "event publish failed" in r.message
|
|
]
|
|
assert not warn_only, "publish failure must not be WARNING; must be ERROR"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# REQ-093 AC-6: consume/RPC paths unaffected
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestConsumePaths:
|
|
"""REQ-093 AC-6 / REQ-091 AC-5: consume/RPC interface signature unchanged."""
|
|
|
|
def test_rabbitmq_transport_has_consume_methods(self):
|
|
t = _make_transport()
|
|
assert callable(getattr(t, "consume_rpc_forever", None))
|
|
assert callable(getattr(t, "consume_events_forever", None))
|
|
assert callable(getattr(t, "call_rpc", None))
|
|
assert callable(getattr(t, "declare_topology", None))
|
|
|
|
def test_keepalive_tick_exists_as_callable(self):
|
|
t = _make_transport()
|
|
assert callable(getattr(t, "keepalive_tick", None))
|