"""Regression: ``GET /cmdb/my-pending-grant-approvals``.

Added 2026-05-19 after a real prod incident — Louis Liu submitted an
AWS PingCAP cloud-access request, the scope owner (Alex Quan) received
the Feishu notification but couldn't find where to act, and the IT
admin helping out couldn't see the approve button on the
``/my/cloud/scopes/{id}`` page (frontend ``isOwner`` gate hid it from
non-owners even though the backend permitted admin approval). The
workaround was to transfer scope ownership to self → approve →
(should transfer back), which is wrong for any audit trail.

The new endpoint is the cross-scope inbox that powers ``MyTodosPage``'s
"云授权审批" tab so the owner has one place to see every pending grant
without per-scope deep-links.

Contract pinned by this test:

* Scope owner sees pending grants on scopes they own — and ONLY those
  (no leak from other scopes).
* Self-requested grants are filtered out — four-eyes rule.
* ``global_admin`` sees every pending grant org-wide.
* Plain employees with no owned scope get an empty list (200, not 403).
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.models import (
    GRANT_STATUS_ACTIVE,
    GRANT_STATUS_PENDING,
    SCOPE_STATUS_ACTIVE,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
)
from app.integrations.models import Integration


async def _integration(db: AsyncSession) -> Integration:
    row = Integration(
        id=uuid.uuid4(),
        kind="cloud",
        provider="aws",
        name=f"aws-test-{uuid.uuid4().hex[:6]}",
        slug=f"aws-test-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=False,
        capabilities=["servers"],
    )
    db.add(row)
    await db.flush()
    return row


async def _scope(
    db: AsyncSession,
    integration: Integration,
    *,
    owner_person_id: str | None = None,
) -> CmdbCloudScope:
    s = CmdbCloudScope(
        id=uuid.uuid4(),
        integration_id=str(integration.id),
        scope_type="aws_account",
        external_id=uuid.uuid4().hex[:12],
        display_name=f"scope-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner_person_id,
        status=SCOPE_STATUS_ACTIVE,
        first_seen_at=datetime.now(UTC),
        last_synced_at=datetime.now(UTC),
        environment="dev",
    )
    db.add(s)
    await db.flush()
    return s


async def _grant(
    db: AsyncSession,
    scope: CmdbCloudScope,
    *,
    person_id: str,
    status: str = GRANT_STATUS_PENDING,
) -> CmdbCloudAccessGrant:
    g = CmdbCloudAccessGrant(
        id=uuid.uuid4(),
        scope_id=str(scope.id),
        person_id=person_id,
        role_key=f"arn:aws:sso:::permissionSet/test/{uuid.uuid4().hex[:8]}",
        role_display_name="ReadOnlyAccess",
        status=status,
        granted_at=datetime.now(UTC),
        detail={"reason": "test reason", "requested_ttl_days": 7},
    )
    db.add(g)
    await db.flush()
    return g


@pytest.mark.asyncio
async def test_owner_sees_pending_on_own_scope(
    client, db_session: AsyncSession, admin_user, employee_user,
):
    """Owner perspective: every pending grant on a scope I own appears.

    Owner is the ``employee_user`` (using employee role to prove the
    endpoint is NOT global_admin-gated). Admin is unused here — we
    only exercise the plain owner path.
    """
    owner, owner_token = employee_user
    _admin, _ = admin_user  # ensure admin row exists; not used as caller
    integ = await _integration(db_session)
    my_scope = await _scope(db_session, integ, owner_person_id=str(owner.id))
    other_scope = await _scope(db_session, integ, owner_person_id=str(_admin.id))
    requester_id = str(uuid.uuid4())
    # Make a requester person row so person_label rendering doesn't
    # NULL out on the orphan path — the endpoint serialises every row
    # through _serialize_grants which expects a real Identity row.
    from app.identity.person_models import IdentityPerson
    requester = IdentityPerson(
        id=uuid.UUID(requester_id),
        work_email="req@test.com",
        display_name="Requester",
        employment_status="active",
        source_system="successfactors",
        hris_person_id="sf-req-001",
    )
    db_session.add(requester)
    await db_session.flush()

    # Pending on owner's scope — should appear.
    my_pending = await _grant(db_session, my_scope, person_id=requester_id)
    # Active on owner's scope — should NOT appear (only pending).
    await _grant(
        db_session, my_scope, person_id=requester_id,
        status=GRANT_STATUS_ACTIVE,
    )
    # Pending on someone else's scope — should NOT appear for plain owner.
    await _grant(db_session, other_scope, person_id=requester_id)
    # Self-requested pending — four-eyes: should NOT appear.
    await _grant(db_session, my_scope, person_id=str(owner.id))
    await db_session.commit()

    resp = await client.get(
        "/api/v1/cmdb/my-pending-grant-approvals",
        headers={"Authorization": f"Bearer {owner_token}"},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["success"] is True
    ids = {row["id"] for row in body["data"]["items"]}
    assert ids == {str(my_pending.id)}
    assert body["data"]["total"] == 1
    # Serializer surfaces the requester reason + TTL — the inbox
    # renders both columns, so this contract must stay stable.
    row = body["data"]["items"][0]
    assert row["request_reason"] == "test reason"
    assert row["requested_ttl_days"] == 7


@pytest.mark.asyncio
async def test_global_admin_sees_org_wide_pending(
    client, db_session: AsyncSession, admin_user,
):
    """Admin perspective: every pending grant, every scope.

    Mirrors the admin "all approvals" surface but scoped to grant
    approvals only. No filter on scope ownership.
    """
    _admin, admin_token = admin_user
    integ = await _integration(db_session)
    s1 = await _scope(db_session, integ)
    s2 = await _scope(db_session, integ)
    from app.identity.person_models import IdentityPerson
    req_id = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=req_id, work_email="r@x.com", display_name="r",
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{req_id.hex[:6]}",
    ))
    await db_session.flush()

    g1 = await _grant(db_session, s1, person_id=str(req_id))
    g2 = await _grant(db_session, s2, person_id=str(req_id))
    await _grant(
        db_session, s1, person_id=str(req_id),
        status=GRANT_STATUS_ACTIVE,
    )  # not pending — excluded
    await db_session.commit()

    resp = await client.get(
        "/api/v1/cmdb/my-pending-grant-approvals",
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    assert resp.status_code == 200, resp.text
    ids = {row["id"] for row in resp.json()["data"]["items"]}
    assert ids == {str(g1.id), str(g2.id)}


@pytest.mark.asyncio
async def test_submit_access_request_notifies_requester(
    client, db_session: AsyncSession, employee_user, admin_user,
):
    """Submission must produce a durable confirmation for the requester.

    Until 2026-05-19 the only acknowledgement was the HTTP 200 + a
    frontend toast — once the tab closed, the requester had no record
    that their submission was even received. Now the requester gets
    a ``notification_outbox`` row titled "[已提交] …" alongside the
    owner's "[待审批] …" row, so the confirmation survives a browser
    refresh and threads against the same ``related_resource_id`` as
    the eventual approve / reject.
    """
    from sqlalchemy import select

    from app.notifications.models import NotificationOutbox

    employee, employee_token = employee_user
    admin, _ = admin_user
    integ = await _integration(db_session)
    # Admin owns the scope; employee is the requester. ``submit`` runs
    # via the HTTP handler so the audit / notify wiring is exercised
    # end-to-end.
    scope = await _scope(db_session, integ, owner_person_id=str(admin.id))
    await db_session.commit()

    resp = await client.post(
        "/api/v1/cmdb/grants/request",
        json={
            "scope_id": str(scope.id),
            "role_key": (
                "arn:aws:sso:::permissionSet/test/"
                + uuid.uuid4().hex[:8]
            ),
            "role_display_name": "ReadOnlyAccess",
            "ttl_days": 7,
            "reason": "submit-notifies test",
        },
        headers={"Authorization": f"Bearer {employee_token}"},
    )
    assert resp.status_code == 201, resp.text

    # Owner notification — pre-existing contract, still must fire.
    owner_rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.recipient_person_id == admin.id,
        )
    )).scalars().all()
    owner_notification = next((r for r in owner_rows if "[待审批]" in r.title), None)
    assert owner_notification is not None, (
        f"owner missing pending-approval notification; got titles="
        f"{[r.title for r in owner_rows]}"
    )
    # Preserve the desktop's role-aware scope route while carrying the
    # precise pending grant id. The Feishu-mobile mapper promotes this to
    # /m/cloud/grant-approval/:id, so a card tap opens the approval itself.
    assert owner_notification.url.endswith(
        "?tab=pending&grant=" + str(resp.json()["data"]["id"])
    )

    # Requester notification — the new contract this test pins.
    req_rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.recipient_person_id == employee.id,
        )
    )).scalars().all()
    assert any("[已提交]" in r.title for r in req_rows), (
        f"requester missing submitted-confirmation; got titles="
        f"{[r.title for r in req_rows]}"
    )


@pytest.mark.asyncio
async def test_employee_with_no_owned_scope_gets_empty_list(
    client, db_session: AsyncSession, employee_user,
):
    """Plain employee — no 403, empty list. The MyTodosPage badge
    polls this in the background for everyone; a 403 would flash an
    error toast on every page load."""
    _, emp_token = employee_user
    integ = await _integration(db_session)
    # Scope owned by someone else, with a pending grant; employee
    # must NOT see it.
    other_owner_id = uuid.uuid4()
    from app.identity.person_models import IdentityPerson
    db_session.add(IdentityPerson(
        id=other_owner_id, work_email="o@x.com", display_name="o",
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{other_owner_id.hex[:6]}",
    ))
    await db_session.flush()
    scope = await _scope(
        db_session, integ, owner_person_id=str(other_owner_id),
    )
    req_id = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=req_id, work_email="r2@x.com", display_name="r2",
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{req_id.hex[:6]}",
    ))
    await db_session.flush()
    await _grant(db_session, scope, person_id=str(req_id))
    await db_session.commit()

    resp = await client.get(
        "/api/v1/cmdb/my-pending-grant-approvals",
        headers={"Authorization": f"Bearer {emp_token}"},
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["items"] == []
    assert resp.json()["data"]["total"] == 0