import json import uuid 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=8")) ).scalar_one() == 1 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=9")) ).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): async def forbidden(*args, **kwargs): raise AssertionError("paid call") monkeypatch.setattr(gemini_text, "call", 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): monkeypatch.setattr(gemini_text, "is_configured", lambda: True) bodies = iter(["가" * 501, "체크인은 15:00입니다."]) async def call(*args, **kwargs): return { "candidates": [ { "content": { "parts": [ { "text": json.dumps( { "body": next(bodies), "fact_keys": ["check_in_time"], } ) } ] } } ] } monkeypatch.setattr(gemini_text, "call", call) 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=9")) ).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")