gahusb
9dd517e82a
- Add _extract_district() helper with DISTRICT_PATTERN regex (서울 only)
- collect_all() now passes RCRIT_PBLANC_DE_FROM param (30-day window) to all detail endpoints
- collect_all() skips announcements where compute_status() returns '완료'
- collect_all() stamps district on each parsed announcement before upsert
- upsert_announcement(): add district to INSERT/VALUES/ON CONFLICT UPDATE; data.setdefault('district', None)
- ANNOUNCEMENT_COLUMNS: add 'district' (closes deferred gap from Task 2 review)
- 9 new tests in realestate-lab/tests/test_collector.py (6 unit + 3 integration)
- Full suite: 22 passed
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
132 lines
4.8 KiB
Python
132 lines
4.8 KiB
Python
"""Tests for collector.py — _extract_district unit tests + collect_all integration tests."""
|
|
from datetime import date, timedelta
|
|
|
|
|
|
# ── _extract_district unit tests ─────────────────────────────────────────────
|
|
|
|
def test_extract_district_seoul_full_address():
|
|
from app.collector import _extract_district
|
|
parsed = {"address": "서울특별시 강남구 도곡동 123-45", "region_name": None}
|
|
assert _extract_district(parsed) == "강남구"
|
|
|
|
|
|
def test_extract_district_seoul_short():
|
|
from app.collector import _extract_district
|
|
parsed = {"address": None, "region_name": "서울 송파구"}
|
|
assert _extract_district(parsed) == "송파구"
|
|
|
|
|
|
def test_extract_district_busan_returns_none():
|
|
from app.collector import _extract_district
|
|
parsed = {"address": "부산광역시 해운대구 우동", "region_name": None}
|
|
assert _extract_district(parsed) is None
|
|
|
|
|
|
def test_extract_district_empty_returns_none():
|
|
from app.collector import _extract_district
|
|
parsed = {"address": "", "region_name": ""}
|
|
assert _extract_district(parsed) is None
|
|
|
|
|
|
def test_extract_district_seoul_county():
|
|
from app.collector import _extract_district
|
|
parsed = {"address": "서울 강서구", "region_name": None}
|
|
assert _extract_district(parsed) == "강서구"
|
|
|
|
|
|
def test_extract_district_prefers_address_over_region():
|
|
from app.collector import _extract_district
|
|
parsed = {"address": "서울특별시 마포구 합정동", "region_name": "서울 강남구"}
|
|
assert _extract_district(parsed) == "마포구"
|
|
|
|
|
|
# ── collect_all integration tests ────────────────────────────────────────────
|
|
|
|
def test_collect_skips_completed_status(monkeypatch):
|
|
"""winner_date가 과거인 응답은 status='완료'로 판정되어 upsert되지 않는다."""
|
|
from app import collector
|
|
from app.db import _conn
|
|
|
|
monkeypatch.setenv("DATA_GO_KR_API_KEY", "TEST")
|
|
monkeypatch.setattr(collector, "API_KEY", "TEST")
|
|
|
|
past_winner = (date.today() - timedelta(days=10)).strftime("%Y-%m-%d")
|
|
|
|
fake_detail_rows = [{
|
|
"HOUSE_MANAGE_NO": "DONE-1",
|
|
"PBLANC_NO": "01",
|
|
"HOUSE_NM": "완료된단지",
|
|
"HSSPLY_ADRES": "서울특별시 강남구",
|
|
"RCEPT_BGNDE": "2026-01-01",
|
|
"RCEPT_ENDDE": "2026-01-05",
|
|
"PRZWNER_PRESNATN_DE": past_winner,
|
|
}]
|
|
|
|
def fake_call(endpoint, params=None):
|
|
if "Detail" in endpoint:
|
|
return fake_detail_rows
|
|
return []
|
|
|
|
monkeypatch.setattr(collector, "_api_call", fake_call)
|
|
collector.collect_all()
|
|
|
|
with _conn() as conn:
|
|
rows = conn.execute("SELECT * FROM announcements WHERE house_manage_no='DONE-1'").fetchall()
|
|
assert len(rows) == 0
|
|
|
|
|
|
def test_collect_stores_district_for_seoul_announcement(monkeypatch):
|
|
from app import collector
|
|
from app.db import _conn
|
|
|
|
monkeypatch.setenv("DATA_GO_KR_API_KEY", "TEST")
|
|
monkeypatch.setattr(collector, "API_KEY", "TEST")
|
|
|
|
future_start = (date.today() + timedelta(days=10)).strftime("%Y-%m-%d")
|
|
future_end = (date.today() + timedelta(days=15)).strftime("%Y-%m-%d")
|
|
future_winner = (date.today() + timedelta(days=30)).strftime("%Y-%m-%d")
|
|
|
|
fake_detail = [{
|
|
"HOUSE_MANAGE_NO": "SEOUL-1",
|
|
"PBLANC_NO": "01",
|
|
"HOUSE_NM": "강남단지",
|
|
"HSSPLY_ADRES": "서울특별시 강남구 도곡동 1",
|
|
"RCEPT_BGNDE": future_start,
|
|
"RCEPT_ENDDE": future_end,
|
|
"PRZWNER_PRESNATN_DE": future_winner,
|
|
}]
|
|
|
|
def fake_call(endpoint, params=None):
|
|
if "Detail" in endpoint:
|
|
return fake_detail
|
|
return []
|
|
|
|
monkeypatch.setattr(collector, "_api_call", fake_call)
|
|
collector.collect_all()
|
|
|
|
with _conn() as conn:
|
|
row = conn.execute("SELECT district, status FROM announcements WHERE house_manage_no='SEOUL-1'").fetchone()
|
|
assert row["district"] == "강남구"
|
|
assert row["status"] in ("청약예정", "청약중")
|
|
|
|
|
|
def test_collect_passes_date_window_param(monkeypatch):
|
|
from app import collector
|
|
|
|
monkeypatch.setenv("DATA_GO_KR_API_KEY", "TEST")
|
|
monkeypatch.setattr(collector, "API_KEY", "TEST")
|
|
|
|
captured_params = []
|
|
|
|
def fake_call(endpoint, params=None):
|
|
captured_params.append(params or {})
|
|
return []
|
|
|
|
monkeypatch.setattr(collector, "_api_call", fake_call)
|
|
collector.collect_all()
|
|
|
|
expected_from = (date.today() - timedelta(days=30)).strftime("%Y%m%d")
|
|
detail_calls = [p for p in captured_params if "RCRIT_PBLANC_DE_FROM" in p]
|
|
assert detail_calls, "detail 엔드포인트 호출에 윈도우 파라미터가 없음"
|
|
assert detail_calls[0]["RCRIT_PBLANC_DE_FROM"] == expected_from
|