gahusb
c2939459e7
- Replace `or 70` fallback with explicit None-check so that
min_match_score=0 ("notify all matches") is no longer silently
coerced to 70
- Add test: 200 OK + sent_ids=[] must not mark matches notified
- Add test: threshold=0 correctly pushes low-score matches
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
from unittest.mock import patch, MagicMock
|
|
|
|
|
|
def _seed_profile_and_match(score, notify_enabled=True, threshold=70):
|
|
from app.db import _conn, upsert_profile
|
|
upsert_profile({
|
|
"name": "u",
|
|
"notify_enabled": notify_enabled,
|
|
"min_match_score": threshold,
|
|
})
|
|
with _conn() as conn:
|
|
conn.execute("""
|
|
INSERT INTO announcements (house_manage_no, pblanc_no, house_nm, status, source)
|
|
VALUES ('NF1', '01', '단지', '청약중', 'manual')
|
|
""")
|
|
ann_id = conn.execute("SELECT id FROM announcements WHERE house_manage_no='NF1'").fetchone()["id"]
|
|
conn.execute("""
|
|
INSERT INTO match_results (announcement_id, model_id, match_score, match_reasons, eligible_types, is_new)
|
|
VALUES (?, NULL, ?, '[]', '[]', 1)
|
|
""", (ann_id, score))
|
|
match_id = conn.execute("SELECT id FROM match_results WHERE announcement_id=?", (ann_id,)).fetchone()["id"]
|
|
return match_id
|
|
|
|
|
|
def test_notify_skips_when_disabled():
|
|
from app import notifier
|
|
_seed_profile_and_match(score=80, notify_enabled=False)
|
|
with patch.object(notifier, "requests") as r:
|
|
result = notifier.notify_new_matches()
|
|
assert r.post.call_count == 0
|
|
assert result["sent"] == 0
|
|
assert result.get("skipped") == "notify_disabled"
|
|
|
|
|
|
def test_notify_filters_below_threshold():
|
|
from app import notifier
|
|
_seed_profile_and_match(score=60, threshold=70)
|
|
with patch.object(notifier, "requests") as r:
|
|
result = notifier.notify_new_matches()
|
|
assert r.post.call_count == 0
|
|
assert result["sent"] == 0
|
|
|
|
|
|
def test_notify_pushes_and_marks_notified():
|
|
from app import notifier
|
|
from app.db import _conn
|
|
|
|
match_id = _seed_profile_and_match(score=80, threshold=70)
|
|
|
|
fake_resp = MagicMock()
|
|
fake_resp.json.return_value = {"sent": 1, "sent_ids": [match_id]}
|
|
fake_resp.raise_for_status.return_value = None
|
|
|
|
with patch.object(notifier.requests, "post", return_value=fake_resp) as post:
|
|
result = notifier.notify_new_matches()
|
|
|
|
assert post.call_count == 1
|
|
args, kwargs = post.call_args
|
|
assert "/api/agent-office/realestate/notify" in args[0]
|
|
assert kwargs["json"]["matches"][0]["id"] == match_id
|
|
|
|
with _conn() as conn:
|
|
row = conn.execute("SELECT notified_at FROM match_results WHERE id=?", (match_id,)).fetchone()
|
|
assert row["notified_at"] is not None
|
|
assert result["sent"] == 1
|
|
|
|
|
|
def test_notify_does_not_mark_on_failure():
|
|
from app import notifier
|
|
from app.db import _conn
|
|
import requests as real_requests
|
|
|
|
match_id = _seed_profile_and_match(score=80, threshold=70)
|
|
|
|
def boom(*a, **k):
|
|
raise real_requests.RequestException("agent-office down")
|
|
|
|
with patch.object(notifier.requests, "post", side_effect=boom):
|
|
result = notifier.notify_new_matches()
|
|
|
|
with _conn() as conn:
|
|
row = conn.execute("SELECT notified_at FROM match_results WHERE id=?", (match_id,)).fetchone()
|
|
assert row["notified_at"] is None
|
|
assert result["sent"] == 0
|
|
assert "error" in result
|
|
|
|
|
|
def test_notify_does_not_mark_when_sent_ids_empty():
|
|
"""agent-office가 200 OK + sent_ids=[]을 반환하면 마킹하지 않고 다음 사이클 재시도 가능."""
|
|
from app import notifier
|
|
from app.db import _conn
|
|
|
|
match_id = _seed_profile_and_match(score=80, threshold=70)
|
|
|
|
fake_resp = MagicMock()
|
|
fake_resp.json.return_value = {"sent": 0, "sent_ids": []}
|
|
fake_resp.raise_for_status.return_value = None
|
|
|
|
with patch.object(notifier.requests, "post", return_value=fake_resp):
|
|
result = notifier.notify_new_matches()
|
|
|
|
with _conn() as conn:
|
|
row = conn.execute("SELECT notified_at FROM match_results WHERE id=?", (match_id,)).fetchone()
|
|
|
|
assert row["notified_at"] is None
|
|
assert result["sent"] == 0
|
|
|
|
|
|
def test_notify_threshold_zero_pushes_all_matches():
|
|
"""min_match_score=0이면 모든 양수 점수 매치를 알림."""
|
|
from app import notifier
|
|
from app.db import _conn
|
|
|
|
match_id = _seed_profile_and_match(score=10, threshold=0)
|
|
|
|
fake_resp = MagicMock()
|
|
fake_resp.json.return_value = {"sent": 1, "sent_ids": [match_id]}
|
|
fake_resp.raise_for_status.return_value = None
|
|
|
|
with patch.object(notifier.requests, "post", return_value=fake_resp) as post:
|
|
result = notifier.notify_new_matches()
|
|
|
|
assert post.call_count == 1
|
|
assert result["sent"] == 1
|