"""Personal access token lifecycle — create, list, authenticate, revoke.

These tests exist to plug a real CI gap: prior to landing PAT support,
the test suite never inserted a PAT row, so the code paths that read
``expires_at`` (``is_active``, ``authenticate``) and the post-commit
serialisation in the create / revoke endpoints never ran. Three
production bugs (naive-vs-aware datetime comparison on MySQL/TiDB,
missing ``eager_defaults`` on the ORM model, ``MissingGreenlet`` after
``db.commit()``) shipped past CI as a result. Each test here asserts
on a specific symptom of those bugs so a regression would fail loudly.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import text

from app.auth.models import AuthUserRole
from app.auth.tokens.models import PersonalAccessToken
from app.auth.tokens.service import (
    MAX_TOKENS_PER_USER,
    PersonalAccessTokenService,
    is_active,
)
from app.core.security import create_access_token
from app.identity.person_models import IdentityPerson


async def _make_employee(db_session, email: str = "pat-user@example.com"):
    """Insert one active employee + the 'employee' role binding so the
    PAT endpoints accept their JWT.
    """
    user_id = uuid.uuid4()
    user = IdentityPerson(
        id=user_id,
        work_email=email,
        display_name="PAT User",
        employment_status="active",
        source_system="manual",
        hris_person_id=f"manual-{user_id}",
    )
    db_session.add(user)
    await db_session.flush()

    role_id_row = await db_session.execute(
        text("SELECT id FROM auth_roles WHERE name = 'employee'")
    )
    role_id = role_id_row.scalar_one()
    if isinstance(role_id, str):
        role_id = uuid.UUID(role_id)
    db_session.add(AuthUserRole(user_id=user_id, role_id=role_id))
    await db_session.flush()
    return user


def _bearer(user_id: uuid.UUID, roles: list[str] | None = None) -> dict[str, str]:
    return {
        "Authorization": (
            f"Bearer {create_access_token(str(user_id), roles or ['employee'])}"
        ),
    }


@pytest.mark.asyncio
async def test_create_returns_plaintext_once_then_list_renders(client, db_session):
    """End-to-end happy path: create a token, the response includes the
    plaintext exactly once, the listing renders the row with
    ``is_active=True`` and the SHA-256-derived prefix.
    """
    user = await _make_employee(db_session)

    # CREATE
    r = await client.post(
        "/api/v1/auth/me/tokens",
        json={"name": "Claude Desktop on test box", "expires_in_days": 30},
        headers=_bearer(user.id),
    )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["success"] is True, body
    plaintext = body["data"]["plaintext"]
    assert plaintext.startswith("helix_pat_")
    assert len(plaintext) > 30
    assert body["data"]["token"]["is_active"] is True
    assert body["data"]["token"]["expires_at"] is not None
    assert body["data"]["token"]["revoked_at"] is None
    prefix = body["data"]["token"]["token_prefix"]
    assert plaintext.startswith(prefix)

    # LIST — the regression gate. Before the fix, this 500'd because
    # ``is_active`` compared a naive datetime against ``datetime.now(UTC)``.
    r = await client.get("/api/v1/auth/me/tokens", headers=_bearer(user.id))
    assert r.status_code == 200, r.text
    rows = r.json()["data"]
    assert len(rows) == 1
    assert rows[0]["token_prefix"] == prefix
    assert rows[0]["is_active"] is True
    # Plaintext must NEVER appear on a list response.
    assert "plaintext" not in rows[0]


@pytest.mark.asyncio
async def test_revoke_then_list_shows_inactive(client, db_session):
    user = await _make_employee(db_session)
    create = await client.post(
        "/api/v1/auth/me/tokens",
        json={"name": "to-revoke", "expires_in_days": 90},
        headers=_bearer(user.id),
    )
    assert create.status_code == 200
    token_id = create.json()["data"]["token"]["id"]

    # REVOKE
    r = await client.delete(
        f"/api/v1/auth/me/tokens/{token_id}", headers=_bearer(user.id),
    )
    assert r.status_code == 200, r.text
    revoked = r.json()["data"]
    assert revoked["is_active"] is False
    assert revoked["revoked_at"] is not None

    # Idempotent — re-revoking returns the same shape, not an error.
    r = await client.delete(
        f"/api/v1/auth/me/tokens/{token_id}", headers=_bearer(user.id),
    )
    assert r.status_code == 200
    assert r.json()["data"]["is_active"] is False


@pytest.mark.asyncio
async def test_create_rejects_over_cap(client, db_session):
    user = await _make_employee(db_session)
    for i in range(MAX_TOKENS_PER_USER):
        r = await client.post(
            "/api/v1/auth/me/tokens",
            json={"name": f"t{i}", "expires_in_days": 30},
            headers=_bearer(user.id),
        )
        assert r.status_code == 200, r.text

    # 5th attempt should fail
    r = await client.post(
        "/api/v1/auth/me/tokens",
        json={"name": "over-cap", "expires_in_days": 30},
        headers=_bearer(user.id),
    )
    assert r.status_code == 409, r.text
    assert r.json()["errors"][0]["code"] == "CONFLICT"


@pytest.mark.asyncio
async def test_create_rejects_invalid_expiry(client, db_session):
    user = await _make_employee(db_session)
    r = await client.post(
        "/api/v1/auth/me/tokens",
        json={"name": "bad", "expires_in_days": 17},
        headers=_bearer(user.id),
    )
    assert r.status_code == 422, r.text


@pytest.mark.asyncio
async def test_pat_authenticates_against_employee_endpoint(client, db_session):
    """A freshly minted PAT can be used as a Bearer to call any other
    employee endpoint. This is the contract the MCP server depends on.
    """
    user = await _make_employee(db_session)
    create = await client.post(
        "/api/v1/auth/me/tokens",
        json={"name": "for-api", "expires_in_days": 30},
        headers=_bearer(user.id),
    )
    plaintext = create.json()["data"]["plaintext"]

    # Use the PAT to call /auth/me
    r = await client.get(
        "/api/v1/auth/me", headers={"Authorization": f"Bearer {plaintext}"},
    )
    assert r.status_code == 200, r.text
    assert r.json()["data"]["email"] == user.work_email


@pytest.mark.asyncio
async def test_pat_cannot_manage_other_pats(client, db_session):
    """Defence-in-depth: a stolen PAT must NOT be able to mint or revoke
    other PATs. Token management requires a JWT (Google SSO web session).
    """
    user = await _make_employee(db_session)
    create = await client.post(
        "/api/v1/auth/me/tokens",
        json={"name": "stolen", "expires_in_days": 30},
        headers=_bearer(user.id),
    )
    plaintext = create.json()["data"]["plaintext"]

    # Attempt to mint another PAT with the stolen one
    r = await client.post(
        "/api/v1/auth/me/tokens",
        json={"name": "perpetuated", "expires_in_days": 30},
        headers={"Authorization": f"Bearer {plaintext}"},
    )
    assert r.status_code == 403, r.text
    assert r.json()["errors"][0]["code"] == "FORBIDDEN"


@pytest.mark.asyncio
async def test_is_active_handles_naive_db_datetime(db_session):
    """Direct unit test for the naive-vs-aware regression. Inserts a row
    whose ``expires_at`` comes back from MySQL/TiDB as a naive datetime,
    then asserts ``is_active`` and ``authenticate`` don't raise
    ``TypeError`` on the comparison.
    """
    user = await _make_employee(db_session, email="naive-dt@example.com")
    service = PersonalAccessTokenService(db_session)
    row, plaintext = await service.create(
        user_id=user.id, name="naive", expires_in_days=30,
    )
    await db_session.flush()
    # Re-fetch — the in-memory row may have a tz-aware datetime; we want
    # to test the read path that goes through the driver.
    fetched = await db_session.get(PersonalAccessToken, row.id)
    assert fetched is not None
    # Pure-Python — the regression site.
    assert is_active(fetched) is True
    # And the auth path, which does its own comparison.
    found = await service.authenticate(plaintext)
    assert found is not None
    assert found.id == row.id


@pytest.mark.asyncio
async def test_count_active_excludes_expired(db_session):
    """Expired tokens must NOT consume a slot against MAX_TOKENS_PER_USER."""
    user = await _make_employee(db_session, email="exp@example.com")
    # Insert a manually-aged row via the ORM (bypass service to set
    # expires_at in the past).
    expired = PersonalAccessToken(
        user_id=user.id,
        name="aged",
        token_hash="0" * 64,
        token_prefix="helix_pat_aa",
        expires_at=datetime.now(UTC) - timedelta(days=1),
    )
    db_session.add(expired)
    await db_session.flush()

    svc = PersonalAccessTokenService(db_session)
    assert await svc.count_active_for_user(user.id) == 0


@pytest.mark.asyncio
async def test_revoke_other_users_token_is_not_found_not_forbidden(db_session):
    """Regression: revoking a token that isn't yours must raise NotFoundError,
    identical to revoking a non-existent id — otherwise a 403-vs-404 split
    lets an authenticated attacker probe which token ids are live (existence
    oracle). Mirrors the api-key /reveal 404/403 collapse.
    """
    from app.core.exceptions import NotFoundError

    owner = await _make_employee(db_session, email="pat-owner@example.com")
    attacker = await _make_employee(db_session, email="pat-attacker@example.com")
    svc = PersonalAccessTokenService(db_session)
    row, _plain = await svc.create(
        user_id=owner.id, name="owner-token", expires_in_days=90,
    )
    await db_session.flush()

    # Attacker (non-admin) trying to revoke the owner's real token …
    with pytest.raises(NotFoundError):
        await svc.revoke(token_id=row.id, actor_user_id=attacker.id, is_admin=False)

    # … is indistinguishable from revoking a token id that doesn't exist.
    with pytest.raises(NotFoundError):
        await svc.revoke(
            token_id=uuid.uuid4(), actor_user_id=attacker.id, is_admin=False,
        )

    # The owner's token was NOT revoked by the failed attempt.
    await db_session.refresh(row)
    assert row.revoked_at is None