feat(agent-office): /watch /unwatch /watchlist 봇 명령

agent-office/tests/test_watch_commands.py

@@ -0,0 +1,93 @@
+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 unittest.mock import AsyncMock, patch
+
+
+@pytest.fixture(autouse=True)
+def _init_db(monkeypatch):
+    import gc
+    gc.collect()
+    # config.DB_PATH는 첫 import 시 1회 고정되므로, 다른 테스트 파일과 조합 실행 시
+    # db가 이 파일의 _TMP가 아닌 다른 경로를 쓸 수 있다. db.DB_PATH를 이 파일 전용으로
+    # 강제해 영속 테이블의 테스트 간 누수를 결정적으로 차단.
+    import app.db as _db
+    monkeypatch.setattr(_db, "DB_PATH", _TMP)
+    for suffix in ("", "-wal", "-shm"):
+        p = _TMP + suffix
+        if os.path.exists(p):
+            os.remove(p)
+    _db.init_db()
+    yield
+    gc.collect()
+
+
+@pytest.mark.asyncio
+async def test_watch_command_calls_add():
+    from app.telegram import webhook
+    msg = {"chat": {"id": 1}, "text": "/watch 005930"}
+    with patch("app.telegram.webhook.service_proxy.watchlist_add",
+               new=AsyncMock(return_value={"ok": True})) as m, \
+         patch("app.telegram.webhook.api_call", new=AsyncMock(return_value={"ok": True})):
+        handled = await webhook.handle_watch_command(msg)
+    assert handled is True
+    m.assert_awaited_once_with("005930")
+
+
+@pytest.mark.asyncio
+async def test_non_watch_text_ignored():
+    from app.telegram import webhook
+    msg = {"chat": {"id": 1}, "text": "안녕"}
+    assert await webhook.handle_watch_command(msg) is False
+
+
+@pytest.mark.asyncio
+async def test_unwatch_command_calls_remove():
+    from app.telegram import webhook
+    msg = {"chat": {"id": 1}, "text": "/unwatch 005930"}
+    with patch("app.telegram.webhook.service_proxy.watchlist_remove",
+               new=AsyncMock(return_value={"ok": True})) as m, \
+         patch("app.telegram.webhook.api_call", new=AsyncMock(return_value={"ok": True})) as sent:
+        handled = await webhook.handle_watch_command(msg)
+    assert handled is True
+    m.assert_awaited_once_with("005930")
+    sent.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_watchlist_command_calls_list_and_formats_items():
+    from app.telegram import webhook
+    msg = {"chat": {"id": 1}, "text": "/watchlist"}
+    items = {"watchlist": [{"ticker": "005930", "name": "삼성전자"}]}
+    with patch("app.telegram.webhook.service_proxy.watchlist_list",
+               new=AsyncMock(return_value=items)) as m, \
+         patch("app.telegram.webhook.api_call", new=AsyncMock(return_value={"ok": True})) as sent:
+        handled = await webhook.handle_watch_command(msg)
+    assert handled is True
+    m.assert_awaited_once_with()
+    text = sent.await_args.args[1]["text"]
+    assert "005930" in text and "삼성전자" in text
+
+
+@pytest.mark.asyncio
+async def test_watch_command_reaches_handle_webhook_before_slash_dispatch():
+    """handle_webhook이 /watch 를 agent_dispatcher 호출 전에 가로채야 한다."""
+    from app.telegram import webhook
+    data = {"message": {"chat": {"id": 1}, "text": "/watch 005930"}}
+    dispatcher = AsyncMock(side_effect=AssertionError("agent_dispatcher가 호출되면 안 됨"))
+    with patch("app.telegram.webhook.service_proxy.watchlist_add",
+               new=AsyncMock(return_value={"ok": True})) as m, \
+         patch("app.telegram.webhook.api_call", new=AsyncMock(return_value={"ok": True})):
+        result = await webhook.handle_webhook(data, agent_dispatcher=dispatcher)
+    assert result is None
+    m.assert_awaited_once_with("005930")
+    dispatcher.assert_not_awaited()