"""API + RBAC tests for the 邮件组管理 estate surface and operation log.

Covers the three things that would silently corrupt the surface if they
regressed:

* the RBAC gate (employee 403 / hr 200 / global_admin 200);
* the tri-state ``is_dynamic`` semantics — a NULL row must never be
  counted or filtered as static, and ``direct_members_count = NULL``
  must never be counted as zero members;
* the operation log's group resolution — including that a soft-deleted
  group's rows stay searchable by the address it used to have, which is
  the whole reason the domain soft-deletes.

No Google calls: every endpoint under test is DB-only.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
from httpx import AsyncClient

from app.audit.models import AuditLog
from app.domains.email_group.models import EmailGroup, EmailGroupOwner

BASE = "/api/v1/email-groups/manage"


async def _mk_group(
    db,
    *,
    email: str,
    name: str | None = None,
    is_dynamic: bool | None = False,
    members: int | None = 3,
    status: str = "active",
    aliases: list[str] | None = None,
    owner_person_id=None,
) -> EmailGroup:
    row = EmailGroup(
        email=email,
        name=name or email.split("@")[0],
        status=status,
        is_dynamic=is_dynamic,
        direct_members_count=members,
        aliases=aliases,
        last_synced_at=datetime.now(UTC),
    )
    db.add(row)
    await db.flush()
    if owner_person_id is not None:
        db.add(EmailGroupOwner(email_group_id=row.id, person_id=owner_person_id))
    await db.commit()
    return row


async def _mk_audit(
    db,
    *,
    action: str,
    group_id: uuid.UUID | None,
    detail: dict | None = None,
    actor_user_id=None,
    actor_type: str = "user",
    created_at: datetime | None = None,
) -> AuditLog:
    row = AuditLog(
        trace_id=uuid.uuid4(),
        actor_user_id=actor_user_id,
        actor_type=actor_type,
        action=action,
        resource_type="email_group",
        resource_id=str(group_id) if group_id else None,
        status="success",
        detail=detail,
    )
    db.add(row)
    await db.flush()
    if created_at is not None:
        # ``created_at`` has a server_default, so it can only be pinned
        # after the INSERT assigned one.
        row.created_at = created_at
    await db.commit()
    return row


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


# ── RBAC ────────────────────────────────────────────────────────────


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path",
    ["/overview", "/groups", "/audit-log", "/audit-log/actions"],
)
async def test_employee_forbidden(
    client: AsyncClient, db_session, employee_user, path,
):
    _user, token = employee_user
    resp = await client.get(f"{BASE}{path}", headers=_auth(token))
    assert resp.status_code == 403, resp.text


@pytest.mark.asyncio
async def test_hr_can_read(client: AsyncClient, db_session, hr_user):
    _user, token = hr_user
    await _mk_group(db_session, email="eng@pingcap.com")
    resp = await client.get(f"{BASE}/groups", headers=_auth(token))
    assert resp.status_code == 200, resp.text
    assert [g["email"] for g in resp.json()["data"]] == ["eng@pingcap.com"]


@pytest.mark.asyncio
async def test_global_admin_can_read(client: AsyncClient, db_session, admin_user):
    _user, token = admin_user
    await _mk_group(db_session, email="eng@pingcap.com")
    resp = await client.get(f"{BASE}/overview", headers=_auth(token))
    assert resp.status_code == 200, resp.text


# ── Overview counters ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_overview_counters(client: AsyncClient, db_session, hr_user):
    user, token = hr_user
    # static, has owner, has members
    await _mk_group(
        db_session, email="owned@pingcap.com", members=5,
        owner_person_id=user.id,
    )
    # static, NO owner, has members → the stewardship gap
    await _mk_group(db_session, email="orphan@pingcap.com", members=7)
    # static, zero members (⇒ no owner, since Google counts owners) →
    # the cleanup pool
    await _mk_group(db_session, email="empty@pingcap.com", members=0)
    # dynamic
    await _mk_group(
        db_session, email="g-all@pingcap.com", is_dynamic=True, members=400,
    )
    # kind never determined — must NOT be counted as static
    await _mk_group(
        db_session, email="mystery@pingcap.com", is_dynamic=None, members=2,
    )
    # tombstone
    await _mk_group(
        db_session, email="gone@pingcap.com", status="deleted", members=0,
    )

    resp = await client.get(f"{BASE}/overview", headers=_auth(token))
    assert resp.status_code == 200, resp.text
    d = resp.json()["data"]
    assert d["total"] == 6
    assert d["active"] == 5
    assert d["static"] == 3
    assert d["dynamic"] == 1
    assert d["kind_unknown"] == 1
    assert d["static_zero_members"] == 1
    assert d["deleted"] == 1
    # orphan(7) + empty(0 → no owner) + g-all + mystery are all ownerless,
    # but only the ones we KNOW have members count as the gap.
    assert d["unowned_with_members"] == 3
    assert d["unowned"] == 4


@pytest.mark.asyncio
async def test_overview_unsynced_member_count_is_not_zero(
    client: AsyncClient, db_session, hr_user,
):
    """``direct_members_count IS NULL`` means never synced.

    Counting it as zero would drop an unsynced group into the cleanup
    pool, where the next bulk action would delete a group nobody has
    ever looked at.
    """
    _user, token = hr_user
    await _mk_group(db_session, email="fresh@pingcap.com", members=None)

    resp = await client.get(f"{BASE}/overview", headers=_auth(token))
    d = resp.json()["data"]
    assert d["static_zero_members"] == 0
    # Unknown membership is not a known stewardship gap either.
    assert d["unowned_with_members"] == 0
    assert d["unowned"] == 1


# ── Group list filters ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_kind_filter_excludes_undetermined(
    client: AsyncClient, db_session, hr_user,
):
    _user, token = hr_user
    await _mk_group(db_session, email="s@pingcap.com", is_dynamic=False)
    await _mk_group(db_session, email="d@pingcap.com", is_dynamic=True)
    await _mk_group(db_session, email="u@pingcap.com", is_dynamic=None)

    async def emails(kind: str) -> list[str]:
        r = await client.get(
            f"{BASE}/groups", params={"kind": kind}, headers=_auth(token),
        )
        assert r.status_code == 200, r.text
        return sorted(g["email"] for g in r.json()["data"])

    assert await emails("static") == ["s@pingcap.com"]
    assert await emails("dynamic") == ["d@pingcap.com"]
    assert await emails("unknown") == ["u@pingcap.com"]
    assert await emails("all") == [
        "d@pingcap.com", "s@pingcap.com", "u@pingcap.com",
    ]


@pytest.mark.asyncio
async def test_members_and_owners_filters(
    client: AsyncClient, db_session, hr_user,
):
    user, token = hr_user
    await _mk_group(
        db_session, email="owned@pingcap.com", members=4,
        owner_person_id=user.id,
    )
    await _mk_group(db_session, email="orphan@pingcap.com", members=4)
    await _mk_group(db_session, email="empty@pingcap.com", members=0)
    await _mk_group(db_session, email="unsynced@pingcap.com", members=None)

    async def emails(**params) -> list[str]:
        r = await client.get(f"{BASE}/groups", params=params, headers=_auth(token))
        assert r.status_code == 200, r.text
        return sorted(g["email"] for g in r.json()["data"])

    assert await emails(members="zero") == ["empty@pingcap.com"]
    assert await emails(members="unsynced") == ["unsynced@pingcap.com"]
    assert await emails(members="nonzero") == [
        "orphan@pingcap.com", "owned@pingcap.com",
    ]
    assert await emails(owners="some") == ["owned@pingcap.com"]
    assert await emails(owners="none") == [
        "empty@pingcap.com", "orphan@pingcap.com", "unsynced@pingcap.com",
    ]
    # The governance view: has members AND nobody accountable.
    assert await emails(members="nonzero", owners="none") == [
        "orphan@pingcap.com",
    ]


@pytest.mark.asyncio
async def test_deleted_hidden_unless_requested(
    client: AsyncClient, db_session, hr_user,
):
    _user, token = hr_user
    await _mk_group(db_session, email="live@pingcap.com")
    await _mk_group(db_session, email="gone@pingcap.com", status="deleted")

    r = await client.get(f"{BASE}/groups", headers=_auth(token))
    assert [g["email"] for g in r.json()["data"]] == ["live@pingcap.com"]

    r = await client.get(
        f"{BASE}/groups", params={"include_deleted": True}, headers=_auth(token),
    )
    assert sorted(g["email"] for g in r.json()["data"]) == [
        "gone@pingcap.com", "live@pingcap.com",
    ]


@pytest.mark.asyncio
async def test_google_test_domain_alias_is_hidden(
    client: AsyncClient, db_session, hr_user,
):
    """Google's ``.test-google-a.com`` mirror is not a real address.

    It exists on essentially every group, so surfacing it would make the
    alias column pure noise and make ~every cleanup candidate look like
    it has a second address worth preserving.
    """
    _user, token = hr_user
    await _mk_group(
        db_session, email="eng@pingcap.com",
        aliases=[
            "eng@pingcap.com.test-google-a.com",
            "engineering@pingcap.com",
        ],
    )
    r = await client.get(f"{BASE}/groups", headers=_auth(token))
    row = r.json()["data"][0]
    assert row["aliases"] == ["engineering@pingcap.com"]
    assert len(row["aliases_raw"]) == 2
    assert row["alias_never_synced"] is False


@pytest.mark.asyncio
async def test_alias_never_synced_flag(client: AsyncClient, db_session, hr_user):
    _user, token = hr_user
    await _mk_group(db_session, email="fresh@pingcap.com", aliases=None)
    r = await client.get(f"{BASE}/groups", headers=_auth(token))
    assert r.json()["data"][0]["alias_never_synced"] is True


@pytest.mark.asyncio
async def test_search_escapes_like_metacharacters(
    client: AsyncClient, db_session, hr_user,
):
    """``g_all`` must not match ``g-all`` via LIKE's single-char wildcard."""
    _user, token = hr_user
    await _mk_group(db_session, email="g-all@pingcap.com")
    r = await client.get(
        f"{BASE}/groups", params={"search": "g_all"}, headers=_auth(token),
    )
    assert r.json()["data"] == []
    r = await client.get(
        f"{BASE}/groups", params={"search": "g-all"}, headers=_auth(token),
    )
    assert len(r.json()["data"]) == 1


@pytest.mark.asyncio
async def test_sort_by_members_desc(client: AsyncClient, db_session, hr_user):
    _user, token = hr_user
    await _mk_group(db_session, email="a@pingcap.com", members=1)
    await _mk_group(db_session, email="b@pingcap.com", members=9)
    await _mk_group(db_session, email="c@pingcap.com", members=5)
    r = await client.get(
        f"{BASE}/groups",
        params={"sort": "members", "order": "desc"},
        headers=_auth(token),
    )
    assert [g["email"] for g in r.json()["data"]] == [
        "b@pingcap.com", "c@pingcap.com", "a@pingcap.com",
    ]


# ── Operation log ───────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_audit_log_enriches_actor_and_group(
    client: AsyncClient, db_session, hr_user,
):
    user, token = hr_user
    group = await _mk_group(db_session, email="eng@pingcap.com", name="Eng")
    await _mk_audit(
        db_session,
        action="email_group.delete",
        group_id=group.id,
        actor_user_id=user.id,
        detail={"group_email": "eng@pingcap.com", "reason": "cleanup"},
    )

    r = await client.get(f"{BASE}/audit-log", headers=_auth(token))
    assert r.status_code == 200, r.text
    row = r.json()["data"][0]
    assert row["action"] == "email_group.delete"
    assert row["action_category"] == "group"
    assert row["group_email"] == "eng@pingcap.com"
    assert row["group_name"] == "Eng"
    assert row["actor_display_name"] == "HR User"
    assert row["detail"]["reason"] == "cleanup"


@pytest.mark.asyncio
async def test_audit_log_only_returns_email_group_rows(
    client: AsyncClient, db_session, hr_user,
):
    _user, token = hr_user
    db_session.add(AuditLog(
        trace_id=uuid.uuid4(), actor_type="user", action="asset.update",
        resource_type="asset", resource_id=str(uuid.uuid4()), status="success",
    ))
    await db_session.commit()
    r = await client.get(f"{BASE}/audit-log", headers=_auth(token))
    assert r.json()["data"] == []
    assert r.json()["meta"]["total"] == 0


@pytest.mark.asyncio
async def test_audit_log_search_finds_deleted_group(
    client: AsyncClient, db_session, hr_user,
):
    """A cleaned-up group's history is still searchable by its address.

    This is the entire point of soft-deleting the ``email_groups`` row
    instead of hard-deleting it: after the cleanup workbench removes a
    group, "what happened to careers@?" must still be answerable.
    """
    _user, token = hr_user
    gone = await _mk_group(
        db_session, email="careers@pingcap.com", status="deleted", members=0,
    )
    live = await _mk_group(db_session, email="eng@pingcap.com")
    await _mk_audit(
        db_session, action="email_group.delete", group_id=gone.id,
        detail={"group_email": "careers@pingcap.com"},
    )
    await _mk_audit(db_session, action="email_group.create", group_id=live.id)

    r = await client.get(
        f"{BASE}/audit-log", params={"search": "careers"}, headers=_auth(token),
    )
    rows = r.json()["data"]
    assert len(rows) == 1
    assert rows[0]["group_email"] == "careers@pingcap.com"
    assert rows[0]["group_status"] == "deleted"


@pytest.mark.asyncio
async def test_audit_log_group_email_falls_back_to_detail(
    client: AsyncClient, db_session, hr_user,
):
    """A row whose group is gone from the DB still names its address."""
    _user, token = hr_user
    await _mk_audit(
        db_session,
        action="email_group.delete",
        group_id=uuid.uuid4(),  # no matching email_groups row
        detail={"group_email": "purged@pingcap.com"},
    )
    r = await client.get(f"{BASE}/audit-log", headers=_auth(token))
    row = r.json()["data"][0]
    assert row["group_email"] == "purged@pingcap.com"
    assert row["group_status"] is None


@pytest.mark.asyncio
async def test_audit_log_handles_non_uuid_resource_id(
    client: AsyncClient, db_session, hr_user,
):
    """``resource_id`` is a free-form string column.

    A row whose value is not a UUID (an address, a composite key) must
    not blow up the group-resolution batch lookup.
    """
    _user, token = hr_user
    db_session.add(AuditLog(
        trace_id=uuid.uuid4(), actor_type="system",
        action="email_group.sync_removed",
        resource_type="email_group", resource_id="not-a-uuid",
        status="success", detail={"group_email": "weird@pingcap.com"},
    ))
    await db_session.commit()
    r = await client.get(f"{BASE}/audit-log", headers=_auth(token))
    assert r.status_code == 200, r.text
    assert r.json()["data"][0]["group_email"] == "weird@pingcap.com"


@pytest.mark.asyncio
async def test_audit_log_filters(client: AsyncClient, db_session, hr_user):
    user, token = hr_user
    group = await _mk_group(db_session, email="eng@pingcap.com")
    await _mk_audit(
        db_session, action="email_group.create", group_id=group.id,
        actor_user_id=user.id,
    )
    await _mk_audit(
        db_session, action="email_group.sync_removed", group_id=group.id,
        actor_type="system",
    )

    async def actions(**params) -> list[str]:
        r = await client.get(
            f"{BASE}/audit-log", params=params, headers=_auth(token),
        )
        assert r.status_code == 200, r.text
        return sorted(x["action"] for x in r.json()["data"])

    assert await actions(actions="email_group.create") == [
        "email_group.create",
    ]
    assert await actions(category="sync") == ["email_group.sync_removed"]
    assert await actions(actor_type="system") == ["email_group.sync_removed"]
    assert await actions(actor_type="user") == ["email_group.create"]
    assert await actions(actor_person_id=str(user.id)) == [
        "email_group.create",
    ]
    # Multi-action OR.
    assert await actions(
        actions="email_group.create,email_group.sync_removed",
    ) == ["email_group.create", "email_group.sync_removed"]


@pytest.mark.asyncio
async def test_audit_log_date_range_is_inclusive(
    client: AsyncClient, db_session, hr_user,
):
    """A single-day range covers the whole calendar day.

    An end bound of 00:00:00 would make ``date_from == date_to`` return
    only rows written in the first microsecond of the day.
    """
    _user, token = hr_user
    group = await _mk_group(db_session, email="eng@pingcap.com")
    now = datetime.now(UTC)
    await _mk_audit(
        db_session, action="email_group.create", group_id=group.id,
        created_at=now.replace(hour=23, minute=59, second=30),
    )
    await _mk_audit(
        db_session, action="email_group.delete", group_id=group.id,
        created_at=now - timedelta(days=10),
    )

    today = now.date().isoformat()
    r = await client.get(
        f"{BASE}/audit-log",
        params={"date_from": today, "date_to": today},
        headers=_auth(token),
    )
    assert [x["action"] for x in r.json()["data"]] == ["email_group.create"]


@pytest.mark.asyncio
async def test_audit_log_search_too_broad_is_rejected(
    client: AsyncClient, db_session, hr_user, monkeypatch,
):
    """An over-broad term errors instead of showing a partial log.

    Silently truncating the resolved group-id IN-list would render a
    log that looks complete but is not — the worst possible failure for
    an audit surface.
    """
    from app.domains.email_group import manage_service

    monkeypatch.setattr(manage_service, "_AUDIT_GROUP_SEARCH_CAP", 1)
    _user, token = hr_user
    await _mk_group(db_session, email="a-team@pingcap.com")
    await _mk_group(db_session, email="b-team@pingcap.com")

    r = await client.get(
        f"{BASE}/audit-log", params={"search": "team"}, headers=_auth(token),
    )
    assert r.status_code == 400, r.text
    assert r.json()["errors"][0]["code"] == "SEARCH_TOO_BROAD"


@pytest.mark.asyncio
async def test_audit_log_unmatched_search_returns_empty_not_error(
    client: AsyncClient, db_session, hr_user,
):
    _user, token = hr_user
    r = await client.get(
        f"{BASE}/audit-log",
        params={"search": "nothing-matches-this"},
        headers=_auth(token),
    )
    assert r.status_code == 200, r.text
    assert r.json()["data"] == []
    assert r.json()["meta"]["total"] == 0


@pytest.mark.asyncio
async def test_audit_actions_catalog(client: AsyncClient, db_session, hr_user):
    _user, token = hr_user
    r = await client.get(f"{BASE}/audit-log/actions", headers=_auth(token))
    assert r.status_code == 200, r.text
    rows = r.json()["data"]
    by_action = {x["action"]: x["category"] for x in rows}
    # Registry-driven, so an action with zero rows in this fresh test DB
    # is still offered — that is the reason it is not a SELECT DISTINCT.
    assert "email_group.sync_owner_changed" in by_action
    assert by_action["email_group.delete"] == "group"
    assert by_action["email_group.sync_removed"] == "sync"
    assert by_action["email_group.dynamic_rule_updated"] == "dynamic"


# ── Per-group history (audit router registration) ───────────────────


@pytest.mark.asyncio
async def test_hr_must_not_reach_generic_email_group_history(
    client: AsyncClient, db_session, hr_user,
):
    """HR gets its per-group timeline from the DESK, not from the
    generic audit endpoint.

    This test used to assert the opposite. ``email_group`` was
    registered in ``_RESOURCE_HISTORY_PREFIX_ACCESS`` with ``HR_ROLE``
    added so the row-level 「历史」 drawer would work for the role that
    owns the desk. That endpoint filters on ``resource_type`` +
    ``resource_id`` and applies NO action filter — and the data-access
    domain writes ``data_access.binding.*`` rows against
    ``resource_type='email_group'``. So the registration handed anyone
    with the HR role a way to walk group ids and read another domain's
    BigQuery dataset topology and revoke history through a mail-group
    screen.

    The fix is not a narrower role list, it is a narrower QUERY:
    ``GET /email-groups/manage/audit-log?group_id=…`` is constrained to
    ``action LIKE 'email_group.%'`` in SQL, so the desk gets the same
    timeline and none of the reach. The assertion below is the
    load-bearing half — a future "let's just register the type" change
    turns it red.
    """
    _user, token = hr_user
    group = await _mk_group(db_session, email="eng@pingcap.com")
    await _mk_audit(
        db_session, action="email_group.create", group_id=group.id,
    )
    await _mk_audit(
        db_session,
        action="data_access.binding.system_revoked_group",
        group_id=group.id,
        detail={"dataset": "prod-analytics.finance_gl", "revoked": 3},
    )
    r = await client.get(
        "/api/v1/audit/resources/history",
        params={"resource_type": "email_group", "resource_id": str(group.id)},
        headers=_auth(token),
    )
    assert r.status_code == 403, r.text

    # …and the desk's own log answers, with the cross-domain row
    # filtered out at the query level rather than by the caller.
    r2 = await client.get(
        f"{BASE}/audit-log",
        params={"group_id": str(group.id)},
        headers=_auth(token),
    )
    assert r2.status_code == 200, r2.text
    actions = [x["action"] for x in r2.json()["data"]]
    assert actions == ["email_group.create"]
    assert not [a for a in actions if a.startswith("data_access.")]


@pytest.mark.asyncio
async def test_employee_cannot_read_per_group_history(
    client: AsyncClient, db_session, employee_user,
):
    _user, token = employee_user
    group = await _mk_group(db_session, email="eng@pingcap.com")
    r = await client.get(
        "/api/v1/audit/resources/history",
        params={"resource_type": "email_group", "resource_id": str(group.id)},
        headers=_auth(token),
    )
    assert r.status_code == 403, r.text


# ── The log is THIS domain's log ────────────────────────────────────


@pytest.mark.asyncio
async def test_log_excludes_other_domains_email_group_rows(
    client: AsyncClient, db_session, hr_user,
):
    """``resource_type='email_group'`` is not owned by this domain.

    ``data_access`` writes it with ``action='data_access.binding.*'``
    whose detail carries BigQuery project / dataset ids and
    per-principal role grants; ``soc2_email_group`` writes it too. This
    surface is readable by plain HR and its export serialises ``detail``
    VERBATIM, so filtering on resource_type alone would hand HR another
    domain's topology. The action prefix is what makes the log mean
    "things the mail-group desk did".
    """
    _user, token = hr_user
    group = await _mk_group(db_session, email="eng@pingcap.com")
    await _mk_audit(
        db_session, action="email_group.create", group_id=group.id,
    )
    # A foreign writer, same resource_type, sensitive detail.
    db_session.add(AuditLog(
        trace_id=uuid.uuid4(), actor_type="system",
        action="data_access.binding.system_revoked_group",
        resource_type="email_group", resource_id=str(group.id),
        status="success",
        detail={"effects": [{"target_project_id": "secret-prod-proj"}]},
    ))
    await db_session.commit()

    r = await client.get(f"{BASE}/audit-log", headers=_auth(token))
    assert r.status_code == 200, r.text
    actions = [x["action"] for x in r.json()["data"]]
    assert actions == ["email_group.create"]
    assert r.json()["meta"]["total"] == 1
    # And the per-group timeline is filtered the same way.
    r = await client.get(
        f"{BASE}/audit-log", params={"group_id": str(group.id)},
        headers=_auth(token),
    )
    assert [x["action"] for x in r.json()["data"]] == ["email_group.create"]


@pytest.mark.asyncio
async def test_other_category_is_reachable(
    client: AsyncClient, db_session, hr_user,
):
    """Applying a category filter must not hide unregistered actions.

    ``category_of`` returns ``other`` for anything the registry does not
    list and the list renders those rows, so the filter has to be able
    to select them. Resolving ``other`` to a registry subset (there is
    none) returned zero rows — i.e. any category filter silently
    dropped every unregistered action.
    """
    _user, token = hr_user
    group = await _mk_group(db_session, email="eng@pingcap.com")
    await _mk_audit(
        db_session, action="email_group.create", group_id=group.id,
    )
    await _mk_audit(
        db_session, action="email_group.brand_new_thing", group_id=group.id,
    )

    r = await client.get(
        f"{BASE}/audit-log", params={"category": "other"},
        headers=_auth(token),
    )
    assert [x["action"] for x in r.json()["data"]] == [
        "email_group.brand_new_thing",
    ]
    assert r.json()["data"][0]["action_category"] == "other"
    # …and the registered one is still selectable by its own category.
    r = await client.get(
        f"{BASE}/audit-log", params={"category": "group"},
        headers=_auth(token),
    )
    assert [x["action"] for x in r.json()["data"]] == ["email_group.create"]


# ── Pagination honesty ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_group_list_total_is_the_filtered_total(
    client: AsyncClient, db_session, hr_user,
):
    """``meta.total`` must be the COUNT of the filter, not of the page.

    Nothing else in the suite passes ``page``/``page_size``, so a total
    computed without the filter conditions — or replaced by
    ``len(rows)`` — would leave the UI's pager reporting one page for a
    760-group estate with every test still green.
    """
    _user, token = hr_user
    for i in range(7):
        await _mk_group(db_session, email=f"g{i}@pingcap.com", members=1)
    await _mk_group(db_session, email="empty@pingcap.com", members=0)

    r = await client.get(
        f"{BASE}/groups", params={"page": 1, "page_size": 3},
        headers=_auth(token),
    )
    assert r.status_code == 200, r.text
    assert len(r.json()["data"]) == 3
    assert r.json()["meta"]["total"] == 8

    # Second page continues rather than restarting.
    r2 = await client.get(
        f"{BASE}/groups", params={"page": 2, "page_size": 3},
        headers=_auth(token),
    )
    first = {g["email"] for g in r.json()["data"]}
    second = {g["email"] for g in r2.json()["data"]}
    assert not (first & second)
    assert r2.json()["meta"]["total"] == 8

    # A filter narrows the total too.
    r3 = await client.get(
        f"{BASE}/groups", params={"members": "zero", "page_size": 3},
        headers=_auth(token),
    )
    assert r3.json()["meta"]["total"] == 1


# ── Exports ─────────────────────────────────────────────────────────


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path",
    [
        "/groups/export.xlsx",
        "/audit-log/export.xlsx",
        "/cleanup-candidates/export.xlsx",
    ],
)
async def test_exports_are_hr_gated(
    client: AsyncClient, db_session, employee_user, path,
):
    _user, token = employee_user
    r = await client.get(f"{BASE}{path}", headers=_auth(token))
    assert r.status_code == 403, r.text


@pytest.mark.asyncio
async def test_group_export_streams_every_matching_row(
    client: AsyncClient, db_session, hr_user,
):
    """The export must not silently ship one interactive page.

    The service used to clamp every caller to the API page maximum, so
    the export's own cap constant was unreachable and a 760-group estate
    exported as 200 rows while reporting 760 as the total.
    """
    _user, token = hr_user
    for i in range(250):
        db_session.add(EmailGroup(
            email=f"g{i:04d}@pingcap.com", name=f"g{i}", status="active",
            is_dynamic=False, direct_members_count=1,
        ))
    await db_session.commit()

    r = await client.get(f"{BASE}/groups/export.xlsx", headers=_auth(token))
    assert r.status_code == 200, r.text
    assert r.headers["x-export-total-matched"] == "250"
    assert r.headers["x-export-returned"] == "250"
    assert r.headers["x-export-truncated"] == "false"


# ── Round-4: the by-id read ─────────────────────────────────────────


@pytest.mark.asyncio
async def test_get_one_group_by_id_is_gated_to_the_desk(
    client: AsyncClient, db_session, hr_user, employee_user,
):
    """`GET /manage/groups/{id}` returns the full owner roster.

    Display name, work email, employee number and employment status for
    an arbitrary group id INCLUDING tombstones — so the gate has to be
    real, and it has to live in the service (a router-body check is
    skipped on an `@idempotent` replay and invisible to any second
    caller of the same logic).

    No test covered this endpoint when it shipped, which is the actual
    risk: the next refactor that folds the read into the service the way
    every sibling does would drop the router-body check with it and the
    whole suite would stay green.
    """
    _u, hr_token = hr_user
    _e, employee_token = employee_user
    group = await _mk_group(db_session, email="eng@pingcap.com")

    ok = await client.get(f"{BASE}/groups/{group.id}", headers=_auth(hr_token))
    assert ok.status_code == 200, ok.text
    assert ok.json()["data"]["email"] == "eng@pingcap.com"

    denied = await client.get(
        f"{BASE}/groups/{group.id}", headers=_auth(employee_token),
    )
    assert denied.status_code == 403, denied.text

    # And an unknown id is a clean 404, not a 500.
    missing = await client.get(
        f"{BASE}/groups/{uuid.uuid4()}", headers=_auth(hr_token),
    )
    assert missing.status_code == 404, missing.text


@pytest.mark.asyncio
async def test_the_by_id_read_does_not_shadow_the_export_route(
    client: AsyncClient, db_session, hr_user,
):
    """Route order: `{group_id}` must not swallow `export.xlsx`.

    FastAPI matches in declaration order, and `group_id` is typed
    `UUID` — so a param route declared first turns the export into a
    422 ("export.xlsx is not a valid uuid"), which reads like a broken
    export rather than a routing mistake. Cheap to assert, and it
    cannot be caught by reading either handler alone.
    """
    _u, token = hr_user
    await _mk_group(db_session, email="eng@pingcap.com")
    r = await client.get(f"{BASE}/groups/export.xlsx", headers=_auth(token))
    assert r.status_code == 200, r.text
    assert "spreadsheet" in r.headers["content-type"]