tests/test_tiktok_foundation.py

"""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()