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,