498 lines
20 KiB
Python
498 lines
20 KiB
Python
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=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")
|
|
|
|
|
|
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, "해제해도 토큰이 남아 있다"
|