155 lines
5.0 KiB
Python
155 lines
5.0 KiB
Python
# agent-office/tests/test_lotto_task_wrap.py
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import gc
|
|
|
|
_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 import db
|
|
db.DB_PATH = _TMP
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fresh_db():
|
|
# Re-patch DB_PATH at the start of every test (cross-file isolation)
|
|
db.DB_PATH = _TMP
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_signal_check_creates_task_row(monkeypatch):
|
|
"""run_signal_check이 agent_tasks에 row를 만들고 result_data를 저장."""
|
|
from app.agents.lotto import LottoAgent
|
|
from app.curator import signal_runner
|
|
|
|
async def fake_run_signal_check(**kwargs):
|
|
return {
|
|
"overall_fire": "normal",
|
|
"results": [
|
|
{"signal_id": 1, "metric": "sim_signal",
|
|
"value": 0.6, "z_score": 1.7, "fire_level": "normal",
|
|
"baseline_mu": 0.5, "baseline_sigma": 0.05, "payload": {}},
|
|
],
|
|
}
|
|
monkeypatch.setattr(signal_runner, "run_signal_check", fake_run_signal_check)
|
|
|
|
from app import service_proxy
|
|
async def fake_latest():
|
|
return 1226
|
|
monkeypatch.setattr(service_proxy, "lotto_latest_draw", fake_latest)
|
|
|
|
from app.notifiers import telegram_lotto
|
|
async def fake_send(_event): pass
|
|
monkeypatch.setattr(telegram_lotto, "send_urgent_signal", fake_send)
|
|
|
|
agent = LottoAgent()
|
|
result = await agent.run_signal_check(source="light")
|
|
assert result["ok"] is True
|
|
|
|
tasks = db.get_agent_tasks("lotto", task_type="signal_check", days=1)
|
|
assert len(tasks) == 1
|
|
t = tasks[0]
|
|
assert t["status"] == "succeeded"
|
|
assert t["result_data"]["source"] == "light"
|
|
assert t["result_data"]["overall_fire"] == "normal"
|
|
assert "sim_signal" in t["result_data"]["fired_metrics"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_signal_check_failure_marks_task_failed(monkeypatch):
|
|
from app.agents.lotto import LottoAgent
|
|
from app.curator import signal_runner
|
|
from app import service_proxy
|
|
|
|
async def boom(**kwargs):
|
|
raise RuntimeError("boom")
|
|
monkeypatch.setattr(signal_runner, "run_signal_check", boom)
|
|
|
|
async def fake_latest():
|
|
return 1226
|
|
monkeypatch.setattr(service_proxy, "lotto_latest_draw", fake_latest)
|
|
|
|
agent = LottoAgent()
|
|
result = await agent.run_signal_check(source="sim")
|
|
assert result["ok"] is False
|
|
|
|
tasks = db.get_agent_tasks("lotto", task_type="signal_check", days=1)
|
|
assert len(tasks) == 1
|
|
assert tasks[0]["status"] == "failed"
|
|
assert "boom" in tasks[0]["result_data"]["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_daily_digest_creates_task(monkeypatch):
|
|
"""run_daily_digest이 agent_tasks에 task 생성 + result_data 저장."""
|
|
from app.agents.lotto import LottoAgent
|
|
from app.notifiers import telegram_lotto
|
|
|
|
async def fake_send(_d): pass
|
|
monkeypatch.setattr(telegram_lotto, "send_signal_summary", fake_send)
|
|
|
|
agent = LottoAgent()
|
|
result = await agent.run_daily_digest()
|
|
assert result["ok"] is True
|
|
|
|
tasks = db.get_agent_tasks("lotto", task_type="daily_digest", days=1)
|
|
assert len(tasks) == 1
|
|
assert tasks[0]["status"] == "succeeded"
|
|
assert "fired" in tasks[0]["result_data"]
|
|
assert "evaluated" in tasks[0]["result_data"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_weekly_evolution_report_creates_task(monkeypatch):
|
|
"""run_weekly_evolution_report이 task 생성 + result_data 저장."""
|
|
from app.agents.lotto import LottoAgent
|
|
from app import service_proxy
|
|
from app.notifiers import telegram_lotto
|
|
|
|
async def fake_eval():
|
|
return {
|
|
"ok": True, "draw_no": 1225,
|
|
"winner": {"day_of_week": 3, "weight": [0.18, 0.32, 0.20, 0.22, 0.08],
|
|
"avg_score": 0.42, "max_correct": 4, "n_picks": 5},
|
|
"new_base": [0.18, 0.32, 0.20, 0.22, 0.08],
|
|
"previous_base": [0.2] * 5,
|
|
"update_reason": "winner_4plus",
|
|
}
|
|
async def fake_status():
|
|
return {"current_base": [0.2] * 5}
|
|
async def fake_send(_e, _b): pass
|
|
|
|
monkeypatch.setattr(service_proxy, "lotto_evolver_evaluate", fake_eval)
|
|
monkeypatch.setattr(service_proxy, "lotto_evolver_status", fake_status)
|
|
monkeypatch.setattr(telegram_lotto, "send_evolution_report", fake_send)
|
|
|
|
agent = LottoAgent()
|
|
result = await agent.run_weekly_evolution_report()
|
|
assert result["ok"] is True
|
|
|
|
tasks = db.get_agent_tasks("lotto", task_type="weekly_evolution_report", days=1)
|
|
assert len(tasks) == 1
|
|
r = tasks[0]["result_data"]
|
|
assert tasks[0]["status"] == "succeeded"
|
|
assert r["draw_no"] == 1225
|
|
assert r["update_reason"] == "winner_4plus"
|
|
assert r["winner_day_of_week"] == 3
|
|
assert r["winner_max_correct"] == 4
|