tests/test_linkedin_publishing.py

from __future__ import annotations

import json
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 app.container import build_container
from app.core.config import Settings
from app.social.domain.errors import (
    SocialAccountNotFoundError,
    SocialCapabilityUnsupportedError,
    SocialIdempotencyConflictError,
    SocialMediaInvalidError,
    SocialPermissionDeniedError,
    SocialPostNotFoundError,
    SocialProviderUnavailableError,
    SocialPublishFailedError,
    SocialRateLimitedError,
    SocialReauthRequiredError,
)
from app.social.models import SocialAccount
from app.social.providers.linkedin import LINKEDIN_API_VERSION, LinkedInProvider
from app.social.schemas.linkedin import LinkedInPostMetadata
from app.social.schemas.posts import SocialPostCreate
from app.social.workers.publisher import SocialPublisher


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


def publishing_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-6b-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,
        "linkedin_publishing_enabled": True,
        "linkedin_media_processing_poll_seconds": 1,
        "linkedin_media_processing_timeout_seconds": 30,
        "temp_dir": tmp_path / "temp",
        "output_dir": tmp_path / "outputs",
        "cleanup_interval_seconds": 3600,
        "whisper_model": "tiny",
    }
    values.update(overrides)
    return Settings(**values)


def image_probe(*, codec: str = "png", frames: int = 1) -> dict[str, object]:
    return {
        "container": "png_pipe",
        "duration": None,
        "fps": 1.0,
        "resolution": {"width": 1200, "height": 675},
        "video_streams": [{"codec": codec, "frame_count": frames}],
        "audio_streams": [],
    }


def video_probe(**overrides: object) -> dict[str, object]:
    values: dict[str, object] = {
        "container": "mov,mp4,m4a,3gp,3g2,mj2",
        "duration": 15.0,
        "fps": 29.97,
        "resolution": {"width": 1280, "height": 720},
        "video_streams": [{"codec": "h264"}],
        "audio_streams": [{"codec": "aac", "sample_rate": 48_000}],
    }
    values.update(overrides)
    return values


def linkedin_post_payload(
    account_id: str,
    *,
    commentary: str = "A production-safe LinkedIn update",
    publish_mode: str = "draft",
    scheduled_at: datetime | None = None,
) -> SocialPostCreate:
    value: dict[str, object] = {
        "publish_mode": publish_mode,
        "targets": [{
            "social_account_id": account_id,
            "caption": {"commentary": commentary},
            "linkedin": {
                "post_type": "text",
                "commentary": commentary,
            },
        }],
    }
    if scheduled_at is not None:
        value.update({"scheduled_at": scheduled_at, "timezone": "Africa/Lagos"})
    return SocialPostCreate.model_validate(value)


async def connected_linkedin_account(
    container: object,
    workspace_id: str,
    *,
    scopes: list[str] | None = None,
) -> SocialAccount:
    social = container.social  # type: ignore[attr-defined]
    account = await social.accounts.repository.create(
        SocialAccount(
            workspace_id=workspace_id,
            provider="linkedin",
            account_type="linkedin_organization",
            external_account_id="5515715",
            username="mediarouter",
            display_name="MediaRouter",
            status="connected",
            metadata_json={
                "organization_urn": "urn:li:organization:5515715",
                "roles": ["ADMINISTRATOR"],
            },
        )
    )
    await social.accounts.tokens.store(
        workspace_id,
        account.id,
        {"access_token": "linkedin-provider-token"},
        expires_at=datetime.now(timezone.utc) + timedelta(hours=2),
        scopes=scopes
        or [
            "openid",
            "profile",
            "rw_organization_admin",
            "w_organization_social",
        ],
        token_type="bearer",
    )
    return account


async def test_capabilities_and_oauth_scopes_are_account_specific_and_gated(
    tmp_path: Path,
) -> None:
    disabled = LinkedInProvider(
        publishing_settings(tmp_path, linkedin_publishing_enabled=False)
    )
    enabled = LinkedInProvider(publishing_settings(tmp_path))
    try:
        assert not disabled.capabilities.direct_publish
        assert disabled.capabilities.account_type_publishing_scopes == {}
        assert enabled.capabilities.text
        assert enabled.capabilities.image
        assert enabled.capabilities.video
        assert enabled.capabilities.link
        assert enabled.capabilities.scheduled_publish
        assert not enabled.capabilities.native_scheduling
        assert enabled.capabilities.delete_post
        assert enabled.publishing_scopes("linkedin_member") == ["w_member_social"]
        assert enabled.publishing_scopes("linkedin_organization") == [
            "w_organization_social"
        ]
        with pytest.raises(SocialCapabilityUnsupportedError):
            enabled.publishing_scopes("unsupported")

        member = await enabled.get_authorization_url(
            state="s" * 43,
            redirect_uri=_REDIRECT_URI,
            additional_scopes=enabled.publishing_scopes("linkedin_member"),
        )
        organization = await enabled.get_authorization_url(
            state="o" * 43,
            redirect_uri=_REDIRECT_URI,
            additional_scopes=[
                *enabled.account_type_scopes("linkedin_organization"),
                *enabled.publishing_scopes("linkedin_organization"),
            ],
        )
        assert parse_qs(urlparse(member).query)["scope"] == [
            "openid profile w_member_social"
        ]
        assert parse_qs(urlparse(organization).query)["scope"] == [
            "openid profile rw_organization_admin w_organization_social"
        ]
    finally:
        await disabled.close()
        await enabled.close()


def test_linkedin_metadata_is_typed_and_media_requirement_is_explicit() -> None:
    text = LinkedInPostMetadata.model_validate(
        {"post_type": "text", "commentary": "Production post"}
    )
    assert text.to_post_body(author_urn="urn:li:person:member1")[
        "lifecycleState"
    ] == "PUBLISHED"
    with pytest.raises(ValidationError):
        LinkedInPostMetadata.model_validate(
            {"post_type": "text", "commentary": "ok", "provider_payload": {}}
        )
    with pytest.raises(ValidationError):
        LinkedInPostMetadata.model_validate({"post_type": "link"})
    with pytest.raises(ValidationError):
        SocialPostCreate.model_validate(
            {
                "targets": [
                    {
                        "social_account_id": "linkedin-account",
                        "linkedin": {"post_type": "video"},
                    }
                ]
            }
        )
    link = SocialPostCreate.model_validate(
        {
            "targets": [
                {
                    "social_account_id": "linkedin-account",
                    "linkedin": {
                        "post_type": "link",
                        "link": {
                            "source": "https://example.com/article",
                            "title": "Explicit title",
                            "description": "Explicit description",
                        },
                    },
                }
            ]
        }
    )
    assert link.media_asset_id is None


@pytest.mark.parametrize(
    ("account_type", "account_id", "expected_author"),
    [
        ("linkedin_member", "member_123", "urn:li:person:member_123"),
        (
            "linkedin_organization",
            "5515715",
            "urn:li:organization:5515715",
        ),
    ],
)
async def test_member_and_organization_text_publishing_use_posts_api(
    tmp_path: Path,
    account_type: str,
    account_id: str,
    expected_author: str,
) -> None:
    requests: list[httpx.Request] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        requests.append(request)
        assert request.url == "https://api.linkedin.com/rest/posts"
        assert request.headers["linkedin-version"] == LINKEDIN_API_VERSION
        assert request.headers["x-restli-protocol-version"] == "2.0.0"
        assert request.headers["authorization"] == "Bearer linkedin-token"
        body = json.loads(request.content)
        assert body["author"] == expected_author
        assert body["commentary"] == "Production post"
        assert body["distribution"]["feedDistribution"] == "MAIN_FEED"
        return httpx.Response(
            201,
            headers={"x-restli-id": "urn:li:share:6844785523593134080"},
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(publishing_settings(tmp_path), http_client=client)
    states: list[dict[str, object] | None] = []

    async def persist(value: dict[str, object] | None) -> None:
        states.append(value)

    try:
        result = await provider.publish(
            {"access_token": "linkedin-token"},
            {
                "provider_account_id": account_id,
                "provider_account_type": account_type,
                "linkedin_post_metadata": {
                    "post_type": "text",
                    "commentary": "Production post",
                },
                "provider_state": {"linkedin_post_submission_attempted": False},
                "persist_provider_state": persist,
                "upload": {"identity_type": "none"},
            },
        )
        assert result["id"] == "urn:li:share:6844785523593134080"
        assert result["status"] == "published"
        assert len(requests) == 1
        assert states[-1]["linkedin_post_submission_attempted"] is True  # type: ignore[index]
    finally:
        await client.aclose()


async def test_image_upload_streams_with_oauth_and_creates_media_urn(
    tmp_path: Path,
) -> None:
    path = tmp_path / "image.png"
    path.write_bytes(b"image" * 1024)
    expires = int((datetime.now(timezone.utc) + timedelta(hours=1)).timestamp() * 1000)
    calls: list[str] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        calls.append(f"{request.method} {request.url.path}")
        if request.url.path == "/rest/images":
            assert request.url.params["action"] == "initializeUpload"
            assert json.loads(request.content)["initializeUploadRequest"]["owner"] == (
                "urn:li:organization:5515715"
            )
            return httpx.Response(
                200,
                json={
                    "value": {
                        "uploadUrlExpiresAt": expires,
                        "uploadUrl": "https://www.linkedin.com/dms-uploads/image/upload",
                        "image": "urn:li:image:C4E10AQFoyyAjHPMQuQ",
                    }
                },
            )
        assert request.url == "https://www.linkedin.com/dms-uploads/image/upload"
        assert request.method == "PUT"
        assert request.headers["authorization"] == "Bearer linkedin-token"
        assert len(request.content) == path.stat().st_size
        return httpx.Response(201)

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(publishing_settings(tmp_path), http_client=client)
    state: dict[str, object] = {}

    async def persist(value: dict[str, object] | None) -> None:
        state.clear()
        state.update(value or {})

    try:
        result = await provider.upload_media(
            {"access_token": "linkedin-token"},
            {
                "path": path,
                "file_size": path.stat().st_size,
                "mime_type": "image/png",
                "probe": image_probe(),
                "provider_account_id": "5515715",
                "provider_account_type": "linkedin_organization",
                "linkedin_post_metadata": {
                    "post_type": "image",
                    "commentary": "Image post",
                    "image_alt_text": "Accessible description",
                },
                "provider_state": {},
                "persist_provider_state": persist,
            },
        )
        assert result["id"] == "urn:li:image:C4E10AQFoyyAjHPMQuQ"
        assert state["linkedin_image_uploaded"] is True
        assert calls == ["POST /rest/images", "PUT /dms-uploads/image/upload"]
        assert "linkedin-token" not in str(state)
    finally:
        await client.aclose()


async def test_video_multipart_upload_streams_ranges_and_finalizes(
    tmp_path: Path,
) -> None:
    path = tmp_path / "video.mp4"
    path.write_bytes(b"a" * 80_000)
    expires = int((datetime.now(timezone.utc) + timedelta(hours=1)).timestamp() * 1000)
    put_bodies: list[bytes] = []
    finalized: list[dict[str, object]] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/rest/videos" and request.url.params.get("action") == "initializeUpload":
            return httpx.Response(
                200,
                json={
                    "value": {
                        "video": "urn:li:video:C4E10AQEfKKMV9a1d-g",
                        "uploadToken": "opaque-upload-token",
                        "uploadUrlsExpireAt": expires,
                        "uploadInstructions": [
                            {
                                "firstByte": 0,
                                "lastByte": 39_999,
                                "uploadUrl": "https://www.linkedin.com/dms-uploads/video/part-0",
                            },
                            {
                                "firstByte": 40_000,
                                "lastByte": 79_999,
                                "uploadUrl": "https://www.linkedin.com/dms-uploads/video/part-1",
                            },
                        ],
                    }
                },
            )
        if request.method == "PUT":
            assert "authorization" not in request.headers
            put_bodies.append(request.content)
            return httpx.Response(200, headers={"ETag": f'"part-{len(put_bodies)}"'})
        assert request.url.params["action"] == "finalizeUpload"
        finalized.append(json.loads(request.content)["finalizeUploadRequest"])
        return httpx.Response(200, json={})

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(publishing_settings(tmp_path), http_client=client)
    state: dict[str, object] = {}

    async def persist(value: dict[str, object] | None) -> None:
        state.clear()
        state.update(value or {})

    try:
        result = await provider.upload_media(
            {
                "access_token": "linkedin-token",
                "_mediarouter_granted_scopes": ["w_member_social"],
            },
            {
                "path": path,
                "file_size": path.stat().st_size,
                "mime_type": "video/mp4",
                "probe": video_probe(),
                "provider_account_id": "member_123",
                "provider_account_type": "linkedin_member",
                "linkedin_post_metadata": {
                    "post_type": "video",
                    "commentary": "Video post",
                },
                "provider_state": {},
                "persist_provider_state": persist,
            },
        )
        assert result["id"] == "urn:li:video:C4E10AQEfKKMV9a1d-g"
        assert [len(body) for body in put_bodies] == [40_000, 40_000]
        assert finalized[0]["uploadedPartIds"] == ["part-1", "part-2"]
        assert state["linkedin_video_finalized"] is True
        assert "linkedin-token" not in str(state)
    finally:
        await client.aclose()


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

    async def handler(request: httpx.Request) -> httpx.Response:
        methods.append(request.method)
        assert request.url.raw_path.decode().split("?", 1)[0].endswith(
            "/urn%3Ali%3Ashare%3A6844785523593134080"
        )
        if request.method == "GET":
            assert request.url.params["viewContext"] == "AUTHOR"
            return httpx.Response(
                200,
                json={
                    "id": "urn:li:share:6844785523593134080",
                    "lifecycleState": "PUBLISH_REQUESTED",
                },
            )
        assert request.headers["x-restli-method"] == "DELETE"
        return httpx.Response(204)

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(publishing_settings(tmp_path), http_client=client)
    token = {
        "access_token": "linkedin-token",
        "_mediarouter_granted_scopes": [
            "w_organization_social",
            "r_organization_social",
        ],
    }
    try:
        status = await provider.get_publish_status(
            token, "urn:li:share:6844785523593134080"
        )
        await provider.delete_post(
            token, "urn:li:share:6844785523593134080"
        )
        assert status["status"] == "processing"
        assert methods == ["GET", "DELETE"]
    finally:
        await client.aclose()


async def test_uncertain_linkedin_create_outcome_never_resubmits(
    tmp_path: Path,
) -> None:
    create_calls = 0
    state: dict[str, object] = {
        "linkedin_post_submission_attempted": False,
        "linkedin_publish_started_at": datetime.now(timezone.utc).isoformat(),
    }

    async def persist(value: dict[str, object] | None) -> None:
        state.clear()
        state.update(value or {})

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal create_calls
        create_calls += 1
        raise httpx.ReadTimeout(
            "response lost after provider acceptance", request=request
        )

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(publishing_settings(tmp_path), http_client=client)
    payload = {
        "provider_account_id": "member_123",
        "provider_account_type": "linkedin_member",
        "linkedin_post_metadata": {
            "post_type": "text",
            "commentary": "One logical post",
        },
        "provider_state": state,
        "persist_provider_state": persist,
        "upload": {"identity_type": "none"},
    }
    try:
        with pytest.raises(SocialProviderUnavailableError):
            await provider.publish(
                {"access_token": "linkedin-token"}, payload
            )
        assert state["linkedin_post_submission_attempted"] is True
        with pytest.raises(SocialProviderUnavailableError, match="uncertain"):
            await provider.reconcile_pending_publish(
                {"access_token": "linkedin-token"},
                {**payload, "provider_state": state},
            )
        with pytest.raises(SocialProviderUnavailableError, match="uncertain"):
            await provider.publish(
                {"access_token": "linkedin-token"},
                {**payload, "provider_state": state},
            )
        assert create_calls == 1
    finally:
        await client.aclose()


async def test_media_validation_status_delete_errors_and_unknown_outcome(
    tmp_path: Path,
) -> None:
    path = tmp_path / "video.mp4"
    path.write_bytes(b"a" * 80_000)
    provider = LinkedInProvider(publishing_settings(tmp_path))
    try:
        with pytest.raises(SocialMediaInvalidError):
            await provider.validate_media(
                {
                    "path": path,
                    "file_size": path.stat().st_size,
                    "mime_type": "video/mp4",
                    "probe": video_probe(fps=30.0),
                    "linkedin_post_metadata": {"post_type": "video"},
                }
            )
        with pytest.raises(SocialProviderUnavailableError):
            await provider.reconcile_pending_publish(
                {"access_token": "secret"},
                {
                    "provider_state": {
                        "linkedin_post_submission_attempted": True
                    }
                },
            )
        disabled = LinkedInProvider(
            publishing_settings(tmp_path, linkedin_publishing_enabled=False)
        )
        try:
            assert disabled.publishing_scopes("linkedin_member") == []
        finally:
            await disabled.close()
    finally:
        await provider.close()


async def test_linkedin_idempotency_scheduling_authorization_and_workspace_isolation(
    tmp_path: Path,
) -> None:
    container = build_container(publishing_settings(tmp_path))
    await container.social.initialize()
    try:
        account = await connected_linkedin_account(container, "workspace-a")
        payload = linkedin_post_payload(account.id)
        first = await container.social.publishing.create(
            workspace_id="workspace-a",
            user_id="user-a",
            payload=payload,
            idempotency_key="linkedin-create-key",
        )
        duplicate = await container.social.publishing.create(
            workspace_id="workspace-a",
            user_id="user-a",
            payload=payload,
            idempotency_key="linkedin-create-key",
        )
        assert duplicate.id == first.id
        with pytest.raises(SocialIdempotencyConflictError):
            await container.social.publishing.create(
                workspace_id="workspace-a",
                user_id="user-a",
                payload=linkedin_post_payload(
                    account.id, commentary="Different payload"
                ),
                idempotency_key="linkedin-create-key",
            )
        scheduled = await container.social.publishing.create(
            workspace_id="workspace-a",
            user_id="user-a",
            payload=linkedin_post_payload(
                account.id,
                publish_mode="schedule",
                scheduled_at=datetime.now(timezone.utc) + timedelta(hours=1),
            ),
            idempotency_key="linkedin-schedule-key",
        )
        assert scheduled.status.value == "scheduled"
        with pytest.raises(SocialAccountNotFoundError):
            await container.social.publishing.create(
                workspace_id="workspace-b",
                user_id="user-b",
                payload=linkedin_post_payload(account.id),
                idempotency_key="workspace-b-key",
            )
        with pytest.raises(SocialPostNotFoundError):
            await container.social.publishing.delete("workspace-b", first.id)

        read_only = await connected_linkedin_account(
            container,
            "workspace-read-only",
            scopes=["openid", "profile", "rw_organization_admin"],
        )
        with pytest.raises(SocialPermissionDeniedError):
            await container.social.publishing.create(
                workspace_id="workspace-read-only",
                user_id="user-read-only",
                payload=linkedin_post_payload(
                    read_only.id, publish_mode="now"
                ),
                idempotency_key="missing-linkedin-write-scope",
            )
    finally:
        await container.social.close()
        await container.security_database.close()


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

    async def handler(request: httpx.Request) -> httpx.Response:
        nonlocal create_calls
        assert request.url.path == "/rest/posts"
        create_calls += 1
        return httpx.Response(
            201,
            headers={"x-restli-id": "urn:li:share:6844785523593134081"},
        )

    adapter._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    adapter._owns_client = True
    try:
        account = await connected_linkedin_account(
            container, "workspace-worker"
        )
        post = await container.social.publishing.create(
            workspace_id="workspace-worker",
            user_id="worker-user",
            payload=linkedin_post_payload(
                account.id,
                commentary="Worker post",
                publish_mode="now",
            ),
            idempotency_key="linkedin-worker-key",
        )
        jobs = await container.social.jobs.list("workspace-worker")
        assert len(jobs) == 1
        await SocialPublisher(container.social).process(
            "workspace-worker", jobs[0].id
        )
        stored = await container.social.publishing.get(
            "workspace-worker", post.id
        )
        stored_job = await container.social.jobs.get(
            "workspace-worker", jobs[0].id
        )
        assert stored.status.value == "published"
        assert (
            stored.targets[0].external_post_id
            == "urn:li:share:6844785523593134081"
        )
        assert stored_job.status.value == "published"
        assert create_calls == 1
        serialized = stored.model_dump_json() + stored_job.model_dump_json()
        assert "linkedin-provider-token" not in serialized
    finally:
        await container.social.close()
        await container.security_database.close()


@pytest.mark.parametrize(
    ("status", "error"),
    [
        (400, SocialPublishFailedError),
        (401, SocialReauthRequiredError),
        (403, SocialPermissionDeniedError),
        (429, SocialRateLimitedError),
        (500, SocialProviderUnavailableError),
        (502, SocialProviderUnavailableError),
        (503, SocialProviderUnavailableError),
        (504, SocialProviderUnavailableError),
    ],
)
async def test_linkedin_error_normalization(
    tmp_path: Path, status: int, error: type[Exception]
) -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status, json={"message": "secret-provider-detail"})

    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
    provider = LinkedInProvider(publishing_settings(tmp_path), http_client=client)
    try:
        with pytest.raises(error) as raised:
            await provider.delete_post(
                {"access_token": "linkedin-token"},
                "urn:li:share:6844785523593134080",
            )
        assert "linkedin-token" not in str(raised.value)
        assert "secret-provider-detail" not in str(raised.value)
    finally:
        await client.aclose()
    SocialIdempotencyConflictError,
    SocialPostNotFoundError,