"""P2 HTTP route tests for the MAF roundtable API.

Covers the full HTTP status code matrix:
- **401 Unauthorized**: missing / invalid auth
- **403 Forbidden**: admin trying to write another user's resource
- **404 Not Found**: non-existent session / run / action / approval
- **409 Conflict**: concurrent run launch, already-decided approval
- **422 Unprocessable Entity**: invalid enum values (mode, write_policy, role)
- **503 Service Unavailable**: feature disabled, mock events disabled

Also verifies that identity comes from ``request.state.user`` (stamped by auth
middleware), not from body parameters —a client cannot forge ``user_id`` in the
request body to impersonate another user.
"""

from __future__ import annotations

import pytest
import pytest_asyncio
from fastapi import FastAPI, Request
from httpx import ASGITransport, AsyncClient
from maf_test_helpers import make_maf_service
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.gateway.routers.maf_roundtable import router as maf_router
from deerflow.config.maf_roundtable_config import MafRoundtableConfig
from deerflow.persistence.base import Base
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store


class FakeUser:
    """Minimal user object for testing (mimics the real User model's shape)."""

    def __init__(self, user_id: str, is_admin: bool = False):
        self.id = user_id
        self.system_role = "admin" if is_admin else "user"


@pytest_asyncio.fixture
async def app_and_client(tmp_path):
    """Build a minimal FastAPI app with the MAF router + a fake auth middleware."""
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_http.db'}")
    async with engine.begin() as conn:
        import deerflow.persistence.maf_roundtable.model  # noqa: F401

        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    store = make_maf_roundtable_store(sf)

    cfg = MafRoundtableConfig(enabled=True, admin_diagnostic_access=True)
    svc = make_maf_service(store, enabled=True, admin_diagnostic_access=True, config=cfg)

    app = FastAPI()
    app.state.maf_roundtable_service = svc
    app.state.maf_auto_dispatch = False
    app.include_router(maf_router)

    # Fake auth middleware: reads X-Test-User / X-Test-Role headers and stamps
    # request.state.user. No header →401 (unauthenticated).
    @app.middleware("http")
    async def fake_auth(request: Request, call_next):
        user_id = request.headers.get("X-Test-User")
        if user_id is None:
            from starlette.responses import JSONResponse

            return JSONResponse(status_code=401, content={"detail": "Authentication required"})
        is_admin = request.headers.get("X-Test-Role") == "admin"
        request.state.user = FakeUser(user_id, is_admin)
        return await call_next(request)

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        yield app, client, store

    await engine.dispose()


# ── 401 Unauthorized ──────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_401_missing_auth(app_and_client):
    _, client, _ = app_and_client
    resp = await client.get("/api/maf-roundtable/sessions")
    assert resp.status_code == 401


# ── 403 Forbidden (admin write cross-user) ────────────────────────────────


@pytest.mark.asyncio
async def test_403_admin_cannot_update_other_user_session(app_and_client):
    _, client, _ = app_and_client
    # alice creates a session
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    session_id = resp.json()["id"]

    # admin tries to update it →403
    resp = await client.put(
        f"/api/maf-roundtable/sessions/{session_id}",
        json={"title": "hijack"},
        headers={"X-Test-User": "admin-1", "X-Test-Role": "admin"},
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_403_admin_cannot_delete_other_user_session(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    resp = await client.delete(
        f"/api/maf-roundtable/sessions/{session_id}",
        headers={"X-Test-User": "admin-1", "X-Test-Role": "admin"},
    )
    assert resp.status_code == 403


# ── 404 Not Found ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_404_nonexistent_session(app_and_client):
    _, client, _ = app_and_client
    resp = await client.get(
        "/api/maf-roundtable/sessions/nonexistent-id",
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_404_other_user_session_not_leaked(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    # bob (non-admin) tries to read alice's session →404 (not 403, so existence is not leaked)
    resp = await client.get(
        f"/api/maf-roundtable/sessions/{session_id}",
        headers={"X-Test-User": "bob"},
    )
    assert resp.status_code == 404


# ── 409 Conflict ──────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_409_second_run_launch_conflicts(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    # First run launch succeeds
    resp = await client.post(
        f"/api/maf-roundtable/sessions/{session_id}/runs",
        json={},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    # Second run launch →409
    resp = await client.post(
        f"/api/maf-roundtable/sessions/{session_id}/runs",
        json={},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 409


@pytest.mark.asyncio
async def test_409_approval_already_decided(app_and_client):
    _, client, store = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    resp = await client.post(
        f"/api/maf-roundtable/sessions/{session_id}/runs",
        json={},
        headers={"X-Test-User": "alice"},
    )
    run_id = resp.json()["id"]
    # Create an approval directly via the store
    appr = await store.approvals.create_approval(
        {"session_id": session_id, "run_id": run_id, "summary": "x", "status": "pending"}
    )
    # First decision succeeds
    resp = await client.post(
        f"/api/maf-roundtable/approvals/{appr['id']}/decision",
        json={"decision": "approved"},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    # Second decision →409
    resp = await client.post(
        f"/api/maf-roundtable/approvals/{appr['id']}/decision",
        json={"decision": "rejected"},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 409


# ── 422 Unprocessable Entity (invalid enums) ──────────────────────────────


@pytest.mark.asyncio
async def test_422_invalid_mode(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t", "mode": "invalid_mode"},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_422_invalid_write_policy(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t", "write_policy": "invalid_policy"},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_422_invalid_participant_role(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t", "participants": [{"agent_id": "a1", "role": "invalid_role"}]},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 422


# ── 503 Service Unavailable ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_503_feature_disabled(tmp_path):
    """When the feature flag is off, write paths return 503."""
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_disabled.db'}")
    async with engine.begin() as conn:
        import deerflow.persistence.maf_roundtable.model  # noqa: F401

        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    store = make_maf_roundtable_store(sf)

    cfg = MafRoundtableConfig(enabled=False)
    svc = make_maf_service(store, enabled=False, admin_diagnostic_access=True, config=cfg)

    app = FastAPI()
    app.state.maf_roundtable_service = svc
    app.state.maf_auto_dispatch = False
    app.include_router(maf_router)

    @app.middleware("http")
    async def fake_auth(request: Request, call_next):
        request.state.user = FakeUser("alice", False)
        return await call_next(request)

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        resp = await client.post("/api/maf-roundtable/sessions", json={"task_text": "t"})
        assert resp.status_code == 503

    await engine.dispose()


@pytest.mark.asyncio
async def test_404_post_events_endpoint_removed(app_and_client):
    """POST /runs/{id}/events is deliberately absent —no public mock write path."""
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    resp = await client.post(
        f"/api/maf-roundtable/sessions/{session_id}/runs",
        json={},
        headers={"X-Test-User": "alice"},
    )
    run_id = resp.json()["id"]
    resp = await client.post(
        f"/api/maf-roundtable/runs/{run_id}/events",
        json={"event_type": "manager.decision"},
        headers={"X-Test-User": "alice"},
    )
    # FastAPI returns 405 when GET exists on the same path but POST is not
    # registered; either 404 or 405 means no public write surface.
    assert resp.status_code in (404, 405)


@pytest.mark.asyncio
async def test_internal_append_event_still_writes(app_and_client):
    """service.append_event remains available for the future MAF executor."""
    app, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    resp = await client.post(
        f"/api/maf-roundtable/sessions/{session_id}/runs",
        json={},
        headers={"X-Test-User": "alice"},
    )
    run_id = resp.json()["id"]
    svc = app.state.maf_roundtable_service
    event = await svc.append_event(run_id=run_id, user_id="alice", event_type="run.started")
    assert event["seq"] == 1
    assert event["event_type"] == "run.started"
    # Public GET still works for the written event.
    resp = await client.get(
        f"/api/maf-roundtable/runs/{run_id}/events",
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    assert len(resp.json()["events"]) == 1


# ── PUT explicit-null rejection ───────────────────────────────────────────


@pytest.mark.asyncio
async def test_422_put_mode_null(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    resp = await client.put(
        f"/api/maf-roundtable/sessions/{session_id}",
        json={"mode": None},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_422_put_title_null(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t", "title": "keep"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    resp = await client.put(
        f"/api/maf-roundtable/sessions/{session_id}",
        json={"title": None},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_422_put_max_rounds_null(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    resp = await client.put(
        f"/api/maf-roundtable/sessions/{session_id}",
        json={"max_rounds": None},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_200_put_task_payload_null_clears(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t", "task_payload": {"k": 1}},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    assert resp.json()["task_payload"] == {"k": 1}
    resp = await client.put(
        f"/api/maf-roundtable/sessions/{session_id}",
        json={"task_payload": None},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    assert resp.json()["task_payload"] is None


@pytest.mark.asyncio
async def test_200_put_model_name_null_clears(app_and_client):
    _, client, _ = app_and_client
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t", "model_name": "glm-5"},
        headers={"X-Test-User": "alice"},
    )
    session_id = resp.json()["id"]
    assert resp.json()["model_name"] == "glm-5"
    resp = await client.put(
        f"/api/maf-roundtable/sessions/{session_id}",
        json={"model_name": None},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    assert resp.json()["model_name"] is None


# ── Identity forgery prevention ───────────────────────────────────────────


@pytest.mark.asyncio
async def test_identity_from_request_state_not_body(app_and_client):
    """A client cannot forge user_id in the request body to impersonate another user.

    The service reads user_id from request.state.user (stamped by auth middleware),
    not from the request body. Even if a malicious client includes a user_id field
    in the JSON body, it is ignored.
    """
    _, client, _ = app_and_client
    # alice creates a session
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    session_id = resp.json()["id"]
    assert resp.json()["user_id"] == "alice"

    # bob tries to create a session with a forged user_id in the body
    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t", "user_id": "alice"},  # forged field
        headers={"X-Test-User": "bob"},
    )
    assert resp.status_code == 200
    # The session is owned by bob (from request.state.user), not alice (from body)
    assert resp.json()["user_id"] == "bob"

    # bob cannot read alice's session
    resp = await client.get(
        f"/api/maf-roundtable/sessions/{session_id}",
        headers={"X-Test-User": "bob"},
    )
    assert resp.status_code == 404


# ── Phase 3 resume / recover ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_resume_and_recover_conflict_when_not_eligible(app_and_client):
    """Resume requires waiting_approval; recover requires running|waiting_approval."""
    app, client, _ = app_and_client
    app.state.maf_auto_dispatch = False

    resp = await client.post(
        "/api/maf-roundtable/sessions",
        json={"task_text": "t"},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    session_id = resp.json()["id"]

    resp = await client.post(
        f"/api/maf-roundtable/sessions/{session_id}/runs",
        json={},
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 200
    run_id = resp.json()["id"]
    # Pending runs (auto-dispatch off) are not eligible for resume/recover.
    assert resp.json()["status"] == "pending"

    resp = await client.post(
        f"/api/maf-roundtable/runs/{run_id}/resume",
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 409

    resp = await client.post(
        f"/api/maf-roundtable/runs/{run_id}/recover",
        headers={"X-Test-User": "alice"},
    )
    assert resp.status_code == 409