100 lines
3.0 KiB
Python
100 lines
3.0 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 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 test_on_new_matches_returns_empty_when_no_matches():
|
|
from app.agents.realestate import RealestateAgent
|
|
|
|
agent = RealestateAgent()
|
|
result = asyncio.run(agent.on_new_matches([]))
|
|
assert result == {"sent": 0, "sent_ids": []}
|
|
|
|
|
|
def test_on_new_matches_sends_telegram_and_returns_ids():
|
|
from app.agents.realestate import RealestateAgent
|
|
from app.telegram import messaging
|
|
|
|
matches = [{
|
|
"id": 7, "match_score": 80, "house_nm": "단지A",
|
|
"region_name": "서울특별시", "district": "강남구",
|
|
"receipt_start": "2026-05-01", "receipt_end": "2026-05-05",
|
|
"match_reasons": [], "eligible_types": [], "pblanc_url": "https://x.test/7",
|
|
}]
|
|
|
|
fake_send = AsyncMock(return_value={"ok": True, "message_id": 123})
|
|
with patch.object(messaging, "send_raw", fake_send):
|
|
agent = RealestateAgent()
|
|
result = asyncio.run(agent.on_new_matches(matches))
|
|
|
|
assert result["sent"] == 1
|
|
assert result["sent_ids"] == [7]
|
|
assert result["message_id"] == 123
|
|
fake_send.assert_awaited_once()
|
|
args, kwargs = fake_send.call_args
|
|
text = args[0]
|
|
assert "단지A" in text
|
|
|
|
|
|
def test_on_new_matches_telegram_failure_returns_zero():
|
|
from app.agents.realestate import RealestateAgent
|
|
from app.telegram import messaging
|
|
|
|
matches = [{
|
|
"id": 8, "match_score": 80, "house_nm": "단지B",
|
|
"region_name": "서울", "district": "송파구",
|
|
"receipt_start": "", "receipt_end": "",
|
|
"match_reasons": [], "eligible_types": [], "pblanc_url": "",
|
|
}]
|
|
|
|
fake_send = AsyncMock(return_value={"ok": False, "description": "401"})
|
|
with patch.object(messaging, "send_raw", fake_send):
|
|
agent = RealestateAgent()
|
|
result = asyncio.run(agent.on_new_matches(matches))
|
|
|
|
assert result["sent"] == 0
|
|
assert result["sent_ids"] == []
|
|
assert "error" in result
|
|
|
|
|
|
def test_endpoint_calls_agent_on_new_matches():
|
|
from fastapi.testclient import TestClient
|
|
from app.main import app
|
|
from app.agents.realestate import RealestateAgent
|
|
|
|
fake = AsyncMock(return_value={"sent": 1, "sent_ids": [99], "message_id": 1})
|
|
with patch.object(RealestateAgent, "on_new_matches", fake):
|
|
with TestClient(app) as client:
|
|
resp = client.post(
|
|
"/api/agent-office/realestate/notify",
|
|
json={"matches": [{"id": 99, "match_score": 80}]},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["sent"] == 1
|
|
assert body["sent_ids"] == [99]
|