247 lines
13 KiB
Python
247 lines
13 KiB
Python
"""빌드 스냅샷 — ★ 사이트에 나가면 안 되는 것이 스냅샷에 들어오지 않는가.
|
|
|
|
스냅샷이 정적 빌드의 경계다. 여기 들어온 것은 그대로 발행되므로, 필터링이 여기서 새면 끝이다.
|
|
"""
|
|
import uuid
|
|
|
|
from sqlalchemy import text
|
|
|
|
from common.enums import FactStatus, MediaStatus, PlaceCategory, SourceType
|
|
from services.snapshot import build_snapshot
|
|
|
|
|
|
async def _seed(db_engine, owner_id, category=PlaceCategory.LODGING):
|
|
pid = uuid.uuid4()
|
|
async with db_engine.begin() as c:
|
|
await c.execute(
|
|
text("INSERT INTO places (place_id, owner_user_id, name, category, status, road_address, phone, verified_at) "
|
|
"VALUES (:p,:c,:n,:cat,3,:addr,:tel,now())"),
|
|
{"p": pid, "c": uuid.UUID(owner_id), "n": "스냅샷펜션", "cat": category.value,
|
|
"addr": "강원특별자치도 양양군 현북면 하조대3길 11", "tel": "033-000-0000"},
|
|
)
|
|
return pid
|
|
|
|
|
|
async def _fact(db_engine, pid, key, value, status, unit_id=None):
|
|
async with db_engine.begin() as c:
|
|
await c.execute(
|
|
text("INSERT INTO place_facts (fact_id, place_id, unit_id, key, value, source_type, status, collected_at) "
|
|
"VALUES (:f,:p,:u,:k,:v,:s,:st,now())"),
|
|
{"f": uuid.uuid4(), "p": pid, "u": unit_id, "k": key, "v": value,
|
|
"s": SourceType.OWNER.value, "st": status.value},
|
|
)
|
|
|
|
|
|
async def _media(db_engine, pid, url, status, alt="설명"):
|
|
async with db_engine.begin() as c:
|
|
await c.execute(
|
|
text("INSERT INTO place_photos (media_id, place_id, url, origin_url, source_type, status, alt_text, sort_order) "
|
|
"VALUES (:m,:p,:u,:u,:s,:st,:a,0)"),
|
|
{"m": uuid.uuid4(), "p": pid, "u": url, "s": SourceType.CRAWL.value, "st": status.value, "a": alt},
|
|
)
|
|
|
|
|
|
class _Place:
|
|
def __init__(self, pid, category=PlaceCategory.LODGING):
|
|
self.place_id = pid
|
|
self.category = category.value
|
|
self.name = "스냅샷펜션"
|
|
self.road_address = "강원특별자치도 양양군 현북면 하조대3길 11"
|
|
self.address = None
|
|
self.phone = "033-000-0000"
|
|
self.latitude = None
|
|
self.longitude = None
|
|
|
|
|
|
async def test_only_publishable_facts_enter_snapshot(db_engine, owner_id):
|
|
"""검증: 여러 상태의 fact 를 섞어 넣는다.
|
|
기대결과: ★ VERIFIED·CORRECTED 만 스냅샷에 담긴다 — 미검증 값이 사이트로 새지 않는다."""
|
|
pid = await _seed(db_engine, owner_id)
|
|
await _fact(db_engine, pid, "check_in_time", "15:00", FactStatus.VERIFIED)
|
|
await _fact(db_engine, pid, "wifi", "true", FactStatus.CORRECTED)
|
|
await _fact(db_engine, pid, "parking", "true", FactStatus.UNVERIFIED)
|
|
await _fact(db_engine, pid, "breakfast", "true", FactStatus.PENDING_OWNER)
|
|
await _fact(db_engine, pid, "bbq_available", "true", FactStatus.REJECTED)
|
|
await _fact(db_engine, pid, "smoking", "false", FactStatus.EXPIRED)
|
|
|
|
snap = await build_snapshot(_Place(pid))
|
|
keys = {f["key"] for f in snap["facts"]}
|
|
assert keys == {"check_in_time", "wifi"}
|
|
|
|
|
|
async def test_only_approved_media_enters_snapshot(db_engine, owner_id):
|
|
"""검증: 승인/확인대기/반려 사진을 섞어 넣는다.
|
|
기대결과: ★ APPROVED 만 담긴다 — Vision 신뢰도가 낮아 확인 큐에 남은 사진은 안 나간다."""
|
|
pid = await _seed(db_engine, owner_id)
|
|
await _media(db_engine, pid, "https://cdn.test/ok.jpg", MediaStatus.APPROVED)
|
|
await _media(db_engine, pid, "https://cdn.test/pending.jpg", MediaStatus.PENDING_REVIEW)
|
|
await _media(db_engine, pid, "https://cdn.test/no.jpg", MediaStatus.REJECTED)
|
|
|
|
snap = await build_snapshot(_Place(pid))
|
|
assert [m["url"] for m in snap["media"]] == ["https://cdn.test/ok.jpg"]
|
|
|
|
|
|
async def test_media_without_alt_is_excluded(db_engine, owner_id):
|
|
"""검증: 승인됐지만 alt 텍스트가 없는 사진.
|
|
기대결과: 빠진다 — alt 없는 이미지는 접근성도 AI 검색 신호도 없다."""
|
|
pid = await _seed(db_engine, owner_id)
|
|
await _media(db_engine, pid, "https://cdn.test/noalt.jpg", MediaStatus.APPROVED, alt="")
|
|
await _media(db_engine, pid, "https://cdn.test/withalt.jpg", MediaStatus.APPROVED, alt="침실 사진")
|
|
|
|
snap = await build_snapshot(_Place(pid))
|
|
assert [m["url"] for m in snap["media"]] == ["https://cdn.test/withalt.jpg"]
|
|
|
|
|
|
async def test_fact_labels_come_from_category_schema(db_engine, owner_id):
|
|
"""검증: 스냅샷의 fact 라벨.
|
|
기대결과: 업종 스키마의 한글 라벨이 붙는다 — 화면이 key 를 그대로 노출하지 않게."""
|
|
pid = await _seed(db_engine, owner_id)
|
|
await _fact(db_engine, pid, "check_in_time", "15:00", FactStatus.VERIFIED)
|
|
|
|
snap = await build_snapshot(_Place(pid))
|
|
f = snap["facts"][0]
|
|
assert f["label"] == "체크인 시간"
|
|
assert f["scope"] == "place"
|
|
|
|
|
|
async def test_unit_scoped_facts_carry_unit_id(db_engine, owner_id):
|
|
"""검증: 객실 단위 fact.
|
|
기대결과: unit_id 가 실려 빌더가 객실별로 묶을 수 있다."""
|
|
pid = await _seed(db_engine, owner_id)
|
|
uid = uuid.uuid4()
|
|
async with db_engine.begin() as c:
|
|
await c.execute(text("INSERT INTO place_units (unit_id, place_id, name, sort_order) VALUES (:u,:p,:n,0)"),
|
|
{"u": uid, "p": pid, "n": "A동"})
|
|
await _fact(db_engine, pid, "max_capacity", "4", FactStatus.VERIFIED, unit_id=uid)
|
|
|
|
snap = await build_snapshot(_Place(pid))
|
|
assert snap["units"][0]["name"] == "A동"
|
|
assert snap["facts"][0]["unit_id"] == str(uid)
|
|
assert snap["facts"][0]["scope"] == "unit"
|
|
|
|
|
|
async def test_empty_place_gives_empty_snapshot(db_engine, owner_id):
|
|
"""검증: 아무것도 없는 사업장.
|
|
기대결과: 빈 스냅샷 — 게이트가 고유 콘텐츠 0건으로 거부할 재료가 된다."""
|
|
pid = await _seed(db_engine, owner_id)
|
|
snap = await build_snapshot(_Place(pid))
|
|
assert snap["facts"] == [] and snap["media"] == [] and snap["faqs"] == []
|
|
assert snap["place"]["name"] == "스냅샷펜션"
|
|
|
|
|
|
# ── 지역 정보 ─────────────────────────────────────────────────────────────
|
|
# ★ 지역 정보가 스냅샷에 담기는 이유: site_payload 는 DB 를 다시 읽지 않는다.
|
|
# 거기서 지역 캐시를 읽으면 발행 시점과 렌더 시점 사이에 값이 바뀌어 '스냅샷과 다른 페이지'가 나온다.
|
|
|
|
async def _local(db_engine, region_code, kind, status, title="지역이야기", **cols):
|
|
"""지역 캐시(`area_contents`)에 **지역 단위** 항목 한 행.
|
|
|
|
★ `external_id` 를 넣지 않는다. 그 값이 있는 행(축제·관광지·맛집)은 업장마다 거리가 달라
|
|
개인화(`site_sections`)를 거쳐 들어오고, 스냅샷의 지역 캐시 경로는 그것들을 일부러
|
|
건너뛴다(`services/snapshot._local_contents`). 예전 이 픽스처는 uuid 를 external_id 로
|
|
넣고 있어서, **읽는 코드가 옳게 걸러내는데도 테스트가 빨개졌다**.
|
|
★ 한 지역에 같은 kind 는 한 행이다(`uq_local_contents_kind`). 여러 행이 필요한 테스트는
|
|
kind 를 달리 준다 — 지역 이야기 다섯 종이 그 자리다.
|
|
"""
|
|
from common.enums import LocalContentType, LocalSource
|
|
|
|
async with db_engine.begin() as c:
|
|
await c.execute(
|
|
text("INSERT INTO area_contents "
|
|
"(local_content_id, region_code, content_type, source, title, body, status, kind, "
|
|
" display_start_at, display_end_at, collected_at) "
|
|
"VALUES (:i,:r,:ct,:src,:t,cast(:b as jsonb),:st,:k,:ds,:de,now())"),
|
|
{"i": uuid.uuid4(), "r": region_code, "ct": LocalContentType.STORY.value,
|
|
"src": LocalSource.LLM.value, "t": title,
|
|
"b": '{"title":"%s"}' % title, "st": status.value, "k": kind,
|
|
"ds": cols.get("display_start_at"), "de": cols.get("display_end_at")},
|
|
)
|
|
|
|
|
|
class _RegionPlace(_Place):
|
|
"""region_code 를 가진 사업장. 지역 캐시의 키는 place_id 가 아니라 region_code 다."""
|
|
|
|
def __init__(self, pid, region_code):
|
|
super().__init__(pid)
|
|
self.region_code = region_code
|
|
|
|
|
|
async def test_only_published_local_content_enters_snapshot(db_engine, owner_id):
|
|
"""검증: 검수대기·종료·발행 지역 정보를 섞어 넣는다.
|
|
기대결과: ★ PUBLISHED 만 담긴다 — 운영자가 검수하지 않은 외부 API 원문이 사이트로 새면
|
|
'미검증 값 노출 금지'가 깨진다(fact 를 VERIFIED 로 거르는 것과 같은 규칙)."""
|
|
from common.enums import LocalContentStatus
|
|
|
|
pid = await _seed(db_engine, owner_id)
|
|
# kind 를 달리 준다 — 한 지역에 같은 kind 는 한 행이다(uq_local_contents_kind).
|
|
await _local(db_engine, "4113500", "songs", LocalContentStatus.PUBLISHED, "발행이야기")
|
|
await _local(db_engine, "4113500", "people", LocalContentStatus.REVIEW, "검수대기이야기")
|
|
await _local(db_engine, "4113500", "chronicle", LocalContentStatus.ENDED, "종료이야기")
|
|
|
|
snap = await build_snapshot(_RegionPlace(pid, "4113500"))
|
|
assert [c["title"] for c in snap["local"]["contents"]] == ["발행이야기"]
|
|
|
|
|
|
async def test_local_content_outside_display_window_is_excluded(db_engine, owner_id):
|
|
"""검증: 발행됐지만 노출 기간을 벗어난 지역 정보.
|
|
기대결과: 빠진다 — 끝난 축제를 '이번 주말 행사'로 걸어두는 것도 틀린 정보다."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from common.enums import LocalContentStatus
|
|
|
|
now = datetime.now(timezone.utc)
|
|
pid = await _seed(db_engine, owner_id)
|
|
await _local(db_engine, "4113500", "songs", LocalContentStatus.PUBLISHED, "지금이야기")
|
|
await _local(db_engine, "4113500", "people", LocalContentStatus.PUBLISHED, "끝난이야기",
|
|
display_end_at=now - timedelta(days=1))
|
|
await _local(db_engine, "4113500", "chronicle", LocalContentStatus.PUBLISHED, "아직이야기",
|
|
display_start_at=now + timedelta(days=1))
|
|
|
|
snap = await build_snapshot(_RegionPlace(pid, "4113500"))
|
|
assert [c["title"] for c in snap["local"]["contents"]] == ["지금이야기"]
|
|
|
|
|
|
async def test_local_content_is_scoped_to_the_places_region(db_engine, owner_id):
|
|
"""검증: 지역 캐시는 region_code 로 묶인다.
|
|
기대결과: 다른 지역의 발행 콘텐츠는 담기지 않는다."""
|
|
from common.enums import LocalContentStatus
|
|
|
|
pid = await _seed(db_engine, owner_id)
|
|
await _local(db_engine, "4113500", "songs", LocalContentStatus.PUBLISHED, "우리지역이야기")
|
|
await _local(db_engine, "5011025", "songs", LocalContentStatus.PUBLISHED, "남의지역이야기")
|
|
|
|
snap = await build_snapshot(_RegionPlace(pid, "4113500"))
|
|
assert [c["title"] for c in snap["local"]["contents"]] == ["우리지역이야기"]
|
|
|
|
|
|
async def test_region_code_is_derived_from_the_address_when_missing(db_engine, owner_id):
|
|
"""검증: region_code 가 비어 있지만 도로명주소는 있는 사업장.
|
|
기대결과: 주소에서 지역 키를 유도해 그 지역 콘텐츠를 담는다 — places.region_code 를 채우는
|
|
코드가 생기기 전에 만들어진 사업장(실측 28곳 중 25곳)이 영영 지역 정보 없이 발행되지 않게 한다."""
|
|
from common.enums import LocalContentStatus
|
|
from services.external.naver import region_key
|
|
|
|
derived = region_key(_Place("x").road_address)
|
|
pid = await _seed(db_engine, owner_id)
|
|
await _local(db_engine, derived, "songs", LocalContentStatus.PUBLISHED, "양양이야기")
|
|
|
|
snap = await build_snapshot(_RegionPlace(pid, ""))
|
|
assert snap["local"]["region_code"] == derived
|
|
assert [c["title"] for c in snap["local"]["contents"]] == ["양양이야기"]
|
|
|
|
|
|
async def test_place_without_any_region_key_gets_no_local_content(db_engine, owner_id):
|
|
"""검증: 지역 코드도 읽을 만한 주소도 없는 사업장.
|
|
기대결과: 빈 목록 — 조회할 캐시 키가 없다. 지어내지 않는다."""
|
|
from common.enums import LocalContentStatus
|
|
|
|
pid = await _seed(db_engine, owner_id)
|
|
await _local(db_engine, "4113500", "songs", LocalContentStatus.PUBLISHED, "어딘가이야기")
|
|
|
|
place = _RegionPlace(pid, "")
|
|
place.road_address = None
|
|
snap = await build_snapshot(place)
|
|
# ★ itineraries 는 항상 있는 키다(2026-09-11) — 이 업장은 place_id 는 있으니
|
|
# place_itineraries 를 조회하고, 아직 생성된 게 없어 빈 목록으로 온다.
|
|
assert snap["local"] == {"region_code": None, "contents": [], "itineraries": []}
|