"""R9 E-2/E-5/E-6: 첫 하트비트 유예 판정, config 키 선언 동일성, 소규모 수정.""" import asyncio import pytest import app.main as m from app.core import heartbeat as hb from app.core.config import settings, Settings class _T: def __init__(self, done=False): self._d = done def done(self): return self._d def _app(alive=True): a = type("A", (), {})() a.state = type("S", (), {})() for attr in ("swing_task", "dt_task"): setattr(a.state, attr, _T(not alive)) return a @pytest.fixture(autouse=True) def _reset_hb(monkeypatch): monkeypatch.setattr(m, "_scheduler_tasks", {"swing": "swing_task", "daytrading": "dt_task"}) hb.reset() yield hb.reset() def test_no_first_beat_after_grace_in_market_is_stale(): over = settings.HB_FIRST_BEAT_GRACE_SEC + 100 h = m._scheduler_health_v3(_app(), now=1.0e6, market_open=True, uptime_sec=over) for name in m._HEARTBEAT_WATCHED: assert h[name]["stale"] is True and h[name]["no_first_beat"] is True def test_no_first_beat_within_grace_is_normal(): under = settings.HB_FIRST_BEAT_GRACE_SEC - 500 h = m._scheduler_health_v3(_app(), now=1.0e6, market_open=True, uptime_sec=under) for name in m._HEARTBEAT_WATCHED: assert h[name]["stale"] is False and h[name]["no_first_beat"] is False def test_no_first_beat_off_hours_is_normal(): over = settings.HB_FIRST_BEAT_GRACE_SEC + 100 h = m._scheduler_health_v3(_app(), now=1.0e6, market_open=False, uptime_sec=over) assert all(v["stale"] is False for v in h.values()) def test_beat_present_not_flagged_as_no_first_beat(): hb.beat("swing") over = settings.HB_FIRST_BEAT_GRACE_SEC + 100 h = m._scheduler_health_v3(_app(), market_open=True, uptime_sec=over) assert h["swing"]["no_first_beat"] is False and h["swing"]["stale"] is False def test_v2_signature_and_behavior_unchanged(): h = m._scheduler_health_v2(_app(), now=1.0e6, market_open=True) assert "no_first_beat" not in h["swing"] and h["swing"]["stale"] is False # ── E-5: 선언된 config 키 값이 소비처 기본값과 동일 ────────────────────────── def _mod_default(modname, const): import importlib return getattr(importlib.import_module(modname), const) @pytest.mark.parametrize("key,const_mod,const", [ ("NEWS_EMPTY_ALERT_THRESHOLD", "app.services.news_sentiment", "_EMPTY_ALERT_DEFAULT"), ]) def test_declared_keys_match_consumer_defaults(key, const_mod, const): assert key in Settings.model_fields assert getattr(settings, key) == _mod_default(const_mod, const) @pytest.mark.parametrize("key,expected", [ ("KIS_WS_MAX_REGISTRATIONS", 41), ("RISK_MAX_DAILY_LOSS_RATIO", 0.03), ("RISK_MAX_CONSECUTIVE_LOSSES", 3), ("SWING_ADAPTIVE_TRAILING_ENABLED", False), ("MEAN_REVERSION_RULE_MERGE_ENABLED", False), ("HB_FIRST_BEAT_GRACE_SEC", 600.0), ]) def test_declared_keys_literal_defaults(key, expected): assert getattr(settings, key) == expected @pytest.mark.parametrize("key", [ "KIS_WS_ORIGIN", "KIS_WS_SUBPROTOCOLS", "BATCH_ANALYSIS_TIMEOUT_SEC", "ZERO_BUY_ALERT_HOURS", ]) def test_declared_none_keys_preserve_none_semantics(key): assert key in Settings.model_fields assert getattr(settings, key) is None def test_cors_origins_stays_undeclared(): """.env 에 값이 있어 선언하면 CORS 동작이 바뀐다 — 선언 금지 회귀 가드.""" assert "CORS_ORIGINS" not in Settings.model_fields # ── E-6 ──────────────────────────────────────────────────────────────────── def test_trailing_stop_uses_resolved_multiplier(monkeypatch): import app.services.trade_helpers as th monkeypatch.setattr(th, "resolve_swing_trailing_atr_mult_v2", lambda: 2.0) import inspect src = inspect.getsource(th) assert "high_price - 1.5 * atr" not in src assert "resolve_swing_trailing_atr_mult_v2() * atr" in src