56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
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)
|
|
# WAL 사이드카(-wal/-shm)까지 지워야 영속 상태가 남지 않음
|
|
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_send_trade_alerts_to_user_and_wife():
|
|
from app.notifiers import telegram_trade
|
|
alerts = [{"ticker": "005930", "name": "삼성전자", "kind": "buy",
|
|
"condition": "buy_breakout", "price": 71500, "detail": {}}]
|
|
with patch("app.notifiers.telegram_trade.send_raw",
|
|
new=AsyncMock(return_value={"ok": True})) as m, \
|
|
patch("app.notifiers.telegram_trade.TELEGRAM_CHAT_ID", "U"), \
|
|
patch("app.notifiers.telegram_trade.TELEGRAM_WIFE_CHAT_ID", "W"):
|
|
res = await telegram_trade.send_trade_alerts(alerts)
|
|
assert res["ok"] is True
|
|
chat_ids = {c.kwargs.get("chat_id") for c in m.await_args_list}
|
|
assert chat_ids == {"U", "W"} # 둘 다 발송
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_format_trade_alert_has_direction():
|
|
from app.notifiers.telegram_trade import format_trade_alert
|
|
txt = format_trade_alert({"ticker": "005930", "name": "삼성전자", "kind": "sell",
|
|
"condition": "sell_stop_loss", "price": 60000, "detail": {}})
|
|
assert "매도" in txt and "삼성전자" in txt
|