tests/test_social_foundation.py

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"