87 lines
3.6 KiB
Python
87 lines
3.6 KiB
Python
import json
|
|
from datetime import date, timedelta
|
|
from app.db import _conn
|
|
|
|
|
|
def _seed_announcement(house_nm, status, winner_date=None, hmno="HM1", pno="P1"):
|
|
with _conn() as conn:
|
|
conn.execute("""
|
|
INSERT INTO announcements (house_manage_no, pblanc_no, house_nm, status, winner_date, source)
|
|
VALUES (?, ?, ?, ?, ?, 'manual')
|
|
""", (hmno, pno, house_nm, status, winner_date))
|
|
return conn.execute("SELECT id FROM announcements WHERE house_manage_no=?", (hmno,)).fetchone()["id"]
|
|
|
|
|
|
def test_delete_old_completed_removes_expired():
|
|
from app.db import delete_old_completed_announcements
|
|
old = (date.today() - timedelta(days=100)).isoformat()
|
|
_seed_announcement("OldA", "완료", old, hmno="OLD", pno="1")
|
|
deleted = delete_old_completed_announcements(grace_days=90)
|
|
assert deleted == 1
|
|
|
|
|
|
def test_delete_old_completed_keeps_recent():
|
|
from app.db import delete_old_completed_announcements
|
|
recent = (date.today() - timedelta(days=30)).isoformat()
|
|
_seed_announcement("RecentA", "완료", recent, hmno="REC", pno="1")
|
|
deleted = delete_old_completed_announcements(grace_days=90)
|
|
assert deleted == 0
|
|
|
|
|
|
def test_delete_old_completed_keeps_active():
|
|
from app.db import delete_old_completed_announcements
|
|
old = (date.today() - timedelta(days=200)).isoformat()
|
|
_seed_announcement("ActiveA", "청약중", old, hmno="ACT", pno="1")
|
|
deleted = delete_old_completed_announcements(grace_days=90)
|
|
assert deleted == 0
|
|
|
|
|
|
def test_delete_old_completed_keeps_null_winner_date():
|
|
from app.db import delete_old_completed_announcements
|
|
_seed_announcement("NullA", "완료", None, hmno="NULL", pno="1")
|
|
deleted = delete_old_completed_announcements(grace_days=90)
|
|
assert deleted == 0 # winner_date NULL은 안전 보존
|
|
|
|
|
|
def test_get_unnotified_matches_filters_by_score_and_null():
|
|
from app.db import get_unnotified_matches
|
|
aid = _seed_announcement("MatchA", "청약중", hmno="MA", pno="1")
|
|
with _conn() as conn:
|
|
# 임계값 미만
|
|
conn.execute("""
|
|
INSERT INTO match_results (announcement_id, model_id, match_score, match_reasons, eligible_types, is_new)
|
|
VALUES (?, NULL, 50, '[]', '[]', 1)
|
|
""", (aid,))
|
|
# 임계값 통과 — 미알림
|
|
conn.execute("""
|
|
INSERT INTO match_results (announcement_id, model_id, match_score, match_reasons, eligible_types, is_new)
|
|
VALUES (?, 1, 80, '[]', '[]', 1)
|
|
""", (aid,))
|
|
# 임계값 통과 — 이미 알림됨
|
|
conn.execute("""
|
|
INSERT INTO match_results (announcement_id, model_id, match_score, match_reasons, eligible_types, is_new, notified_at)
|
|
VALUES (?, 2, 90, '[]', '[]', 1, '2026-04-01T00:00:00.000Z')
|
|
""", (aid,))
|
|
|
|
matches = get_unnotified_matches(min_score=70)
|
|
assert len(matches) == 1
|
|
assert matches[0]["match_score"] == 80
|
|
assert matches[0]["house_nm"] == "MatchA"
|
|
|
|
|
|
def test_mark_matches_notified_sets_timestamp():
|
|
from app.db import mark_matches_notified
|
|
aid = _seed_announcement("NotifyA", "청약중", hmno="NT", pno="1")
|
|
with _conn() as conn:
|
|
cur = conn.execute("""
|
|
INSERT INTO match_results (announcement_id, model_id, match_score, match_reasons, eligible_types, is_new)
|
|
VALUES (?, NULL, 80, '[]', '[]', 1)
|
|
""", (aid,))
|
|
match_id = cur.lastrowid
|
|
|
|
mark_matches_notified([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
|