from __future__ import annotations from datetime import datetime, timedelta, timezone from pathlib import Path from uuid import uuid4 import pytest from pydantic import ValidationError from sqlalchemy import select from starlette.requests import Request from app.container import build_container from app.core.config import Settings from app.security.policy import ScopePolicy from app.social.database import SocialDatabase from app.social.domain.enums import JobStatus from app.social.domain.errors import ( SocialAccountNotFoundError, SocialIdempotencyConflictError, SocialJobNotFoundError, SocialMediaInvalidError, SocialOAuthStateError, SocialPermissionDeniedError, SocialPostNotFoundError, SocialReauthRequiredError, SocialTransitionError, ) from app.social.domain.retry import classify_retry from app.social.domain.state_machine import validate_transition from app.social.models import OAuthState, SocialAccount, SocialAccountToken, SocialJob, SocialMediaAsset from app.social.schemas.posts import SocialPostCreate from app.social.schemas.scheduling import SocialScheduleCreate def social_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", temp_dir=tmp_path / "temp", output_dir=tmp_path / "outputs", cleanup_interval_seconds=3600, whisper_model="tiny", ) @pytest.fixture async def social_container(tmp_path: Path): container = build_container(social_settings(tmp_path)) await container.social.initialize() try: yield container finally: await container.social.close() await container.security_database.close() async def connected_account(container, workspace_id: str, provider: str = "youtube") -> SocialAccount: return await container.social.accounts.repository.create( SocialAccount( workspace_id=workspace_id, provider=provider, account_type="channel", external_account_id=f"external-{workspace_id}-{provider}", display_name="Test channel", status="connected", ) ) async def registered_output_asset(container, workspace_id: str) -> str: request_id = str(uuid4()) output = container.settings.output_dir / request_id output.mkdir(parents=True, exist_ok=True) (output / "video.mp4").write_bytes(b"test") record = await container.social.media_assets.repository.create( SocialMediaAsset( workspace_id=workspace_id, request_id=request_id, filename="video.mp4", mime_type="video/mp4", file_size=4, ) ) return record.id def post_payload(account_id: str, *, title: str = "Example") -> SocialPostCreate: return SocialPostCreate.model_validate( { "media_asset_id": "asset-owned-by-workspace", "publish_mode": "draft", "targets": [ { "social_account_id": account_id, "caption": {"title": title, "description": "Description"}, "youtube": { "title": title, "description": "Description", "privacy_status": "private", "made_for_kids": False, }, } ], } ) async def test_provider_registry_and_capability_matrix(social_container) -> None: providers = social_container.social.accounts.list_providers() assert {item.provider.value for item in providers} == { "youtube", "facebook", "instagram", "tiktok", "x", "linkedin", "telegram", "whatsapp", } assert all(item.available is False for item in providers) youtube = next(item for item in providers if item.provider.value == "youtube") assert youtube.capabilities.implementation_status == "implemented" assert youtube.capabilities.video_upload assert youtube.capabilities.video_status assert youtube.capabilities.channel_metadata assert youtube.configured is False assert next(item for item in providers if item.provider.value == "telegram").connection_strategy.value == "token_bot" assert next(item for item in providers if item.provider.value == "whatsapp").connection_strategy.value == "business_api" def test_job_state_machine_and_retry_classification() -> None: assert validate_transition(JobStatus.DRAFT, JobStatus.SCHEDULED) == JobStatus.SCHEDULED assert validate_transition(JobStatus.RETRYING, JobStatus.PROCESSING) == JobStatus.PROCESSING with pytest.raises(SocialTransitionError): validate_transition(JobStatus.PUBLISHED, JobStatus.QUEUED) assert classify_retry(status_code=429, attempt=3).retryable assert classify_retry(status_code=401, attempt=1).refresh_token_first assert not classify_retry(status_code=403, attempt=1).retryable assert not classify_retry(status_code=400, attempt=1).retryable def test_only_the_exact_provider_callback_route_is_public() -> None: policy = ScopePolicy() def request_for(path: str) -> Request: return Request( {"type": "http", "method": "GET", "path": path, "headers": []} ) assert policy.is_public( request_for("/v1/social/accounts/youtube/callback") ) assert not policy.is_public( request_for("/v1/social/accounts/youtube/untrusted/callback") ) assert not policy.is_public( Request( { "type": "http", "method": "POST", "path": "/v1/social/accounts/youtube/callback", "headers": [], } ) ) async def test_social_auto_migrate_false_does_not_mutate_schema(tmp_path: Path) -> None: settings = social_settings(tmp_path) settings.social_auto_migrate = False database = SocialDatabase(settings) try: await database.initialize() assert not await database.schema_ready() assert "social_accounts" in await database.missing_tables() finally: await database.close() async def test_oauth_state_is_random_expiring_single_use_and_tenant_bound( social_container, ) -> None: state = await social_container.social.oauth.states.create( provider="youtube", workspace_id="workspace-a", user_id="user-a", redirect_uri="https://api.example/v1/social/accounts/youtube/callback", ) assert len(state.state) >= 32 consumed = await social_container.social.oauth.states.consume( state=state.state, provider="youtube" ) assert consumed.workspace_id == "workspace-a" with pytest.raises(SocialOAuthStateError): await social_container.social.oauth.states.consume( state=state.state, provider="youtube" ) expired = OAuthState( state="expired-state", provider="youtube", workspace_id="workspace-a", user_id="user-a", redirect_uri="https://api.example/callback", expires_at=datetime.now(timezone.utc) - timedelta(seconds=1), ) async with social_container.social.database.session() as session: session.add(expired) await session.commit() with pytest.raises(SocialOAuthStateError): await social_container.social.oauth.states.consume( state="expired-state", provider="youtube" ) wrong_provider = await social_container.social.oauth.states.create( provider="youtube", workspace_id="workspace-a", user_id="user-a", redirect_uri="https://api.example/v1/social/accounts/youtube/callback", ) with pytest.raises(SocialOAuthStateError): await social_container.social.oauth.states.consume( state=wrong_provider.state, provider="linkedin" ) assert ( await social_container.social.oauth.states.consume( state=wrong_provider.state, provider="youtube" ) ).workspace_id == "workspace-a" with pytest.raises(SocialOAuthStateError): await social_container.social.oauth.states.consume( state="not-a-valid-state", provider="youtube" ) async def test_oauth_redirect_uri_is_backend_owned(social_container) -> None: oauth = social_container.social.oauth with pytest.raises(SocialPermissionDeniedError): oauth._redirect_uri( "youtube", "https://attacker.example/v1/social/accounts/youtube/callback" ) social_container.settings.social_oauth_redirect_base_url = "https://api.example" expected = "https://api.example/v1/social/accounts/youtube/callback" assert oauth._redirect_uri("youtube", None) == expected with pytest.raises(SocialPermissionDeniedError): oauth._redirect_uri( "youtube", "https://attacker.example/v1/social/accounts/youtube/callback" ) async def test_token_service_encrypts_and_never_returns_storage_metadata( social_container, ) -> None: account = await connected_account(social_container, "workspace-token") secret = "provider-access-token-that-must-not-leak" await social_container.social.accounts.tokens.store( "workspace-token", account.id, {"access_token": secret, "refresh_token": "refresh-secret"}, scopes=["upload"], ) async with social_container.social.database.session() as session: row = await session.scalar( select(SocialAccountToken).where( SocialAccountToken.social_account_id == account.id ) ) assert row is not None assert secret not in (row.encrypted_payload or "") assert await social_container.social.accounts.tokens.retrieve( "workspace-token", account.id ) == { "access_token": secret, "refresh_token": "refresh-secret", } view = await social_container.social.accounts.get("workspace-token", account.id) assert "token" not in view.model_dump_json().lower() with pytest.raises(SocialReauthRequiredError): await social_container.social.accounts.tokens.retrieve("workspace-other", account.id) async def test_workspace_ownership_idempotency_and_multi_target_foundation( social_container, ) -> None: account = await connected_account(social_container, "workspace-a") payload = post_payload(account.id) first = await social_container.social.publishing.create( workspace_id="workspace-a", user_id="user-a", payload=payload, idempotency_key="create-post-key", ) replay = await social_container.social.publishing.create( workspace_id="workspace-a", user_id="user-a", payload=payload, idempotency_key="create-post-key", ) assert replay.id == first.id assert len(first.targets) == 1 with pytest.raises(SocialIdempotencyConflictError): await social_container.social.publishing.create( workspace_id="workspace-a", user_id="user-a", payload=post_payload(account.id, title="Different"), idempotency_key="create-post-key", ) with pytest.raises(SocialAccountNotFoundError): await social_container.social.publishing.create( workspace_id="workspace-b", user_id="user-b", payload=payload, idempotency_key="cross-workspace-key", ) async def test_cross_workspace_asset_cannot_be_queued_for_publishing(social_container) -> None: asset_id = await registered_output_asset(social_container, "workspace-a") account_b = await connected_account(social_container, "workspace-b") post = await social_container.social.publishing.create( workspace_id="workspace-b", user_id="user-b", payload=SocialPostCreate.model_validate( { **post_payload(account_b.id).model_dump(mode="json"), "media_asset_id": asset_id, } ), idempotency_key="cross-workspace-asset-post", ) with pytest.raises(SocialMediaInvalidError): await social_container.social.publishing.queue( "workspace-b", post.id, idempotency_key="cross-workspace-asset-publish" ) async def test_cross_workspace_records_cannot_be_read_or_modified(social_container) -> None: account_a = await connected_account(social_container, "workspace-a") account_b = await connected_account(social_container, "workspace-b") post_b = await social_container.social.publishing.create( workspace_id="workspace-b", user_id="user-b", payload=post_payload(account_b.id), idempotency_key="workspace-b-post", ) job_b = ( await social_container.social.jobs.repository.create_many( [ SocialJob( workspace_id="workspace-b", social_post_id=post_b.id, social_post_target_id=post_b.targets[0].id, provider="youtube", status="queued", idempotency_key="workspace-b-job", ) ] ) )[0] await social_container.social.accounts.tokens.store( "workspace-b", account_b.id, {"access_token": "workspace-b-secret"} ) with pytest.raises(SocialAccountNotFoundError): await social_container.social.accounts.get("workspace-a", account_b.id) with pytest.raises(SocialAccountNotFoundError): await social_container.social.accounts.repository.set_status( "workspace-a", account_b.id, "disconnected" ) with pytest.raises(SocialPostNotFoundError): await social_container.social.publishing.get("workspace-a", post_b.id) with pytest.raises(SocialPostNotFoundError): await social_container.social.publishing.posts.set_status( "workspace-a", post_b.id, "cancelled" ) with pytest.raises(SocialJobNotFoundError): await social_container.social.jobs.get("workspace-a", job_b.id) with pytest.raises(SocialJobNotFoundError): await social_container.social.jobs.repository.transition( "workspace-a", job_b.id, "cancelled" ) with pytest.raises(SocialReauthRequiredError): await social_container.social.accounts.tokens.retrieve("workspace-a", account_b.id) with pytest.raises(SocialAccountNotFoundError): await social_container.social.analytics.account("workspace-a", account_b.id) with pytest.raises(SocialPostNotFoundError): await social_container.social.analytics.post("workspace-a", post_b.id) assert account_a.id != account_b.id async def test_token_401_is_refreshed_and_retried_once(social_container, monkeypatch) -> None: account = await connected_account(social_container, "workspace-refresh") await social_container.social.accounts.tokens.store( "workspace-refresh", account.id, {"access_token": "expired-access", "refresh_token": "refresh-token"}, ) adapter = social_container.social.accounts.providers.get("youtube") async def refreshed(_: dict[str, object]) -> dict[str, object]: return {"access_token": "fresh-access", "expires_in": 3600} monkeypatch.setattr(adapter, "refresh_token", refreshed) received: list[str] = [] async def protected_call(token: dict[str, object]) -> str: received.append(str(token["access_token"])) if len(received) == 1: raise SocialReauthRequiredError("first credential was rejected") return "ok" assert await social_container.social.oauth.execute_with_reauth_retry( workspace_id="workspace-refresh", account_id=account.id, operation=protected_call, ) == "ok" assert received == ["expired-access", "fresh-access"] async def test_resumable_upload_state_is_encrypted_and_excluded_from_job_views( social_container, ) -> None: account = await connected_account(social_container, "workspace-upload-state") post = await social_container.social.publishing.create( workspace_id="workspace-upload-state", user_id="user", payload=post_payload(account.id), idempotency_key="upload-state-post", ) job = ( await social_container.social.jobs.repository.create_many( [ SocialJob( workspace_id="workspace-upload-state", social_post_id=post.id, social_post_target_id=post.targets[0].id, provider="youtube", status="queued", idempotency_key="upload-state-job", ) ] ) )[0] session_url = "https://www.googleapis.com/upload/youtube/v3/videos?upload_id=bearer-like" await social_container.social.jobs.repository.set_provider_state( "workspace-upload-state", job.id, {"youtube_upload_session_url": session_url} ) async with social_container.social.database.session("workspace-upload-state") as session: stored = await session.scalar(select(SocialJob).where(SocialJob.id == job.id)) assert stored is not None and session_url not in (stored.provider_state_encrypted or "") assert await social_container.social.jobs.repository.get_provider_state( "workspace-upload-state", job.id ) == {"youtube_upload_session_url": session_url} view = await social_container.social.jobs.get("workspace-upload-state", job.id) assert session_url not in view.model_dump_json() async def test_scheduling_normalizes_to_utc_and_preserves_iana_timezone( social_container, ) -> None: account = await connected_account(social_container, "workspace-schedule") asset_id = await registered_output_asset(social_container, "workspace-schedule") post = await social_container.social.publishing.create( workspace_id="workspace-schedule", user_id="user", payload=SocialPostCreate.model_validate( { **post_payload(account.id).model_dump(mode="json"), "media_asset_id": asset_id, } ), idempotency_key="schedule-create-key", ) payload = SocialScheduleCreate.model_validate( { "scheduled_at": "2030-02-01T14:00:00+01:00", "timezone": "Africa/Lagos", } ) schedule = await social_container.social.scheduling.schedule( "workspace-schedule", post.id, payload ) assert schedule.timezone == "Africa/Lagos" assert schedule.scheduled_at.astimezone(timezone.utc).hour == 13 replacement = await social_container.social.scheduling.schedule( "workspace-schedule", post.id, SocialScheduleCreate.model_validate( { "scheduled_at": "2030-02-01T15:00:00+01:00", "timezone": "Africa/Lagos", } ), ) assert replacement.id == schedule.id assert replacement.scheduled_at.astimezone(timezone.utc).hour == 14 def test_schedule_rejects_past_naive_and_invalid_timezone_values() -> None: with pytest.raises(ValidationError): SocialScheduleCreate.model_validate( {"scheduled_at": "2000-01-01T00:00:00+00:00", "timezone": "UTC"} ) with pytest.raises(ValidationError): SocialScheduleCreate.model_validate( {"scheduled_at": "2030-03-10T01:30:00", "timezone": "America/New_York"} ) with pytest.raises(ValidationError): SocialScheduleCreate.model_validate( { "scheduled_at": "2030-03-10T01:30:00-05:00", "timezone": "Not/A_Timezone", } ) assert SocialScheduleCreate.model_validate( { "scheduled_at": "2030-03-10T01:30:00-05:00", "timezone": "America/New_York", } ).timezone == "America/New_York"