web-page-backend/agent-office/tests/test_stock_screener_job.py

"""StockAgent.on_screener_schedule — 평일 16:30 KST 자동 잡 단위 테스트.

stock-lab HTTP 호출은 service_proxy mock, 텔레그램은 messaging.send_raw mock.
"""
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 asyncio
from unittest.mock import AsyncMock, patch
import pytest


@pytest.fixture(autouse=True)
def _init_db():
    import gc
    gc.collect()
    if os.path.exists(_TMP):
        os.remove(_TMP)
    from app.db import init_db
    init_db()
    yield
    gc.collect()


def _success_body(asof="2026-05-12"):
    return {
        "asof": asof,
        "mode": "auto",
        "status": "success",
        "run_id": 42,
        "survivors_count": 600,
        "top_n": 20,
        "results": [],
        "telegram_payload": {
            "chat_target": "default",
            "parse_mode": "MarkdownV2",
            "text": "*KRX 강세주 스크리너* test body",
        },
        "warnings": [],
    }


def _holiday_body(asof="2026-05-05"):
    return {
        "asof": asof,
        "mode": "auto",
        "status": "skipped_holiday",
        "run_id": None,
        "survivors_count": None,
        "top_n": 0,
        "results": [],
        "telegram_payload": None,
        "warnings": [f"{asof} is a holiday — skipped"],
    }


def test_screener_success_sends_markdownv2_telegram():
    from app.agents.stock import StockAgent
    from app import service_proxy
    from app.telegram import messaging

    fake_snap = AsyncMock(return_value={"status": "ok"})
    fake_run = AsyncMock(return_value=_success_body())
    fake_send = AsyncMock(return_value={"ok": True, "message_id": 7777})

    with patch.object(service_proxy, "refresh_screener_snapshot", fake_snap), \
         patch.object(service_proxy, "run_stock_screener", fake_run), \
         patch.object(messaging, "send_raw", fake_send):
        agent = StockAgent()
        asyncio.run(agent.on_screener_schedule())

    fake_snap.assert_awaited_once()
    fake_run.assert_awaited_once_with(mode="auto")
    fake_send.assert_awaited_once()
    args, kwargs = fake_send.call_args
    # 첫 인자(text) 또는 kwargs로 전달
    text = args[0] if args else kwargs.get("text")
    assert "KRX 강세주 스크리너" in text
    assert kwargs.get("parse_mode") == "MarkdownV2"
    assert agent.state == "idle"


def test_screener_holiday_skips_telegram():
    from app.agents.stock import StockAgent
    from app import service_proxy
    from app.telegram import messaging

    fake_snap = AsyncMock(return_value={"status": "skipped_weekend"})
    fake_run = AsyncMock(return_value=_holiday_body())
    fake_send = AsyncMock(return_value={"ok": True, "message_id": 1})

    with patch.object(service_proxy, "refresh_screener_snapshot", fake_snap), \
         patch.object(service_proxy, "run_stock_screener", fake_run), \
         patch.object(messaging, "send_raw", fake_send):
        agent = StockAgent()
        asyncio.run(agent.on_screener_schedule())

    fake_run.assert_awaited_once()
    # 휴일이면 텔레그램 미발신
    fake_send.assert_not_awaited()
    assert agent.state == "idle"


def test_screener_snapshot_failure_still_runs_screener():
    """스냅샷 실패는 경고만 남기고 screener 호출은 계속됨."""
    from app.agents.stock import StockAgent
    from app import service_proxy
    from app.telegram import messaging

    fake_snap = AsyncMock(side_effect=RuntimeError("snapshot upstream down"))
    fake_run = AsyncMock(return_value=_success_body())
    fake_send = AsyncMock(return_value={"ok": True, "message_id": 8888})

    with patch.object(service_proxy, "refresh_screener_snapshot", fake_snap), \
         patch.object(service_proxy, "run_stock_screener", fake_run), \
         patch.object(messaging, "send_raw", fake_send):
        agent = StockAgent()
        asyncio.run(agent.on_screener_schedule())

    fake_snap.assert_awaited_once()
    fake_run.assert_awaited_once_with(mode="auto")
    fake_send.assert_awaited_once()


def test_screener_run_failure_notifies_operator():
    """screener/run 실패 시 운영자 알림 텔레그램 발송."""
    from app.agents.stock import StockAgent
    from app import service_proxy
    from app.telegram import messaging

    fake_snap = AsyncMock(return_value={"status": "ok"})
    fake_run = AsyncMock(side_effect=RuntimeError("stock-lab 500"))
    fake_send = AsyncMock(return_value={"ok": True, "message_id": 1})

    with patch.object(service_proxy, "refresh_screener_snapshot", fake_snap), \
         patch.object(service_proxy, "run_stock_screener", fake_run), \
         patch.object(messaging, "send_raw", fake_send):
        agent = StockAgent()
        asyncio.run(agent.on_screener_schedule())

    # 운영자 알림 1회는 호출
    assert fake_send.await_count == 1
    args, kwargs = fake_send.call_args
    text = args[0] if args else kwargs.get("text")
    assert "스크리너 실패" in text
    assert agent.state == "idle"


def test_screener_unexpected_status_treated_as_failure():
    from app.agents.stock import StockAgent
    from app import service_proxy
    from app.telegram import messaging

    fake_snap = AsyncMock(return_value={"status": "ok"})
    fake_run = AsyncMock(return_value={"status": "weird", "asof": "2026-05-12"})
    fake_send = AsyncMock(return_value={"ok": True, "message_id": 1})

    with patch.object(service_proxy, "refresh_screener_snapshot", fake_snap), \
         patch.object(service_proxy, "run_stock_screener", fake_run), \
         patch.object(messaging, "send_raw", fake_send):
        agent = StockAgent()
        asyncio.run(agent.on_screener_schedule())

    # 운영자 알림 1회 + screener payload 미발송
    assert fake_send.await_count == 1
    args, kwargs = fake_send.call_args
    text = args[0] if args else kwargs.get("text")
    assert "스크리너 실패" in text