copilot-swe-agent[bot]andgenichin
1ca9e60c98
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
195 lines
7.3 KiB
Python
195 lines
7.3 KiB
Python
"""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
|