gahusb
b885d02ac4
db.DB_PATH = _TMP를 from app import db 직후에 주입해 타 테스트 파일이 app.db를 먼저 import해 DB_PATH가 동결된 경우에도 올바른 임시 경로를 사용하도록 수정. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
import gc
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
|
|
_fd, _TMP = tempfile.mkstemp(suffix=".db")
|
|
os.close(_fd)
|
|
os.unlink(_TMP)
|
|
os.environ["AGENT_OFFICE_DB_PATH"] = _TMP
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import pytest
|
|
|
|
from app.curator import signal_runner
|
|
from app import db
|
|
|
|
db.DB_PATH = _TMP # patch frozen module-level DB_PATH (import order safety)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fresh_db():
|
|
gc.collect()
|
|
if os.path.exists(_TMP):
|
|
os.remove(_TMP)
|
|
db.init_db()
|
|
yield
|
|
gc.collect()
|
|
if os.path.exists(_TMP):
|
|
try:
|
|
os.remove(_TMP)
|
|
except PermissionError:
|
|
pass # Windows: WAL-mode file locked; DB is ephemeral anyway
|
|
|
|
|
|
def test_evaluate_and_persist_cold_start():
|
|
"""첫 호출은 warmup으로 기록되고 baseline에 값이 들어간다."""
|
|
result = signal_runner.evaluate_metric_and_persist(
|
|
source="light",
|
|
metric="sim_signal",
|
|
value=1.5,
|
|
draw_no=None,
|
|
z_normal=1.5,
|
|
z_urgent=2.5,
|
|
push_to_window=True,
|
|
)
|
|
assert result["fire_level"] == "warmup"
|
|
assert result["z_score"] is None
|
|
|
|
bl = db.get_baseline("sim_signal")
|
|
assert bl is not None
|
|
assert bl["window_values"] == [1.5]
|
|
|
|
|
|
def test_evaluate_after_window_filled_normal_fire():
|
|
"""8회 push 후 정상 운영, 평균 대비 z≥1.5면 normal."""
|
|
for v in [1.0, 1.1, 0.9, 1.0, 1.0, 1.1, 0.9, 1.0]:
|
|
signal_runner.evaluate_metric_and_persist(
|
|
source="sim",
|
|
metric="sim_signal",
|
|
value=v,
|
|
draw_no=None,
|
|
z_normal=1.5,
|
|
z_urgent=2.5,
|
|
push_to_window=True,
|
|
)
|
|
|
|
result = signal_runner.evaluate_metric_and_persist(
|
|
source="sim",
|
|
metric="sim_signal",
|
|
value=1.12,
|
|
draw_no=None,
|
|
z_normal=1.5,
|
|
z_urgent=2.5,
|
|
push_to_window=True,
|
|
)
|
|
assert result["fire_level"] in ("normal", "urgent")
|
|
assert result["z_score"] is not None and result["z_score"] >= 1.5
|
|
|
|
|
|
def test_evaluate_drift_skips_same_draw_push():
|
|
"""drift는 회차 단위. 같은 회차에서 두 번 호출하면 두 번째는 window push X."""
|
|
signal_runner.evaluate_metric_and_persist(
|
|
source="sim", metric="drift", value=0.05, draw_no=1100,
|
|
z_normal=1.5, z_urgent=2.5, push_to_window=True,
|
|
)
|
|
bl_before = db.get_baseline("drift")
|
|
assert bl_before["window_values"] == [0.05]
|
|
assert bl_before["last_pushed_draw_no"] == 1100
|
|
|
|
signal_runner.evaluate_metric_and_persist(
|
|
source="sim", metric="drift", value=0.08, draw_no=1100,
|
|
z_normal=1.5, z_urgent=2.5, push_to_window=True,
|
|
)
|
|
bl_after = db.get_baseline("drift")
|
|
assert bl_after["window_values"] == [0.05]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_signal_check_aggregates_three_metrics(monkeypatch):
|
|
"""run_signal_check이 3종 메트릭 모두 평가하고 overall fire를 반환."""
|
|
async def fake_lotto_best():
|
|
return [{"numbers": [1,2,3,4,5,6], "scores": [10,10,10,10,10]}] * 20
|
|
|
|
async def fake_lotto_strategy_weights():
|
|
return {"gap_focus": 0.4, "hot_focus": 0.3, "pair_bias": 0.3}
|
|
|
|
monkeypatch.setattr(signal_runner, "_fetch_best_picks", fake_lotto_best)
|
|
monkeypatch.setattr(signal_runner, "_fetch_strategy_weights", fake_lotto_strategy_weights)
|
|
|
|
out = await signal_runner.run_signal_check(source="light", curate_result=None, current_draw_no=1101)
|
|
assert "overall_fire" in out
|
|
assert "results" in out
|
|
assert any(r["metric"] == "sim_signal" for r in out["results"])
|
|
# light_check는 confidence 평가 안 함
|
|
assert not any(r["metric"] == "confidence" for r in out["results"])
|