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

import json
import uuid
from urllib.parse import parse_qs, urlparse
from datetime import datetime, timedelta, timezone
import httpx
import pytest
from sqlalchemy import text
from crud import social_crud
from services import social_service as service
from services.external import gemini_text, threads
from services.external.social import SocialOutcomeUnknown, weighted_length


async def seed(client, auth_headers, db_engine):
    h = await auth_headers("social-owner")
    pid = uuid.UUID(
        (
            await client.post(
                "/v1/place", headers=h, json={"name": "소식숙소", "category": 1}
            )
        ).json()["place"]["place_id"]
    )
    async with db_engine.begin() as c:
        uid = (
            await c.execute(
                text("SELECT owner_user_id FROM places WHERE place_id=:p"), {"p": pid}
            )
        ).scalar_one()
        version = uuid.uuid4()
        await c.execute(
            text(
                "INSERT INTO sites(site_id,place_id,domain,status,current_version_id) VALUES (:id,:p,'social-stay',3,:v)"
            ),
            {"id": uuid.uuid4(), "p": pid, "v": version},
        )
        await c.execute(
            text(
                "INSERT INTO place_facts(fact_id,place_id,key,value,source_type,status) VALUES (:id,:p,'check_in_time','15:00',1,3)"
            ),
            {"id": uuid.uuid4(), "p": pid},
        )
    return h, pid, uid, version


async def pending(db_engine, pid, uid, version, expired=False):
    post_id, account_id = uuid.uuid4(), uuid.uuid4()
    token = "a" * 43
    async with db_engine.begin() as c:
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked')"
            ),
            {"a": account_id, "u": uid},
        )
        await c.execute(
            text("""INSERT INTO place_social_posts(post_id,place_id,user_id,site_version_id,account_id,provider,body,link_url,status,approval_token_sha,approval_expires_at)
            VALUES (:id,:p,:u,:v,:a,2,'체크인은 15:00입니다.',:url,'PENDING_APPROVAL',:sha,:expires)"""),
            {
                "id": post_id,
                "p": pid,
                "u": uid,
                "v": version,
                "a": account_id,
                "sha": service.sha(token),
                "url": service.site_payload.publish_origin() + "/s/social-stay",
                "expires": datetime.now(timezone.utc)
                + timedelta(hours=-1 if expired else 1),
            },
        )
    return post_id, token


async def test_draft_dedup_owner_scope(client, auth_headers, db_engine):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    first = await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    assert first.status_code == 200, first.text
    second = await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    assert second.json()["post_id"] == first.json()["post_id"]
    assert (
        first.json()["link_url"]
        == service.site_payload.publish_origin() + "/s/social-stay"
    )
    other = await auth_headers("social-other")
    response = await client.get(f"/v1/social/place/{pid}", headers=other)
    assert (
        response.status_code == 404 and response.json()["detail"] == "PLACE_NOT_FOUND"
    )
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM jobs WHERE job_type=9"))
        ).scalar_one() == 1


async def test_social_job_types_dispatch_to_the_right_handler():
    """실측(2026-09-22): 큐에 넣는 쪽(social_crud.decide, social_service.create_draft/
    publish_reused_text)은 job_type 숫자를 하드코딩(8/9)하고, 디스패처(worker/handlers.py)는
    JobType.SOCIAL_DRAFT(9)/SOCIAL_POST(10) enum 값으로 등록돼 있었다. 번호가 어긋나 있어서
    "쓰레드에 게시" 잡이 run_draft 로, "초안 생성" 잡이 run_rollback 으로 잘못 배달됐다 —
    잡은 에러 없이 DONE 으로 끝나지만 아무 일도 안 일어나는 조용한 실패였다. 큐 삽입 값만
    보던 기존 테스트들(job_type=N 카운트)은 그 N 이 실제 핸들러와 맞는지는 확인하지
    않아서 이 어긋남을 못 잡았다. 여기서는 워커가 실제로 쓰는 배달 경로
    (worker.handlers.HANDLERS)가 큐 삽입 쪽이 쓰는 것과 같은 JobType enum 값을 가리키는지
    직접 대조한다 — 숫자가 다시 어긋나면(둘 중 하나가 하드코딩으로 되돌아가면) 여기서 잡힌다."""
    from common.enums import JobType
    from services.social_service import run_draft, run_post
    from worker.handlers import HANDLERS

    assert HANDLERS[JobType.SOCIAL_DRAFT.value] is run_draft
    assert HANDLERS[JobType.SOCIAL_POST.value] is run_post


async def test_draft_and_post_jobs_enqueue_with_dispatchable_job_types(
    client, auth_headers, db_engine, monkeypatch
):
    """create_draft 가 넣는 job_type 이 실제로 run_draft 로, decide 가 넣는 job_type 이
    실제로 run_post 로 배달되는지 엔드투엔드로 확인한다(위 테스트의 정적 대조를 실제
    큐 삽입 값으로 한 번 더 검증). 같은 사업장에 site_version 을 새로 하나 더 발급해
    (place_id, site_version_id) 유니크 인덱스와 안 부딪히게 한다 — seed() 를 두 번 부르면
    도메인('social-stay') 유니크 인덱스와 부딪힌다."""
    from worker.handlers import HANDLERS

    monkeypatch.setattr(service, "posting_enabled", lambda: True)

    h, pid, uid, v1 = await seed(client, auth_headers, db_engine)
    draft_res = await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    assert draft_res.status_code == 200, draft_res.text
    draft_post_id = draft_res.json()["post_id"]

    async with db_engine.begin() as c:
        draft_job_type = (
            await c.execute(
                text("SELECT job_type FROM jobs WHERE dedupe_key LIKE :k"),
                {"k": f"social:%:{draft_post_id}"},
            )
        ).scalar_one()
    assert HANDLERS[draft_job_type] is service.run_draft

    v2 = uuid.uuid4()
    async with db_engine.begin() as c:
        await c.execute(
            text("UPDATE sites SET current_version_id=:v WHERE place_id=:p"),
            {"v": v2, "p": pid},
        )
    post_id, token = await pending(db_engine, pid, uid, v2)
    approve = await client.post(
        f"/v1/social/approval/{post_id}/decision", json={"t": token, "approve": True}
    )
    assert approve.json()["applied"] is True

    async with db_engine.begin() as c:
        post_job_type = (
            await c.execute(
                text("SELECT job_type FROM jobs WHERE dedupe_key LIKE :k"),
                {"k": f"social:%:{post_id}"},
            )
        ).scalar_one()
    assert HANDLERS[post_job_type] is service.run_post


async def test_requires_fixed_domain(client, auth_headers, db_engine):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    async with db_engine.begin() as c:
        await c.execute(
            text("UPDATE sites SET domain=NULL WHERE place_id=:p"), {"p": pid}
        )
    assert (
        await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    ).status_code == 409


async def test_prefetch_read_only_one_time_cas(client, auth_headers, db_engine):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    post_id, token = await pending(db_engine, pid, uid, v)
    for _ in range(2):
        res = await client.get(f"/v1/social/approval/{post_id}?t={token}")
        assert res.status_code == 200 and res.json()["status"] == "PENDING_APPROVAL"
        assert res.headers["cache-control"] == "no-store"
        assert "approval_token_sha" not in res.text
    first = await client.post(
        f"/v1/social/approval/{post_id}/decision", json={"t": token, "approve": True}
    )
    second = await client.post(
        f"/v1/social/approval/{post_id}/decision", json={"t": token, "approve": True}
    )
    assert first.json()["applied"] is True, first.text
    assert second.json()["applied"] is False
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM jobs WHERE job_type=10"))
        ).scalar_one() == 1


async def test_expired_and_stale_sweep(client, auth_headers, db_engine):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    post_id, token = await pending(db_engine, pid, uid, v, expired=True)
    res = await client.post(
        f"/v1/social/approval/{post_id}/decision", json={"t": token, "approve": False}
    )
    assert res.json()["applied"] is False
    await social_crud.sweep()
    async with db_engine.begin() as c:
        row = (
            await c.execute(
                text("SELECT status,body FROM place_social_posts WHERE post_id=:p"),
                {"p": post_id},
            )
        ).first()
        assert row.status == "EXPIRED" and row.body
        await c.execute(
            text(
                "UPDATE place_social_posts SET status='POSTING',updated_at=now()-interval '11 minutes' WHERE post_id=:p"
            ),
            {"p": post_id},
        )
    await social_crud.sweep()
    async with db_engine.begin() as c:
        assert (
            await c.execute(
                text("SELECT status FROM place_social_posts WHERE post_id=:p"),
                {"p": post_id},
            )
        ).scalar_one() == "UNKNOWN"


async def test_no_facts_no_paid_call(monkeypatch):
    from services.llm import provider as llm_provider

    def forbidden():
        raise AssertionError("paid call")

    monkeypatch.setattr(llm_provider, "active", forbidden)
    with pytest.raises(gemini_text.GeminiInvalidOutput, match="NO_GROUNDED_FACTS"):
        await gemini_text.generate_social_post("숙소", [], "https://example.com/s/stay")


async def test_long_draft_regenerates(monkeypatch):
    """generate_social_post 는 LLM_PROVIDER 추상화(services/llm/provider.py)를 탄다
    (2026-09-21, Gemini 하드코딩 제거) — 여기서는 provider.active() 가 돌려주는 공급자
    자체를 가짜로 바꿔 길이 초과 → 재요청 → 통과 흐름만 검증한다."""
    from services.llm import provider as llm_provider
    from services.llm.types import LlmResult, Usage

    bodies = iter(["가" * 501, "체크인은 15:00입니다."])

    class _FakeLlm:
        __name__ = "services.llm.fake"
        DEFAULT_MODEL = "fake-model"

        @staticmethod
        def is_configured():
            return True

        @staticmethod
        async def generate(client, model, *, prompt, response_schema=None, temperature=0.2, max_retries=0, images=None):
            body = next(bodies)
            return LlmResult(
                json={"body": body, "fact_keys": ["check_in_time"]},
                text=json.dumps({"body": body, "fact_keys": ["check_in_time"]}),
                usage=Usage(input_tokens=0, output_tokens=0),
            )

    monkeypatch.setattr(llm_provider, "active", lambda: _FakeLlm())
    result = await gemini_text.generate_social_post(
        "숙소",
        [gemini_text.FactInput(key="check_in_time", label="체크인", value="15:00")],
        "https://example.com/s/stay",
    )
    assert result == "체크인은 15:00입니다.\n\nhttps://example.com/s/stay"


def test_lengths():
    assert weighted_length("한글", 1) == 4
    assert weighted_length("https://example.com/" + "a" * 200, 1) == 23
    assert weighted_length("한글", 2) == 2
    assert weighted_length("https://example.com/" + "a" * 200, 2) == 220


async def test_threads_timeout_no_retry():
    calls = []

    def handler(req):
        calls.append(req)
        if req.url.path.endswith("/me/threads"):
            assert (
                b"media_type=TEXT" in req.content
                and b"auto_publish_text=false" in req.content
            )
            assert b"image_url" not in req.content
            return httpx.Response(200, json={"id": "11"})
        raise httpx.ReadTimeout("lost", request=req)

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        with pytest.raises(SocialOutcomeUnknown):
            await threads.publish("소개", "secret", client=client)
    assert len(calls) == 2


async def test_published_permalink_failure():
    def handler(req):
        if req.url.path.endswith("/me/threads"):
            return httpx.Response(200, json={"id": "11"})
        if req.url.path.endswith("/me/threads_publish"):
            return httpx.Response(200, json={"id": "22"})
        return httpx.Response(500, json={"error": {}})

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        result = await threads.publish("소개", "secret", client=client)
    assert result == {"id": "22", "permalink": None}


async def test_screen_approval_without_contract_never_queues_post(
    client, auth_headers, db_engine
):
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    result = await client.post(f"/v1/social/place/{pid}/draft", headers=h, json={})
    post_id = uuid.UUID(result.json()["post_id"])
    async with db_engine.begin() as c:
        await c.execute(
            text(
                "UPDATE place_social_posts SET body='작성된 원고',status='DRAFT' WHERE post_id=:p"
            ),
            {"p": post_id},
        )
    response = await client.post(
        f"/v1/social/posts/{post_id}/request-approval", headers=h, json={}
    )
    assert response.status_code == 200, response.text
    assert response.json()["post"]["account_bound"] is False
    approved = await client.post(
        f"/v1/social/posts/{post_id}/decision", headers=h, json={"approve": True}
    )
    assert approved.json()["applied"] is True
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM jobs WHERE job_type=10"))
        ).scalar_one() == 0


async def test_token_cipher_and_oauth_browser_binding(monkeypatch):
    from cryptography.fernet import Fernet
    from services import social_account_service as accounts
    from services.external.social import SocialError

    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", "")
    assert not accounts.configured()
    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    monkeypatch.setenv("THREADS_APP_ID", "test")
    monkeypatch.setenv("THREADS_APP_SECRET", "test")
    monkeypatch.setenv("THREADS_REDIRECT_URI", "https://example.com/callback")
    encrypted = accounts.encrypt("owner-token")
    assert (
        "owner-token" not in encrypted and accounts.decrypt(encrypted) == "owner-token"
    )
    url, browser = accounts.begin(uuid.uuid4(), 2)
    from urllib.parse import parse_qs, urlparse

    state = parse_qs(urlparse(url).query)["state"][0]
    with pytest.raises(SocialError, match="INVALID_OAUTH_STATE"):
        await accounts.finish(state, "wrong-browser", "unused-code")


@pytest.mark.parametrize("unknown", [False, True])
async def test_post_claim_prevents_second_external_write(
    client, auth_headers, db_engine, monkeypatch, unknown
):
    from cryptography.fernet import Fernet
    from services import social_account_service as accounts

    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    post_id, _ = await pending(db_engine, pid, uid, v)
    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    monkeypatch.setattr(service, "posting_enabled", lambda: True)
    async with db_engine.begin() as c:
        await c.execute(
            text("UPDATE place_social_posts SET status='APPROVED' WHERE post_id=:p"),
            {"p": post_id},
        )
        await c.execute(
            text(
                "UPDATE owner_social_accounts SET access_token=:t,access_expires_at=now()+interval '1 day' WHERE user_id=:u"
            ),
            {"t": accounts.encrypt("secret"), "u": uid},
        )
    calls = []

    class Adapter:
        @staticmethod
        def weighted_limit():
            return 500

        @staticmethod
        async def me(*args, **kwargs):
            return {"id": "22"}

        @staticmethod
        async def publish(*args, **kwargs):
            calls.append(1)
            if unknown:
                raise SocialOutcomeUnknown("POST_RESULT_UNKNOWN")
            return {"id": "33", "permalink": "https://www.threads.com/@host/post/abc"}

    monkeypatch.setattr(service, "adapter", lambda provider: Adapter)
    job = {"payload": {"post_id": str(post_id)}}
    if unknown:
        with pytest.raises(SocialOutcomeUnknown):
            await service.run_post(job)
    else:
        await service.run_post(job)
    assert (await service.run_post(job)) == {"skipped": True}
    assert calls == [1]
    async with db_engine.begin() as c:
        assert (
            await c.execute(
                text("SELECT status FROM place_social_posts WHERE post_id=:p"),
                {"p": post_id},
            )
        ).scalar_one() == ("UNKNOWN" if unknown else "POSTED")


async def test_test_post_requires_connected_account(client, auth_headers):
    h = await auth_headers("social-test-owner")
    res = await client.post("/v1/social/test-post", headers=h, json={"text": "연동 확인"})
    assert (
        res.status_code == 409 and res.json()["detail"] == "ACCOUNT_CONNECTION_REQUIRED"
    )


async def test_test_post_requires_reauth_when_expired(client, auth_headers, db_engine, monkeypatch):
    from cryptography.fernet import Fernet

    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    h = await auth_headers("social-reauth-owner")
    async with db_engine.begin() as c:
        uid = (
            await c.execute(
                text("SELECT user_id FROM users WHERE id=:i"), {"i": "social-reauth-owner"}
            )
        ).scalar_one()
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) "
                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','needs_reauth')"
            ),
            {"a": uuid.uuid4(), "u": uid},
        )
    res = await client.post("/v1/social/test-post", headers=h, json={"text": "연동 확인"})
    assert (
        res.status_code == 409 and res.json()["detail"] == "ACCOUNT_NEEDS_REAUTH"
    )


async def test_test_post_publishes_immediately_bypassing_approval(
    client, auth_headers, db_engine, monkeypatch
):
    """연동 확인용 게시는 posting_enabled 게이트·승인 절차 없이 바로 나간다."""
    from cryptography.fernet import Fernet
    from services import social_account_service as accounts

    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    monkeypatch.setattr(service, "posting_enabled", lambda: False)
    h = await auth_headers("social-verify-owner")
    async with db_engine.begin() as c:
        uid = (
            await c.execute(
                text("SELECT user_id FROM users WHERE id=:i"), {"i": "social-verify-owner"}
            )
        ).scalar_one()
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status,access_token,access_expires_at) "
                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked',:t,now()+interval '1 day')"
            ),
            {"a": uuid.uuid4(), "u": uid, "t": accounts.encrypt("secret-token")},
        )

    calls = []

    class Adapter:
        @staticmethod
        async def publish(text_, token, *, client):
            calls.append((text_, token))
            return {"id": "99", "permalink": "https://www.threads.com/@host/post/99"}

    monkeypatch.setattr(service, "adapter", lambda provider: Adapter)
    res = await client.post(
        "/v1/social/test-post", headers=h, json={"text": "연동 확인용 테스트"}
    )
    assert res.status_code == 200, res.text
    assert res.json() == {"id": "99", "permalink": "https://www.threads.com/@host/post/99"}
    assert calls == [("연동 확인용 테스트", "secret-token")]


async def test_oauth_roundtrip_saves_encrypted_account(db_engine, monkeypatch):
    """검증: 인가 코드를 받아 계정을 연결하고, 해제까지 한 바퀴 돈다.

    ★ 왜 가짜 서버로 미리 도는가 — 실제 연결은 Meta 앱 등록(리디렉션 URI·권한·테스터 추가)이
      끝나야 시험할 수 있다. 그때 실패하면 우리 코드가 틀린 건지 앱 설정이 틀린 건지 구별이
      안 된다. 우리 쪽 왕복(코드 교환 → 장기토큰 → 검증 → 저장 → 해제)은 여기서 먼저 못 박는다.
    ★ 이 검사가 지키는 것 셋:
      1. 저장된 것은 **암호문**이다 — 토큰 원문이 DB 에 남으면 안 된다.
      2. 장기 토큰 교환과 `debug_token` 검증을 건너뛰지 않는다(권한이 모자란 연결을 만들지 않는다).
      3. 해제하면 토큰이 **지워진다** — 행만 남기고 토큰을 두면 지운 줄 알고 계속 쓰게 된다.
    """
    import json as _json
    import uuid as _uuid

    from cryptography.fernet import Fernet
    from sqlalchemy import text as _text

    from services import social_account_service as accounts

    monkeypatch.setenv("SOCIAL_TOKEN_SECRET", Fernet.generate_key().decode())
    monkeypatch.setenv("THREADS_APP_ID", "app-1")
    monkeypatch.setenv("THREADS_APP_SECRET", "secret-1")
    monkeypatch.setenv("THREADS_REDIRECT_URI", "https://example.com/v1/social/oauth/callback")

    long_lived = "long-lived-token"

    def handler(request: httpx.Request) -> httpx.Response:
        path = request.url.path
        if path.endswith("/oauth/access_token"):
            return httpx.Response(200, json={"access_token": "short-token", "user_id": "1"})
        if path.endswith("/access_token"):
            # 장기 토큰 교환. 60일짜리를 준다 — 하루 미만이면 코드가 거절해야 한다.
            return httpx.Response(200, json={"access_token": long_lived, "expires_in": 5184000})
        if path.endswith("/debug_token"):
            return httpx.Response(200, json={"data": {
                "is_valid": True, "app_id": "app-1",
                "scopes": ["threads_basic", "threads_content_publish"]}})
        if path.endswith("/me"):
            return httpx.Response(200, json={
                "id": "th-1", "username": "mumum", "threads_profile_picture_url": ""})
        return httpx.Response(404, json={"error": {"message": "unexpected " + path}})

    transport = httpx.MockTransport(handler)
    original = httpx.AsyncClient

    def fake_client(*args, **kwargs):
        kwargs["transport"] = transport
        return original(*args, **kwargs)

    monkeypatch.setattr(httpx, "AsyncClient", fake_client)

    user_id = _uuid.uuid4()
    url, browser = accounts.begin(user_id, 2)
    state = parse_qs(urlparse(url).query)["state"][0]
    await accounts.finish(state, browser, "auth-code")

    async with db_engine.begin() as conn:
        row = (await conn.execute(_text(
            "SELECT handle, access_token, status, scopes FROM owner_social_accounts "
            "WHERE user_id = :u AND deleted = false"), {"u": user_id})).first()
    assert row is not None, "연결이 저장되지 않았다"
    assert row.handle == "mumum"
    assert row.status == "linked"
    # ★ 원문이 DB 에 있으면 안 된다.
    assert long_lived not in row.access_token
    assert accounts.decrypt(row.access_token) == long_lived
    scopes = row.scopes if isinstance(row.scopes, list) else _json.loads(row.scopes)
    assert set(scopes) == {"threads_basic", "threads_content_publish"}

    await accounts.disconnect(user_id, 2)
    async with db_engine.begin() as conn:
        after = (await conn.execute(_text(
            "SELECT status, access_token FROM owner_social_accounts WHERE user_id = :u"),
            {"u": user_id})).first()
    assert after.status == "revoked" and after.access_token is None, "해제해도 토큰이 남아 있다"


async def test_publish_reused_text_skips_when_no_threads_account(db_engine, auth_headers, client):
    from services import social_service as svc

    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    result = await svc.publish_reused_text(uid, pid, "오늘도 마당이 조용합니다.")
    assert result is None
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM place_social_posts WHERE place_id=:p"), {"p": pid})
        ).scalar_one() == 0


async def test_publish_reused_text_skips_when_posting_disabled(db_engine, auth_headers, client, monkeypatch):
    from services import social_service as svc

    monkeypatch.setattr(svc, "posting_enabled", lambda: False)
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    async with db_engine.begin() as c:
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) "
                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked')"
            ),
            {"a": uuid.uuid4(), "u": uid},
        )
    result = await svc.publish_reused_text(uid, pid, "오늘도 마당이 조용합니다.")
    assert result is None
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM place_social_posts WHERE place_id=:p"), {"p": pid})
        ).scalar_one() == 0


async def test_publish_reused_text_skips_when_domain_not_fixed(db_engine, auth_headers, client, monkeypatch):
    from services import social_service as svc

    monkeypatch.setattr(svc, "posting_enabled", lambda: True)
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    async with db_engine.begin() as c:
        await c.execute(text("UPDATE sites SET domain=NULL WHERE place_id=:p"), {"p": pid})
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) "
                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked')"
            ),
            {"a": uuid.uuid4(), "u": uid},
        )
    result = await svc.publish_reused_text(uid, pid, "오늘도 마당이 조용합니다.")
    assert result is None


async def test_publish_reused_text_inserts_approved_post_with_link_and_enqueues_run_post(
    db_engine, auth_headers, client, monkeypatch
):
    from services import social_service as svc

    monkeypatch.setattr(svc, "posting_enabled", lambda: True)
    h, pid, uid, v = await seed(client, auth_headers, db_engine)
    account_id = uuid.uuid4()
    async with db_engine.begin() as c:
        await c.execute(
            text(
                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) "
                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked')"
            ),
            {"a": account_id, "u": uid},
        )

    result = await svc.publish_reused_text(uid, pid, "오늘도 마당이 조용합니다.")

    assert result is not None
    expected_url = svc.site_payload.publish_origin() + "/s/social-stay"
    async with db_engine.begin() as c:
        row = (
            await c.execute(
                text(
                    "SELECT status, body, link_url, decided_via, account_id FROM place_social_posts WHERE post_id=:p"
                ),
                {"p": result},
            )
        ).first()
        assert row.status == "APPROVED"
        assert row.body == f"오늘도 마당이 조용합니다.\n\n{expected_url}"
        assert row.link_url == expected_url
        assert row.decided_via == "mini_blog"
        assert row.account_id == account_id
        job_count = (
            await c.execute(text("SELECT count(*) FROM jobs WHERE job_type=10"))
        ).scalar_one()
        assert job_count == 1

    # 같은 site_version 에 두 번째 호출 — 유니크 인덱스가 중복 삽입을 막는다
    second = await svc.publish_reused_text(uid, pid, "다른 문구")
    assert second is None
    async with db_engine.begin() as c:
        assert (
            await c.execute(text("SELECT count(*) FROM place_social_posts WHERE place_id=:p"), {"p": pid})
        ).scalar_one() == 1