shuim
9fab4018d4
하네스 규칙 검사 전역 위반 117→0, 회차별 검증(code-verifier opus 4도메인)으로 Critical/High 해소. 주문·리스크 - MDD 강제축소 반복 매도(종목별 완료 집합·재시도 상한·날짜 롤오버·잔고 오류 보류·epoch·장외 미주문) - KIS 거부 FAILED 기록, 결과 불명(전송 후) 재전송 금지·PENDING, 미전송 실패만 재시도(msg_cd 분류) - 국내 전량·손실 매도 시장가, 부분 익절 직전 현재가, 지정가만 미체결 취소(cancel_order_v2: TR 0013U·조직번호) + 0체결 롤백 - 누적 4%·현금·레버리지(신규분 포함)·사이클 내 매수 누적, 수동매수 공용 가드(buy_guards), 수동매도 DT 차감·락 - 가드 우회 엔드포인트 차단(/recommendations/auto-trade 위임, US /execute 매수 410·매도 가드) - 매수 슬롯=성공 매수, 평가 상한·회전, 업무거부/계좌 쿨다운, 프로세스 내 in-flight 주문 중복 차단 데이트레이딩 - 진입·청산 shield, 청산 진행 중 표식(소유 토큰), 2단계 청산 체크, 결과 불명 → 보유 감소 판정·비DT 차감 상한 - 전일 포지션 재시도(일시 실패 미산정), 분할매도 영속, ORB 구간 손절, /cycle 락, 사이클 내 현금 차감, 체결 수량 보정 스윙 - 손절 공백(판정 예외 시 하드 바닥), 단일 손절 소스·범위 검증, 매도 직전 KIS 잔고 상한(min(주문가능,보유)−DT) - 결과 불명 체결 대조·장부 반영, 대체가(나이·진입 이후·ATR 캐시), 분할 단계 기록, 국면 실패 fail-closed - 듀얼 플레이북·서킷브레이커 게이트, 평균회귀 우선·듀얼 독립, 시간손절 거래일 - 거래 확대(사용자 결정, 백테스트 미검증): RSI 46/35 정본 경로, 모멘텀 v2 ON, SIDEWAYS ADX AND, 트레일링 1차 익절 후, 20일 시간손절 ON 인프라·보안 - 계좌·손익 GET 인증, 세션 is_active·관리자 경로, 서명키 파생, realip, WS 토큰 마스킹 - KIS 토큰 즉시 실패(sleep 제거)·무효 시 캐시 삭제, 주문 5xx+JSON 본문 반환 - db_worker 종료 drain(멱등 재실행·커밋 불명 격리·예산·밴 커밋), stop_grace 40s, Redis 타임아웃·volatile-lru, /health DLQ·캐시 - KRX 휴장일, py3.11 비호환 f-string 수정 + Jenkins 이미지 내 compileall·헬스 실패 시 빌드 실패 하네스: rules_check 에 PEP 701 f-string 검사, 마커(allow-*), deploy 스모크 401 기준 CLAUDE.md: 실제 동작·플래그·백로그(§8.5-27~38) 갱신 backend 1451 passed / hooks 112 passed / tsc ok / py3.9 compile 0 errors Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
384 lines
18 KiB
Python
384 lines
18 KiB
Python
"""스윙 도메인 4회차 검증 잔여 결함 회귀 테스트 (2026-10-02).
|
||
|
||
L-4 현재가 조회 실패(예외·빈 문자열)여도 대체 가격으로 폴백 손절 판정
|
||
L-6 손절률·ATR 배수 합리 범위 검증(config 키)
|
||
L-3 시간손절 원 진입일 탐색을 마지막 전량 청산 이후 BUY 로 제한
|
||
L-7 KRX_EXTRA_HOLIDAYS 파싱 캐시 + 형식 오류 warning 설정값당 1회
|
||
|
||
규약(CLAUDE.md §4.1): 주문 경로 목은 예외 발생형(_Reached, BaseException). 기대값은 settings.*.
|
||
"""
|
||
import logging
|
||
from datetime import date, datetime, timedelta
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
from sqlalchemy.dialects import postgresql
|
||
|
||
import app.services.mean_reversion_strategy as mr
|
||
from app.core.config import settings
|
||
|
||
|
||
class _Reached(BaseException):
|
||
pass
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset(monkeypatch):
|
||
monkeypatch.setattr(mr, "_previous_regime", None)
|
||
from app.core import log_throttle
|
||
log_throttle.reset_for_tests()
|
||
import app.services.swing.exit_service as es
|
||
monkeypatch.setattr(es, "_last_live_prices", {})
|
||
yield
|
||
|
||
|
||
def _no_adaptive(monkeypatch):
|
||
import app.services.adaptive_strategy as ad
|
||
monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace(fallback_used=True))
|
||
monkeypatch.setattr(mr.strategy_cfg, "get", lambda k, d=None: None)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# L-4
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def _harness(monkeypatch, *, holding_price, avg=10000, price_resp=None, price_exc=None,
|
||
holding_age_sec=60, entry_days_ago=90):
|
||
import app.services.swing.exit_service as es
|
||
import app.services.trade_executor as te
|
||
from app.core.timeutil import now_kst
|
||
|
||
# [H-1/M-1] Portfolio 대체가는 updated_at 기준 나이 상한 안이어야 쓰인다
|
||
holding = SimpleNamespace(stock_code="005930", stock_name="T", quantity=10,
|
||
average_price=avg, current_price=holding_price, market="KRX",
|
||
updated_at=now_kst() - timedelta(seconds=holding_age_sec))
|
||
monkeypatch.setattr(es, "get_swing_holdings", AsyncMock(return_value=[holding]))
|
||
monkeypatch.setattr(es, "check_market_regime", AsyncMock(return_value={"regime_200d": "bull"}))
|
||
at = SimpleNamespace(strategy_name="swing", remaining_pct=1.0, highest_since_entry=None,
|
||
created_at=now_kst() - timedelta(days=entry_days_ago), entry_reason=None)
|
||
monkeypatch.setattr(es, "get_swing_active_trade", AsyncMock(return_value=at))
|
||
api = MagicMock()
|
||
if price_exc is not None:
|
||
api.get_current_price = AsyncMock(side_effect=price_exc)
|
||
else:
|
||
api.get_current_price = AsyncMock(return_value=price_resp)
|
||
bars = [{"stck_clpr": str(avg), "stck_hgpr": str(avg + 100), "stck_lwpr": str(avg - 100)}
|
||
for _ in range(30)]
|
||
api.get_daily_chart = AsyncMock(return_value={"output2": bars})
|
||
# [C-1 6회차] 국내 매도 직전 KIS 실보유 확인 — 보유 10주 전부 스윙(DT 없음)
|
||
api.get_balance = AsyncMock(return_value={"rt_cd": "0", "output1": [
|
||
{"pdno": "005930", "hldg_qty": "10", "ord_psbl_qty": "10"}]})
|
||
monkeypatch.setattr(es, "korea_investment_api", api)
|
||
monkeypatch.setattr(es, "calculate_atr", lambda h, l, c: avg * 0.01)
|
||
monkeypatch.setattr(es, "calculate_rsi", lambda c: 45.0)
|
||
monkeypatch.setattr(es, "calculate_bollinger", lambda c: (avg * 1.2, avg * 1.1, avg * 0.9))
|
||
sig_calls = []
|
||
|
||
def _sig(**kw):
|
||
sig_calls.append(kw)
|
||
return mr.ExitSignal("SELL_PARTIAL", "익절", 0.5)
|
||
monkeypatch.setattr(es, "check_exit_signal", _sig)
|
||
placed = []
|
||
|
||
async def _spo(**kw):
|
||
placed.append(kw)
|
||
raise _Reached("sell")
|
||
monkeypatch.setattr(te, "safe_place_order", _spo)
|
||
from app.services.daytrading.position_manager import day_trading_state
|
||
monkeypatch.setattr(day_trading_state, "positions", {})
|
||
es._recent_exits.clear()
|
||
return es, api, placed, sig_calls
|
||
|
||
|
||
async def _run(es):
|
||
try:
|
||
await es.check_and_execute_exit_v2(MagicMock())
|
||
except _Reached:
|
||
pass
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize("kw", [
|
||
dict(price_exc=RuntimeError("KIS down")),
|
||
dict(price_resp={"output": {"stck_prpr": ""}}),
|
||
dict(price_resp={"output": {}}),
|
||
dict(price_resp=None),
|
||
])
|
||
async def test_l4_price_failure_still_runs_fallback_stop(monkeypatch, kw):
|
||
_no_adaptive(monkeypatch)
|
||
es, api, placed, sig_calls = _harness(monkeypatch, holding_price=8000, **kw) # -20%
|
||
await _run(es)
|
||
assert len(placed) == 1 and placed[0]["quantity"] == 10 and placed[0]["order_type"] == "SELL"
|
||
assert sig_calls == [] # 낡은 가격으로 지표 판정(익절)은 하지 않는다
|
||
api.get_daily_chart.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l4_stale_price_never_takes_profit_or_time_stop(monkeypatch):
|
||
_no_adaptive(monkeypatch)
|
||
# 90일 보유 + 대체 가격 -1% → 실시세라면 시간손절, 낡은 가격이면 판정 안 함
|
||
es, _, placed, sig_calls = _harness(monkeypatch, holding_price=9900,
|
||
price_exc=RuntimeError("down"))
|
||
await _run(es)
|
||
assert placed == [] and sig_calls == []
|
||
|
||
|
||
def _cache(es, price, age_sec):
|
||
"""실시세 캐시 항목을 age_sec 초 전에 기록된 것으로 심는다."""
|
||
import time as _t
|
||
from app.core.timeutil import now_kst
|
||
es._last_live_prices["005930"] = (float(price), _t.monotonic() - age_sec,
|
||
now_kst() - timedelta(seconds=age_sec))
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l4_newer_cache_beats_older_portfolio(monkeypatch):
|
||
"""[H-1/M-1 사양 교체] 캐시·Portfolio 중 **더 최신** 값을 쓴다 — 캐시가 최신."""
|
||
_no_adaptive(monkeypatch)
|
||
es, _, placed, _ = _harness(monkeypatch, holding_price=10000, holding_age_sec=300,
|
||
price_exc=RuntimeError("down"))
|
||
_cache(es, 8000, 10)
|
||
await _run(es)
|
||
assert len(placed) == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l4_newer_portfolio_beats_older_cache(monkeypatch):
|
||
_no_adaptive(monkeypatch)
|
||
es, _, placed, _ = _harness(monkeypatch, holding_price=8000, holding_age_sec=10,
|
||
price_exc=RuntimeError("down"))
|
||
_cache(es, 10000, 300) # 더 오래된 캐시(보합) — 최신 Portfolio(-20%)를 가리면 안 된다
|
||
await _run(es)
|
||
assert len(placed) == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l4_aged_out_prices_skip_with_error(monkeypatch, caplog):
|
||
_no_adaptive(monkeypatch)
|
||
too_old = float(settings.SWING_FALLBACK_PRICE_MAX_AGE_SEC) + 60
|
||
es, _, placed, _ = _harness(monkeypatch, holding_price=8000, holding_age_sec=too_old,
|
||
price_exc=RuntimeError("down"))
|
||
_cache(es, 8000, too_old)
|
||
with caplog.at_level(logging.ERROR, logger=es.logger.name):
|
||
await _run(es)
|
||
assert placed == []
|
||
assert any(r.levelno >= logging.ERROR and "수동 확인" in r.getMessage() for r in caplog.records)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l4_portfolio_without_updated_at_is_not_used(monkeypatch):
|
||
_no_adaptive(monkeypatch)
|
||
es, _, placed, _ = _harness(monkeypatch, holding_price=8000, price_exc=RuntimeError("down"))
|
||
es.get_swing_holdings.return_value[0].updated_at = None
|
||
await _run(es)
|
||
assert placed == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_h1_rebuy_does_not_use_previous_position_price(monkeypatch):
|
||
"""[재현] 예전 보유분 실시세 9,300 이 캐시에 남은 채 같은 종목 재매수(진입 11,000).
|
||
시세 조회 1회 실패 → 이전: 9,300 으로 −15% 판정 → 시장가 전량 오손절 + loss_tracker 오염.
|
||
이제: 진입 이전 값은 거부, 재매수 이후 Portfolio 평가가(11,000)로 판정 → 손절 없음."""
|
||
_no_adaptive(monkeypatch)
|
||
from app.core.timeutil import now_kst
|
||
es, _, placed, _ = _harness(monkeypatch, holding_price=11000, avg=11000, holding_age_sec=30,
|
||
price_exc=RuntimeError("down"), entry_days_ago=0)
|
||
# 진입(ActiveTrade.created_at) = 2분 전, 캐시 = 5분 전(이전 보유분)
|
||
import app.services.swing.exit_service as _es
|
||
at = await _es.get_swing_active_trade(None, "005930")
|
||
at.created_at = now_kst() - timedelta(seconds=120)
|
||
_cache(es, 9300, 300)
|
||
await _run(es)
|
||
assert placed == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_h1_rebuy_only_stale_cache_skips(monkeypatch):
|
||
"""재매수 후 Portfolio 도 아직 갱신 전(진입 이전 updated_at)이면 판정 skip."""
|
||
_no_adaptive(monkeypatch)
|
||
from app.core.timeutil import now_kst
|
||
es, _, placed, _ = _harness(monkeypatch, holding_price=9300, avg=11000, holding_age_sec=300,
|
||
price_exc=RuntimeError("down"), entry_days_ago=0)
|
||
at = await es.get_swing_active_trade(None, "005930")
|
||
at.created_at = now_kst() - timedelta(seconds=120)
|
||
_cache(es, 9300, 300)
|
||
await _run(es)
|
||
assert placed == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l4_no_price_at_all_logs_error(monkeypatch, caplog):
|
||
_no_adaptive(monkeypatch)
|
||
es, _, placed, _ = _harness(monkeypatch, holding_price=0, price_exc=RuntimeError("down"))
|
||
with caplog.at_level(logging.ERROR, logger=es.logger.name):
|
||
await _run(es)
|
||
assert placed == []
|
||
assert any(r.levelno >= logging.ERROR and "005930" in r.getMessage() for r in caplog.records)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l4_live_price_success_is_remembered(monkeypatch):
|
||
_no_adaptive(monkeypatch)
|
||
es, _, placed, sig_calls = _harness(monkeypatch, holding_price=1,
|
||
price_resp={"output": {"stck_prpr": "10500"}})
|
||
await _run(es)
|
||
assert es._last_live_prices["005930"][0] == 10500 # (가격, monotonic, wall)
|
||
assert sig_calls and sig_calls[0]["close"] == 10500 # 실시세면 지표 판정 정상 수행
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# L-6
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def _cfg(monkeypatch, values):
|
||
import app.services.adaptive_strategy as ad
|
||
monkeypatch.setattr(mr.strategy_cfg, "get", lambda k, d=None: values.get(k))
|
||
monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace(fallback_used=True))
|
||
|
||
|
||
@pytest.mark.parametrize("bad", [-0.5, 0.5, -35.0, 500.0])
|
||
def test_l6_stop_loss_out_of_range_falls_back(monkeypatch, bad):
|
||
_cfg(monkeypatch, {"STOP_LOSS_PERCENT": bad})
|
||
assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[0] == -abs(settings.STOP_LOSS_PERCENT)
|
||
|
||
|
||
@pytest.mark.parametrize("ok", [-3.0, -settings.SWING_STOP_LOSS_PCT_MIN, -settings.SWING_STOP_LOSS_PCT_MAX])
|
||
def test_l6_stop_loss_in_range_accepted(monkeypatch, ok):
|
||
_cfg(monkeypatch, {"STOP_LOSS_PERCENT": ok})
|
||
assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[0] == -abs(ok)
|
||
|
||
|
||
@pytest.mark.parametrize("bad", [0.4, 10.5, -2.0])
|
||
def test_l6_atr_mult_out_of_range_falls_back(monkeypatch, bad):
|
||
_cfg(monkeypatch, {"ATR_STOP_MULTIPLIER": bad})
|
||
assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[1] == float(settings.ATR_STOP_MULTIPLIER)
|
||
|
||
|
||
def test_l6_atr_mult_in_range_accepted(monkeypatch):
|
||
_cfg(monkeypatch, {"ATR_STOP_MULTIPLIER": 1.5})
|
||
assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[1] == 1.5
|
||
|
||
|
||
def test_l6_bounds_come_from_settings(monkeypatch):
|
||
monkeypatch.setattr(settings, "SWING_STOP_LOSS_PCT_MAX", 5.0)
|
||
_cfg(monkeypatch, {"STOP_LOSS_PERCENT": -6.0})
|
||
assert mr.resolve_swing_stop_params_v2(100.0, 10000.0)[0] == -abs(settings.STOP_LOSS_PERCENT)
|
||
|
||
|
||
def test_l6_adaptive_out_of_range_uses_lower_source(monkeypatch):
|
||
import app.services.adaptive_strategy as ad
|
||
monkeypatch.setattr(mr.strategy_cfg, "get", lambda k, d=None: None)
|
||
monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace(
|
||
fallback_used=False, stop_loss_pct=-0.01, stop_loss_high_vol=-0.01, atr_stop_multiplier=50.0))
|
||
assert mr.resolve_swing_stop_params_v2(100.0, 10000.0) == (
|
||
-abs(settings.STOP_LOSS_PERCENT), float(settings.ATR_STOP_MULTIPLIER))
|
||
|
||
|
||
@pytest.mark.parametrize("mode", ["aggressive", "normal", "defensive", "crisis"])
|
||
def test_l6_all_adaptive_modes_pass_validation(monkeypatch, mode):
|
||
"""현행 _MODE_PARAMS 값이 범위 검증에 걸려 조용히 config 로 대체되지 않는다."""
|
||
import app.services.adaptive_strategy as ad
|
||
p = ad._MODE_PARAMS[ad.TradingMode(mode)] if hasattr(ad, "TradingMode") else ad._MODE_PARAMS[mode]
|
||
monkeypatch.setattr(mr.strategy_cfg, "get", lambda k, d=None: None)
|
||
monkeypatch.setattr(ad, "get_current_adaptive_params", lambda: SimpleNamespace(
|
||
fallback_used=False, stop_loss_pct=p["stop_loss_pct"],
|
||
stop_loss_high_vol=p["stop_loss_high_vol"], atr_stop_multiplier=p["atr_stop_multiplier"]))
|
||
assert mr.resolve_swing_stop_params_v2(100.0, 10000.0) == (
|
||
-abs(p["stop_loss_pct"]), float(p["atr_stop_multiplier"]))
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# L-3
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
class _SeqDB:
|
||
"""execute 호출 순서대로 scalar 값을 돌려주고 문장을 기록한다."""
|
||
|
||
def __init__(self, *values):
|
||
self.values = list(values)
|
||
self.statements = []
|
||
|
||
async def execute(self, stmt):
|
||
self.statements.append(stmt)
|
||
v = self.values.pop(0) if self.values else None
|
||
return SimpleNamespace(scalar=lambda: v)
|
||
|
||
|
||
def _sql(stmt) -> str:
|
||
return str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
|
||
|
||
|
||
_SYNC = "동기화 유입 포지션 (부분매도 시 자동 생성)"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l3_buy_search_bounded_by_last_full_close():
|
||
import app.services.swing.exit_service as es
|
||
created = datetime(2026, 3, 2, 10, 0)
|
||
last_close = datetime(2026, 2, 1, 15, 0)
|
||
buy = datetime(2026, 2, 10, 9, 30)
|
||
db = _SeqDB(last_close, buy)
|
||
at = SimpleNamespace(created_at=created, entry_reason=_SYNC)
|
||
basis = await es._resolve_time_stop_entry_v2(db, "005930", at)
|
||
assert basis.created_at == buy
|
||
buy_sql = _sql(db.statements[1])
|
||
assert "trade_history.created_at > '2026-02-01 15:00:00'" in buy_sql
|
||
closed_sql = _sql(db.statements[0])
|
||
assert "active_trades.status = 'closed'" in closed_sql or "status = 'closed'" in closed_sql
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l3_no_buy_after_close_keeps_created_at():
|
||
import app.services.swing.exit_service as es
|
||
at = SimpleNamespace(created_at=datetime(2026, 3, 2, 10, 0), entry_reason=_SYNC)
|
||
db = _SeqDB(datetime(2026, 2, 1), None) # 전량 청산 이후 BUY 없음
|
||
assert await es._resolve_time_stop_entry_v2(db, "005930", at) is at
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l3_no_closed_row_has_no_lower_bound():
|
||
import app.services.swing.exit_service as es
|
||
at = SimpleNamespace(created_at=datetime(2026, 3, 2, 10, 0), entry_reason=_SYNC)
|
||
db = _SeqDB(None, datetime(2026, 1, 5))
|
||
basis = await es._resolve_time_stop_entry_v2(db, "005930", at)
|
||
assert basis.created_at == datetime(2026, 1, 5)
|
||
assert "trade_history.created_at >" not in _sql(db.statements[1])
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_l3_query_failure_keeps_created_at():
|
||
import app.services.swing.exit_service as es
|
||
|
||
class _Boom:
|
||
async def execute(self, stmt):
|
||
raise RuntimeError("db down")
|
||
at = SimpleNamespace(created_at=datetime(2026, 3, 2, 10, 0), entry_reason=_SYNC)
|
||
assert await es._resolve_time_stop_entry_v2(_Boom(), "005930", at) is at
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# L-7
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_l7_extra_holidays_parsed_once_and_warns_once(monkeypatch, caplog):
|
||
from app.core import market_schedule as ms
|
||
monkeypatch.setattr(ms, "_extra_holidays_cache", None)
|
||
monkeypatch.setattr(settings, "KRX_EXTRA_HOLIDAYS", "2026-10-06,not-a-date", raising=False)
|
||
with caplog.at_level(logging.WARNING, logger=ms.__name__):
|
||
assert ms.is_krx_holiday(date(2026, 10, 6))
|
||
assert ms.is_krx_holiday(date(2026, 10, 6))
|
||
ms.count_krx_trading_days_between(date(2026, 9, 1), date(2026, 10, 30))
|
||
warns = [r for r in caplog.records if "KRX_EXTRA_HOLIDAYS" in r.getMessage()]
|
||
assert len(warns) == 1
|
||
|
||
|
||
def test_l7_cache_follows_setting_changes(monkeypatch):
|
||
from app.core import market_schedule as ms
|
||
monkeypatch.setattr(ms, "_extra_holidays_cache", None)
|
||
monkeypatch.setattr(settings, "KRX_EXTRA_HOLIDAYS", "2026-10-06", raising=False)
|
||
assert ms.is_krx_holiday(date(2026, 10, 6))
|
||
monkeypatch.setattr(settings, "KRX_EXTRA_HOLIDAYS", "2026-10-07", raising=False)
|
||
assert not ms.is_krx_holiday(date(2026, 10, 6))
|
||
assert ms.is_krx_holiday(date(2026, 10, 7))
|