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