import hashlib, hmac, json from datetime import datetime, timedelta, timezone from pathlib import Path from fastapi.testclient import TestClient from app.main import app from app.db import init_db from app.models import PageContext, Channel, RiskLevel from app.services.event_normalizer import normalize_meta_payload from app.services.risk_rules import classify_risk, requires_human_approval init_db() client = TestClient(app) def sig(body: bytes, secret="example-secret") -> str: return "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() def test_healthz(): assert client.get("/healthz").json() == {"status": "ok"} def test_webhook_verify(): r = client.get("/meta/webhook", params={"hub.mode":"subscribe","hub.verify_token":"verify-test","hub.challenge":"12345"}) assert r.status_code == 200 assert r.json() == 12345 def test_webhook_rejects_missing_signature(): r = client.post("/meta/webhook", json={"entry":[]}) assert r.status_code == 403 def test_webhook_accepts_valid_signature_and_stores(): body = json.dumps({"entry":[{"id":"page-healthcare","messaging":[{"sender":{"id":"u1"},"recipient":{"id":"page-healthcare"},"message":{"mid":"m1","text":"hello"}}]}]}, separators=(",", ":")).encode() r = client.post("/meta/webhook", content=body, headers={"X-Hub-Signature-256": sig(body), "Content-Type":"application/json"}) assert r.status_code == 200 assert r.json()["stored_event_ids"] def test_unknown_page_fails_closed(): try: normalize_meta_payload({"entry":[{"id":"unknown","messaging":[]}]}) except ValueError as e: assert "Unrecognized" in str(e) else: raise AssertionError("unknown page accepted") def test_internal_auth_rejection(): r = client.post("/tools/meta/draft", json={}) assert r.status_code == 401 def test_ccm_clinical_risk_high(): risk = classify_risk(PageContext.healthcare, Channel.messenger, "Should I change my medication dose?", False) assert risk == RiskLevel.high assert requires_human_approval(PageContext.healthcare, Channel.messenger, risk, False) def test_civic_public_comment_requires_approval(): risk = classify_risk(PageContext.civic, Channel.comment, "Where do I vote?", True) assert risk == RiskLevel.high assert requires_human_approval(PageContext.civic, Channel.comment, risk, True) def test_page_post_media_required_refuses_missing_media(): r = client.post("/tools/meta/draft", headers={"X-Meta-Bridge-Key":"internal-test"}, json={ "page_context": "civic", "channel": "page_post", "draft_text": "Approved civic card caption", "media_required": True, "risk_level": "medium", }) assert r.status_code == 400 assert "media_required" in r.json()["detail"] def test_page_post_draft_accepts_local_png_media(tmp_path: Path): png = tmp_path / "card.png" png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00\x00\x00\rIHDR" + b"\x00\x00\x04\xb0\x00\x00\x04\xb0" + b"\x08\x02\x00\x00\x00" + b"\x00\x00\x00\x00") r = client.post("/tools/meta/draft", headers={"X-Meta-Bridge-Key":"internal-test"}, json={ "page_context": "civic", "channel": "page_post", "draft_text": "Approved civic card caption", "media_required": True, "media_attachments": [{"media_path": str(png), "media_type": "image", "alt_text": "Civic card"}], "risk_level": "medium", }) assert r.status_code == 200 assert r.json()["status"] == "needs_review" def test_reels_validation_refuses_non_video_file(tmp_path: Path): not_video = tmp_path / "card.png" not_video.write_bytes(b"not a video") r = client.post("/tools/meta/reels/publish", headers={"X-Meta-Bridge-Key":"internal-test"}, json={ "page_context": "civic", "video_path": str(not_video), "description": "test", "approved_by": "test", "validation_only": True, }) assert r.status_code == 400 assert "unsupported Reels video type" in r.json()["detail"] def test_read_only_page_posts_endpoint(monkeypatch): from app.services.meta_client import MetaClient calls = {} async def fake_list_page_posts(self, page_context, edge="posts", limit=10, after=None, before=None): calls.update({"page_context": page_context.value, "edge": edge, "limit": limit, "after": after, "before": before}) return {"data": [{"id": "post-1", "message": "hello"}], "paging": {"cursors": {"after": "next"}}} monkeypatch.setattr(MetaClient, "list_page_posts", fake_list_page_posts) r = client.get( "/tools/meta/page/posts", headers={"X-Meta-Bridge-Key": "internal-test"}, params={"page_context": "civic", "edge": "published_posts", "limit": 5, "after": "cursor-a"}, ) assert r.status_code == 200 assert r.json()["data"][0]["id"] == "post-1" assert calls == {"page_context": "civic", "edge": "published_posts", "limit": 5, "after": "cursor-a", "before": None} def test_read_only_insights_rejects_unverified_metric(): r = client.get( "/tools/meta/insights", headers={"X-Meta-Bridge-Key": "internal-test"}, params={"page_context": "civic", "metrics": "page_fans"}, ) assert r.status_code == 400 assert "page_fans" in r.json()["detail"]["invalid_metrics"] def test_read_only_media_endpoint_can_return_all_edges(monkeypatch): from app.services.meta_client import MetaClient async def fake_list_page_media(self, page_context, media_type="all", limit=10, after=None, before=None): return {"media": {"photos": {"data": []}, "videos": {"data": []}, "video_reels": {"data": []}}} monkeypatch.setattr(MetaClient, "list_page_media", fake_list_page_media) r = client.get( "/tools/meta/page/media", headers={"X-Meta-Bridge-Key": "internal-test"}, params={"page_context": "healthcare", "media_type": "all"}, ) assert r.status_code == 200 assert sorted(r.json()["media"]) == ["photos", "video_reels", "videos"] def test_read_only_endpoints_require_internal_auth(): r = client.get("/tools/meta/page/scheduled_posts", params={"page_context": "civic"}) assert r.status_code == 401 def test_page_post_validation_only_returns_payload_without_publish(): scheduled = (datetime.now(timezone.utc) + timedelta(minutes=20)).isoformat() r = client.post("/tools/meta/page_post/publish", headers={"X-Meta-Bridge-Key":"internal-test"}, json={ "page_context": "civic", "message": "Validation only caption", "approved_by": "test", "publish_mode": "validation_only", "scheduled_publish_time": scheduled, }) assert r.status_code == 200 body = r.json() assert body["status"] == "validated" assert body["meta_response"]["publish_attempted"] is False assert body["meta_response"]["payload"]["message"] == "Validation only caption" assert body["meta_response"]["payload"]["published"] == "false" assert "scheduled_publish_time" in body["meta_response"]["payload"] def test_scheduled_page_post_payload_has_no_immediate_publish_flag(): from app.services.meta_client import MetaClient scheduled = datetime.now(timezone.utc) + timedelta(minutes=20) payload = MetaClient().build_page_post_payload("Scheduled caption", publish_mode="scheduled", scheduled_publish_time=scheduled) assert payload["message"] == "Scheduled caption" assert payload["published"] == "false" assert int(payload["scheduled_publish_time"]) >= int(scheduled.timestamp()) - 1 def test_draft_detail_edit_history_and_validation_publish(): create = client.post("/tools/meta/draft", headers={"X-Meta-Bridge-Key":"internal-test"}, json={ "page_context": "civic", "channel": "page_post", "draft_text": "Draft lifecycle caption", "risk_level": "medium", "publish_mode": "validation_only", "created_by": "test", }) assert create.status_code == 200 draft_id = create.json()["draft_id"] detail = client.get(f"/approvals/drafts/{draft_id}", headers={"X-Meta-Bridge-Key":"internal-test"}) assert detail.status_code == 200 assert detail.json()["publish_mode"] == "validation_only" edit = client.post(f"/approvals/drafts/{draft_id}/edit", headers={"X-Meta-Bridge-Key":"internal-test"}, json={ "edited_by": "test-editor", "draft_text": "Edited lifecycle caption", }) assert edit.status_code == 200 assert edit.json()["draft_text"] == "Edited lifecycle caption" approve = client.post(f"/approvals/drafts/{draft_id}/approve", headers={"X-Meta-Bridge-Key":"internal-test"}, json={"approved_by": "test-approver"}) assert approve.status_code == 200 publish = client.post(f"/approvals/drafts/{draft_id}/publish", headers={"X-Meta-Bridge-Key":"internal-test"}) assert publish.status_code == 200 assert publish.json()["meta_response"]["publish_attempted"] is False history = client.get(f"/approvals/drafts/{draft_id}/history", headers={"X-Meta-Bridge-Key":"internal-test"}) assert history.status_code == 200 actions = {row["action"] for row in history.json()} assert {"draft.created", "draft.edited", "draft.approved", "draft.validated"} <= actions def test_comment_thread_read_endpoint(monkeypatch): from app.services.meta_client import MetaClient async def fake_read_comment_thread(self, page_context, object_id, limit=10, after=None, before=None): return {"data": [{"id": "comment-1", "message": "test"}], "paging": {"cursors": {"after": "next"}}} monkeypatch.setattr(MetaClient, "read_comment_thread", fake_read_comment_thread) r = client.get( "/tools/meta/comment/thread", headers={"X-Meta-Bridge-Key":"internal-test"}, params={"page_context": "civic", "object_id": "post-1", "limit": 5}, ) assert r.status_code == 200 assert r.json()["data"][0]["id"] == "comment-1" def test_comment_moderation_requires_approval_reason(): r = client.post("/tools/meta/comment/moderate", headers={"X-Meta-Bridge-Key":"internal-test"}, json={ "page_context": "civic", "comment_id": "comment-1", "action": "hide", "approved_by": "", "reason": "", }) assert r.status_code == 403 def test_comment_moderation_calls_meta_with_audit(monkeypatch): from app.services.meta_client import MetaClient async def fake_moderate_comment(self, page_context, comment_id, action): return {"success": True, "comment_id": comment_id, "action": action} monkeypatch.setattr(MetaClient, "moderate_comment", fake_moderate_comment) r = client.post("/tools/meta/comment/moderate", headers={"X-Meta-Bridge-Key":"internal-test"}, json={ "page_context": "civic", "comment_id": "comment-1", "action": "hide", "approved_by": "test", "reason": "policy test", }) assert r.status_code == 200 assert r.json()["status"] == "hide" assert r.json()["meta_response"]["success"] is True def test_messenger_conversation_and_response_window_endpoints(monkeypatch): from app.services.meta_client import MetaClient async def fake_read_conversation_detail(self, page_context, conversation_id, limit=10, after=None, before=None): return {"id": conversation_id, "messages": {"data": [{"id": "m1", "created_time": "2026-05-24T18:00:00+0000"}]}} async def fake_response_window(self, page_context, conversation_id): return {"conversation_id": conversation_id, "response_window_status": "inspect_latest_message_time"} monkeypatch.setattr(MetaClient, "read_conversation_detail", fake_read_conversation_detail) monkeypatch.setattr(MetaClient, "messenger_response_window", fake_response_window) detail = client.get("/tools/meta/messenger/conversation", headers={"X-Meta-Bridge-Key":"internal-test"}, params={"page_context":"healthcare", "conversation_id":"t_1"}) assert detail.status_code == 200 assert detail.json()["id"] == "t_1" window = client.get("/tools/meta/messenger/response_window", headers={"X-Meta-Bridge-Key":"internal-test"}, params={"page_context":"healthcare", "conversation_id":"t_1"}) assert window.status_code == 200 assert window.json()["response_window_status"] == "inspect_latest_message_time"