"""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