tests/test_linkedin_foundation.py

"""Phase 6A LinkedIn OIDC and organization-discovery coverage.

All LinkedIn traffic is mocked. Normal CI needs no developer application,
member credential, organization role, or interactive authorization flow.
"""

from __future__ import annotations

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 func, select

from app.container import build_container
from app.core.config import Settings
from app.social.domain.errors import (
    SocialAccountNotFoundError,
    SocialCapabilityUnsupportedError,
    SocialOAuthStateError,
    SocialPermissionDeniedError,
    SocialReauthRequiredError,
)
from app.social.models import OAuthState, SocialAccountToken
from app.social.providers.linkedin import LINKEDIN_API_VERSION, LinkedInProvider
from app.social.schemas.accounts import SocialAccountConnectRequest

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


def linkedin_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-6a-linkedin-test-encryption-material",
        social_oauth_redirect_base_url="https://api.example.com",
        linkedin_client_id="linkedin-client-id",
        linkedin_client_secret="linkedin-client-secret",
        linkedin_redirect_uri=_REDIRECT_URI,
        temp_dir=tmp_path / "temp",
        output_dir=tmp_path / "outputs",
        cleanup_interval_seconds=3600,
        whisper_model="tiny",
    )


async def test_linkedin_member_authorization_uses_official_oidc_without_pkce(
    tmp_path: Path,
) -> None:
    provider = LinkedInProvider(linkedin_settings(tmp_path))
    try:
        url = await provider.get_authorization_url(
            state="s" * 43,
            redirect_uri=_REDIRECT_URI,
        )
        with pytest.raises(SocialPermissionDeniedError):
            await provider.get_authorization_url(
                state="s" * 43,
                redirect_uri=_REDIRECT_URI,
                code_challenge="undocumented-pkce-challenge",
            )
        with pytest.raises(SocialCapabilityUnsupportedError):
            await provider.get_authorization_url(
                state="s" * 43,
                redirect_uri=_REDIRECT_URI,
                additional_scopes=["w_member_social"],
            )
    finally:
        await provider.close()

    parsed = urlparse(url)
    query = parse_qs(parsed.query)
    assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == (
        "https://www.linkedin.com/oauth/v2/authorization"
    )
    assert query == {
        "client_id": ["linkedin-client-id"],
        "redirect_uri": [_REDIRECT_URI],
        "response_type": ["code"],
        "state": ["s" * 43],
        "scope": ["openid profile"],
    }
    assert "code_challenge" not in query


async def test_linkedin_exchange_member_and_organization_discovery_use_official_apis(
    tmp_path: Path,
) -> None:
    calls: list[str] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        calls.append(request.url.path)
        if request.url.path == "/oauth/v2/accessToken":
            assert request.url.host == "www.linkedin.com"
            form = parse_qs(request.content.decode())
            assert form == {
                "grant_type": ["authorization_code"],
                "code": ["authorization-code"],
                "redirect_uri": [_REDIRECT_URI],
                "client_id": ["linkedin-client-id"],
                "client_secret": ["linkedin-client-secret"],
            }
            return httpx.Response(
                200,
                json={
                    "access_token": "linkedin-access-token",
                    "expires_in": 5184000,
                    "token_type": "Bearer",
                },
            )
        assert request.headers["authorization"] == "Bearer linkedin-access-token"
        if request.url.path == "/v2/userinfo":
            return httpx.Response(
                200,
                json={
                    "sub": "oidc-member-subject_123",
                    "name": "Ada Lovelace",
                    "given_name": "Ada",
                    "family_name": "Lovelace",
                    "picture": "https://media.licdn.com/member.jpg",
                    "locale": {"country": "US", "language": "en"},
                    "email": "not-persisted@example.com",
                    "email_verified": True,
                },
            )
        assert request.headers["linkedin-version"] == LINKEDIN_API_VERSION
        assert request.headers["x-restli-protocol-version"] == "2.0.0"
        if request.url.path == "/rest/organizationAcls":
            assert parse_qs(request.url.query.decode()) == {
                "q": ["roleAssignee"],
                "role": ["ADMINISTRATOR"],
                "state": ["APPROVED"],
                "count": ["100"],
                "start": ["0"],
            }
            return httpx.Response(
                200,
                json={
                    "elements": [
                        {"organization": "urn:li:organization:123456"},
                        {
                            "organizationTarget": "urn:li:organization:789012"
                        },
                    ],
                    "paging": {"start": 0, "count": 2, "total": 2, "links": []},
                },
            )
        organization_id = request.url.path.rsplit("/", 1)[-1]
        return httpx.Response(
            200,
            json={
                "id": int(organization_id),
                "localizedName": f"Organization {organization_id}",
                "vanityName": f"organization-{organization_id}",
                "logoV2": {
                    "digitalmediaAsset": "urn:li:digitalmediaAsset:logo_asset",
                    "original~": {
                        "elements": [
                            {
                                "identifiers": [
                                    {
                                        "identifier": f"https://media.licdn.com/{organization_id}.png"
                                    }
                                ]
                            }
                        ]
                    },
                },
                "primaryOrganizationType": "NONE",
                "defaultLocale": {"country": "US", "language": "en"},
                "localizedWebsite": "https://example.com",
            },
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(linkedin_settings(tmp_path), http_client=client)
    try:
        token = await provider.exchange_code(
            code="authorization-code",
            redirect_uri=_REDIRECT_URI,
        )
        accounts = await provider.discover_accounts(
            token, account_type="linkedin_organization"
        )
    finally:
        await client.aclose()

    member, first, second = accounts
    assert member["external_account_id"] == "oidc-member-subject_123"
    assert member["account_type"] == "linkedin_member"
    assert member["connection_status"] == "connected"
    assert member["metadata"]["email_verified"] is True
    assert "email" not in member["metadata"]
    assert [first["external_account_id"], second["external_account_id"]] == [
        "123456",
        "789012",
    ]
    assert first["account_type"] == "linkedin_organization"
    assert first["connection_status"] == "pending"
    assert first["avatar_url"] == "https://media.licdn.com/123456.png"
    assert first["metadata"]["parent_member_id"] == "oidc-member-subject_123"
    assert first["metadata"]["logo_asset"] == (
        "urn:li:digitalmediaAsset:logo_asset"
    )
    assert calls == [
        "/oauth/v2/accessToken",
        "/v2/userinfo",
        "/rest/organizationAcls",
        "/rest/organizations/123456",
        "/rest/organizations/789012",
    ]


async def test_linkedin_organization_scope_is_explicit_and_bound_to_state(
    tmp_path: Path,
) -> None:
    container = build_container(linkedin_settings(tmp_path))
    await container.social.initialize()
    try:
        member = await container.social.oauth.connect(
            provider="linkedin",
            workspace_id="workspace-a",
            user_id="user-a",
            payload=SocialAccountConnectRequest(),
        )
        member_query = parse_qs(urlparse(member.authorization_url or "").query)
        assert member_query["scope"] == ["openid profile"]
        assert "code_challenge" not in member_query

        organization = await container.social.oauth.connect(
            provider="linkedin",
            workspace_id="workspace-a",
            user_id="user-a",
            payload=SocialAccountConnectRequest(
                account_type="linkedin_organization"
            ),
        )
        organization_query = parse_qs(
            urlparse(organization.authorization_url or "").query
        )
        assert organization_query["scope"] == [
            "openid profile rw_organization_admin"
        ]
        assert "w_organization_social" not in organization_query["scope"][0]
        state = await container.social.oauth.states.consume(
            state=organization_query["state"][0], provider="linkedin"
        )
        assert state.workspace_id == "workspace-a"
        assert state.user_id == "user-a"
        assert state.requested_account_type == "linkedin_organization"
        assert state.requested_scopes == [
            "openid",
            "profile",
            "rw_organization_admin",
        ]

        with pytest.raises(SocialCapabilityUnsupportedError):
            await container.social.oauth.connect(
                provider="linkedin",
                workspace_id="workspace-a",
                user_id="user-a",
                payload=SocialAccountConnectRequest(account_type="organization"),
            )
    finally:
        await container.social.close()
        await container.security_database.close()


async def test_linkedin_callback_is_duplicate_safe_selectable_and_workspace_bound(
    tmp_path: Path,
) -> None:
    container = build_container(linkedin_settings(tmp_path))
    await container.social.initialize()
    adapter = container.social.accounts.providers.get("linkedin")
    assert isinstance(adapter, LinkedInProvider)
    await adapter._client.aclose()

    async def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/oauth/v2/accessToken":
            return httpx.Response(
                200,
                json={
                    "access_token": "linkedin-token-that-must-remain-encrypted",
                    "expires_in": 3600,
                    "token_type": "Bearer",
                },
            )
        if request.url.path == "/v2/userinfo":
            return httpx.Response(
                200,
                json={"sub": "stable-member-sub", "name": "Workspace Member"},
            )
        if request.url.path == "/rest/organizationAcls":
            return httpx.Response(
                200,
                json={
                    "elements": [
                        {"organization": "urn:li:organization:123456"}
                    ],
                    "paging": {"start": 0, "count": 1, "total": 1},
                },
            )
        return httpx.Response(
            200,
            json={
                "id": 123456,
                "localizedName": "Workspace Organization",
                "vanityName": "workspace-organization",
            },
        )

    adapter._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    adapter._owns_client = True
    try:
        connect = await container.social.oauth.connect(
            provider="linkedin",
            workspace_id="workspace-a",
            user_id="user-a",
            payload=SocialAccountConnectRequest(
                account_type="linkedin_organization"
            ),
        )
        state = parse_qs(urlparse(connect.authorization_url or "").query)["state"][0]
        member = await container.social.oauth.callback(
            provider="linkedin", state=state, code="first-code"
        )
        assert member.account_type == "linkedin_member"
        assert member.status.value == "connected"
        with pytest.raises(SocialOAuthStateError):
            await container.social.oauth.callback(
                provider="linkedin", state=state, code="replayed-code"
            )

        accounts = await container.social.accounts.list("workspace-a")
        assert len(accounts) == 2
        organization = next(
            item
            for item in accounts
            if item.account_type == "linkedin_organization"
        )
        assert organization.status.value == "pending"
        assert "linkedin-token-that-must-remain-encrypted" not in (
            organization.model_dump_json()
        )

        with pytest.raises(SocialAccountNotFoundError):
            await container.social.accounts.select_discovered(
                "workspace-b", [organization.id]
            )
        selected = await container.social.accounts.select_discovered(
            "workspace-a", [organization.id, organization.id]
        )
        assert len(selected) == 1
        assert selected[0].status.value == "connected"

        second_connect = await container.social.oauth.connect(
            provider="linkedin",
            workspace_id="workspace-a",
            user_id="user-a",
            payload=SocialAccountConnectRequest(
                account_type="linkedin_organization"
            ),
        )
        second_state = parse_qs(
            urlparse(second_connect.authorization_url or "").query
        )["state"][0]
        await container.social.oauth.callback(
            provider="linkedin", state=second_state, code="second-code"
        )
        duplicate_safe = await container.social.accounts.list("workspace-a")
        assert len(duplicate_safe) == 2
        assert next(
            item
            for item in duplicate_safe
            if item.account_type == "linkedin_organization"
        ).status.value == "connected"

        async with container.social.database.session("workspace-a") as session:
            token_count = await session.scalar(select(func.count(SocialAccountToken.id)))
            encrypted_payloads = list(
                (
                    await session.scalars(
                        select(SocialAccountToken.encrypted_payload)
                    )
                ).all()
            )
        assert token_count == 2
        assert all(encrypted_payloads)
        assert all(
            "linkedin-token-that-must-remain-encrypted" not in str(payload)
            for payload in encrypted_payloads
        )
    finally:
        await container.social.close()
        await container.security_database.close()


async def test_linkedin_state_redirect_expiry_and_provider_binding(
    tmp_path: Path,
) -> None:
    container = build_container(linkedin_settings(tmp_path))
    await container.social.initialize()
    try:
        assert container.social.oauth._redirect_uri("linkedin", None) == _REDIRECT_URI
        with pytest.raises(SocialPermissionDeniedError):
            container.social.oauth._redirect_uri(
                "linkedin",
                "https://attacker.example/v1/social/accounts/linkedin/callback",
            )
        state = await container.social.oauth.states.create(
            provider="linkedin",
            workspace_id="workspace-a",
            user_id="user-a",
            redirect_uri=_REDIRECT_URI,
            requested_account_type="linkedin_member",
            requested_scopes=["openid", "profile"],
        )
        with pytest.raises(SocialOAuthStateError):
            await container.social.oauth.states.consume(
                state=state.state, provider="x"
            )
        consumed = await container.social.oauth.states.consume(
            state=state.state, provider="linkedin"
        )
        assert consumed.workspace_id == "workspace-a"

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


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

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

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(linkedin_settings(tmp_path), http_client=client)
    try:
        with pytest.raises(SocialPermissionDeniedError) as raised:
            await provider.exchange_code(
                code=secret_code, redirect_uri=_REDIRECT_URI
            )
        with pytest.raises(SocialReauthRequiredError):
            await provider.refresh_token({"access_token": "expired"})
    finally:
        await client.aclose()
    assert secret_code not in str(raised.value)
    assert "linkedin-client-secret" not in str(raised.value)


async def test_linkedin_refresh_is_used_only_when_provider_issued_it(
    tmp_path: Path,
) -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        form = parse_qs(request.content.decode())
        assert form == {
            "grant_type": ["refresh_token"],
            "refresh_token": ["partner-refresh-token"],
            "client_id": ["linkedin-client-id"],
            "client_secret": ["linkedin-client-secret"],
        }
        return httpx.Response(
            200,
            json={
                "access_token": "refreshed-access-token",
                "expires_in": 5184000,
                "token_type": "Bearer",
            },
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(linkedin_settings(tmp_path), http_client=client)
    try:
        refreshed = await provider.refresh_token(
            {
                "access_token": "expired-access-token",
                "refresh_token": "partner-refresh-token",
            }
        )
    finally:
        await client.aclose()

    assert refreshed["access_token"] == "refreshed-access-token"
    assert refreshed["refresh_token"] == "partner-refresh-token"


async def test_linkedin_capabilities_are_discovery_only(tmp_path: Path) -> None:
    container = build_container(linkedin_settings(tmp_path))
    try:
        linkedin = container.social.accounts.get_provider("linkedin")
        assert linkedin.available
        assert linkedin.configured
        assert linkedin.capabilities.implementation_status == "implemented"
        assert linkedin.capabilities.account_types == [
            "linkedin_member",
            "linkedin_organization",
        ]
        assert linkedin.capabilities.required_scopes == ["openid", "profile"]
        assert linkedin.capabilities.optional_scopes == ["rw_organization_admin"]
        assert not linkedin.capabilities.video
        assert not linkedin.capabilities.image
        assert not linkedin.capabilities.direct_publish
        assert not linkedin.capabilities.draft_upload
        assert not linkedin.capabilities.scheduled_publish
        assert not linkedin.capabilities.analytics
        assert not linkedin.capabilities.delete_post
        assert not linkedin.capabilities.personal_publishing
        assert not linkedin.capabilities.organization_publishing
        assert linkedin.capabilities.publish_metadata_schema == {}
    finally:
        await container.social.close()
        await container.security_database.close()


def test_linkedin_redirect_configuration_is_fail_closed() -> None:
    invalid = [
        "https://attacker.example/not-the-linkedin-callback",
        "http://api.example.com/v1/social/accounts/linkedin/callback",
        "https://api.example.com/v1/social/accounts/linkedin/callback?next=bad",
        "ftp://localhost/v1/social/accounts/linkedin/callback",
    ]
    for redirect in invalid:
        with pytest.raises(ValidationError):
            Settings(_env_file=None, linkedin_redirect_uri=redirect)
    local = Settings(
        _env_file=None,
        linkedin_redirect_uri=(
            "http://localhost/v1/social/accounts/linkedin/callback"
        ),
    )
    assert local.linkedin_redirect_uri.startswith("http://localhost/")