"""HTTP surface for owner-approved cloud grant renewal.

The service-level state machine is covered in
``test_grant_renewal_approval.py``. This file pins the things only the
HTTP layer can get wrong, and the ones that matter most if they regress:

* the old self-service ``POST /grants/{id}/renew`` no longer works for a
  regular holder — that route was the whole problem, and leaving it open
  would make the approval gate cosmetic;
* it still works for a cloud admin as an explicitly-audited break-glass;
* the expiry reminder card's magic link now FILES A REQUEST instead of
  renewing, while still accepting tokens minted under the old purpose
  (live cards in people's Feishu have TTLs up to 44 days);
* the approval queue is not readable by an unrelated employee.
"""
from __future__ import annotations

import uuid
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, patch

import pytest
from httpx import AsyncClient
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthUserRole
from app.domains.cmdb.models import (
    GRANT_STATUS_ACTIVE,
    RENEWAL_STATUS_APPROVED,
    RENEWAL_STATUS_PENDING_OWNER,
    SCOPE_STATUS_ACTIVE,
    CmdbCloudAccessGrant,
    CmdbCloudGrantRenewalRequest,
    CmdbCloudScope,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

API = "/api/v1/cmdb"


async def _role_id(db: AsyncSession, name: str) -> uuid.UUID:
    row = await db.execute(
        text("SELECT id FROM auth_roles WHERE name = :n"), {"n": name},
    )
    return uuid.UUID(str(row.scalar_one()))


async def _user(
    db: AsyncSession, *, email: str, roles: tuple[str, ...] = ("employee",),
) -> tuple[IdentityPerson, str]:
    from app.core.security import create_access_token

    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@")[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.flush()
    for r in roles:
        db.add(AuthUserRole(user_id=p.id, role_id=await _role_id(db, r)))
    await db.commit()
    token = create_access_token(str(p.id), list(roles))
    return p, token


async def _scope_with_grant(
    db: AsyncSession, *, holder: IdentityPerson, owner: IdentityPerson,
    environment: str = "dev",
):
    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aws",
        name=f"aws-http-{uuid.uuid4().hex[:6]}",
        slug=f"aws-http-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False, capabilities=["servers"],
    )
    db.add(integ)
    await db.flush()
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integ.id),
        scope_type="aws_account", external_id=uuid.uuid4().hex[:12],
        display_name="HTTP Renew Scope", status=SCOPE_STATUS_ACTIVE,
        environment=environment, owner_person_id=str(owner.id),
        default_grant_ttl_days=90,
    )
    db.add(scope)
    await db.flush()
    grant = CmdbCloudAccessGrant(
        id=uuid.uuid4(), scope_id=str(scope.id), person_id=str(holder.id),
        role_key="role/reader", status=GRANT_STATUS_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=60),
        expires_at=datetime.now(UTC) + timedelta(days=5),
        source="discovered",
    )
    db.add(grant)
    await db.commit()
    return scope, grant


def _patch_notify():
    async def _noop(_env, **_kw):
        return None
    return patch(
        "app.notifications.service.NotificationService.dispatch",
        new=AsyncMock(side_effect=_noop),
    )


def _patch_locks():
    @asynccontextmanager
    async def _noop(*_a, **_k):
        yield
    return (
        patch(
            "app.domains.cmdb.lifecycle_service.distributed_lock", new=_noop,
        ),
        patch(
            "app.domains.cmdb.renewal_service.distributed_lock", new=_noop,
        ),
    )


@pytest.mark.asyncio
async def test_holder_cannot_self_renew_via_http(
    client: AsyncClient, db_session: AsyncSession,
) -> None:
    """The route that caused the problem must refuse the holder.

    763 renewals in three months were executed this way, by the holder
    alone. Leaving the route open to holders would make the new owner
    approval purely decorative.
    """
    holder, holder_token = await _user(db_session, email="h-http@test.com")
    owner, _ = await _user(db_session, email="o-http@test.com")
    scope, grant = await _scope_with_grant(
        db_session, holder=holder, owner=owner,
    )

    resp = await client.post(
        f"{API}/grants/{grant.id}/renew",
        headers={"Authorization": f"Bearer {holder_token}"},
    )
    assert resp.status_code == 403, resp.text
    body = resp.json()
    assert body["errors"][0]["code"] == "CLOUD_RENEWAL_NEEDS_APPROVAL"


@pytest.mark.asyncio
async def test_cloud_admin_break_glass_renew_is_audited(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_user,
) -> None:
    """The break-glass path stays available but leaves its own trail, so
    "which renewals skipped approval?" is one query."""
    from app.audit.models import AuditLog

    admin, admin_token = cloud_admin_user
    holder, _ = await _user(db_session, email="h-bg@test.com")
    owner, _ = await _user(db_session, email="o-bg@test.com")
    scope, grant = await _scope_with_grant(
        db_session, holder=holder, owner=owner,
    )
    lock_a, lock_b = _patch_locks()

    with _patch_notify(), lock_a, lock_b:
        resp = await client.post(
            f"{API}/grants/{grant.id}/renew",
            headers={"Authorization": f"Bearer {admin_token}"},
        )
    assert resp.status_code == 200, resp.text

    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "cloud.grant.renew_admin_override",
            AuditLog.resource_id == str(grant.id),
        )
    )).scalars().all()
    assert len(rows) == 1, "break-glass renewal must be separately audited"
    assert rows[0].detail.get("bypassed") == "owner_approval"


@pytest.mark.asyncio
async def test_submit_then_owner_approves_over_http(
    client: AsyncClient, db_session: AsyncSession,
) -> None:
    """Full round trip through the HTTP surface."""
    holder, holder_token = await _user(db_session, email="h-rt@test.com")
    owner, owner_token = await _user(db_session, email="o-rt@test.com")
    scope, grant = await _scope_with_grant(
        db_session, holder=holder, owner=owner,
    )
    scope_id, gid = str(scope.id), grant.id
    lock_a, lock_b = _patch_locks()

    with _patch_notify(), lock_a, lock_b:
        submit = await client.post(
            f"{API}/scopes/{scope_id}/renewal-requests",
            headers={"Authorization": f"Bearer {holder_token}"},
            json={"ttl_days": 90, "reason": "压测还没做完"},
        )
        assert submit.status_code == 200, submit.text
        group_id = submit.json()["data"]["request_group_id"]

        # The holder sees it in their own history.
        mine = await client.get(
            f"{API}/me/renewal-requests",
            headers={"Authorization": f"Bearer {holder_token}"},
        )
        assert mine.status_code == 200
        assert any(
            r["request_group_id"] == group_id
            and r["status"] == RENEWAL_STATUS_PENDING_OWNER
            for r in mine.json()["data"]
        )

        # The owner sees it in their queue.
        queue = await client.get(
            f"{API}/scopes/{scope_id}/renewal-requests",
            headers={"Authorization": f"Bearer {owner_token}"},
        )
        assert queue.status_code == 200, queue.text
        groups = queue.json()["data"]
        assert len(groups) == 1

        decide = await client.post(
            f"{API}/renewal-requests/{group_id}/decide",
            headers={"Authorization": f"Bearer {owner_token}"},
            json={"approve": True, "note": "ok"},
        )
        assert decide.status_code == 200, decide.text
        assert decide.json()["data"]["status"] == RENEWAL_STATUS_APPROVED

    db_session.expire_all()
    after = await db_session.get(CmdbCloudAccessGrant, gid)
    assert after is not None
    assert after.last_renewed_at is not None


@pytest.mark.asyncio
async def test_unrelated_employee_cannot_read_the_approval_queue(
    client: AsyncClient, db_session: AsyncSession,
) -> None:
    """The queue names colleagues and their access — not public."""
    holder, _ = await _user(db_session, email="h-priv@test.com")
    owner, _ = await _user(db_session, email="o-priv@test.com")
    stranger, stranger_token = await _user(
        db_session, email="x-priv@test.com",
    )
    scope, grant = await _scope_with_grant(
        db_session, holder=holder, owner=owner,
    )

    resp = await client.get(
        f"{API}/scopes/{scope.id}/renewal-requests",
        headers={"Authorization": f"Bearer {stranger_token}"},
    )
    assert resp.status_code == 403, resp.text


@pytest.mark.asyncio
@pytest.mark.parametrize("purpose", [
    "cloud_grant_renewal_request",
    # Cards minted before 2026-07-29 carry the legacy purpose and have
    # TTLs up to 44 days — a live card must not become a dead end.
    "cloud_grant_renew",
])
async def test_reminder_card_link_files_a_request(
    client: AsyncClient, db_session: AsyncSession, purpose: str,
) -> None:
    """The magic link no longer renews on click; it asks."""
    from app.core.security import create_action_token

    holder, _ = await _user(db_session, email=f"h-tok-{purpose}@test.com")
    owner, _ = await _user(db_session, email=f"o-tok-{purpose}@test.com")
    scope, grant = await _scope_with_grant(
        db_session, holder=holder, owner=owner,
    )
    before_expiry = grant.expires_at
    gid = grant.id
    token = create_action_token(
        purpose=purpose,
        subject_id=str(holder.id),
        resource_id=str(gid),
        ttl_days=30,
    )
    lock_a, lock_b = _patch_locks()

    with _patch_notify(), lock_a, lock_b:
        resp = await client.post(
            f"{API}/grants/renew-by-token", params={"token": token},
        )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["renewal_requested"] is True
    assert data["request_group_id"]

    db_session.expire_all()
    after = await db_session.get(CmdbCloudAccessGrant, gid)
    assert after is not None
    # Clicking must NOT have extended anything by itself.
    assert after.expires_at == before_expiry or abs(
        (after.expires_at - before_expiry.replace(tzinfo=None)).total_seconds()
        if after.expires_at.tzinfo is None
        else (after.expires_at - before_expiry).total_seconds(),
    ) < 2
    rows = (await db_session.execute(
        select(CmdbCloudGrantRenewalRequest).where(
            CmdbCloudGrantRenewalRequest.grant_id == str(gid),
        )
    )).scalars().all()
    assert len(rows) == 1
    assert rows[0].status == RENEWAL_STATUS_PENDING_OWNER


@pytest.mark.asyncio
async def test_ttl_over_hard_cap_is_rejected_at_the_boundary(
    client: AsyncClient, db_session: AsyncSession,
) -> None:
    """422 from the schema, before any DB work."""
    holder, holder_token = await _user(db_session, email="h-cap@test.com")
    owner, _ = await _user(db_session, email="o-cap@test.com")
    scope, grant = await _scope_with_grant(
        db_session, holder=holder, owner=owner,
    )

    resp = await client.post(
        f"{API}/scopes/{scope.id}/renewal-requests",
        headers={"Authorization": f"Bearer {holder_token}"},
        json={"ttl_days": 365, "reason": "想要一年"},
    )
    assert resp.status_code == 422, resp.text


@pytest.mark.asyncio
async def test_pending_approvals_list_matches_the_inbox_provider(
    client: AsyncClient, db_session: AsyncSession,
) -> None:
    """The owner's "awaiting me" list and the mobile to-do badge must agree.

    They are two implementations of the same authority rule (owner sees own
    scopes, cloud admin sees all, self-submitted excluded). A badge that
    shows work the page then says does not exist is worse than no badge, so
    this asserts both answer identically for the same caller.
    """
    from app.mobile.action_inbox import _provider_cloud_renewal_approval

    holder, holder_token = await _user(db_session, email="h-inbox@test.com")
    owner, owner_token = await _user(db_session, email="o-inbox@test.com")
    stranger, stranger_token = await _user(db_session, email="x-inbox@test.com")
    scope, grant = await _scope_with_grant(
        db_session, holder=holder, owner=owner,
    )
    scope_id = str(scope.id)
    lock_a, lock_b = _patch_locks()

    with _patch_notify(), lock_a, lock_b:
        submit = await client.post(
            f"{API}/scopes/{scope_id}/renewal-requests",
            headers={"Authorization": f"Bearer {holder_token}"},
            json={"ttl_days": 90, "reason": "还要用"},
        )
        assert submit.status_code == 200, submit.text
        group_id = submit.json()["data"]["request_group_id"]

    # Owner: endpoint sees exactly one, and so does the inbox provider.
    resp = await client.get(
        f"{API}/my-pending-renewal-approvals",
        headers={"Authorization": f"Bearer {owner_token}"},
    )
    assert resp.status_code == 200, resp.text
    rows = resp.json()["data"]
    assert [r["request_group_id"] for r in rows] == [group_id]
    # Plain scope owner is not a privileged viewer → 工号 withheld, same
    # rule ``_serialize_grants`` applies to grant rows.
    assert rows[0]["person_employee_no"] is None

    cands = await _provider_cloud_renewal_approval(
        db_session, owner.id, ["employee"],
    )
    assert [c.item.id for c in cands] == [
        f"cloud_renewal_approval:{group_id}"
    ]
    assert cands[0].item.deep_link == (
        f"/m/cloud/renewal-approval/{group_id}"
    )

    # An unrelated employee owes nothing, on both surfaces.
    resp = await client.get(
        f"{API}/my-pending-renewal-approvals",
        headers={"Authorization": f"Bearer {stranger_token}"},
    )
    assert resp.status_code == 200
    assert resp.json()["data"] == []
    assert await _provider_cloud_renewal_approval(
        db_session, stranger.id, ["employee"],
    ) == []

    # The requester must not see their own submission as a to-do — the
    # decide path refuses self-approval, so it would be un-completable.
    assert await _provider_cloud_renewal_approval(
        db_session, holder.id, ["employee"],
    ) == []


@pytest.mark.asyncio
async def test_access_ledger_shows_renewal_events_and_is_gated(
    client: AsyncClient, db_session: AsyncSession,
) -> None:
    """The ledger must show what the old 「申请记录」 tab could not.

    That tab re-rendered grant rows sorted by ``granted_at``, so a renewal —
    which mutates an existing row and adds none — left no trace at all (92
    renewals on aws-account 385595570414 were invisible). The ledger is
    audit-derived, so the request and the decision both appear.
    """
    holder, holder_token = await _user(db_session, email="h-led@test.com")
    owner, owner_token = await _user(db_session, email="o-led@test.com")
    stranger, stranger_token = await _user(db_session, email="x-led@test.com")
    scope, grant = await _scope_with_grant(
        db_session, holder=holder, owner=owner,
    )
    scope_id = str(scope.id)
    lock_a, lock_b = _patch_locks()

    with _patch_notify(), lock_a, lock_b:
        submit = await client.post(
            f"{API}/scopes/{scope_id}/renewal-requests",
            headers={"Authorization": f"Bearer {holder_token}"},
            json={"ttl_days": 90, "reason": "压测未完"},
        )
        assert submit.status_code == 200, submit.text
        group_id = submit.json()["data"]["request_group_id"]
        decide = await client.post(
            f"{API}/renewal-requests/{group_id}/decide",
            headers={"Authorization": f"Bearer {owner_token}"},
            json={"approve": True},
        )
        assert decide.status_code == 200, decide.text

    resp = await client.get(
        f"{API}/scopes/{scope_id}/access-ledger",
        headers={"Authorization": f"Bearer {owner_token}"},
    )
    assert resp.status_code == 200, resp.text
    kinds = [e["event_kind"] for e in resp.json()["data"]]
    # Both halves of the story, newest first.
    assert "renewal_requested" in kinds
    assert "renewal_approved" in kinds
    assert "renewed" in kinds
    # NOT asserting that approve sorts above request: ``audit_logs.created_at``
    # is second-granular and a submit-then-approve burst lands in the same
    # second, so their relative order is not something the data can express.
    # What MUST hold is determinism — a list that reshuffles between
    # refreshes reads as "events changed" to whoever is watching it.
    again = await client.get(
        f"{API}/scopes/{scope_id}/access-ledger",
        headers={"Authorization": f"Bearer {owner_token}"},
    )
    assert [e["id"] for e in again.json()["data"]] == [
        e["id"] for e in resp.json()["data"]
    ], "ledger order must be stable across calls"

    # Not readable by an unrelated employee — it names colleagues and the
    # reasons they gave.
    denied = await client.get(
        f"{API}/scopes/{scope_id}/access-ledger",
        headers={"Authorization": f"Bearer {stranger_token}"},
    )
    assert denied.status_code == 403, denied.text