"""The admin roster and the password flows, end to end (spec admin-user-management).

Needs a real database: the last-super-admin invariant is a locking read, the username
uniqueness is a functional index, and the owned-events guard is a foreign key. None of the
three exists in a fake, and all three are the kind of rule that looks fine in the handler and
fails under a second concurrent request.
"""

import asyncio
import uuid
from collections.abc import AsyncIterator
from typing import Any

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.models.enums import AdminRole, AuthMethod
from app.services.auth import CurrentAdmin, get_current_admin

from .conftest import requires_db

pytestmark = [requires_db, pytest.mark.asyncio]


@pytest_asyncio.fixture
async def acting_super_admin(
    session: AsyncSession, app_engine_per_loop: None
) -> AsyncIterator[uuid.UUID]:
    from app.main import app

    admin_id = uuid.uuid4()
    email = f"roster-actor-{admin_id.hex[:8]}@example.test"
    await session.execute(
        text(
            "INSERT INTO admin_user (id, email, name, role, status, auth_method, "
            "                        must_change_password, failed_login_count) "
            "VALUES (:id, :email, 'Roster Actor', 'super_admin', 'active', 'google', "
            "        false, 0)"
        ),
        {"id": admin_id, "email": email},
    )
    await session.commit()

    app.dependency_overrides[get_current_admin] = lambda: CurrentAdmin(
        id=admin_id,
        email=email,
        name="Roster Actor",
        role=AdminRole.SUPER_ADMIN,
        auth_method=AuthMethod.GOOGLE,
    )
    yield admin_id
    app.dependency_overrides.pop(get_current_admin, None)
    await session.execute(text("DELETE FROM audit_log WHERE admin_user_id = :id"), {"id": admin_id})
    await session.execute(text("DELETE FROM admin_user WHERE id = :id"), {"id": admin_id})
    await session.commit()


@pytest_asyncio.fixture
async def client() -> AsyncIterator[AsyncClient]:
    from app.main import app

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


def _unique_email(tag: str) -> str:
    return f"{tag}-{uuid.uuid4().hex[:8]}@example.com"


async def _cleanup(session: AsyncSession, admin_id: str | uuid.UUID) -> None:
    # Coerced because ids arrive from JSON as strings, and asyncpg will not compare a
    # varchar to a uuid column — it raises rather than silently casting.
    ident = uuid.UUID(str(admin_id))
    await session.execute(text("DELETE FROM audit_log WHERE admin_user_id = :id"), {"id": ident})
    await session.execute(text("DELETE FROM admin_user WHERE id = :id"), {"id": ident})
    await session.commit()


# ------------------------------------------------------------------ creating accounts


async def test_creating_a_password_account_returns_it_marked_temporary(
    client: AsyncClient, acting_super_admin: uuid.UUID, session: AsyncSession
) -> None:
    response = await client.post(
        "/api/admin/users",
        json={
            "email": _unique_email("pw-host"),
            "name": "Password Host",
            "role": "host",
            "auth_method": "password",
            "username": f"host{uuid.uuid4().hex[:6]}",
            "temporary_password": "mango-tree-42-sunset",
        },
    )
    assert response.status_code == 201, response.text
    body = response.json()
    assert body["auth_method"] == "password"
    assert body["status"] == "active"
    assert body["temporary_password_outstanding"] is True

    # No credential comes back. Checked by key and by hash prefix rather than by searching
    # for the word "password", which legitimately appears in `auth_method` and in the
    # outstanding-temporary flag.
    assert "password_hash" not in body
    assert "temporary_password" not in body
    assert "$argon2" not in str(body)
    assert "mango-tree-42-sunset" not in str(body)
    await _cleanup(session, body["id"])


async def test_a_password_account_without_a_password_is_refused(
    client: AsyncClient, acting_super_admin: uuid.UUID
) -> None:
    """An account that cannot sign in has no reason to exist yet."""
    response = await client.post(
        "/api/admin/users",
        json={
            "email": _unique_email("no-pw"),
            "auth_method": "password",
            "username": f"nopw{uuid.uuid4().hex[:6]}",
        },
    )
    assert response.status_code == 400


async def test_a_google_account_refuses_a_password(
    client: AsyncClient, acting_super_admin: uuid.UUID
) -> None:
    response = await client.post(
        "/api/admin/users",
        json={
            "email": _unique_email("goog"),
            "auth_method": "google",
            "temporary_password": "mango-tree-42-sunset",
        },
    )
    assert response.status_code == 400


async def test_a_weak_temporary_password_is_refused(
    client: AsyncClient, acting_super_admin: uuid.UUID
) -> None:
    """The quality rules apply at every point a password is set, including by an admin."""
    response = await client.post(
        "/api/admin/users",
        json={
            "email": _unique_email("weak"),
            "auth_method": "password",
            "username": f"weak{uuid.uuid4().hex[:6]}",
            "temporary_password": "password123",
        },
    )
    assert response.status_code == 400


async def test_usernames_collide_case_insensitively(
    client: AsyncClient, acting_super_admin: uuid.UUID, session: AsyncSession
) -> None:
    """Two logins differing only by capitalisation is a phishing affordance, not a feature."""
    username = f"Fatima{uuid.uuid4().hex[:6]}"
    first = await client.post(
        "/api/admin/users",
        json={
            "email": _unique_email("dup-a"),
            "auth_method": "password",
            "username": username,
            "temporary_password": "mango-tree-42-sunset",
        },
    )
    assert first.status_code == 201

    second = await client.post(
        "/api/admin/users",
        json={
            "email": _unique_email("dup-b"),
            "auth_method": "password",
            "username": username.lower(),
            "temporary_password": "mango-tree-42-sunset",
        },
    )
    assert second.status_code == 409
    await _cleanup(session, first.json()["id"])


async def test_an_accounts_kind_cannot_be_edited(
    client: AsyncClient, acting_super_admin: uuid.UUID, session: AsyncSession
) -> None:
    """Conversion would silently change what evidence is accepted for that identity (D16)."""
    created = await client.post(
        "/api/admin/users", json={"email": _unique_email("kind"), "auth_method": "google"}
    )
    admin_id = created.json()["id"]

    # The field does not exist on the update model, so this is ignored rather than applied.
    response = await client.patch(
        f"/api/admin/users/{admin_id}", json={"auth_method": "password", "name": "Renamed"}
    )
    assert response.status_code == 200
    assert response.json()["auth_method"] == "google"
    assert response.json()["name"] == "Renamed"

    # And a username cannot be bolted onto a Google account either.
    refused = await client.patch(f"/api/admin/users/{admin_id}", json={"username": "sneaky"})
    assert refused.status_code == 400
    await _cleanup(session, admin_id)


# ------------------------------------------------------------- temporary passwords


async def test_issuing_a_temporary_password_returns_it_once_and_never_again(
    client: AsyncClient, acting_super_admin: uuid.UUID, session: AsyncSession
) -> None:
    created = await client.post(
        "/api/admin/users",
        json={
            "email": _unique_email("reset"),
            "auth_method": "password",
            "username": f"reset{uuid.uuid4().hex[:6]}",
            "temporary_password": "mango-tree-42-sunset",
        },
    )
    admin_id = created.json()["id"]

    issued = await client.post(f"/api/admin/users/{admin_id}/temporary-password")
    assert issued.status_code == 200
    secret = issued.json()["password"]
    assert len(secret) == 16

    # Not retrievable afterwards, from the roster or from the record.
    roster = await client.get("/api/admin/users")
    assert secret not in roster.text

    # And never written to the audit trail — the entry records the fact, not the value.
    trail = await session.scalar(
        text(
            "SELECT string_agg(coalesce(after::text, '') || coalesce(before::text, ''), ' ') "
            "FROM audit_log WHERE entity_id = :id"
        ).bindparams(id=uuid.UUID(admin_id))
    )
    assert secret not in (trail or "")
    await _cleanup(session, admin_id)


async def test_a_temporary_password_cannot_be_issued_for_a_google_account(
    client: AsyncClient, acting_super_admin: uuid.UUID, session: AsyncSession
) -> None:
    created = await client.post(
        "/api/admin/users", json={"email": _unique_email("g-noreset"), "auth_method": "google"}
    )
    admin_id = created.json()["id"]
    response = await client.post(f"/api/admin/users/{admin_id}/temporary-password")
    assert response.status_code == 400
    await _cleanup(session, admin_id)


async def test_issuing_a_temporary_password_drops_existing_sessions(
    client: AsyncClient, acting_super_admin: uuid.UUID, session: AsyncSession
) -> None:
    """The usual reason for issuing one is that the old credential was compromised."""
    created = await client.post(
        "/api/admin/users",
        json={
            "email": _unique_email("drop"),
            "auth_method": "password",
            "username": f"drop{uuid.uuid4().hex[:6]}",
            "temporary_password": "mango-tree-42-sunset",
        },
    )
    admin_id = created.json()["id"]
    await client.post(f"/api/admin/users/{admin_id}/temporary-password")

    epoch = await session.scalar(
        text("SELECT session_epoch FROM admin_user WHERE id = :id").bindparams(
            id=uuid.UUID(admin_id)
        )
    )
    # Created with one temporary password and issued a second: two bumps, so anything held
    # from before either of them is now refused.
    assert epoch >= 2
    await _cleanup(session, admin_id)


# --------------------------------------------------------- pending, activate, withdraw


async def test_activating_a_pending_account_sets_its_role(
    client: AsyncClient, acting_super_admin: uuid.UUID, session: AsyncSession
) -> None:
    pending_id = uuid.uuid4()
    await session.execute(
        text(
            "INSERT INTO admin_user (id, email, name, role, status, auth_method, "
            "                        must_change_password, failed_login_count) "
            "VALUES (:id, :email, 'Arrival', 'host', 'pending', 'google', false, 0)"
        ),
        {"id": pending_id, "email": _unique_email("pending")},
    )
    await session.commit()

    response = await client.post(f"/api/admin/users/{pending_id}/activate", json={"role": "host"})
    assert response.status_code == 200
    assert response.json()["status"] == "active"
    await _cleanup(session, pending_id)


async def test_rejecting_a_pending_account_withdraws_rather_than_deletes(
    client: AsyncClient, acting_super_admin: uuid.UUID, session: AsyncSession
) -> None:
    """A deleted row would be recreated as a fresh pending account by the next sign-in."""
    pending_id = uuid.uuid4()
    await session.execute(
        text(
            "INSERT INTO admin_user (id, email, name, role, status, auth_method, "
            "                        must_change_password, failed_login_count) "
            "VALUES (:id, :email, 'Unwanted', 'host', 'pending', 'google', false, 0)"
        ),
        {"id": pending_id, "email": _unique_email("reject")},
    )
    await session.commit()

    response = await client.post(f"/api/admin/users/{pending_id}/reject")
    assert response.status_code == 200
    assert response.json()["status"] == "withdrawn"

    still_there = await session.scalar(
        text("SELECT count(*) FROM admin_user WHERE id = :id").bindparams(id=pending_id)
    )
    assert still_there == 1
    await _cleanup(session, pending_id)


# ------------------------------------------------------ the last-super-admin invariant


@pytest_asyncio.fixture
async def sole_super_admin(
    session: AsyncSession, acting_super_admin: uuid.UUID
) -> AsyncIterator[uuid.UUID]:
    """Make the acting account the only active super admin, and put the others back.

    The invariant under test is global — "no operation may leave zero active super admins" —
    so there is no way to exercise it without the system holding exactly one. That makes these
    the only tests here that touch rows they did not create, and it is why the restore is a
    fixture rather than three copies of a cleanup block at the end of a test body.

    **This was a real bug, not a hypothetical one.** The three tests below each withdrew every
    other super admin inline; two restored them and one restored only the role. Run against a
    dev database with a real operator account in it, that logged the operator out of their own
    system — and `make test` against the dev database is the documented way to run this suite.

    Two things fix it. The ids are captured before anything is written, so only rows this
    fixture actually changed are restored — never "every super admin", which is what silently
    swept up the operator. And the restore is teardown, so it runs even when the test fails
    part-way, which is exactly when a half-applied withdrawal would otherwise be left behind.
    """
    withdrawn = list(
        (
            await session.scalars(
                text(
                    "SELECT id FROM admin_user "
                    "WHERE role = 'super_admin' AND status = 'active' AND id <> :id"
                ).bindparams(id=acting_super_admin)
            )
        ).all()
    )
    if withdrawn:
        await session.execute(
            text("UPDATE admin_user SET status = 'withdrawn' WHERE id = ANY(:ids)"),
            {"ids": withdrawn},
        )
    await session.commit()

    yield acting_super_admin

    await session.rollback()
    if withdrawn:
        await session.execute(
            text("UPDATE admin_user SET status = 'active' WHERE id = ANY(:ids)"),
            {"ids": withdrawn},
        )
    # The acting account too: a test may have demoted or withdrawn it before failing.
    await session.execute(
        text("UPDATE admin_user SET status = 'active', role = 'super_admin' WHERE id = :id"),
        {"id": acting_super_admin},
    )
    await session.commit()


async def test_the_last_active_super_admin_cannot_be_withdrawn(
    client: AsyncClient, sole_super_admin: uuid.UUID
) -> None:
    response = await client.post(f"/api/admin/users/{sole_super_admin}/withdraw")
    assert response.status_code == 409
    assert "last active Super Admin" in response.json()["detail"]


async def test_the_last_active_super_admin_cannot_be_demoted(
    client: AsyncClient, sole_super_admin: uuid.UUID
) -> None:
    response = await client.patch(f"/api/admin/users/{sole_super_admin}", json={"role": "host"})
    assert response.status_code == 409


async def test_two_super_admins_demoting_each_other_cannot_both_win(
    client: AsyncClient,
    sole_super_admin: uuid.UUID,
    session: AsyncSession,
    sessions: async_sessionmaker[AsyncSession],
) -> None:
    """The concurrency case a count-then-write loses (spec admin-user-management).

    Both requests read "two active super admins" and both proceed, leaving none. The locking
    read in `_assert_not_last_super_admin` is what serialises them; this asserts the outcome
    rather than the mechanism, so a different correct implementation still passes.
    """
    acting_super_admin = sole_super_admin
    other_id = uuid.uuid4()
    await session.execute(
        text(
            "INSERT INTO admin_user (id, email, name, role, status, auth_method, "
            "                        must_change_password, failed_login_count) "
            "VALUES (:id, :email, 'Other Super', 'super_admin', 'active', 'google', false, 0)"
        ),
        {"id": other_id, "email": _unique_email("other-super")},
    )
    await session.commit()

    from app.main import app

    async with (
        AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as a,
        AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as b,
    ):
        results = await asyncio.gather(
            a.patch(f"/api/admin/users/{other_id}", json={"role": "host"}),
            b.patch(f"/api/admin/users/{acting_super_admin}", json={"role": "host"}),
            return_exceptions=True,
        )

    del results  # the outcome that matters is the roster, not which request won

    remaining = await session.scalar(
        text("SELECT count(*) FROM admin_user WHERE role = 'super_admin' AND status = 'active'")
    )
    assert remaining >= 1, "both demotions succeeded and the system has no super admin left"

    # Only the row this test created. The acting account is restored by `sole_super_admin`'s
    # teardown, which is also what puts back any super admin the fixture withdrew — this test
    # used to do that itself, and did it incompletely.
    await _cleanup(session, other_id)


# ------------------------------------------------------------------ ownership transfer


async def test_transferring_an_event_moves_its_guests_with_it(
    client: AsyncClient,
    acting_super_admin: uuid.UUID,
    fixture_event: dict[str, Any],
    session: AsyncSession,
) -> None:
    created = await client.post(
        "/api/admin/users", json={"email": _unique_email("new-owner"), "auth_method": "google"}
    )
    new_owner_id = created.json()["id"]

    response = await client.post(
        f"/api/admin/events/{fixture_event['event_id']}/transfer",
        json={"new_owner_id": new_owner_id},
    )
    assert response.status_code == 200
    assert response.json()["owned_event_count"] == 1

    owner = await session.scalar(
        text("SELECT owner_admin_id FROM event WHERE id = :id").bindparams(
            id=fixture_event["event_id"]
        )
    )
    assert str(owner) == new_owner_id

    # Hand it back so the fixture's teardown can remove the account it created.
    await session.execute(
        text("UPDATE event SET owner_admin_id = :o WHERE id = :id"),
        {"o": fixture_event["owner_admin_id"], "id": fixture_event["event_id"]},
    )
    await session.commit()
    await _cleanup(session, new_owner_id)


async def test_an_event_cannot_be_transferred_to_a_pending_account(
    client: AsyncClient,
    acting_super_admin: uuid.UUID,
    fixture_event: dict[str, Any],
    session: AsyncSession,
) -> None:
    pending_id = uuid.uuid4()
    await session.execute(
        text(
            "INSERT INTO admin_user (id, email, name, role, status, auth_method, "
            "                        must_change_password, failed_login_count) "
            "VALUES (:id, :email, 'Not Yet', 'host', 'pending', 'google', false, 0)"
        ),
        {"id": pending_id, "email": _unique_email("pending-owner")},
    )
    await session.commit()

    response = await client.post(
        f"/api/admin/events/{fixture_event['event_id']}/transfer",
        json={"new_owner_id": str(pending_id)},
    )
    assert response.status_code == 400
    await _cleanup(session, pending_id)


async def test_an_account_owning_events_cannot_be_removed(
    client: AsyncClient,
    acting_super_admin: uuid.UUID,
    fixture_event: dict[str, Any],
) -> None:
    """The FK would refuse anyway, but with an integrity error rather than a sentence."""
    response = await client.delete(f"/api/admin/users/{fixture_event['owner_admin_id']}")
    assert response.status_code == 409
    assert "owns 1 event" in response.json()["detail"]