gahusb
c6b969443f
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EqCYBhvTcdeCTUDX3RhWx9
79 lines
4.4 KiB
Python
79 lines
4.4 KiB
Python
"""매물 알림 텔레그램 렌더+전송 테스트. 기존 test_trade_alert_notify.py 패턴(async mock + chat_id patch) 준수."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_format_listing_rent_and_sale():
|
|
from app.notifiers.telegram_realestate_listing import format_listing_alert
|
|
|
|
rent = format_listing_alert({"id": 1, "category": "임차", "deal_type": "전세",
|
|
"complex_name": "OO", "deposit": 29000, "area_exclusive": 42.0, "floor": "5/15",
|
|
"safety_tier": "안전", "jeonse_ratio": 0.68, "market_median": 43000, "sample_size": 7,
|
|
"dong": "신대방동", "url": "http://x", "regulation_flags": []})
|
|
assert "전세" in rent and "안전" in rent and "OO" in rent and "68" in rent
|
|
|
|
sale = format_listing_alert({"id": 2, "category": "매매", "deal_type": "매매",
|
|
"complex_name": "PP", "sale_price": 89000, "area_exclusive": 59.0, "floor": "12/20",
|
|
"valuation_tier": "시세", "price_ratio": 1.01, "market_median": 88000, "sample_size": 7,
|
|
"dong": "상도동", "url": "http://y", "budget_ok": 1, "regulation_flags": []})
|
|
assert "매매" in sale and "시세" in sale
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_format_listing_pending_sample_and_regulation_flags():
|
|
"""표본<3(보류)·토허 등 regulation_flags 경고 라인이 포함되는지."""
|
|
from app.notifiers.telegram_realestate_listing import format_listing_alert
|
|
|
|
pending = format_listing_alert({"id": 3, "category": "임차", "deal_type": "전세",
|
|
"complex_name": "QQ", "deposit": 28000, "area_exclusive": 40.0, "floor": "3/10",
|
|
"safety_tier": "보류", "jeonse_ratio": None, "market_median": None, "sample_size": 1,
|
|
"dong": "봉천동", "url": "http://z", "regulation_flags": []})
|
|
assert "보류" in pending
|
|
|
|
toheo = format_listing_alert({"id": 4, "category": "매매", "deal_type": "매매",
|
|
"complex_name": "RR", "sale_price": 150000, "area_exclusive": 84.0, "floor": "7/15",
|
|
"valuation_tier": "고가", "price_ratio": 1.1, "market_median": 136000, "sample_size": 5,
|
|
"dong": "대치동", "url": "http://w", "budget_ok": 0, "regulation_flags": ["토허"]})
|
|
assert "토허" in toheo
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_listing_alerts_returns_sent_ids():
|
|
from app.notifiers import telegram_realestate_listing as t
|
|
items = [{"id": 5, "category": "임차", "deal_type": "전세", "complex_name": "OO",
|
|
"deposit": 29000, "area_exclusive": 42.0, "safety_tier": "안전",
|
|
"jeonse_ratio": 0.68, "market_median": 43000, "sample_size": 7,
|
|
"dong": "신대방동", "url": "http://x", "regulation_flags": []}]
|
|
with patch("app.notifiers.telegram_realestate_listing.send_raw",
|
|
new=AsyncMock(return_value={"ok": True})) as m, \
|
|
patch("app.notifiers.telegram_realestate_listing.TELEGRAM_CHAT_ID", "U"), \
|
|
patch("app.notifiers.telegram_realestate_listing.TELEGRAM_WIFE_CHAT_ID", "W"):
|
|
res = await t.send_listing_alerts(items)
|
|
assert res["sent_ids"] == [5] and res["ok"] is True
|
|
assert res["sent"] == 2 # 너+아내 둘 다
|
|
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_send_listing_alerts_partial_failure_not_marked_sent():
|
|
"""한쪽 chat_id 전송이 예외를 던져도 나머지는 계속 발송되고, 최소 1건 성공이면 sent_ids에 포함."""
|
|
from app.notifiers import telegram_realestate_listing as t
|
|
items = [{"id": 6, "category": "매매", "deal_type": "매매", "complex_name": "PP",
|
|
"sale_price": 89000, "area_exclusive": 59.0, "valuation_tier": "시세",
|
|
"price_ratio": 1.01, "market_median": 88000, "sample_size": 7,
|
|
"dong": "상도동", "url": "http://y", "budget_ok": 1, "regulation_flags": []}]
|
|
|
|
async def _flaky(text, chat_id=None, **kw):
|
|
if chat_id == "U":
|
|
raise RuntimeError("network error")
|
|
return {"ok": True}
|
|
|
|
with patch("app.notifiers.telegram_realestate_listing.send_raw", side_effect=_flaky), \
|
|
patch("app.notifiers.telegram_realestate_listing.TELEGRAM_CHAT_ID", "U"), \
|
|
patch("app.notifiers.telegram_realestate_listing.TELEGRAM_WIFE_CHAT_ID", "W"):
|
|
res = await t.send_listing_alerts(items)
|
|
assert res["sent_ids"] == [6]
|
|
assert res["ok"] is False # 일부 실패했으므로 all_ok=False
|