o2o-site-AEO/solution/backend/tests/test_snapshot.py

"""빌드 스냅샷 — ★ 사이트에 나가면 안 되는 것이 스냅샷에 들어오지 않는가.

스냅샷이 정적 빌드의 경계다. 여기 들어온 것은 그대로 발행되므로, 필터링이 여기서 새면 끝이다.
"""
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, company_id, category=PlaceCategory.LODGING):
    pid = uuid.uuid4()
    async with db_engine.begin() as c:
        await c.execute(
            text("INSERT INTO places (place_id, company_id, name, category, status, road_address, phone, verified_at) "
                 "VALUES (:p,:c,:n,:cat,3,:addr,:tel,now())"),
            {"p": pid, "c": uuid.UUID(company_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 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 media (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, company_id):
    """검증: 여러 상태의 fact 를 섞어 넣는다.
    기대결과: ★ VERIFIED·CORRECTED 만 스냅샷에 담긴다 — 미검증 값이 사이트로 새지 않는다."""
    pid = await _seed(db_engine, company_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, company_id):
    """검증: 승인/확인대기/반려 사진을 섞어 넣는다.
    기대결과: ★ APPROVED 만 담긴다 — Vision 신뢰도가 낮아 확인 큐에 남은 사진은 안 나간다."""
    pid = await _seed(db_engine, company_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, company_id):
    """검증: 승인됐지만 alt 텍스트가 없는 사진.
    기대결과: 빠진다 — alt 없는 이미지는 접근성도 AI 검색 신호도 없다."""
    pid = await _seed(db_engine, company_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, company_id):
    """검증: 스냅샷의 fact 라벨.
    기대결과: 업종 스키마의 한글 라벨이 붙는다 — 화면이 key 를 그대로 노출하지 않게."""
    pid = await _seed(db_engine, company_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, company_id):
    """검증: 객실 단위 fact.
    기대결과: unit_id 가 실려 빌더가 객실별로 묶을 수 있다."""
    pid = await _seed(db_engine, company_id)
    uid = uuid.uuid4()
    async with db_engine.begin() as c:
        await c.execute(text("INSERT INTO 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, company_id):
    """검증: 아무것도 없는 사업장.
    기대결과: 빈 스냅샷 — 게이트가 고유 콘텐츠 0건으로 거부할 재료가 된다."""
    pid = await _seed(db_engine, company_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, content_type, status, title="지역행사", **cols):
    from common.enums import LocalSource

    async with db_engine.begin() as c:
        await c.execute(
            text("INSERT INTO local_contents "
                 "(local_content_id, region_code, content_type, source, external_id, title, body, status, "
                 " display_start_at, display_end_at, collected_at) "
                 "VALUES (:i,:r,:ct,:src,:ext,:t,cast(:b as jsonb),:st,:ds,:de,now())"),
            {"i": uuid.uuid4(), "r": region_code, "ct": content_type.value, "src": LocalSource.TOUR_API.value,
             "ext": uuid.uuid4().hex, "t": title, "b": '{"title":"%s"}' % title, "st": status.value,
             "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, company_id):
    """검증: 검수대기·종료·발행 지역 정보를 섞어 넣는다.
    기대결과: ★ PUBLISHED 만 담긴다 — 운영자가 검수하지 않은 외부 API 원문이 사이트로 새면
    '미검증 값 노출 금지'가 깨진다(fact 를 VERIFIED 로 거르는 것과 같은 규칙)."""
    from common.enums import LocalContentStatus, LocalContentType

    pid = await _seed(db_engine, company_id)
    await _local(db_engine, "4113500", LocalContentType.FESTIVAL, LocalContentStatus.PUBLISHED, "발행축제")
    await _local(db_engine, "4113500", LocalContentType.FESTIVAL, LocalContentStatus.REVIEW, "검수대기축제")
    await _local(db_engine, "4113500", LocalContentType.FESTIVAL, 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, company_id):
    """검증: 발행됐지만 노출 기간을 벗어난 지역 정보.
    기대결과: 빠진다 — 끝난 축제를 '이번 주말 행사'로 걸어두는 것도 틀린 정보다."""
    from datetime import datetime, timedelta, timezone

    from common.enums import LocalContentStatus, LocalContentType

    now = datetime.now(timezone.utc)
    pid = await _seed(db_engine, company_id)
    await _local(db_engine, "4113500", LocalContentType.FESTIVAL, LocalContentStatus.PUBLISHED, "지금축제")
    await _local(db_engine, "4113500", LocalContentType.FESTIVAL, LocalContentStatus.PUBLISHED, "끝난축제",
                 display_end_at=now - timedelta(days=1))
    await _local(db_engine, "4113500", LocalContentType.FESTIVAL, 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, company_id):
    """검증: 지역 캐시는 region_code 로 묶인다.
    기대결과: 다른 지역의 발행 콘텐츠는 담기지 않는다."""
    from common.enums import LocalContentStatus, LocalContentType

    pid = await _seed(db_engine, company_id)
    await _local(db_engine, "4113500", LocalContentType.FESTIVAL, LocalContentStatus.PUBLISHED, "우리지역축제")
    await _local(db_engine, "5011025", LocalContentType.FESTIVAL, LocalContentStatus.PUBLISHED, "남의지역축제")

    snap = await build_snapshot(_RegionPlace(pid, "4113500"))
    assert [c["title"] for c in snap["local"]["contents"]] == ["우리지역축제"]


async def test_place_without_region_code_gets_no_local_content(db_engine, company_id):
    """검증: region_code 가 비어 있는 사업장(수집이 지역을 특정하지 못한 경우).
    기대결과: 빈 목록 — 조회할 캐시 키가 없다. 지어내지 않는다."""
    from common.enums import LocalContentStatus, LocalContentType

    pid = await _seed(db_engine, company_id)
    await _local(db_engine, "4113500", LocalContentType.FESTIVAL, LocalContentStatus.PUBLISHED, "어딘가축제")

    snap = await build_snapshot(_RegionPlace(pid, ""))
    assert snap["local"] == {"region_code": None, "contents": []}