"""Database integration for the authorization loader and schema (Workstreams A, C).

Runs only when P2HS_AUTHZ_TEST_DATABASE_URL points at a non-production database that
is already migrated to head (for example the scratch database). Every write is rolled back.
"""

import json
import os
from uuid import uuid4

import pytest
import pytest_asyncio
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from src.domain.project_auth import permissions as perms
from src.domain.project_auth.authorization import (
    Action,
    AssignmentTier,
    authorize_role_change,
    authorize_suspension,
)
from src.domain.project_auth.authorization_loader import load_membership, load_role

URL = os.environ.get("P2HS_AUTHZ_TEST_DATABASE_URL", "")
pytestmark = pytest.mark.skipif(not URL, reason="P2HS_AUTHZ_TEST_DATABASE_URL not set")


@pytest_asyncio.fixture
async def db():
    engine = create_async_engine(URL)
    async with engine.connect() as conn:
        trans = await conn.begin()
        session = AsyncSession(bind=conn, expire_on_commit=False)
        try:
            yield session
        finally:
            await session.close()
            await trans.rollback()
    await engine.dispose()


async def _project(db):
    ws = uuid4()
    project = uuid4()
    await db.execute(
        text("insert into workspaces (id, name, slug) values (:id, 'ws', :slug)"),
        {"id": ws, "slug": f"ws-{ws}"},
    )
    ptype = (await db.execute(text(
        "select enumlabel from pg_enum e join pg_type t on t.oid = e.enumtypid "
        "where t.typname = (select udt_name from information_schema.columns "
        "where table_name = 'projects' and column_name = 'type') order by enumsortorder limit 1"
    ))).scalar_one()
    await db.execute(
        text(f"insert into projects (id, name, slug, type, workspace_id) "
             f"values (:id, 'p', :slug, '{ptype}', :ws)"),
        {"id": project, "slug": f"p-{project}", "ws": ws},
    )
    return project


async def _role(db, project, name, tier, permissions, protected=False):
    role_id = uuid4()
    await db.execute(
        text(
            "insert into project_auth_roles (id, project_id, name, created_by, permissions_json, "
            "assignment_tier, protected) values (:id, :p, :n, 'user', cast(:perms as jsonb), "
            "cast(:tier as p2hs_project_auth_assignment_tier), :protected)"
        ),
        {"id": role_id, "p": project, "n": name, "perms": json.dumps(permissions),
         "tier": tier, "protected": protected},
    )
    return role_id


async def _membership(db, project, email, *, is_owner=False, roles=(), status="active"):
    mid = uuid4()
    await db.execute(
        text(
            "insert into project_memberships "
            "(id, project_id, email_normalized, source, status, is_owner) "
            "values (:id, :p, :e, 'local', cast(:s as p2hs_project_membership_status), :o)"
        ),
        {"id": mid, "p": project, "e": email, "s": status, "o": is_owner},
    )
    for role_id in roles:
        await db.execute(
            text("insert into project_membership_roles (membership_id, role_id, source) "
            "values (:m, :r, 'local')"),
            {"m": mid, "r": role_id},
        )
    return mid


@pytest.mark.asyncio
async def test_loader_builds_actor_target_and_role_from_canonical_rows(db):
    project = await _project(db)
    admin_role = await _role(db, project, "admin", "admin", list(perms.ADMIN_PRESET))
    owner_role = await _role(db, project, "super_admin", None, ["super_admin"], protected=True)
    owner = await _membership(db, project, "owner@x.test", is_owner=True, roles=[owner_role])
    admin = await _membership(db, project, "admin@x.test", roles=[admin_role])

    owner_view = await load_membership(db, project, owner)
    admin_view = await load_membership(db, project, admin)
    assert owner_view.is_owner and not owner_view.holds_admin
    assert admin_view.holds_admin and not admin_view.is_owner
    assert perms.ROLE_GRANT_ADMIN in admin_view.permissions

    role = await load_role(db, project, admin_role)
    assert role.assignment_tier is AssignmentTier.admin and not role.protected


@pytest.mark.asyncio
async def test_engine_decisions_use_loaded_state(db):
    project = await _project(db)
    admin_role = await _role(db, project, "admin", "admin", list(perms.ADMIN_PRESET))
    staff_role = await _role(db, project, "staff", "staff", ["staff"])
    owner_role = await _role(db, project, "super_admin", None, ["super_admin"], protected=True)
    owner = await _membership(db, project, "owner@x.test", is_owner=True, roles=[owner_role])
    admin = await _membership(db, project, "admin@x.test", roles=[admin_role])
    admin_peer = await _membership(db, project, "peer@x.test", roles=[admin_role])
    staff = await _membership(db, project, "staff@x.test", roles=[staff_role])

    actor = (await load_membership(db, project, admin)).as_actor()
    peer = (await load_membership(db, project, admin_peer)).as_target()
    owner_t = (await load_membership(db, project, owner)).as_target()
    staff_t = (await load_membership(db, project, staff)).as_target()
    grant_staff = await load_role(db, project, staff_role)

    peer_decision = authorize_suspension(actor, peer, project_id=project, action=Action.suspend)
    assert peer_decision.allowed is False
    assert authorize_suspension(actor, staff_t, project_id=project, action=Action.suspend).allowed
    assert authorize_role_change(
        actor, owner_t, grant_staff, project_id=project, action=Action.grant
    ).code == "role_protected"


@pytest.mark.asyncio
async def test_second_owner_rejected_by_database(db):
    project = await _project(db)
    await _membership(db, project, "owner@x.test", is_owner=True)
    with pytest.raises(IntegrityError):
        await _membership(db, project, "second@x.test", is_owner=True)


@pytest.mark.asyncio
async def test_unclassified_role_loads_with_null_tier(db):
    project = await _project(db)
    role_id = await _role(db, project, "legacy_custom", None, ["billing"])
    role = await load_role(db, project, role_id)
    assert role.assignment_tier is None


@pytest.mark.asyncio
async def test_loader_is_project_bound(db):
    project = await _project(db)
    other = await _project(db)
    role_id = await _role(db, project, "admin", "admin", list(perms.ADMIN_PRESET))
    membership = await _membership(db, project, "a@x.test", roles=[role_id])
    assert await load_membership(db, other, membership) is None
    assert await load_role(db, other, role_id) is None


@pytest.mark.asyncio
async def test_protected_role_cannot_carry_an_assignment_tier(db):
    project = await _project(db)
    with pytest.raises(IntegrityError):
        await _role(db, project, "owner_like", "admin", ["super_admin"], protected=True)