tests/test_linkedin_production.py

"""Phase 6C LinkedIn analytics, security, tenancy, and certification tests.

Normal CI uses SQLite and mocked official LinkedIn REST traffic. Destructive
live verification is isolated in ``test_linkedin_live.py`` and is opt-in.
"""

from __future__ import annotations

import json
import logging
from datetime import datetime, timedelta, timezone
from pathlib import Path
from types import SimpleNamespace
from urllib.parse import parse_qs, urlparse

import httpx
import pytest
from sqlalchemy import select

from app.container import build_container
from app.core.config import Settings
from app.core.logger import JsonFormatter
from app.mcp.registry import MCPRegistry
from app.mcp.server import create_mcp_server
from app.security.context import AuthContext, auth_context, http_auth_applied
from app.social.domain.errors import (
    SocialAccountNotFoundError,
    SocialJobNotFoundError,
    SocialMediaInvalidError,
    SocialPostNotFoundError,
    SocialProviderUnavailableError,
    SocialReauthRequiredError,
)
from app.social.domain.retry import classify_retry
from app.social.models import (
    SocialAccount,
    SocialAuditEvent,
    SocialJob,
    SocialMediaAsset,
    SocialPost,
    SocialPostMetric,
    SocialPostTarget,
)
from app.social.providers.linkedin import LINKEDIN_API_VERSION, LinkedInProvider
from app.social.schemas.accounts import SocialAccountConnectRequest, SocialAccountView
from app.social.schemas.jobs import SocialJobView
from app.social.workers.publisher import SocialPublisher


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


def phase6c_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": "phase-6c-linkedin-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,
        "linkedin_publishing_enabled": True,
        "temp_dir": tmp_path / "temp",
        "output_dir": tmp_path / "outputs",
        "cleanup_interval_seconds": 3600,
        "whisper_model": "tiny",
    }
    values.update(overrides)
    return Settings(**values)


@pytest.fixture
async def phase6c_container(tmp_path: Path):
    container = build_container(phase6c_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: object,
    workspace_id: str,
    *,
    account_type: str,
    external_id: str,
    scopes: list[str],
) -> SocialAccount:
    social = container.social  # type: ignore[attr-defined]
    account = await social.accounts.repository.create(
        SocialAccount(
            workspace_id=workspace_id,
            provider="linkedin",
            account_type=account_type,
            external_account_id=external_id,
            display_name="LinkedIn production test",
            status="connected",
            metadata_json=(
                {"roles": ["ADMINISTRATOR"]}
                if account_type == "linkedin_organization"
                else {}
            ),
        )
    )
    await social.accounts.tokens.store(
        workspace_id,
        account.id,
        {
            "access_token": "linkedin-provider-secret",
            "refresh_token": "linkedin-refresh-secret",
        },
        expires_at=datetime.now(timezone.utc) + timedelta(hours=2),
        scopes=scopes,
        token_type="bearer",
    )
    return account


async def test_linkedin_member_analytics_uses_official_endpoint_and_normalizes(
    tmp_path: Path,
) -> None:
    secret = "member-analytics-secret"
    external_id = "urn:li:share:7325786486870552578"
    counts = {
        "IMPRESSION": 101,
        "MEMBERS_REACHED": 88,
        "REACTION": 22,
        "COMMENT": 3,
        "RESHARE": 4,
    }
    requested: list[str] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.method == "GET"
        assert request.url.path == "/rest/memberCreatorPostAnalytics"
        assert request.headers["authorization"] == f"Bearer {secret}"
        assert request.headers["linkedin-version"] == LINKEDIN_API_VERSION
        assert request.headers["x-restli-protocol-version"] == "2.0.0"
        query_type = request.url.params["queryType"]
        requested.append(query_type)
        assert request.url.params["q"] == "entity"
        assert request.url.params["entity"] == f"(share:{external_id})"
        assert request.url.params["aggregation"] == "TOTAL"
        return httpx.Response(
            200,
            json={
                "elements": [{
                    "count": counts[query_type],
                    "targetEntity": {"share": external_id},
                    "metricType": {"type": query_type},
                    "access_token": secret,
                }],
                "paging": {"count": 10, "start": 0},
            },
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(phase6c_settings(tmp_path), http_client=client)
    try:
        result = await provider.get_metrics(
            {
                "access_token": secret,
                "_mediarouter_account_type": "linkedin_member",
                "_mediarouter_external_account_id": "member_123",
            },
            external_id,
        )
    finally:
        await client.aclose()

    assert requested == [
        "IMPRESSION",
        "MEMBERS_REACHED",
        "REACTION",
        "COMMENT",
        "RESHARE",
    ]
    assert result["status"] == "available"
    assert result["impressions"] == 101
    assert result["likes"] == 22
    assert result["comments"] == 3
    assert result["shares"] == 4
    assert result["raw_metrics"]["members_reached"] == 88
    assert secret not in json.dumps(result)


async def test_linkedin_member_analytics_never_invents_an_omitted_metric(
    tmp_path: Path,
) -> None:
    async def handler(_: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"elements": []})

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(phase6c_settings(tmp_path), http_client=client)
    try:
        with pytest.raises(SocialProviderUnavailableError):
            await provider.get_metrics(
                {
                    "access_token": "member-missing-metric-secret",
                    "_mediarouter_account_type": "linkedin_member",
                    "_mediarouter_external_account_id": "member_123",
                },
                "urn:li:share:7325786486870552578",
            )
    finally:
        await client.aclose()


@pytest.mark.parametrize(
    ("external_id", "query_key"),
    [
        ("urn:li:share:7132564752928563200", "shares"),
        ("urn:li:ugcPost:7132564752928563201", "ugcPosts[0]"),
    ],
)
async def test_linkedin_organization_analytics_uses_official_share_statistics(
    tmp_path: Path,
    external_id: str,
    query_key: str,
) -> None:
    secret = "organization-analytics-secret"
    organization_urn = "urn:li:organization:5515715"

    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/rest/organizationalEntityShareStatistics"
        assert request.headers["authorization"] == f"Bearer {secret}"
        assert request.url.params["q"] == "organizationalEntity"
        assert request.url.params["organizationalEntity"] == organization_urn
        if query_key == "shares":
            assert request.url.params[query_key] == f"List({external_id})"
        else:
            assert request.url.params[query_key] == external_id
        field = "share" if external_id.startswith("urn:li:share:") else "ugcPost"
        return httpx.Response(
            200,
            json={
                "elements": [{
                    "organizationalEntity": organization_urn,
                    field: external_id,
                    "totalShareStatistics": {
                        "clickCount": 7,
                        "commentCount": 3,
                        "engagement": 0.125,
                        "impressionCount": 101,
                        "likeCount": 22,
                        "shareCount": 4,
                        "refresh_token": secret,
                    },
                }],
                "paging": {"count": 10, "start": 0},
            },
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(phase6c_settings(tmp_path), http_client=client)
    try:
        result = await provider.get_metrics(
            {
                "access_token": secret,
                "_mediarouter_account_type": "linkedin_organization",
                "_mediarouter_external_account_id": "5515715",
            },
            external_id,
        )
    finally:
        await client.aclose()

    assert result["status"] == "available"
    assert result["impressions"] == 101
    assert result["likes"] == 22
    assert result["comments"] == 3
    assert result["shares"] == 4
    assert result["engagement_rate"] == 0.125
    assert result["raw_metrics"]["click_count"] == 7
    assert secret not in json.dumps(result)


async def test_linkedin_analytics_scopes_are_explicit_and_account_specific(
    phase6c_container,
) -> None:
    social = phase6c_container.social
    normal = await social.oauth.connect(
        provider="linkedin",
        workspace_id="workspace-scope",
        user_id="user-scope",
        payload=SocialAccountConnectRequest(account_type="linkedin_member"),
    )
    member = await social.oauth.connect(
        provider="linkedin",
        workspace_id="workspace-scope",
        user_id="user-scope",
        payload=SocialAccountConnectRequest(
            account_type="linkedin_member",
            authorization_purpose="analytics",
        ),
    )
    organization = await social.oauth.connect(
        provider="linkedin",
        workspace_id="workspace-scope",
        user_id="user-scope",
        payload=SocialAccountConnectRequest(
            account_type="linkedin_organization",
            authorization_purpose="analytics",
        ),
    )
    normal_scopes = parse_qs(urlparse(str(normal.authorization_url)).query)["scope"][0].split()
    member_scopes = parse_qs(urlparse(str(member.authorization_url)).query)["scope"][0].split()
    organization_scopes = parse_qs(
        urlparse(str(organization.authorization_url)).query
    )["scope"][0].split()

    assert normal_scopes == ["openid", "profile"]
    assert member_scopes == ["openid", "profile", "r_member_postAnalytics"]
    assert organization_scopes == ["openid", "profile", "rw_organization_admin"]
    capabilities = social.accounts.providers.get("linkedin").capabilities
    assert capabilities.analytics
    assert capabilities.account_type_analytics_scopes == {
        "linkedin_member": ["r_member_postAnalytics"],
        "linkedin_organization": ["rw_organization_admin"],
    }


async def test_linkedin_analytics_persists_normalized_metrics_and_raw_data(
    phase6c_container,
) -> None:
    social = phase6c_container.social
    account = await _connected_account(
        phase6c_container,
        "workspace-analytics",
        account_type="linkedin_organization",
        external_id="5515715",
        scopes=["openid", "profile", "rw_organization_admin"],
    )
    post, targets = await social.publishing.posts.create(
        SocialPost(
            workspace_id="workspace-analytics",
            status="published",
            publish_mode="now",
        ),
        [SocialPostTarget(
            social_post_id="",
            social_account_id=account.id,
            provider="linkedin",
            status="published",
            external_post_id="urn:li:share:7132564752928563200",
        )],
    )
    adapter = social.accounts.providers.get("linkedin")
    assert isinstance(adapter, LinkedInProvider)
    await adapter._client.aclose()

    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.headers["authorization"] == "Bearer linkedin-provider-secret"
        return httpx.Response(
            200,
            json={
                "elements": [{
                    "organizationalEntity": "urn:li:organization:5515715",
                    "share": targets[0].external_post_id,
                    "totalShareStatistics": {
                        "clickCount": 9,
                        "commentCount": 4,
                        "engagement": 0.25,
                        "impressionCount": 120,
                        "likeCount": 30,
                        "shareCount": 5,
                    },
                }],
            },
        )

    adapter._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    adapter._owns_client = True
    result = await social.analytics.post("workspace-analytics", post.id)

    assert result["unavailable"] == []
    assert result["metrics"][0]["impressions"] == 120
    assert result["metrics"][0]["engagement_rate"] == 0.25
    async with social.database.session("workspace-analytics") as session:
        persisted = await session.scalar(
            select(SocialPostMetric).where(
                SocialPostMetric.social_post_target_id == targets[0].id
            )
        )
    assert persisted is not None
    assert persisted.likes == 30
    assert persisted.raw_metrics["click_count"] == 9
    assert "linkedin-provider-secret" not in json.dumps(result, default=str)


async def test_linkedin_missing_analytics_scope_fails_closed_without_provider_call(
    phase6c_container,
) -> None:
    social = phase6c_container.social
    account = await _connected_account(
        phase6c_container,
        "workspace-no-analytics",
        account_type="linkedin_member",
        external_id="member_analytics",
        scopes=["openid", "profile", "w_member_social"],
    )
    post, _ = await social.publishing.posts.create(
        SocialPost(
            workspace_id="workspace-no-analytics",
            status="published",
            publish_mode="now",
        ),
        [SocialPostTarget(
            social_post_id="",
            social_account_id=account.id,
            provider="linkedin",
            status="published",
            external_post_id="urn:li:share:7132564752928563202",
        )],
    )
    result = await social.analytics.post("workspace-no-analytics", post.id)
    assert result["metrics"] == []
    assert result["unavailable"] == [{
        "provider": "linkedin",
        "status": "unavailable",
        "reason": "LINKEDIN_ANALYTICS_ADDITIONAL_AUTHORIZATION_REQUIRED",
        "required_scopes": ["r_member_postAnalytics"],
    }]


async def test_linkedin_workspace_isolation_covers_all_phase6c_resources(
    phase6c_container,
) -> None:
    social = phase6c_container.social
    member = await _connected_account(
        phase6c_container,
        "workspace-a",
        account_type="linkedin_member",
        external_id="member_a",
        scopes=["openid", "profile", "r_member_postAnalytics"],
    )
    organization = await _connected_account(
        phase6c_container,
        "workspace-a",
        account_type="linkedin_organization",
        external_id="5515715",
        scopes=["openid", "profile", "rw_organization_admin"],
    )
    post, targets = await social.publishing.posts.create(
        SocialPost(workspace_id="workspace-a", status="published", publish_mode="now"),
        [SocialPostTarget(
            social_post_id="",
            social_account_id=organization.id,
            provider="linkedin",
            status="published",
            external_post_id="urn:li:share:7132564752928563203",
        )],
    )
    asset = await social.media_assets.repository.create(
        SocialMediaAsset(
            workspace_id="workspace-a",
            request_id="11111111-1111-4111-8111-111111111111",
            filename="owned.mp4",
            mime_type="video/mp4",
            file_size=80_000,
        )
    )
    jobs = await social.jobs.repository.create_many([
        SocialJob(
            workspace_id="workspace-a",
            social_post_id=post.id,
            social_post_target_id=targets[0].id,
            provider="linkedin",
            status="queued",
            idempotency_key="workspace-a-job",
        )
    ])
    async with social.database.session("workspace-a") as session:
        session.add(SocialPostMetric(
            social_post_id=post.id,
            social_post_target_id=targets[0].id,
            provider="linkedin",
            impressions=1,
            raw_metrics={"source": "official"},
        ))
        await session.commit()

    for account_id in (member.id, organization.id):
        with pytest.raises(SocialAccountNotFoundError):
            await social.accounts.repository.get("workspace-b", account_id)
    with pytest.raises(SocialPostNotFoundError):
        await social.publishing.posts.get("workspace-b", post.id)
    with pytest.raises(SocialPostNotFoundError):
        await social.publishing.posts.set_target_status(
            "workspace-b", targets[0].id, "failed"
        )
    with pytest.raises(SocialJobNotFoundError):
        await social.jobs.repository.get("workspace-b", jobs[0].id)
    with pytest.raises(SocialMediaInvalidError):
        await social.media_assets.repository.get("workspace-b", asset.id)
    with pytest.raises(SocialPostNotFoundError):
        await social.analytics.post("workspace-b", post.id)


async def test_linkedin_revoked_and_expired_credentials_require_reauthorization(
    phase6c_container,
) -> None:
    social = phase6c_container.social
    revoked = await _connected_account(
        phase6c_container,
        "workspace-token",
        account_type="linkedin_member",
        external_id="member_revoked",
        scopes=["openid", "profile", "r_member_postAnalytics"],
    )
    await social.accounts.tokens.revoke("workspace-token", revoked.id)
    with pytest.raises(SocialReauthRequiredError):
        await social.oauth.token_for_request(
            workspace_id="workspace-token", account_id=revoked.id
        )

    expired = await _connected_account(
        phase6c_container,
        "workspace-token",
        account_type="linkedin_member",
        external_id="member_expired",
        scopes=["openid", "profile", "r_member_postAnalytics"],
    )
    await social.accounts.tokens.store(
        "workspace-token",
        expired.id,
        {"access_token": "expired-linkedin-token"},
        expires_at=datetime.now(timezone.utc) - timedelta(minutes=1),
        scopes=["openid", "profile", "r_member_postAnalytics"],
    )
    with pytest.raises(SocialReauthRequiredError):
        await social.oauth.token_for_request(
            workspace_id="workspace-token", account_id=expired.id
        )


@pytest.mark.parametrize(
    ("status_code", "retryable", "reauth"),
    [
        (429, True, False),
        (500, True, False),
        (502, True, False),
        (503, True, False),
        (504, True, False),
        (401, True, True),
        (403, False, False),
        (400, False, False),
    ],
)
def test_linkedin_retry_policy_is_bounded_and_classified(
    status_code: int, retryable: bool, reauth: bool
) -> None:
    decision = classify_retry(status_code=status_code, attempt=1)
    assert decision.retryable is retryable
    assert decision.refresh_token_first is reauth
    if status_code == 401:
        assert not classify_retry(status_code=401, attempt=2).retryable


async def test_linkedin_transient_retry_stops_at_job_attempt_limit() -> None:
    transitions: list[str] = []

    class Jobs:
        async def complete_attempt(self, *_: object, **__: object) -> None:
            return None

        async def transition(
            self, _: str, __: str, status: str, **___: object
        ) -> SocialJob:
            transitions.append(status)
            return job

    class Audit:
        async def record(self, **_: object) -> None:
            return None

    job = SocialJob(
        id="linkedin-job-limit",
        workspace_id="workspace-limit",
        social_post_id="linkedin-post-limit",
        provider="linkedin",
        status="publishing",
        attempt_count=5,
        max_attempts=5,
    )
    publisher = SocialPublisher(
        SimpleNamespace(
            jobs=SimpleNamespace(repository=Jobs()),
            audit=Audit(),
        )
    )
    await publisher._handle_failure(
        "workspace-limit",
        job,
        "linkedin-attempt-limit",
        SocialProviderUnavailableError("temporary LinkedIn failure"),
    )
    assert transitions == ["failed"]


async def test_linkedin_status_reconciliation_covers_all_normalized_states(
    tmp_path: Path,
) -> None:
    external_ids = {
        "urn:li:share:7132564752928563210": "published",
        "urn:li:share:7132564752928563211": "processing",
        "urn:li:share:7132564752928563212": "failed",
        "urn:li:share:7132564752928563213": "deleted",
        "urn:li:share:7132564752928563214": "unavailable",
    }

    async def handler(request: httpx.Request) -> httpx.Response:
        encoded = request.url.path.rsplit("/", 1)[-1]
        external_id = next(key for key in external_ids if key.split(":")[-1] in encoded)
        expected = external_ids[external_id]
        if expected == "deleted":
            return httpx.Response(404, json={"message": "not found"})
        lifecycle = {
            "published": "PUBLISHED",
            "processing": "PUBLISH_REQUESTED",
            "failed": "PUBLISH_FAILED",
            "unavailable": "UNKNOWN_PROVIDER_STATE",
        }[expected]
        return httpx.Response(200, json={"lifecycleState": lifecycle})

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(phase6c_settings(tmp_path), http_client=client)
    token = {
        "access_token": "linkedin-status-secret",
        "_mediarouter_granted_scopes": ["r_organization_social"],
    }
    try:
        results = {
            external_id: await provider.get_publish_status(token, external_id)
            for external_id in external_ids
        }
    finally:
        await client.aclose()
    assert {key: value["status"] for key, value in results.items()} == external_ids
    assert "linkedin-status-secret" not in json.dumps(results)


async def test_linkedin_public_views_logs_and_audit_boundaries_redact_credentials(
    phase6c_container,
) -> None:
    secret = "linkedin-secret-never-expose"
    account = SocialAccount(
        workspace_id="workspace-security",
        provider="linkedin",
        account_type="linkedin_member",
        external_account_id="member_secure",
        status="connected",
        metadata_json={"access_token": secret, "name": "Safe member"},
    )
    job = SocialJob(
        workspace_id="workspace-security",
        social_post_id="post-security",
        provider="linkedin",
        status="queued",
        payload_json={"refresh_token": secret},
        provider_state_encrypted=secret,
    )
    assert secret not in SocialAccountView.from_record(account).model_dump_json()
    assert secret not in SocialJobView.from_record(job).model_dump_json()
    record = logging.LogRecord(
        "linkedin-security",
        logging.ERROR,
        __file__,
        1,
        f"Authorization: Bearer {secret}",
        (),
        None,
    )
    record.provider_payload = {
        "refresh_token": secret,
        "message": f"access_token={secret}",
    }
    assert secret not in JsonFormatter().format(record)

    await phase6c_container.social.audit.record(
        workspace_id="workspace-security",
        event_type="SOCIAL_LINKEDIN_SECURITY_TEST",
        provider="linkedin",
        metadata={
            "client_secret": secret,
            "message": f"Authorization: Bearer {secret}",
        },
    )
    async with phase6c_container.social.database.session(
        "workspace-security"
    ) as session:
        audit = await session.scalar(
            select(SocialAuditEvent).where(
                SocialAuditEvent.event_type
                == "SOCIAL_LINKEDIN_SECURITY_TEST"
            )
        )
    assert audit is not None
    assert secret not in json.dumps(audit.metadata_json)


async def test_linkedin_mcp_contract_enforces_scope_and_never_exposes_credentials(
    phase6c_container,
) -> None:
    server = create_mcp_server(phase6c_container)
    tools = {tool.name for tool in await server.list_tools()}
    assert {
        "social.list_providers",
        "social.get_capabilities",
        "social.list_accounts",
        "social.create_post",
        "social.publish_post",
        "social.schedule_post",
        "social.get_job",
        "social.get_analytics",
    } <= tools
    context = AuthContext(
        api_key_id="workspace-linkedin",
        key_name="phase-6c",
        key_prefix="mp_test",
        environment="test",
        role="viewer",
        scopes=frozenset({"social:accounts:read"}),
        requests_per_minute=100,
        concurrent_jobs=2,
        uploads_per_hour=10,
        processing_bytes_per_day=1_000_000,
        expires_at=None,
    )
    auth_token = auth_context.set(context)
    http_token = http_auth_applied.set(True)
    called = False

    async def forbidden_action() -> dict[str, object]:
        nonlocal called
        called = True
        return {"access_token": "must-not-appear"}

    try:
        result = await MCPRegistry(phase6c_container).run_metadata_tool(
            "social.get_analytics",
            forbidden_action,
            required_scope="social:analytics:read",
        )
    finally:
        http_auth_applied.reset(http_token)
        auth_context.reset(auth_token)
    assert result["success"] is False
    assert result["error"]["code"] == "FORBIDDEN"
    assert not called
    assert "must-not-appear" not in json.dumps(result)


async def test_linkedin_analytics_timeout_is_retryable_and_secret_safe(
    tmp_path: Path,
) -> None:
    secret = "linkedin-timeout-secret"

    async def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout(
            f"Authorization: Bearer {secret}", request=request
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(phase6c_settings(tmp_path), http_client=client)
    try:
        with pytest.raises(SocialProviderUnavailableError) as raised:
            await provider.get_metrics(
                {
                    "access_token": secret,
                    "_mediarouter_account_type": "linkedin_organization",
                    "_mediarouter_external_account_id": "5515715",
                },
                "urn:li:share:7132564752928563220",
            )
    finally:
        await client.aclose()
    assert secret not in str(raised.value)
    assert classify_retry(
        status_code=raised.value.status_code, attempt=1
    ).retryable