gahusb
32e021cfc7
- scheduler.py: remove _run_realestate_schedule() and its 09:15 cron job
- service_proxy.py: add realestate_bookmark_toggle() helper (PATCH bookmark endpoint)
- webhook.py: add _handle_realestate_bookmark() dispatcher before DB-lookup path;
realestate_bookmark_{id} callbacks are handled inline without a DB entry
- tests/test_realestate_callback.py: 4 new unit tests covering happy path,
invalid id, proxy error, and regression that approve/reject still uses DB path
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
import os
|
|
import sys
|
|
import tempfile
|
|
import gc
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
_TMP = tempfile.mktemp(suffix=".db")
|
|
os.environ["AGENT_OFFICE_DB_PATH"] = _TMP
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import asyncio
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _init_db():
|
|
gc.collect()
|
|
if os.path.exists(_TMP):
|
|
try:
|
|
os.remove(_TMP)
|
|
except PermissionError:
|
|
pass
|
|
from app.db import init_db
|
|
init_db()
|
|
yield
|
|
|
|
|
|
def test_callback_realestate_bookmark_calls_proxy():
|
|
"""callback_data 'realestate_bookmark_42' 가 service_proxy.realestate_bookmark_toggle(42) 를 호출."""
|
|
from app import service_proxy
|
|
from app.telegram import webhook
|
|
|
|
fake_toggle = AsyncMock(return_value={"bookmarked": True})
|
|
fake_send = AsyncMock(return_value={"ok": True})
|
|
fake_api_call = AsyncMock(return_value={"ok": True})
|
|
|
|
update = {
|
|
"callback_query": {
|
|
"id": "cb1",
|
|
"from": {"id": 1},
|
|
"data": "realestate_bookmark_42",
|
|
}
|
|
}
|
|
|
|
with patch.object(service_proxy, "realestate_bookmark_toggle", fake_toggle), \
|
|
patch("app.telegram.messaging.send_raw", fake_send), \
|
|
patch("app.telegram.webhook.api_call", fake_api_call):
|
|
result = asyncio.run(webhook.handle_webhook(update))
|
|
|
|
fake_toggle.assert_awaited_once_with(42)
|
|
assert result == {"ok": True, "announcement_id": 42}
|
|
|
|
|
|
def test_callback_realestate_bookmark_invalid_id():
|
|
"""callback_data 'realestate_bookmark_abc' 는 ValueError를 처리하고 에러 응답 반환."""
|
|
from app import service_proxy
|
|
from app.telegram import webhook
|
|
|
|
fake_toggle = AsyncMock(return_value={"bookmarked": True})
|
|
fake_send = AsyncMock(return_value={"ok": True})
|
|
fake_api_call = AsyncMock(return_value={"ok": True})
|
|
|
|
update = {
|
|
"callback_query": {
|
|
"id": "cb2",
|
|
"from": {"id": 1},
|
|
"data": "realestate_bookmark_abc",
|
|
}
|
|
}
|
|
|
|
with patch.object(service_proxy, "realestate_bookmark_toggle", fake_toggle), \
|
|
patch("app.telegram.messaging.send_raw", fake_send), \
|
|
patch("app.telegram.webhook.api_call", fake_api_call):
|
|
result = asyncio.run(webhook.handle_webhook(update))
|
|
|
|
fake_toggle.assert_not_awaited()
|
|
assert result is not None
|
|
assert result.get("ok") is False
|
|
assert result.get("error") == "invalid_callback_data"
|
|
|
|
|
|
def test_callback_realestate_bookmark_proxy_error():
|
|
"""service_proxy 가 예외를 던질 때 에러 응답 반환."""
|
|
from app import service_proxy
|
|
from app.telegram import webhook
|
|
|
|
fake_toggle = AsyncMock(side_effect=Exception("connection refused"))
|
|
fake_send = AsyncMock(return_value={"ok": True})
|
|
fake_api_call = AsyncMock(return_value={"ok": True})
|
|
|
|
update = {
|
|
"callback_query": {
|
|
"id": "cb3",
|
|
"from": {"id": 1},
|
|
"data": "realestate_bookmark_99",
|
|
}
|
|
}
|
|
|
|
with patch.object(service_proxy, "realestate_bookmark_toggle", fake_toggle), \
|
|
patch("app.telegram.messaging.send_raw", fake_send), \
|
|
patch("app.telegram.webhook.api_call", fake_api_call):
|
|
result = asyncio.run(webhook.handle_webhook(update))
|
|
|
|
fake_toggle.assert_awaited_once_with(99)
|
|
assert result is not None
|
|
assert result.get("ok") is False
|
|
assert "connection refused" in result.get("error", "")
|
|
|
|
|
|
def test_non_realestate_callback_uses_db_path():
|
|
"""approve_*/reject_* 콜백은 기존 DB 조회 경로를 사용 (realestate 분기를 타지 않음)."""
|
|
from app.telegram import webhook
|
|
|
|
fake_api_call = AsyncMock(return_value={"ok": True})
|
|
|
|
update = {
|
|
"callback_query": {
|
|
"id": "cb4",
|
|
"from": {"id": 1},
|
|
"data": "approve_abcd1234",
|
|
}
|
|
}
|
|
|
|
# DB에 등록되지 않은 콜백이므로 None 반환 — 기존 로직 진입 확인
|
|
with patch("app.telegram.webhook.api_call", fake_api_call):
|
|
result = asyncio.run(webhook.handle_webhook(update))
|
|
|
|
assert result is None # DB에 없으면 None 반환 (기존 동작 유지)
|