"""P1 admin write-boundary + P2 config-default tests for the MAF roundtable.

P1 —admin diagnostic access is **read-only**:
- An admin with ``admin_diagnostic_access=True`` can *read* another user's
  session / run / action / approval (cross-user diagnostics).
- The same admin **cannot** *write* another user's resources: update / delete
  session, start / cancel run, append event, decide approval all raise
  ``MafForbiddenError`` (→HTTP 403), never silently succeeding.
- Normal users remain fully isolated from each other.

P2 —config-driven create defaults:
- Omitted ``mode`` / ``write_policy`` / ``max_rounds`` fall back to the deployed
  ``MafRoundtableConfig`` values.
- An explicit request value overrides the config default.
"""

from __future__ import annotations

import pytest
import pytest_asyncio
from maf_test_helpers import make_maf_service
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.maf_roundtable.service import MafForbiddenError, MafNotFoundError
from deerflow.config.maf_roundtable_config import MafRoundtableConfig
from deerflow.persistence.base import Base
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store


@pytest_asyncio.fixture
async def store(tmp_path):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_adm.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)
    yield make_maf_roundtable_store(sf)
    await engine.dispose()


def _svc(store, *, enabled=True, admin_diagnostic_access=True, config=None):
    return make_maf_service(
        store,
        enabled=enabled,
        admin_diagnostic_access=admin_diagnostic_access,
        config=config,
    )


# ── P1: admin can READ cross-user ─────────────────────────────────────────


@pytest.mark.asyncio
async def test_admin_can_read_other_user_session(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    fetched = await svc.get_session(session_id=s["id"], user_id="admin-1", is_admin=True)
    assert fetched["id"] == s["id"]
    assert fetched["user_id"] == "alice"


@pytest.mark.asyncio
async def test_admin_can_read_other_user_run_and_action(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="alice")
    action = await store.actions.create_action(
        {"session_id": s["id"], "run_id": run["id"], "tool_name": "t", "idempotency_key": "k", "status": "pending"}
    )
    # Admin reads run + action cross-user without error.
    assert (await svc.get_run(run_id=run["id"], user_id="admin-1", is_admin=True))["id"] == run["id"]
    assert (await svc.get_action(action_id=action["id"], user_id="admin-1", is_admin=True))["id"] == action["id"]
    assert len(await svc.list_runs(session_id=s["id"], user_id="admin-1", is_admin=True)) == 1


# ── P1: admin CANNOT WRITE cross-user ─────────────────────────────────────


@pytest.mark.asyncio
async def test_admin_cannot_update_other_user_session(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    with pytest.raises(MafForbiddenError):
        await svc.update_session(session_id=s["id"], user_id="admin-1", fields={"title": "hijack"}, is_admin=True)
    # Unchanged.
    assert (await svc.get_session(session_id=s["id"], user_id="alice"))["title"] == ""


@pytest.mark.asyncio
async def test_admin_cannot_delete_other_user_session(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    with pytest.raises(MafForbiddenError):
        await svc.delete_session(session_id=s["id"], user_id="admin-1", is_admin=True)
    # Still present.
    assert (await svc.get_session(session_id=s["id"], user_id="alice"))["id"] == s["id"]


@pytest.mark.asyncio
async def test_admin_cannot_start_run_on_other_user_session(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    with pytest.raises(MafForbiddenError):
        await svc.create_run(session_id=s["id"], user_id="admin-1", is_admin=True)
    # No run was created.
    assert await svc.list_runs(session_id=s["id"], user_id="alice") == []


@pytest.mark.asyncio
async def test_admin_cannot_cancel_other_user_run(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="alice")
    with pytest.raises(MafForbiddenError):
        await svc.cancel_run(run_id=run["id"], user_id="admin-1", is_admin=True)
    assert (await svc.get_run(run_id=run["id"], user_id="alice"))["status"] == "pending"


@pytest.mark.asyncio
async def test_non_owner_cannot_cancel_run_gets_not_found(store):
    """A non-admin user cancelling another user's run gets 404 (not 403)."""
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="alice")
    with pytest.raises(MafNotFoundError):
        await svc.cancel_run(run_id=run["id"], user_id="bob")
    assert (await svc.get_run(run_id=run["id"], user_id="alice"))["status"] == "pending"


@pytest.mark.asyncio
async def test_admin_cannot_append_event_to_other_user_run(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="alice")
    with pytest.raises(MafForbiddenError):
        await svc.append_event(run_id=run["id"], user_id="admin-1", event_type="run.started", is_admin=True)


@pytest.mark.asyncio
async def test_admin_cannot_decide_other_user_approval(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    run = await svc.create_run(session_id=s["id"], user_id="alice")
    appr = await store.approvals.create_approval(
        {"session_id": s["id"], "run_id": run["id"], "summary": "x", "status": "pending"}
    )
    with pytest.raises(MafForbiddenError):
        await svc.decide_approval(approval_id=appr["id"], decision="approved", user_id="admin-1", is_admin=True)
    assert (await store.approvals.get_approval(appr["id"]))["status"] == "pending"


@pytest.mark.asyncio
async def test_admin_diagnostic_off_cannot_even_read_cross_user(store):
    svc = _svc(store, admin_diagnostic_access=False)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    with pytest.raises(MafNotFoundError):
        await svc.get_session(session_id=s["id"], user_id="admin-1", is_admin=True)


@pytest.mark.asyncio
async def test_normal_users_remain_fully_isolated(store):
    svc = _svc(store, admin_diagnostic_access=True)
    s = await svc.create_session(user_id="alice", body={"task_text": "t"})
    # bob is a non-admin →isolated on both read and write.
    with pytest.raises(MafNotFoundError):
        await svc.get_session(session_id=s["id"], user_id="bob")
    with pytest.raises(MafNotFoundError):
        await svc.update_session(session_id=s["id"], user_id="bob", fields={"title": "x"})


# ── P2: config-driven create defaults ─────────────────────────────────────


@pytest.mark.asyncio
async def test_create_session_uses_config_defaults_when_omitted(store):
    cfg = MafRoundtableConfig(
        default_mode="group_chat",
        default_write_policy="confirm_all_writes",
        default_max_rounds=12,
    )
    svc = _svc(store, config=cfg)
    s = await svc.create_session(user_id="u1", body={"task_text": "t"})
    assert s["mode"] == "group_chat"
    assert s["write_policy"] == "confirm_all_writes"
    assert s["max_rounds"] == 12


@pytest.mark.asyncio
async def test_create_session_explicit_values_override_config(store):
    cfg = MafRoundtableConfig(
        default_mode="group_chat",
        default_write_policy="confirm_all_writes",
        default_max_rounds=12,
    )
    svc = _svc(store, config=cfg)
    s = await svc.create_session(
        user_id="u1",
        body={"task_text": "t", "mode": "structured", "write_policy": "auto", "max_rounds": 3},
    )
    assert s["mode"] == "structured"
    assert s["write_policy"] == "auto"
    assert s["max_rounds"] == 3