"""Phase 3C unit coverage for Meta analytics and boundary hardening. All Graph calls use MockTransport. Live tests remain opt-in so normal CI never requires a Page, professional account, browser consent, or Meta credentials. """ from __future__ import annotations import os from pathlib import Path from urllib.parse import parse_qs, urlparse import httpx import pytest from app.container import build_container from app.core.config import Settings from app.social.domain.errors import SocialReauthRequiredError from app.social.models import SocialAccount, SocialJob, SocialPost, SocialPostTarget from app.social.providers.facebook import FacebookProvider from app.social.providers.instagram import InstagramProvider from app.social.schemas.accounts import SocialAccountView from app.social.schemas.jobs import SocialJobView def meta_settings(tmp_path: Path) -> Settings: return Settings( _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="test-only-encryption-material", meta_app_id="meta-app-id", meta_app_secret="meta-app-secret", temp_dir=tmp_path / "temp", output_dir=tmp_path / "outputs", cleanup_interval_seconds=3600, whisper_model="tiny", ) @pytest.fixture async def meta_container(tmp_path: Path): container = build_container(meta_settings(tmp_path)) await container.social.initialize() try: yield container finally: await container.social.close() await container.security_database.close() async def test_facebook_page_metrics_use_v25_bearer_auth_and_normalize(tmp_path: Path) -> None: async def handler(request: httpx.Request) -> httpx.Response: assert request.url.path == "/v25.0/page-post-id" assert request.headers["authorization"] == "Bearer token-that-must-not-enter-url" assert "access_token" not in request.url.query.decode() return httpx.Response( 200, json={ "created_time": "2026-07-31T00:00:00+0000", "insights": { "data": [ {"name": "post_impressions", "values": [{"value": 42}]}, {"name": "post_video_views", "values": [{"value": 11}]}, ] }, "reactions": {"summary": {"total_count": 7}}, "comments": {"summary": {"total_count": 3}}, "shares": {"count": 2}, }, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = FacebookProvider(meta_settings(tmp_path), http_client=client) try: result = await provider.get_metrics({"access_token": "token-that-must-not-enter-url"}, "page-post-id") finally: await client.aclose() assert result["status"] == "available" assert result["impressions"] == 42 assert result["views"] == 11 assert result["likes"] == 7 assert result["comments"] == 3 assert result["shares"] == 2 async def test_instagram_reel_metrics_are_media_type_specific(tmp_path: Path) -> None: requests: list[httpx.Request] = [] async def handler(request: httpx.Request) -> httpx.Response: requests.append(request) assert request.headers["authorization"] == "Bearer meta-token" if request.url.path == "/v25.0/ig-media-id": return httpx.Response( 200, json={ "media_product_type": "REELS", "media_type": "VIDEO", "timestamp": "2026-07-31T00:00:00+0000", "permalink": "https://www.instagram.com/reel/example/", }, ) assert request.url.path == "/v25.0/ig-media-id/insights" assert parse_qs(request.url.query.decode())["metric"] == ["views,reach,likes,comments,shares,saved"] return httpx.Response( 200, json={ "data": [ {"name": "views", "values": [{"value": 100}]}, {"name": "reach", "values": [{"value": 80}]}, {"name": "likes", "values": [{"value": 20}]}, {"name": "comments", "values": [{"value": 4}]}, {"name": "shares", "values": [{"value": 2}]}, {"name": "saved", "values": [{"value": 9}]}, ] }, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = InstagramProvider(meta_settings(tmp_path), http_client=client) try: result = await provider.get_metrics({"access_token": "meta-token"}, "ig-media-id") finally: await client.aclose() assert len(requests) == 2 assert result["status"] == "available" assert result["views"] == 100 assert result["shares"] == 2 assert result["url"] == "https://www.instagram.com/reel/example/" async def test_meta_graph_authentication_error_is_safe_and_reauth_required(tmp_path: Path) -> None: secret = "never-return-this-access-token" async def handler(_: httpx.Request) -> httpx.Response: return httpx.Response(400, json={"error": {"code": 190, "message": secret}}) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = FacebookProvider(meta_settings(tmp_path), http_client=client) try: with pytest.raises(SocialReauthRequiredError) as raised: await provider.get_metrics({"access_token": secret}, "page-post-id") finally: await client.aclose() assert secret not in str(raised.value) async def test_meta_analytics_requires_explicit_authorization_and_persists_safe_snapshot(meta_container) -> None: account = await meta_container.social.accounts.repository.create( SocialAccount( workspace_id="workspace-meta", provider="facebook", account_type="facebook_page", external_account_id="page-id", status="connected", ) ) await meta_container.social.accounts.tokens.store( "workspace-meta", account.id, {"access_token": "stored-token"}, scopes=["pages_read_engagement"] ) readiness = await meta_container.social.analytics.account("workspace-meta", account.id) assert readiness["status"] == "unavailable" assert readiness["reason"] == "META_ANALYTICS_ADDITIONAL_AUTHORIZATION_REQUIRED" assert readiness["required_scopes"] == ["read_insights"] await meta_container.social.accounts.tokens.store( "workspace-meta", account.id, {"access_token": "stored-token"}, scopes=["pages_read_engagement", "read_insights"], ) post, targets = await meta_container.social.publishing.posts.create( SocialPost(workspace_id="workspace-meta", media_asset_id="owned-asset"), [ SocialPostTarget( social_post_id="", social_account_id=account.id, provider="facebook", status="published", external_post_id="page-post-id", ) ], ) assert targets adapter = meta_container.social.accounts.providers.get("facebook") async def metrics(_: dict[str, object], __: str) -> dict[str, object]: return { "status": "available", "views": 8, "impressions": 12, "likes": 3, "comments": 1, "shares": 2, "raw_metrics": {"access_token": "must-not-leak", "provider_value": 8}, } adapter.get_metrics = metrics # type: ignore[method-assign] result = await meta_container.social.analytics.post("workspace-meta", post.id) assert result["metrics"][0]["views"] == 8 assert "access_token" not in str(result) assert result["metrics"][0]["raw_metrics"] == {"provider_value": 8} async def test_meta_analytics_consent_is_explicit_and_never_added_to_normal_connection(tmp_path: Path) -> None: provider = FacebookProvider(meta_settings(tmp_path)) try: normal = await provider.get_authorization_url( state="s" * 32, redirect_uri="https://api.example/callback" ) analytics = await provider.get_authorization_url( state="a" * 32, redirect_uri="https://api.example/callback", additional_scopes=provider.capabilities.analytics_required_scopes, ) finally: await provider.close() assert "read_insights" not in parse_qs(urlparse(normal).query).get("scope", [""])[0] requested = parse_qs(urlparse(analytics).query)["scope"][0].split() assert requested == ["pages_read_engagement", "read_insights"] def test_public_social_views_remove_token_like_data() -> None: secret = "never-expose-me" account = SocialAccount( workspace_id="workspace-a", provider="facebook", account_type="facebook_page", external_account_id="page-a", status="connected", metadata_json={"access_token": secret, "nested": {"client_secret": secret}, "name": "Page"}, ) job = SocialJob( workspace_id="workspace-a", social_post_id="post-a", provider="facebook", status="queued", payload_json={"access_token": secret, "media_asset_id": "asset-a"}, ) assert secret not in SocialAccountView.from_record(account).model_dump_json() assert secret not in SocialJobView.from_record(job).model_dump_json() @pytest.mark.skipif( os.getenv("RUN_META_INTEGRATION_TESTS") != "true", reason="Set RUN_META_INTEGRATION_TESTS=true with dedicated Meta test credentials.", ) async def test_live_meta_page_post_insights() -> None: """Optional live smoke test; OAuth/publishing require separate manual consent setup. Required CI-secret variables are deliberately not named or logged by the application. This test uses a dedicated Page post and never publishes. """ token = os.environ.get("META_TEST_PAGE_ACCESS_TOKEN") post_id = os.environ.get("META_TEST_PAGE_POST_ID") if not token or not post_id: pytest.skip("META_TEST_PAGE_ACCESS_TOKEN and META_TEST_PAGE_POST_ID are not configured.") settings = Settings(_env_file=None, meta_graph_api_version="v25.0", whisper_model="tiny") provider = FacebookProvider(settings) try: result = await provider.get_metrics({"access_token": token}, post_id) finally: await provider.close() assert result["status"] in {"available", "unavailable"}