tests/test_linkedin_live.py

"""Opt-in, destructive LinkedIn integration verification.

Normal CI always skips this module. Run it only with a dedicated LinkedIn
member and, when organization coverage is required, a dedicated organization.
Provider credentials are read from the process environment and never logged.
"""

from __future__ import annotations

import asyncio
import os
from pathlib import Path
from urllib.parse import parse_qs, urlparse

import pytest


pytestmark = pytest.mark.skipif(
    os.getenv("RUN_LINKEDIN_INTEGRATION_TESTS", "").lower() != "true",
    reason=(
        "LinkedIn live integration is NOT VERIFIED; set "
        "RUN_LINKEDIN_INTEGRATION_TESTS=true with dedicated credentials."
    ),
)


def _required(name: str) -> str:
    value = os.getenv(name, "").strip()
    if not value:
        pytest.skip(f"LinkedIn live integration is NOT VERIFIED; missing {name}.")
    return value


def _granted_scopes() -> list[str]:
    value = os.getenv("LINKEDIN_LIVE_TEST_GRANTED_SCOPES", "")
    return list(dict.fromkeys(value.replace(",", " ").split()))


def _require_base_configuration() -> None:
    for name in (
        "LINKEDIN_CLIENT_ID",
        "LINKEDIN_CLIENT_SECRET",
        "LINKEDIN_REDIRECT_URI",
        "LINKEDIN_LIVE_TEST_ACCESS_TOKEN",
    ):
        _required(name)


def _require_publish_consent() -> None:
    _require_base_configuration()
    if os.getenv("LINKEDIN_LIVE_TEST_ALLOW_PUBLISH", "").lower() != "true":
        pytest.skip(
            "Set LINKEDIN_LIVE_TEST_ALLOW_PUBLISH=true to create test posts."
        )
    if os.getenv("LINKEDIN_LIVE_TEST_DELETE", "").lower() != "true":
        pytest.skip(
            "Set LINKEDIN_LIVE_TEST_DELETE=true to require deletion of test posts."
        )


def _settings():
    from app.core.config import Settings

    return Settings(
        _env_file=None,
        auth_enabled=False,
        linkedin_client_id=_required("LINKEDIN_CLIENT_ID"),
        linkedin_client_secret=_required("LINKEDIN_CLIENT_SECRET"),
        linkedin_redirect_uri=_required("LINKEDIN_REDIRECT_URI"),
        linkedin_publishing_enabled=True,
        whisper_model="tiny",
    )


def _token() -> dict[str, object]:
    return {
        "access_token": _required("LINKEDIN_LIVE_TEST_ACCESS_TOKEN"),
        "_mediarouter_granted_scopes": _granted_scopes(),
    }


async def _identity(provider, token: dict[str, object]) -> tuple[str, str]:
    account_type = os.getenv(
        "LINKEDIN_LIVE_TEST_ACCOUNT_TYPE", "linkedin_member"
    ).strip()
    if account_type == "linkedin_member":
        member = await provider.get_account(token)
        return account_type, str(member["external_account_id"])
    if account_type != "linkedin_organization":
        pytest.skip(
            "LINKEDIN_LIVE_TEST_ACCOUNT_TYPE must be linkedin_member or "
            "linkedin_organization."
        )
    organization_id = _required("LINKEDIN_LIVE_TEST_ORGANIZATION_ID")
    discovered = await provider.discover_accounts(
        token, account_type="linkedin_organization"
    )
    organizations = {
        str(account["external_account_id"]): account
        for account in discovered
        if account.get("account_type") == "linkedin_organization"
    }
    if organization_id not in organizations:
        pytest.fail(
            "The configured LinkedIn organization was not returned by official "
            "organization-access discovery."
        )
    return account_type, organization_id


async def _publish_text(provider, token: dict[str, object]) -> dict[str, object]:
    account_type, account_id = await _identity(provider, token)
    state: dict[str, object] = {}

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

    return await provider.publish(
        token,
        {
            "provider_account_id": account_id,
            "provider_account_type": account_type,
            "linkedin_post_metadata": {
                "post_type": "text",
                "commentary": (
                    "MediaRouter Phase 6C live integration verification"
                ),
            },
            "upload": {"identity_type": "none"},
            "provider_state": state,
            "persist_provider_state": persist,
        },
    )


def test_linkedin_live_configuration_requires_explicit_opt_in() -> None:
    assert os.getenv("RUN_LINKEDIN_INTEGRATION_TESTS", "").lower() == "true"
    _require_base_configuration()


async def test_linkedin_live_authorization_url_and_account_discovery() -> None:
    """Verify the official authorization contract and current identity token."""

    _require_base_configuration()
    from app.social.providers.linkedin import LinkedInProvider

    provider = LinkedInProvider(_settings())
    token = _token()
    account_type = os.getenv(
        "LINKEDIN_LIVE_TEST_ACCOUNT_TYPE", "linkedin_member"
    ).strip()
    additional_scopes = provider.account_type_scopes(account_type)
    try:
        authorization_url = await provider.get_authorization_url(
            state="phase6c-live-linkedin-state-value-that-is-long-enough",
            redirect_uri=_required("LINKEDIN_REDIRECT_URI"),
            additional_scopes=additional_scopes,
        )
        parsed = urlparse(authorization_url)
        assert parsed.scheme == "https"
        assert parsed.netloc == "www.linkedin.com"
        assert parsed.path == "/oauth/v2/authorization"
        assert "code_challenge" not in parse_qs(parsed.query)
        discovered_type, external_id = await _identity(provider, token)
        assert discovered_type == account_type
        assert external_id
    finally:
        await provider.close()


async def test_linkedin_live_authorization_code_exchange_when_supplied() -> None:
    """A fresh one-time browser code is optional and never required by CI."""

    _require_base_configuration()
    code = os.getenv("LINKEDIN_LIVE_TEST_AUTHORIZATION_CODE", "").strip()
    if not code:
        pytest.skip(
            "LinkedIn OAuth code exchange is NOT VERIFIED; provide a fresh "
            "LINKEDIN_LIVE_TEST_AUTHORIZATION_CODE."
        )
    from app.social.providers.linkedin import LinkedInProvider

    provider = LinkedInProvider(_settings())
    try:
        token = await provider.exchange_code(
            code=code,
            redirect_uri=_required("LINKEDIN_REDIRECT_URI"),
        )
        account = await provider.get_account(token)
        assert account["external_account_id"]
    finally:
        await provider.close()


async def test_linkedin_live_text_publish_status_and_delete() -> None:
    _require_publish_consent()
    from app.social.providers.linkedin import LinkedInProvider

    provider = LinkedInProvider(_settings())
    token = _token()
    account_type = os.getenv(
        "LINKEDIN_LIVE_TEST_ACCOUNT_TYPE", "linkedin_member"
    ).strip()
    read_scope = {
        "linkedin_member": "r_member_social",
        "linkedin_organization": "r_organization_social",
    }.get(account_type)
    if read_scope is None or read_scope not in _granted_scopes():
        await provider.close()
        pytest.skip(
            "LinkedIn live status reconciliation is NOT VERIFIED; declare the "
            f"approved {read_scope or 'account read'} scope."
        )
    external_id: str | None = None
    try:
        result = await _publish_text(provider, token)
        external_id = str(result["id"])
        status: dict[str, object] | None = None
        for _ in range(12):
            status = await provider.get_publish_status(token, external_id)
            if status.get("status") in {"published", "failed", "deleted"}:
                break
            await asyncio.sleep(5)
        assert status is not None and status.get("status") == "published"
    finally:
        if external_id:
            await provider.delete_post(token, external_id)
        await provider.close()


async def test_linkedin_live_media_publish_when_asset_is_supplied() -> None:
    _require_publish_consent()
    media_value = os.getenv("LINKEDIN_LIVE_TEST_MEDIA_PATH", "").strip()
    if not media_value:
        pytest.skip(
            "LinkedIn media publishing is NOT VERIFIED; set "
            "LINKEDIN_LIVE_TEST_MEDIA_PATH to a dedicated image or MP4 asset."
        )
    from app.services.ffprobe_service import FFprobeService
    from app.services.validator import MediaValidator
    from app.social.providers.linkedin import LinkedInProvider

    media_path = Path(media_value).expanduser().resolve()
    if not media_path.is_file():
        pytest.skip("LINKEDIN_LIVE_TEST_MEDIA_PATH is not a readable file.")
    settings = _settings()
    provider = LinkedInProvider(settings)
    token = _token()
    state: dict[str, object] = {}

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

    external_id: str | None = None
    try:
        account_type, account_id = await _identity(provider, token)
        probe = await FFprobeService(settings).probe(media_path)
        mime_type = MediaValidator(settings).infer_mime(media_path)
        post_type = "image" if mime_type.startswith("image/") else "video"
        media = {
            "path": media_path,
            "mime_type": mime_type,
            "file_size": media_path.stat().st_size,
            "probe": probe,
            "provider_account_id": account_id,
            "provider_account_type": account_type,
            "linkedin_post_metadata": {
                "post_type": post_type,
                "commentary": "MediaRouter Phase 6C media verification",
            },
            "provider_state": state,
            "persist_provider_state": persist,
        }
        await provider.validate_media(media)
        uploaded = await provider.upload_media(token, media)
        result = await provider.publish(
            token,
            {
                "provider_account_id": account_id,
                "provider_account_type": account_type,
                "linkedin_post_metadata": media["linkedin_post_metadata"],
                "upload": uploaded,
                "provider_state": state,
                "persist_provider_state": persist,
            },
        )
        external_id = str(result["id"])
        assert external_id.startswith("urn:li:")
    finally:
        if external_id:
            await provider.delete_post(token, external_id)
        await provider.close()


async def test_linkedin_live_analytics_when_authorized() -> None:
    _require_publish_consent()
    from app.social.providers.linkedin import LinkedInProvider

    provider = LinkedInProvider(_settings())
    token = _token()
    external_id: str | None = None
    try:
        account_type, account_id = await _identity(provider, token)
        required_scopes = provider.analytics_scopes(account_type)
        if not required_scopes or required_scopes[0] not in _granted_scopes():
            pytest.skip(
                "LinkedIn analytics are NOT VERIFIED; the dedicated token does "
                "not declare the required analytics grant."
            )
        result = await _publish_text(provider, token)
        external_id = str(result["id"])
        metrics = await provider.get_metrics(
            {
                **token,
                "_mediarouter_account_type": account_type,
                "_mediarouter_external_account_id": account_id,
            },
            external_id,
        )
        assert metrics["status"] == "available"
        assert isinstance(metrics.get("raw_metrics"), dict)
    finally:
        if external_id:
            await provider.delete_post(token, external_id)
        await provider.close()