gahusb
fef5f7a835
daily_digest에 create_task/update_task_status/add_log task_id wrap 적용. test_run_daily_digest_creates_task 추가 (75 passed). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
115 lines
3.5 KiB
Python
115 lines
3.5 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():
|
|
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"]
|