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

"""미니 블로그 — 문구 필터와 메일 승인.

★ 이 파일이 지키는 것:
   - 확인되지 않은 주장(요금·시간·인원·전화)이 든 문구는 화면에 못 간다
   - 같은 링크를 두 번 눌러도 두 번 올라가지 않는다
   - 만료된 링크는 오류가 아니라 안내로 끝난다

★ 2026-09-17 이후 GET 이 곧 승인이다(사장님 지시: "승인은 바로 승인 되게 그 링크만
  클릭하면"). 메일 클라이언트의 링크 미리 열기에 그대로 노출된다는 걸 알고도 택한 것 —
  예전처럼 "GET 은 확인 화면만" 을 기대하는 테스트를 두지 않는다.
"""
import uuid
from datetime import timezone

from sqlalchemy import text

from common.enums import JobType, PlaceCategory, PostStatus, PostTopicKind, SiteStatus
from services import blog_service, site_payload

BODY = (
    "비가 한 차례 지나간 뒤 마당 돌이 검게 젖었습니다. 이런 날에는 대청마루에 앉아 빗소리만 들어도 "
    "하루가 지나갑니다. 우산은 현관에 넉넉히 두었으니 편하게 다녀오세요. 젖은 길은 미끄러우니 "
    "편한 신발을 권합니다. 마당 평상은 비가 그치면 금세 마릅니다."
)


def test_filter_rejects_unverifiable_claims():
    for bad, word in [
        (BODY[:120] + " 주중 198,000원입니다.", "198,000원"),
        (BODY[:120] + " 입실은 15:00 입니다.", "15:00"),
        (BODY[:120] + " 문의는 010-1234-5678 로.", "010-1234-5678"),
        (BODY[:120] + " 군산 최고의 숙소입니다.", "최고"),
    ]:
        ok, reason = blog_service.is_publishable_body(bad)
        assert ok is False
        assert word in reason


def test_filter_accepts_a_plain_scene():
    ok, reason = blog_service.is_publishable_body(BODY)
    assert ok is True, reason


def test_filter_drops_same_topic_twice_in_one_batch():
    rows = [
        {"body": BODY, "topic_key": "festival:야행", "topic_kind": PostTopicKind.FESTIVAL.value},
        {"body": BODY, "topic_key": "festival:야행", "topic_kind": PostTopicKind.FESTIVAL.value},
    ]
    kept, dropped = blog_service.filter_drafts(rows)
    assert len(kept) == 1
    assert "주제 중복" in dropped[0][1]


def test_build_prompt_falls_back_to_stay_wording_for_unmapped_category():
    """업종별로 분기할 수 있는 구조만 갖춘다 — 숙소 외 업종은 아직 실구현 전이라 숙소 문구로 대체한다."""
    stay = blog_service.build_prompt(
        place_name="테스트펜션", region="군산", topic_kind=PostTopicKind.WEATHER.value,
        material="맑음인 날", used_topics=[], place_category=PlaceCategory.LODGING.value,
    )
    restaurant = blog_service.build_prompt(
        place_name="테스트식당", region="군산", topic_kind=PostTopicKind.WEATHER.value,
        material="맑음인 날", used_topics=[], place_category=PlaceCategory.RESTAURANT.value,
    )
    assert "숙소" in stay
    assert "숙소" in restaurant  # 음식점 전용 문구가 아직 없어 숙소 문구로 폴백한다


async def test_generate_one_forwards_place_category_into_the_prompt(monkeypatch):
    """generate_one 은 place_category 를 build_prompt 로 그대로 넘긴다 — 링크는 여기서
    붙이지 않는다(호출부의 길이 게이트가 이 반환값에 그대로 걸리기 때문에, 붙이면
    게이트가 링크까지 세어 정상 문구도 버려진다). 공급자는 provider.active() 를 통해
    고르므로(Gemini 로 고정하지 않는다) 가짜 공급자로 검증한다."""
    from services.llm import provider
    from services.llm.types import LlmResult, Usage

    captured = {}

    class FakeLlm:
        DEFAULT_MODEL = "fake-model"

        @staticmethod
        def is_configured():
            return True

        @staticmethod
        async def generate(client, model, *, prompt, temperature=0.2, **kwargs):
            captured["prompt"] = prompt
            return LlmResult(json=None, text=BODY, usage=Usage())

    monkeypatch.setattr(provider, "active", lambda: FakeLlm)

    result = await blog_service.generate_one(
        place_name="테스트식당", region="군산", topic_kind=PostTopicKind.WEATHER.value,
        material="맑음인 날", used_topics=[], place_category=PlaceCategory.RESTAURANT.value,
    )

    assert result == (BODY, "fake-model")
    assert "숙소" in captured["prompt"]  # 음식점 전용 문구가 아직 없어 숙소 문구로 폴백한다


async def _seed_post(db_engine, owner_id, *, status=PostStatus.SENT, expires_days=14):
    place_id, post_id = uuid.uuid4(), uuid.uuid4()
    token, token_hash, expires = blog_service.issue_token()
    if expires_days < 0:
        from common.utils.gtime import GTime
        expires = GTime.AddDays(expires_days)
    # ★ 함정: 이 시각은 naive-UTC 다. raw text() 로 바인딩하면 asyncpg 가 컬럼 타입 정보 없이
    #   드라이버 기본값(로컬 시스템 시간대)으로 해석해 KST(UTC+9) 서버에서는 9시간 밀린다 —
    #   14일짜리 옛 만료값에서는 안 드러났는데(그래도 미래), 자정 단위(issue_token 2026-09-17
    #   개편)로 정밀해지자 자정 만료가 "9시간 전에 이미 만료"로 뒤집혔다. ORM 경로(update()/
    #   insert())는 컬럼의 DateTime(timezone=True) 프로세서를 타서 이 문제가 없다 — raw text()
    #   로 timestamptz 컬럼에 naive datetime 을 바인딩할 때만 명시적으로 UTC 를 달아야 한다.
    aware_expires = expires.replace(tzinfo=timezone.utc)
    async with db_engine.begin() as conn:
        await conn.execute(
            text("INSERT INTO places (place_id, owner_user_id, name, category, status) "
                 "VALUES (:pid, :uid, :name, :cat, 1)"),
            {"pid": place_id, "uid": owner_id, "name": "스테이,머뭄", "cat": PlaceCategory.LODGING.value},
        )
        await conn.execute(
            text("INSERT INTO place_posts (post_id, place_id, body, topic_kind, topic_key, status, "
                 "approve_token_hash, token_expires_at) "
                 "VALUES (:id, :pid, :body, :kind, :key, :st, :hash, :exp)"),
            {"id": post_id, "pid": place_id, "body": BODY, "kind": PostTopicKind.WEATHER.value,
             "key": f"weather:{post_id.hex[:6]}", "st": status.value, "hash": token_hash, "exp": aware_expires},
        )
    return token, post_id


async def _status(db_engine, post_id) -> int:
    async with db_engine.begin() as conn:
        row = await conn.execute(text("SELECT status FROM place_posts WHERE post_id = :id"), {"id": post_id})
        return row.scalar()


async def test_get_approves_immediately(client, db_engine, owner_id):
    token, post_id = await _seed_post(db_engine, owner_id)

    res = await client.get(f"/v1/site/post/approve?t={token}")

    assert res.status_code == 200
    assert "올렸습니다" in res.text
    assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
    # 발행된 사이트가 없으면(이 테스트) 자동 이동 없이 확인 문구만 보여준다.
    assert "meta http-equiv=\"refresh\"" not in res.text


async def test_approve_page_redirects_to_the_published_blog_section(client, db_engine, owner_id):
    """승인 확인 화면은 5초 뒤 그 업장의 발행된 사이트(미니 블로그 자리)로 자동 이동한다
    (2026-09-22, 사장님 지시: "이메일 승인후에 블로그가 올라가는 페이지에 5초 정도
    이후에 연결되도록"). 자동 이동을 못 믿어도 되게 같은 주소를 링크로도 남긴다."""
    token, post_id = await _seed_post(db_engine, owner_id)
    async with db_engine.begin() as conn:
        place_id = (
            await conn.execute(text("SELECT place_id FROM place_posts WHERE post_id=:p"), {"p": post_id})
        ).scalar_one()
        await conn.execute(
            text(
                "INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"
            ),
            {"sid": uuid.uuid4(), "pid": place_id, "dom": "blog-redirect-test", "st": SiteStatus.PUBLISHED.value},
        )

    res = await client.get(f"/v1/site/post/approve?t={token}")

    assert res.status_code == 200
    assert "올렸습니다" in res.text
    expected_url = site_payload.publish_origin() + "/s/blog-redirect-test#blog"
    assert f'<meta http-equiv="refresh" content="5;url={expected_url}">' in res.text
    assert f"href='{expected_url}'" in res.text


async def test_approve_link_is_single_use(client, db_engine, owner_id):
    token, post_id = await _seed_post(db_engine, owner_id)

    first = await client.get(f"/v1/site/post/approve?t={token}")
    second = await client.get(f"/v1/site/post/approve?t={token}")

    assert "올렸습니다" in first.text
    assert "처리할 수 없는 링크" in second.text
    assert await _status(db_engine, post_id) == PostStatus.APPROVED.value


async def test_expired_link_is_an_answer_not_an_error(client, db_engine, owner_id):
    token, post_id = await _seed_post(db_engine, owner_id, expires_days=-1)

    res = await client.get(f"/v1/site/post/approve?t={token}")

    assert res.status_code == 200
    assert "처리할 수 없는 링크" in res.text
    assert await _status(db_engine, post_id) == PostStatus.SENT.value


async def test_approve_enqueues_build_with_owner_user_id(client, db_engine, owner_id):
    """회귀: BUILD 잡 payload 에 owner_user_id 가 없으면 build_service.run_build 가
    payload["owner_user_id"] 에서 KeyError 로 죽는다 — 승인 클릭이 실제로는 사이트를
    재발행하지 않는 채로 '올렸습니다' 만 보여주고 있었다."""
    token, post_id = await _seed_post(db_engine, owner_id)

    res = await client.get(f"/v1/site/post/approve?t={token}")
    assert "올렸습니다" in res.text

    async with db_engine.begin() as conn:
        row = await conn.execute(
            text("SELECT payload FROM jobs WHERE job_type = :jt ORDER BY created_at DESC LIMIT 1"),
            {"jt": JobType.BUILD.value},
        )
        payload = row.scalar()
    assert payload["owner_user_id"] == owner_id