"""AlarmEmitter unit tests (dcy-033 / REQ-100). Coverage: - 8 CRITICAL code × (a) normal publish + envelope validation × (b) TransportUnavailableError → no re-raise + ERROR log - arbitrary exception swallowing - transport=None skip (reset.py / CLI path) - routing key format: alarm.critical.trex-core """ from __future__ import annotations import logging import pytest from trex_core.adapters.messaging.in_memory import InMemoryTransport from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.messaging import TransportUnavailableError # ------------------------------------------------------------------ helpers CRITICAL_CODES = [ "fill_price_unresolvable", "sell_skipped_no_balance", "core_holding_broker_flat", "ledger_delivery_failed", "kill_switch_fired", "publish_unavailable_after_reconnect", "dlq_publish_failed", "portfolio_hub_unavailable", ] _ROUTING_KEY = "alarm.critical.trex-core" def _make_emitter() -> tuple[AlarmEmitter, InMemoryTransport]: transport = InMemoryTransport() return AlarmEmitter(transport), transport # ------------------------------------------------------------------ envelope validation @pytest.mark.parametrize("code", CRITICAL_CODES) def test_publish_alarm_envelope_fields(code: str): """(a) Normal publish: envelope contains all 7 required fields with correct values.""" emitter, transport = _make_emitter() fields_in = {"slot_id": "abc", "symbol": "005930"} emitter.publish_alarm(code, f"Test alarm for {code}", fields=fields_in) alarms = transport.alarm_events() assert len(alarms) == 1, f"Expected 1 alarm event for code={code}" rk, payload = alarms[0] assert rk == _ROUTING_KEY, f"routing key mismatch: {rk}" # REQ-098: all 7 envelope fields assert payload["source"] == "trex-core" assert payload["severity"] == "critical" assert payload["code"] == code assert payload["title"] == f"Test alarm for {code}" assert payload["fields"] == fields_in assert "occurred_at" in payload assert payload["schema_version"] == "1" @pytest.mark.parametrize("code", CRITICAL_CODES) def test_publish_alarm_transport_unavailable_no_reraise( code: str, caplog: pytest.LogCaptureFixture ): """(b) TransportUnavailableError must not propagate; ERROR log must appear.""" emitter, transport = _make_emitter() transport.set_available(False) with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"): # Must NOT raise emitter.publish_alarm(code, f"Test alarm for {code}") # No alarms recorded (transport was down) assert transport.alarm_events() == [] # ERROR log must have been emitted assert any("alarm publish" in r.message for r in caplog.records), ( f"Expected ERROR log for code={code}" ) # ------------------------------------------------------------------ additional edge cases @pytest.mark.parametrize("code", CRITICAL_CODES) def test_publish_alarm_arbitrary_exception_swallowed(code: str, caplog: pytest.LogCaptureFixture): """Any exception (not just TransportUnavailableError) is swallowed + ERROR logged.""" class BrokenTransport: def publish_alarm_event(self, routing_key, payload): raise RuntimeError("boom") def health(self): from trex_core.application.messaging import TransportHealth return TransportHealth.CONNECTED emitter = AlarmEmitter(BrokenTransport()) # type: ignore[arg-type] with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"): emitter.publish_alarm(code, "boom test") # Must NOT raise assert any("alarm publish" in r.message for r in caplog.records) def test_publish_alarm_none_transport_skip(caplog: pytest.LogCaptureFixture): """transport=None: publish is skipped with INFO log (reset.py / CLI path, REQ-100 KD-2).""" emitter = AlarmEmitter(None) with caplog.at_level(logging.INFO, logger="trex_core.application.alarm_emitter"): emitter.publish_alarm("kill_switch_fired", "no transport test") assert any("alarm skipped" in r.message for r in caplog.records) def test_publish_alarm_empty_fields_default(): """fields defaults to {} when not provided — not None in payload.""" emitter, transport = _make_emitter() emitter.publish_alarm("kill_switch_fired", "no fields") _, payload = transport.alarm_events()[0] assert payload["fields"] == {} def test_publish_alarm_routing_key_format(): """routing key must be alarm.critical.trex-core for default severity.""" emitter, transport = _make_emitter() emitter.publish_alarm("kill_switch_fired", "routing key test") rk, _ = transport.alarm_events()[0] assert rk == "alarm.critical.trex-core" # ------------------------------------------------------------------ domain purity sweep def test_alarm_emitter_no_pika_import(): """AlarmEmitter module must not import pika (domain/application purity).""" import importlib import sys # Ensure pika is not in alarm_emitter's import chain at module level. mod = sys.modules.get("trex_core.application.alarm_emitter") if mod is None: mod = importlib.import_module("trex_core.application.alarm_emitter") # pika should NOT be in the module's global namespace assert "pika" not in dir(mod), "alarm_emitter must not import pika directly" def test_alarm_emitter_no_telegram_reference(): """Verify no TELEGRAM_* references in alarm_emitter source (REQ-100 AC-5).""" import inspect import trex_core.application.alarm_emitter as ae_mod source = inspect.getsource(ae_mod) assert "TELEGRAM" not in source, "alarm_emitter must not reference TELEGRAM_*" assert "telegram" not in source.lower(), "alarm_emitter must not reference telegram" def test_publish_alarm_transport_unavailable_log_is_sanitized( caplog: pytest.LogCaptureFixture, ): class UnavailableTransport: def publish_alarm_event(self, routing_key, payload): raise TransportUnavailableError("secret-token-value") def health(self): from trex_core.application.messaging import TransportHealth return TransportHealth.DISCONNECTED emitter = AlarmEmitter(UnavailableTransport()) # type: ignore[arg-type] with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"): emitter.publish_alarm("strategy_intent_rejected", "alarm title") assert any("transport unavailable" in r.message for r in caplog.records) assert "secret-token-value" not in caplog.text def test_publish_alarm_generic_exception_log_is_sanitized(caplog: pytest.LogCaptureFixture): class BrokenTransport: def publish_alarm_event(self, routing_key, payload): raise RuntimeError("******") def health(self): from trex_core.application.messaging import TransportHealth return TransportHealth.CONNECTED emitter = AlarmEmitter(BrokenTransport()) # type: ignore[arg-type] with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"): emitter.publish_alarm("strategy_intent_rejected", "alarm title") assert any("alarm publish failed" in r.message for r in caplog.records) assert "RuntimeError" in caplog.text assert "******" not in caplog.text