tests/test_x_foundation.py

"""Phase 5A X API v2 OAuth and account-discovery coverage.

All provider traffic is mocked. Normal CI never needs X credentials, API
credits, or an interactive browser authorization flow.
"""

from __future__ import annotations

import base64
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import parse_qs, urlparse

import httpx
import pytest
from pydantic import ValidationError
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,
    SocialProviderUnavailableError,
    SocialReauthRequiredError,
)
from app.social.models import OAuthState, SocialAccountToken
from app.social.providers.x import XProvider
from app.social.schemas.accounts import SocialAccountConnectRequest

_REDIRECT_URI = "https://api.example.com/v1/social/accounts/x/callback"


def x_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="phase-5a-test-encryption-material",
        social_oauth_redirect_base_url="https://api.example.com",
        x_client_id="x-client-id",
        x_client_secret="x-client-secret",
        x_redirect_uri=_REDIRECT_URI,
        x_publishing_enabled=True,
        temp_dir=tmp_path / "temp",
        output_dir=tmp_path / "outputs",
        cleanup_interval_seconds=3600,
        whisper_model="tiny",
    )


def assert_confidential_client(request: httpx.Request) -> None:
    scheme, encoded = request.headers["authorization"].split(" ", 1)
    assert scheme == "Basic"
    assert base64.b64decode(encoded).decode() == "x-client-id:x-client-secret"


async def test_x_authorization_uses_official_url_minimum_scopes_and_s256_pkce(
    tmp_path: Path,
) -> None:
    provider = XProvider(x_settings(tmp_path))
    try:
        url = await provider.get_authorization_url(
            state="s" * 43,
            redirect_uri=_REDIRECT_URI,
            code_challenge="s256-code-challenge",
        )
        with pytest.raises(SocialPermissionDeniedError):
            await provider.get_authorization_url(
                state="s" * 43,
                redirect_uri=_REDIRECT_URI,
                code_challenge=None,
            )
    finally:
        await provider.close()

    parsed = urlparse(url)
    query = parse_qs(parsed.query)
    assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == (
        "https://x.com/i/oauth2/authorize"
    )
    assert query["client_id"] == ["x-client-id"]
    assert query["redirect_uri"] == [_REDIRECT_URI]
    assert query["response_type"] == ["code"]
    assert query["scope"] == ["tweet.read users.read offline.access"]
    assert query["state"] == ["s" * 43]
    assert query["code_challenge"] == ["s256-code-challenge"]
    assert query["code_challenge_method"] == ["S256"]
    assert "tweet.write" not in query["scope"][0]
    assert "media.write" not in query["scope"][0]


async def test_x_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)
        assert request.url.host == "api.x.com"
        if request.url.path == "/2/oauth2/token":
            assert_confidential_client(request)
            form = parse_qs(request.content.decode())
            assert "client_secret" not in form
            assert "client_id" not in form
            if form["grant_type"] == ["authorization_code"]:
                assert form == {
                    "code": ["authorization-code"],
                    "grant_type": ["authorization_code"],
                    "redirect_uri": [_REDIRECT_URI],
                    "code_verifier": ["pkce-verifier"],
                }
            else:
                assert form == {
                    "refresh_token": ["refresh-token"],
                    "grant_type": ["refresh_token"],
                }
            return httpx.Response(
                200,
                json={
                    "access_token": "x-access-token",
                    "refresh_token": "x-rotated-refresh-token",
                    "expires_in": 7200,
                    "scope": "tweet.read users.read offline.access",
                    "token_type": "bearer",
                },
            )
        if request.url.path == "/2/users/me":
            assert request.headers["authorization"] == "Bearer x-access-token"
            assert parse_qs(request.url.query.decode()) == {
                "user.fields": [
                    "created_at,description,profile_image_url,protected,verified"
                ]
            }
            return httpx.Response(
                200,
                json={
                    "data": {
                        "id": "2244994945",
                        "username": "XDevelopers",
                        "name": "X Developers",
                        "profile_image_url": "https://pbs.twimg.com/profile.jpg",
                        "created_at": "2013-12-14T04:35:55.000Z",
                        "description": "Official developer account",
                        "protected": False,
                        "verified": True,
                    }
                },
            )
        assert request.url.path == "/2/oauth2/revoke"
        assert_confidential_client(request)
        assert parse_qs(request.content.decode()) == {
            "token": ["x-rotated-refresh-token"]
        }
        return httpx.Response(200)

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = XProvider(x_settings(tmp_path), http_client=client)
    try:
        token = await provider.exchange_code(
            code="authorization-code",
            redirect_uri=_REDIRECT_URI,
            code_verifier="pkce-verifier",
        )
        account = await provider.get_account(token)
        refreshed = await provider.refresh_token(
            {"access_token": "old-token", "refresh_token": "refresh-token"}
        )
        await provider.revoke_token(refreshed)
    finally:
        await client.aclose()

    assert account == {
        "external_account_id": "2244994945",
        "account_type": "user",
        "username": "XDevelopers",
        "display_name": "X Developers",
        "avatar_url": "https://pbs.twimg.com/profile.jpg",
        "metadata": {
            "x_user_id": "2244994945",
            "created_at": "2013-12-14T04:35:55.000Z",
            "verified": True,
            "protected": False,
            "description": "Official developer account",
        },
    }
    assert refreshed["refresh_token"] == "x-rotated-refresh-token"
    assert calls == [
        "/2/oauth2/token",
        "/2/users/me",
        "/2/oauth2/token",
        "/2/oauth2/revoke",
    ]


async def test_x_invalid_code_and_pkce_failure_are_normalized_without_secrets(
    tmp_path: Path,
) -> None:
    secret_code = "x-code-that-must-not-leak"

    async def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(
            400,
            json={
                "error": "invalid_grant",
                "error_description": f"invalid code {secret_code}",
            },
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = XProvider(x_settings(tmp_path), http_client=client)
    try:
        with pytest.raises(SocialPermissionDeniedError):
            await provider.exchange_code(
                code=secret_code,
                redirect_uri=_REDIRECT_URI,
                code_verifier=None,
            )
        with pytest.raises(SocialReauthRequiredError) as raised:
            await provider.exchange_code(
                code=secret_code,
                redirect_uri=_REDIRECT_URI,
                code_verifier="incorrect-verifier",
            )
    finally:
        await client.aclose()
    assert secret_code not in str(raised.value)


async def test_x_invalid_client_is_configuration_failure_not_consent_loop(
    tmp_path: Path,
) -> None:
    async def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(
            401,
            json={
                "error": "invalid_client",
                "error_description": "client secret is not accepted",
            },
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = XProvider(x_settings(tmp_path), http_client=client)
    try:
        with pytest.raises(SocialProviderUnavailableError) as raised:
            await provider.exchange_code(
                code="authorization-code",
                redirect_uri=_REDIRECT_URI,
                code_verifier="pkce-verifier",
            )
    finally:
        await client.aclose()

    assert "client secret is not accepted" not in str(raised.value)


async def test_x_account_discovery_rejects_non_ascii_or_oversized_user_ids(
    tmp_path: Path,
) -> None:
    invalid_ids = ["٢٢٤٤٩٩٤٩٤٥", "12345678901234567890"]
    for user_id in invalid_ids:

        async def handler(_: httpx.Request, value: str = user_id) -> httpx.Response:
            return httpx.Response(
                200,
                json={"data": {"id": value, "username": "invalid"}},
            )

        client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
        provider = XProvider(x_settings(tmp_path), http_client=client)
        try:
            with pytest.raises(SocialProviderUnavailableError):
                await provider.get_account({"access_token": "x-access-token"})
        finally:
            await client.aclose()


async def test_x_callback_is_single_use_duplicate_safe_and_workspace_bound(
    tmp_path: Path,
) -> None:
    container = build_container(x_settings(tmp_path))
    await container.social.initialize()
    adapter = container.social.accounts.providers.get("x")
    assert isinstance(adapter, XProvider)
    await adapter._client.aclose()

    async def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/2/oauth2/token":
            form = parse_qs(request.content.decode())
            assert form.get("code_verifier", [""])[0]
            return httpx.Response(
                200,
                json={
                    "access_token": "x-token-that-must-stay-encrypted",
                    "refresh_token": "x-refresh-that-must-stay-encrypted",
                    "expires_in": 7200,
                    "scope": "tweet.read users.read offline.access",
                    "token_type": "bearer",
                },
            )
        return httpx.Response(
            200,
            json={
                "data": {
                    "id": "2244994945",
                    "username": "workspace_user",
                    "name": "Workspace User",
                }
            },
        )

    adapter._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    adapter._owns_client = True
    try:
        first_connect = await container.social.oauth.connect(
            provider="x",
            workspace_id="workspace-a",
            user_id="user-a",
            payload=SocialAccountConnectRequest(),
        )
        first_query = parse_qs(urlparse(first_connect.authorization_url or "").query)
        first_state = first_query["state"][0]
        assert first_query["code_challenge_method"] == ["S256"]
        assert first_query["code_challenge"][0]

        first = await container.social.oauth.callback(
            provider="x",
            state=first_state,
            code="first-code",
        )
        with pytest.raises(SocialOAuthStateError):
            await container.social.oauth.callback(
                provider="x",
                state=first_state,
                code="replayed-code",
            )

        second_connect = await container.social.oauth.connect(
            provider="x",
            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="x",
            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 if account.provider.value == "x"] == [
            first.id
        ]
        assert "x-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.expires_at is not None
        assert stored.encrypted_payload
        assert "x-token-that-must-stay-encrypted" not in stored.encrypted_payload
    finally:
        await container.social.close()
        await container.security_database.close()


async def test_x_state_redirect_provider_and_expiry_validation(tmp_path: Path) -> None:
    container = build_container(x_settings(tmp_path))
    await container.social.initialize()
    try:
        assert container.social.oauth._redirect_uri("x", None) == _REDIRECT_URI
        with pytest.raises(SocialPermissionDeniedError):
            container.social.oauth._redirect_uri(
                "x",
                "https://attacker.example/v1/social/accounts/x/callback",
            )

        state = await container.social.oauth.states.create(
            provider="x",
            workspace_id="workspace-a",
            user_id="user-a",
            redirect_uri=_REDIRECT_URI,
        )
        with pytest.raises(SocialOAuthStateError):
            await container.social.oauth.states.consume(
                state=state.state,
                provider="linkedin",
            )
        consumed = await container.social.oauth.states.consume(
            state=state.state,
            provider="x",
        )
        assert consumed.workspace_id == "workspace-a"
        assert consumed.user_id == "user-a"

        expired = OAuthState(
            state="expired-x-state-value-that-is-long-enough",
            provider="x",
            workspace_id="workspace-a",
            user_id="user-a",
            redirect_uri=_REDIRECT_URI,
            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="x",
            )
    finally:
        await container.social.close()
        await container.security_database.close()


def test_x_redirect_configuration_is_fail_closed() -> None:
    invalid_redirects = [
        "https://attacker.example/not-the-x-callback",
        "ftp://localhost/v1/social/accounts/x/callback",
        "http://api.example.com/v1/social/accounts/x/callback",
        "https://api.example.com/v1/social/accounts/x/callback?next=attacker",
    ]
    for redirect in invalid_redirects:
        with pytest.raises(ValidationError):
            Settings(_env_file=None, x_redirect_uri=redirect)

    settings = Settings(
        _env_file=None,
        x_redirect_uri="http://localhost/v1/social/accounts/x/callback",
    )
    assert settings.x_redirect_uri.startswith("http://localhost/")


async def test_x_capability_discovery_advertises_implemented_publishing(
    tmp_path: Path,
) -> None:
    container = build_container(x_settings(tmp_path))
    try:
        provider = container.social.accounts.get_provider("x")
        assert provider.available
        assert provider.configured
        assert provider.capabilities.implementation_status == "implemented"
        assert provider.capabilities.account_types == ["user"]
        assert provider.capabilities.required_scopes == [
            "tweet.read",
            "users.read",
            "offline.access",
        ]
        assert provider.capabilities.video
        assert provider.capabilities.video_upload
        assert provider.capabilities.video_status
        assert provider.capabilities.image
        assert provider.capabilities.direct_publish
        assert not provider.capabilities.draft_upload
        assert provider.capabilities.scheduled_publish
        assert not provider.capabilities.native_scheduling
        assert provider.capabilities.delete_post
        assert provider.capabilities.publishing_required_scopes == [
            "tweet.write",
            "media.write",
        ]
        assert provider.capabilities.analytics
        assert provider.capabilities.analytics_required_scopes == ["tweet.read"]
    finally:
        await container.social.close()
        await container.security_database.close()


async def test_x_publishing_capabilities_are_fail_closed_without_operator_gate(
    tmp_path: Path,
) -> None:
    settings = x_settings(tmp_path).model_copy(
        update={"x_publishing_enabled": False}
    )
    provider = XProvider(settings)
    try:
        assert provider.configuration_ready
        assert not provider.publishing_ready
        assert not provider.capabilities.direct_publish
        assert not provider.capabilities.video_upload
        assert not provider.capabilities.delete_post
        assert provider.capabilities.publishing_required_scopes == []
    finally:
        await provider.close()