"""R3: KODEX200 봉수 부족 시 변동비 공식 폴백 대신 갱신 실패(staleness 경로).""" import pytest import app.services.adaptive_strategy as ad def _rows(n): return [{"stck_clpr": "100", "stck_hgpr": "101", "stck_lwpr": "99", "acml_vol": "1"} for _ in range(n)] @pytest.fixture(autouse=True) def _isolate(monkeypatch): async def _no(*a, **k): raise AssertionError("must not run") monkeypatch.setattr(ad, "_save_to_redis", _no) monkeypatch.setattr(ad, "_save_data_source_v2", _no) monkeypatch.setattr(ad, "_current_params", None) @pytest.mark.asyncio async def test_short_history_is_failure(monkeypatch): async def short(): return _rows(30), "kis_range" monkeypatch.setattr(ad, "_fetch_kodex_chart_v2", short) called = [] monkeypatch.setattr(ad, "enforce_staleness_v2", lambda: called.append(1) or None) def _boom(*a, **k): raise AssertionError("analyze must not run") monkeypatch.setattr(ad, "analyze_market_condition", _boom) before = ad._last_success_ts await ad.refresh_adaptive_params() assert called == [1] and ad._last_success_ts == before @pytest.mark.asyncio async def test_fetch_falls_through_to_kis_range(monkeypatch): import app.services.naver_finance as nf import app.services.korea_investment_api as k async def naver(*a, **kw): return _rows(70) monkeypatch.setattr(nf, "get_naver_daily_chart", naver) class Api: async def get_daily_ohlcv_range(self, *a): return _rows(ad._MIN_BARS_FOR_ATR_PCT + 5) monkeypatch.setattr(k, "real_market_api", Api()) data, src = await ad._fetch_kodex_chart_v2() assert src == "kis_range" and len(data) == ad._MIN_BARS_FOR_ATR_PCT + 5 @pytest.mark.asyncio async def test_fetch_naver_sufficient_skips_kis(monkeypatch): import app.services.naver_finance as nf import app.services.korea_investment_api as k async def naver(*a, **kw): return _rows(200) monkeypatch.setattr(nf, "get_naver_daily_chart", naver) class Api: async def get_daily_ohlcv_range(self, *a): raise AssertionError("no kis") monkeypatch.setattr(k, "real_market_api", Api()) data, src = await ad._fetch_kodex_chart_v2() assert src == "naver" and len(data) == 200