"""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" # --------------------------------------------------------------------------- # 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))