tests/test_tiktok_publishing.py

"""Phase 4B TikTok Direct Post coverage using only mocked official endpoints."""

from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import AsyncMock
from urllib.parse import parse_qs, urlparse
from uuid import uuid4

import httpx
import pytest
from pydantic import ValidationError

from app.container import build_container
from app.core.config import Settings
from app.social.domain.errors import (
    SocialAccountNotFoundError,
    SocialCapabilityUnsupportedError,
    SocialIdempotencyConflictError,
    SocialMediaInvalidError,
    SocialPermissionDeniedError,
    SocialPostNotFoundError,
    SocialProviderUnavailableError,
    SocialPublishFailedError,
    SocialRateLimitedError,
    SocialReauthRequiredError,
)
from app.social.models import SocialAccount, SocialMediaAsset
from app.social.providers.tiktok import TikTokProvider
from app.social.schemas.posts import SocialPostCreate
from app.social.schemas.tiktok import TikTokPostMetadata
from app.social.workers.publisher import SocialPublisher


def publishing_settings(tmp_path: Path, **overrides: object) -> Settings:
    values: dict[str, object] = {
        "_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",
        "tiktok_client_key": "tiktok-client-key",
        "tiktok_client_secret": "tiktok-client-secret",
        "tiktok_redirect_uri": (
            "https://api.example.com/v1/social/accounts/tiktok/callback"
        ),
        "tiktok_direct_post_enabled": True,
        "tiktok_upload_chunk_bytes": 5_000_000,
        "temp_dir": tmp_path / "temp",
        "output_dir": tmp_path / "outputs",
        "cleanup_interval_seconds": 3600,
        "whisper_model": "tiny",
    }
    values.update(overrides)
    return Settings(**values)


def valid_probe(*, duration: float = 15.0) -> dict[str, object]:
    return {
        "container": "mov,mp4,m4a,3gp,3g2,mj2",
        "duration": duration,
        "fps": 30.0,
        "resolution": {"width": 1080, "height": 1920},
        "video_streams": [{"codec": "h264"}],
        "audio_streams": [{"codec": "aac"}],
    }


def valid_metadata(**overrides: object) -> dict[str, object]:
    values: dict[str, object] = {
        "title": "A production-safe TikTok post",
        "privacy_level": "SELF_ONLY",
        "disable_comment": False,
        "disable_duet": False,
        "disable_stitch": False,
        "brand_content_toggle": False,
        "brand_organic_toggle": False,
        "is_aigc": False,
        "music_usage_confirmed": True,
    }
    values.update(overrides)
    return values


async def test_direct_post_capabilities_are_fail_closed_and_approval_gated(
    tmp_path: Path,
) -> None:
    disabled = TikTokProvider(
        publishing_settings(tmp_path, tiktok_direct_post_enabled=False)
    )
    enabled = TikTokProvider(publishing_settings(tmp_path))
    try:
        assert not disabled.capabilities.direct_publish
        assert not disabled.capabilities.video_upload
        assert disabled.capabilities.publishing_required_scopes == []
        assert enabled.capabilities.direct_publish
        assert enabled.capabilities.video_upload
        assert enabled.capabilities.video_status
        assert enabled.capabilities.scheduled_publish
        assert not enabled.capabilities.native_scheduling
        assert not enabled.capabilities.delete_post
        assert enabled.capabilities.publishing_required_scopes == ["video.publish"]
        with pytest.raises(SocialCapabilityUnsupportedError):
            await enabled.delete_post({"access_token": "access-token"}, "post-id")
    finally:
        await disabled.close()
        await enabled.close()


async def test_publishing_oauth_scope_is_requested_only_by_explicit_elevation(
    tmp_path: Path,
) -> None:
    provider = TikTokProvider(publishing_settings(tmp_path))
    try:
        connection_url = await provider.get_authorization_url(
            state="s" * 43,
            redirect_uri=provider.redirect_uri,
        )
        publishing_url = await provider.get_authorization_url(
            state="s" * 43,
            redirect_uri=provider.redirect_uri,
            additional_scopes=["video.publish"],
        )
    finally:
        await provider.close()
    assert parse_qs(urlparse(connection_url).query)["scope"] == ["user.info.basic"]
    assert parse_qs(urlparse(publishing_url).query)["scope"] == [
        "user.info.basic,video.publish"
    ]


async def test_direct_post_queries_creator_initializes_streams_and_reconciles(
    tmp_path: Path,
) -> None:
    video = tmp_path / "video.mp4"
    video.write_bytes(b"streamed-tiktok-video")
    calls: list[str] = []
    persisted: list[dict[str, object]] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        calls.append(f"{request.method} {request.url.path}")
        if request.url.path.endswith("/creator_info/query/"):
            assert request.headers["authorization"] == "Bearer access-token"
            return httpx.Response(200, json={
                "data": {
                    "privacy_level_options": ["SELF_ONLY", "PUBLIC_TO_EVERYONE"],
                    "comment_disabled": False,
                    "duet_disabled": False,
                    "stitch_disabled": False,
                    "max_video_post_duration_sec": 300,
                },
                "error": {"code": "ok", "message": ""},
            })
        if request.url.path.endswith("/video/init/"):
            payload = json.loads(request.content)
            assert payload["source_info"] == {
                "source": "FILE_UPLOAD",
                "video_size": video.stat().st_size,
                "chunk_size": video.stat().st_size,
                "total_chunk_count": 1,
            }
            assert payload["post_info"]["privacy_level"] == "SELF_ONLY"
            assert "music_usage_confirmed" not in payload["post_info"]
            return httpx.Response(200, json={
                "data": {
                    "publish_id": "publish-id",
                    "upload_url": "https://open-upload.tiktokapis.com/video/session",
                },
                "error": {"code": "ok", "message": ""},
            })
        if request.method == "PUT":
            assert request.headers["content-range"] == (
                f"bytes 0-{video.stat().st_size - 1}/{video.stat().st_size}"
            )
            assert request.content == video.read_bytes()
            return httpx.Response(201)
        if request.url.path.endswith("/status/fetch/"):
            return httpx.Response(200, json={
                "data": {
                    "status": "PUBLISH_COMPLETE",
                    "publicaly_available_post_id": ["public-video-id"],
                    "uploaded_bytes": video.stat().st_size,
                },
                "error": {"code": "ok", "message": ""},
            })
        raise AssertionError(f"Unexpected request {request.method} {request.url}")

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = TikTokProvider(publishing_settings(tmp_path), http_client=client)

    async def persist(value: dict[str, object]) -> None:
        persisted.append(dict(value))

    try:
        uploaded = await provider.upload_media(
            {"access_token": "access-token"},
            {
                "path": video,
                "mime_type": "video/mp4",
                "file_size": video.stat().st_size,
                "probe": valid_probe(),
                "tiktok_post_info": TikTokPostMetadata.model_validate(
                    valid_metadata()
                ).to_post_info(),
                "persist_provider_state": persist,
            },
        )
        published = await provider.publish(
            {"access_token": "access-token"}, {"upload": uploaded}
        )
        status = await provider.get_publish_status(
            {"access_token": "access-token"}, "publish-id"
        )
    finally:
        await client.aclose()

    assert uploaded == {"id": "publish-id"}
    assert published == uploaded
    assert status["status"] == "published"
    assert status["metadata"]["public_post_ids"] == ["public-video-id"]
    assert persisted[0] == {
        "tiktok_init_started": True,
        "tiktok_video_size": video.stat().st_size,
    }
    assert persisted[-1]["tiktok_uploaded_bytes"] == video.stat().st_size
    assert calls == [
        "POST /v2/post/publish/creator_info/query/",
        "POST /v2/post/publish/video/init/",
        "PUT /video/session",
        "POST /v2/post/publish/status/fetch/",
    ]


async def test_media_validation_rejects_incompatible_video_before_provider_call(
    tmp_path: Path,
) -> None:
    video = tmp_path / "video.avi"
    video.write_bytes(b"invalid")
    client = httpx.AsyncClient(
        transport=httpx.MockTransport(
            lambda request: pytest.fail(f"Unexpected provider call {request.url}")
        )
    )
    provider = TikTokProvider(publishing_settings(tmp_path), http_client=client)
    try:
        with pytest.raises(SocialMediaInvalidError):
            await provider.validate_media({
                "path": video,
                "mime_type": "video/x-msvideo",
                "file_size": video.stat().st_size,
                "probe": {
                    **valid_probe(),
                    "container": "avi",
                    "video_streams": [{"codec": "mpeg4"}],
                },
            })
    finally:
        await client.aclose()


@pytest.mark.parametrize(
    ("provider_status", "expected"),
    [
        ("PROCESSING_UPLOAD", "processing"),
        ("PROCESSING_DOWNLOAD", "processing"),
        ("SEND_TO_USER_INBOX", "processing"),
        ("PUBLISH_COMPLETE", "published"),
        ("FAILED", "failed"),
        ("UNKNOWN_PROVIDER_STATE", "unavailable"),
    ],
)
async def test_tiktok_status_reconciliation_normalizes_official_states(
    tmp_path: Path, provider_status: str, expected: str
) -> None:
    async def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={
            "data": {"status": provider_status, "fail_reason": "internal"},
            "error": {"code": "ok"},
        })

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = TikTokProvider(publishing_settings(tmp_path), http_client=client)
    try:
        result = await provider.get_publish_status(
            {"access_token": "access-token"}, "publish-id"
        )
    finally:
        await client.aclose()
    assert result["status"] == expected


async def test_tiktok_metadata_and_chunk_planning_enforce_current_contract(
    tmp_path: Path,
) -> None:
    with pytest.raises(ValidationError):
        TikTokPostMetadata.model_validate(
            valid_metadata(music_usage_confirmed=False)
        )
    with pytest.raises(ValidationError):
        TikTokPostMetadata.model_validate(
            valid_metadata(title="\U0001f600" * 1101)
        )
    provider = TikTokProvider(publishing_settings(tmp_path))
    try:
        assert provider._chunk_plan(4_000_000) == (4_000_000, 1)
        assert provider._chunk_plan(70_000_000) == (5_000_000, 14)
    finally:
        await provider.close()


async def test_unknown_init_outcome_never_creates_a_second_tiktok_post(
    tmp_path: Path,
) -> None:
    video = tmp_path / "video.mp4"
    video.write_bytes(b"video")
    init_calls = 0
    durable_state: dict[str, object] = {}

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal init_calls
        if request.url.path.endswith("/creator_info/query/"):
            return httpx.Response(200, json={
                "data": {
                    "privacy_level_options": ["SELF_ONLY"],
                    "comment_disabled": False,
                    "duet_disabled": False,
                    "stitch_disabled": False,
                    "max_video_post_duration_sec": 300,
                },
                "error": {"code": "ok"},
            })
        if request.url.path.endswith("/video/init/"):
            init_calls += 1
            return httpx.Response(200, json={
                "data": {
                    "publish_id": "accepted-but-not-durable",
                    "upload_url": "https://open-upload.tiktokapis.com/video/session",
                },
                "error": {"code": "ok"},
            })
        raise AssertionError("No upload is safe after provider-state persistence fails")

    async def fail_after_marker(value: dict[str, object]) -> None:
        if "tiktok_publish_id" in value:
            raise RuntimeError("simulated database outage")
        durable_state.clear()
        durable_state.update(value)

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = TikTokProvider(publishing_settings(tmp_path), http_client=client)
    media = {
        "path": video,
        "mime_type": "video/mp4",
        "file_size": video.stat().st_size,
        "probe": valid_probe(),
        "tiktok_post_info": TikTokPostMetadata.model_validate(
            valid_metadata()
        ).to_post_info(),
        "persist_provider_state": fail_after_marker,
    }
    try:
        with pytest.raises(RuntimeError):
            await provider.upload_media(
                {"access_token": "access-token"}, media
            )
        with pytest.raises(SocialPublishFailedError) as raised:
            await provider.upload_media(
                {"access_token": "access-token"},
                {**media, "provider_state": durable_state},
            )
    finally:
        await client.aclose()
    assert init_calls == 1
    assert "duplicate publishing was prevented" in str(raised.value)


@pytest.mark.parametrize(
    ("status_code", "error_code", "exception_type"),
    [
        (401, "access_token_expired", SocialReauthRequiredError),
        (403, "scope_not_authorized", SocialPermissionDeniedError),
        (429, "rate_limit_exceeded", SocialRateLimitedError),
        (500, "internal_error", SocialProviderUnavailableError),
        (400, "invalid_file_upload", SocialMediaInvalidError),
    ],
)
async def test_tiktok_error_normalization_is_safe_and_retry_classifiable(
    tmp_path: Path,
    status_code: int,
    error_code: str,
    exception_type: type[Exception],
) -> None:
    secret = "token-that-must-not-leak"
    client = httpx.AsyncClient(transport=httpx.MockTransport(
        lambda request: httpx.Response(
            status_code,
            json={"error": {"code": error_code, "message": secret}},
        )
    ))
    provider = TikTokProvider(publishing_settings(tmp_path), http_client=client)
    try:
        with pytest.raises(exception_type) as raised:
            await provider.get_publish_options({"access_token": secret})
    finally:
        await client.aclose()
    assert secret not in str(raised.value)


async def test_tiktok_worker_lifecycle_idempotency_scope_and_workspace_isolation(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    container = build_container(publishing_settings(tmp_path))
    await container.social.initialize()
    workspace = "workspace-tiktok"
    other_workspace = "workspace-other"
    request_id = str(uuid4())
    output = container.settings.output_dir / request_id
    output.mkdir(parents=True, exist_ok=True)
    video = output / "video.mp4"
    video.write_bytes(b"video")
    account = await container.social.accounts.repository.create(SocialAccount(
        workspace_id=workspace,
        provider="tiktok",
        account_type="creator",
        external_account_id="creator-open-id",
        display_name="Creator",
        status="connected",
    ))
    asset = await container.social.media_assets.repository.create(SocialMediaAsset(
        workspace_id=workspace,
        request_id=request_id,
        filename=video.name,
        mime_type="video/mp4",
        file_size=video.stat().st_size,
        metadata_json=valid_probe(),
    ))
    await container.social.accounts.tokens.store(
        workspace,
        account.id,
        {"access_token": "encrypted-token", "refresh_token": "encrypted-refresh"},
        scopes=["user.info.basic", "video.publish"],
    )
    adapter = container.social.accounts.providers.get("tiktok")
    statuses = iter([
        {"id": "publish-id", "status": "processing", "metadata": {"provider_status": "PROCESSING_UPLOAD"}},
        {"id": "publish-id", "status": "published", "metadata": {"provider_status": "PUBLISH_COMPLETE"}},
    ])
    monkeypatch.setattr(adapter, "validate_media", AsyncMock(return_value=None))
    monkeypatch.setattr(adapter, "upload_media", AsyncMock(return_value={"id": "publish-id"}))
    monkeypatch.setattr(adapter, "publish", AsyncMock(return_value={"id": "publish-id"}))
    monkeypatch.setattr(adapter, "get_publish_status", AsyncMock(side_effect=lambda *_: next(statuses)))

    async def resolve(*_: object, **__: object) -> dict[str, object]:
        return {
            "path": video,
            "mime_type": "video/mp4",
            "file_size": video.stat().st_size,
            "probe": valid_probe(),
        }

    monkeypatch.setattr(container.social.media_assets, "resolve_for_publish", resolve)
    payload = SocialPostCreate.model_validate({
        "media_asset_id": asset.id,
        "publish_mode": "now",
        "targets": [{
            "social_account_id": account.id,
            "caption": {"caption": "TikTok caption"},
            "tiktok": valid_metadata(),
        }],
    })
    try:
        post = await container.social.publishing.create(
            workspace_id=workspace,
            user_id="user",
            payload=payload,
            idempotency_key="one-logical-publish",
        )
        replay = await container.social.publishing.create(
            workspace_id=workspace,
            user_id="user",
            payload=payload,
            idempotency_key="one-logical-publish",
        )
        assert replay.id == post.id
        with pytest.raises(SocialIdempotencyConflictError):
            await container.social.publishing.create(
                workspace_id=workspace,
                user_id="user",
                payload=SocialPostCreate.model_validate({
                    **payload.model_dump(mode="json"),
                    "targets": [{
                        **payload.targets[0].model_dump(mode="json"),
                        "tiktok": valid_metadata(title="different"),
                    }],
                }),
                idempotency_key="one-logical-publish",
            )
        job = (await container.social.jobs.repository.list_for_post(
            workspace, post.id
        ))[0]
        worker = SocialPublisher(container.social)
        await worker.process(workspace, job.id)
        processing = await container.social.jobs.get(workspace, job.id)
        assert processing.status == "publishing"
        await worker.process(workspace, job.id)
        published = await container.social.jobs.get(workspace, job.id)
        assert published.status == "published"
        assert adapter.upload_media.await_count == 1
        with pytest.raises(SocialPostNotFoundError):
            await container.social.publishing.get(other_workspace, post.id)
        with pytest.raises(SocialAccountNotFoundError):
            await container.social.publishing.publish_options(
                other_workspace, account.id
            )
        with pytest.raises(SocialMediaInvalidError):
            await container.social.media_assets.repository.get(
                other_workspace, asset.id
            )
    finally:
        await container.social.close()
        await container.security_database.close()


async def test_tiktok_publish_requires_explicit_video_publish_scope(
    tmp_path: Path,
) -> None:
    container = build_container(publishing_settings(tmp_path))
    await container.social.initialize()
    account = await container.social.accounts.repository.create(SocialAccount(
        workspace_id="workspace",
        provider="tiktok",
        account_type="creator",
        external_account_id="open-id",
        status="connected",
    ))
    await container.social.accounts.tokens.store(
        "workspace",
        account.id,
        {"access_token": "foundation-only"},
        scopes=["user.info.basic"],
    )
    try:
        with pytest.raises(SocialPermissionDeniedError):
            await container.social.publishing.publish_options(
                "workspace", account.id
            )
    finally:
        await container.social.close()
        await container.security_database.close()