"""Phase 4A TikTok Login Kit foundation coverage. All provider traffic uses MockTransport. Normal CI never needs TikTok credentials or an interactive browser consent flow. """ from __future__ import annotations from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import parse_qs, urlparse import httpx import pytest from sqlalchemy import select from app.container import build_container from app.core.config import Settings from app.social.domain.errors import ( SocialAccountNotFoundError, SocialOAuthStateError, SocialPermissionDeniedError, SocialReauthRequiredError, ) from app.social.models import OAuthState, SocialAccountToken from app.social.providers.tiktok import TikTokProvider from app.social.schemas.accounts import SocialAccountConnectRequest def tiktok_settings(tmp_path: Path) -> Settings: return Settings( _env_file=None, auth_enabled=False, database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}", social_database_url=f"sqlite+aiosqlite:///{tmp_path / 'social.db'}", social_auto_migrate=True, social_worker_enabled=False, social_oauth_encryption_key="test-only-encryption-material", social_oauth_redirect_base_url="https://api.example.com", tiktok_client_key="tiktok-client-key", tiktok_client_secret="tiktok-client-secret", tiktok_redirect_uri=( "https://api.example.com/v1/social/accounts/tiktok/callback" ), temp_dir=tmp_path / "temp", output_dir=tmp_path / "outputs", cleanup_interval_seconds=3600, whisper_model="tiny", ) async def test_tiktok_web_authorization_uses_minimum_scope_and_no_unsupported_pkce( tmp_path: Path, ) -> None: provider = TikTokProvider(tiktok_settings(tmp_path)) try: url = await provider.get_authorization_url( state="s" * 43, redirect_uri="https://api.example.com/v1/social/accounts/tiktok/callback", code_challenge="challenge-that-web-login-kit-does-not-support", ) finally: await provider.close() parsed = urlparse(url) query = parse_qs(parsed.query) assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == ( "https://www.tiktok.com/v2/auth/authorize/" ) assert query["client_key"] == ["tiktok-client-key"] assert query["response_type"] == ["code"] assert query["scope"] == ["user.info.basic"] assert query["state"] == ["s" * 43] assert "code_challenge" not in query assert "code_challenge_method" not in query async def test_tiktok_exchange_refresh_discovery_and_revoke_use_official_v2_endpoints( tmp_path: Path, ) -> None: calls: list[str] = [] async def handler(request: httpx.Request) -> httpx.Response: calls.append(request.url.path) if request.url.path == "/v2/oauth/token/": form = parse_qs(request.content.decode()) assert form["client_key"] == ["tiktok-client-key"] assert form["client_secret"] == ["tiktok-client-secret"] if form["grant_type"] == ["authorization_code"]: assert form["code"] == ["authorization-code"] assert form["redirect_uri"] == [ "https://api.example.com/v1/social/accounts/tiktok/callback" ] assert "code_verifier" not in form else: assert form["grant_type"] == ["refresh_token"] assert form["refresh_token"] == ["refresh-token"] return httpx.Response( 200, json={ "access_token": "access-token", "refresh_token": "rotated-refresh-token", "expires_in": 86400, "refresh_expires_in": 31536000, "open_id": "open-id", "scope": "user.info.basic", "token_type": "Bearer", }, ) if request.url.path == "/v2/user/info/": assert request.headers["authorization"] == "Bearer access-token" assert parse_qs(request.url.query.decode())["fields"] == [ "open_id,union_id,avatar_url,display_name" ] return httpx.Response( 200, json={ "data": { "user": { "open_id": "open-id", "union_id": "union-id", "display_name": "TikTok Creator", "avatar_url": "https://example.com/avatar.jpg", } }, "error": {"code": "ok", "message": ""}, }, ) assert request.url.path == "/v2/oauth/revoke/" form = parse_qs(request.content.decode()) assert form == { "client_key": ["tiktok-client-key"], "client_secret": ["tiktok-client-secret"], "token": ["access-token"], } return httpx.Response(200) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = TikTokProvider(tiktok_settings(tmp_path), http_client=client) try: token = await provider.exchange_code( code="authorization-code", redirect_uri="https://api.example.com/v1/social/accounts/tiktok/callback", code_verifier="unused-web-verifier", ) account = await provider.get_account(token) refreshed = await provider.refresh_token( {"access_token": "old-access", "refresh_token": "refresh-token"} ) await provider.revoke_token({"access_token": "access-token"}) finally: await client.aclose() assert account == { "external_account_id": "open-id", "account_type": "creator", "username": None, "display_name": "TikTok Creator", "avatar_url": "https://example.com/avatar.jpg", "metadata": { "tiktok_open_id": "open-id", "tiktok_union_id": "union-id", }, } assert refreshed["refresh_token"] == "rotated-refresh-token" assert calls == [ "/v2/oauth/token/", "/v2/user/info/", "/v2/oauth/token/", "/v2/oauth/revoke/", ] async def test_tiktok_invalid_code_is_normalized_without_provider_secret( tmp_path: Path, ) -> None: secret = "authorization-code-that-must-not-leak" async def handler(_: httpx.Request) -> httpx.Response: return httpx.Response( 400, json={ "error": "invalid_grant", "error_description": f"bad code {secret}", "log_id": "provider-log-id", }, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = TikTokProvider(tiktok_settings(tmp_path), http_client=client) try: with pytest.raises(SocialReauthRequiredError) as raised: await provider.exchange_code( code=secret, redirect_uri="https://api.example.com/v1/social/accounts/tiktok/callback", ) finally: await client.aclose() assert secret not in str(raised.value) assert "provider-log-id" not in str(raised.value) async def test_tiktok_oauth_callback_is_single_use_duplicate_safe_and_workspace_bound( tmp_path: Path, ) -> None: settings = tiktok_settings(tmp_path) container = build_container(settings) await container.social.initialize() adapter = container.social.accounts.providers.get("tiktok") assert isinstance(adapter, TikTokProvider) await adapter._client.aclose() async def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/v2/oauth/token/": return httpx.Response( 200, json={ "access_token": "token-that-must-stay-encrypted", "refresh_token": "refresh-that-must-stay-encrypted", "expires_in": 86400, "scope": "user.info.basic", "token_type": "Bearer", }, ) return httpx.Response( 200, json={ "data": { "user": { "open_id": "stable-open-id", "display_name": "Workspace Creator", } }, "error": {"code": "ok", "message": ""}, }, ) adapter._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) adapter._owns_client = True try: first_connect = await container.social.oauth.connect( provider="tiktok", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest(), ) first_state = parse_qs(urlparse(first_connect.authorization_url or "").query)[ "state" ][0] assert "code_challenge" not in parse_qs( urlparse(first_connect.authorization_url or "").query ) first = await container.social.oauth.callback( provider="tiktok", state=first_state, code="first-code" ) with pytest.raises(SocialOAuthStateError): await container.social.oauth.callback( provider="tiktok", state=first_state, code="replayed-code" ) second_connect = await container.social.oauth.connect( provider="tiktok", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest(), ) second_state = parse_qs( urlparse(second_connect.authorization_url or "").query )["state"][0] second = await container.social.oauth.callback( provider="tiktok", state=second_state, code="second-code" ) assert first.id == second.id accounts = await container.social.accounts.list("workspace-a") assert [account.id for account in accounts] == [first.id] assert "token-that-must-stay-encrypted" not in first.model_dump_json() with pytest.raises(SocialAccountNotFoundError): await container.social.accounts.get("workspace-b", first.id) async with container.social.database.session("workspace-a") as session: stored = await session.scalar( select(SocialAccountToken).where( SocialAccountToken.social_account_id == first.id ) ) assert stored is not None assert stored.encrypted_payload assert "token-that-must-stay-encrypted" not in stored.encrypted_payload finally: await container.social.close() await container.security_database.close() async def test_tiktok_state_expiry_provider_binding_and_redirect_validation( tmp_path: Path, ) -> None: container = build_container(tiktok_settings(tmp_path)) await container.social.initialize() try: assert container.social.oauth._redirect_uri("tiktok", None) == ( "https://api.example.com/v1/social/accounts/tiktok/callback" ) with pytest.raises(SocialPermissionDeniedError): container.social.oauth._redirect_uri( "tiktok", "https://attacker.example/v1/social/accounts/tiktok/callback", ) state = await container.social.oauth.states.create( provider="tiktok", workspace_id="workspace-a", user_id="user-a", redirect_uri=settings_redirect(container.settings), ) with pytest.raises(SocialOAuthStateError): await container.social.oauth.states.consume( state=state.state, provider="youtube" ) consumed = await container.social.oauth.states.consume( state=state.state, provider="tiktok" ) assert consumed.workspace_id == "workspace-a" assert consumed.user_id == "user-a" expired = OAuthState( state="expired-tiktok-state-value-that-is-long-enough", provider="tiktok", workspace_id="workspace-a", user_id="user-a", redirect_uri=settings_redirect(container.settings), expires_at=datetime.now(timezone.utc) - timedelta(seconds=1), ) async with container.social.database.session("workspace-a") as session: session.add(expired) await session.commit() with pytest.raises(SocialOAuthStateError): await container.social.oauth.states.consume( state=expired.state, provider="tiktok" ) finally: await container.social.close() await container.security_database.close() def settings_redirect(settings: Settings) -> str: return settings.tiktok_redirect_uri async def test_tiktok_capability_discovery_does_not_advertise_publishing( tmp_path: Path, ) -> None: container = build_container(tiktok_settings(tmp_path)) try: tiktok = container.social.accounts.get_provider("tiktok") assert tiktok.available assert tiktok.configured assert tiktok.capabilities.implementation_status == "implemented" assert tiktok.capabilities.required_scopes == ["user.info.basic"] assert tiktok.capabilities.account_types == ["creator"] assert not tiktok.capabilities.video assert not tiktok.capabilities.video_upload assert not tiktok.capabilities.direct_publish assert not tiktok.capabilities.draft_upload assert not tiktok.capabilities.scheduled_publish assert not tiktok.capabilities.delete_post assert tiktok.capabilities.analytics assert tiktok.capabilities.analytics_required_scopes == ["video.list"] finally: await container.social.close() await container.security_database.close()