tests/test_bridge_core.py

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"