"""Router-level tests for the email-group directory.

Covers the new filter surface (GET /email-groups with suffix / owner /
only_mine / only_unowned / created_from / created_to) and the
on-demand transfer-owner endpoint (POST /email-groups/{id}/transfer-owner)
including RBAC.

Google driver is mocked — these tests never make a real HTTP call.
Google is exercised by the unit tests in test_integrations/.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from unittest.mock import patch

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

from app.domains.email_group.models import EmailGroup, EmailGroupOwner
from app.identity.person_models import IdentityPerson

# ── Helpers ────────────────────────────────────────────────────


async def _create_group(
    db: AsyncSession,
    *,
    email: str,
    name: str | None = None,
    description: str | None = None,
    owners: list[IdentityPerson] | None = None,
    created_at: datetime | None = None,
) -> EmailGroup:
    g = EmailGroup(
        id=uuid.uuid4(),
        email=email,
        name=name or email.split("@", 1)[0],
        description=description,
        status="active",
    )
    if created_at is not None:
        g.created_at = created_at
    db.add(g)
    await db.flush()
    for p in owners or []:
        db.add(EmailGroupOwner(
            id=uuid.uuid4(),
            email_group_id=g.id,
            person_id=p.id,
        ))
    await db.commit()
    return g


async def _make_person(
    db: AsyncSession,
    *,
    email: str,
    name: str,
) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=name,
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(p)
    await db.commit()
    return p


class _StubDriver:
    """Replaces the real GoogleWorkspaceDriver for router tests.

    Every op returns ``ok=True`` so the service layer lands its local
    writes. Tests that want to exercise Google-failure branches should
    override individual methods directly.
    """

    def __init__(self) -> None:
        self.calls: list[tuple[str, dict]] = []

    async def _ok(self, **kwargs):
        from app.integrations.drivers.google_workspace import WorkspaceOpResult
        return WorkspaceOpResult(ok=True, data={"already_absent": False})

    async def delete_group(self, **kwargs):
        self.calls.append(("delete_group", kwargs))
        return await self._ok(**kwargs)

    async def set_group_owner(self, **kwargs):
        self.calls.append(("set_group_owner", kwargs))
        return await self._ok(**kwargs)

    async def remove_group_owner(self, **kwargs):
        self.calls.append(("remove_group_owner", kwargs))
        return await self._ok(**kwargs)

    async def create_group(self, **kwargs):
        self.calls.append(("create_group", kwargs))
        from app.integrations.drivers.google_workspace import WorkspaceOpResult
        return WorkspaceOpResult(
            ok=True,
            data={"email": kwargs.get("email"), "id": "g-stub-id"},
        )


def _patch_driver():
    """Context manager replacing EmailGroupService._get_driver with a stub.

    Returns the stub so tests can introspect the recorded calls.
    """
    stub = _StubDriver()

    async def _fake_get_driver(self):
        return stub

    ctx = patch(
        "app.domains.email_group.service.EmailGroupService._get_driver",
        new=_fake_get_driver,
    )
    return ctx, stub


# ── List filters ───────────────────────────────────────────────


class TestListFilters:
    @pytest.mark.asyncio
    async def test_suffix_filter_matches_only_exact_domain(
        self,
        client: AsyncClient,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        await _create_group(db_session, email="a@pingcap.com")
        await _create_group(db_session, email="b@pingcap.com")
        await _create_group(db_session, email="c@pingcap.com.cn")
        await _create_group(db_session, email="d@tidb.io")

        resp = await client.get(
            "/api/v1/email-groups",
            params={"suffix": "pingcap.com"},
            headers=employee_headers,
        )
        assert resp.status_code == 200
        emails = sorted(g["email"] for g in resp.json()["data"])
        # pingcap.com matches ONLY @pingcap.com — not the .cn subdomain,
        # not tidb.io. This is the whole point of the rhs-anchored LIKE.
        assert emails == ["a@pingcap.com", "b@pingcap.com"]

    @pytest.mark.asyncio
    async def test_owner_person_id_filter(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        emp, _ = employee_user
        stranger = await _make_person(
            db_session, email="stranger@test.com", name="Stranger",
        )
        await _create_group(db_session, email="mine@pingcap.com", owners=[emp])
        await _create_group(
            db_session, email="theirs@pingcap.com", owners=[stranger],
        )
        await _create_group(db_session, email="orphan@pingcap.com")

        resp = await client.get(
            "/api/v1/email-groups",
            params={"owner_person_id": str(emp.id)},
            headers=employee_headers,
        )
        assert resp.status_code == 200
        emails = [g["email"] for g in resp.json()["data"]]
        assert emails == ["mine@pingcap.com"]

    @pytest.mark.asyncio
    async def test_only_mine_resolves_current_user(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        emp, _ = employee_user
        stranger = await _make_person(
            db_session, email="stranger2@test.com", name="Stranger 2",
        )
        await _create_group(db_session, email="mine@pingcap.com", owners=[emp])
        await _create_group(
            db_session, email="theirs@pingcap.com", owners=[stranger],
        )

        resp = await client.get(
            "/api/v1/email-groups",
            params={"only_mine": "true"},
            headers=employee_headers,
        )
        assert resp.status_code == 200
        emails = [g["email"] for g in resp.json()["data"]]
        assert emails == ["mine@pingcap.com"]

    @pytest.mark.asyncio
    async def test_only_mine_overrides_owner_person_id(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        emp, _ = employee_user
        stranger = await _make_person(
            db_session, email="stranger3@test.com", name="Stranger 3",
        )
        await _create_group(db_session, email="mine@pingcap.com", owners=[emp])
        await _create_group(
            db_session, email="theirs@pingcap.com", owners=[stranger],
        )

        resp = await client.get(
            "/api/v1/email-groups",
            params={
                "only_mine": "true",
                # Intentionally pass somebody else's id — only_mine wins.
                "owner_person_id": str(stranger.id),
            },
            headers=employee_headers,
        )
        assert resp.status_code == 200
        emails = [g["email"] for g in resp.json()["data"]]
        assert emails == ["mine@pingcap.com"]

    @pytest.mark.asyncio
    async def test_only_unowned_returns_orphan_groups(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        emp, _ = employee_user
        await _create_group(db_session, email="mine@pingcap.com", owners=[emp])
        await _create_group(db_session, email="orphan-a@pingcap.com")
        await _create_group(db_session, email="orphan-b@pingcap.com")

        resp = await client.get(
            "/api/v1/email-groups",
            params={"only_unowned": "true"},
            headers=employee_headers,
        )
        assert resp.status_code == 200
        emails = sorted(g["email"] for g in resp.json()["data"])
        assert emails == ["orphan-a@pingcap.com", "orphan-b@pingcap.com"]

    @pytest.mark.asyncio
    async def test_created_at_range_is_inclusive(
        self,
        client: AsyncClient,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        base = datetime(2026, 4, 15, tzinfo=UTC)
        await _create_group(
            db_session, email="early@pingcap.com",
            created_at=base - timedelta(days=5),
        )
        in_range = await _create_group(
            db_session, email="midday@pingcap.com",
            created_at=base + timedelta(hours=12),
        )
        await _create_group(
            db_session, email="late@pingcap.com",
            created_at=base + timedelta(days=5),
        )

        resp = await client.get(
            "/api/v1/email-groups",
            params={
                "created_from": "2026-04-15",
                "created_to": "2026-04-15",
            },
            headers=employee_headers,
        )
        assert resp.status_code == 200
        emails = [g["email"] for g in resp.json()["data"]]
        assert emails == [in_range.email]


# ── Owners-directory (search-helper for the owner-filter picker) ──


class TestOwnersDirectory:
    """``GET /email-groups/owners-directory`` returns DISTINCT active
    owners across all mail groups, with name + email. Same data the
    ``/email-groups`` listing already exposes via
    ``EmailGroupOwnerResponse.work_email``, just aggregated for picker
    search.
    """

    @pytest.mark.asyncio
    async def test_distinct_collapses_multi_group_owner(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        """A person owning N groups appears exactly once."""
        emp, _ = employee_user
        # Same person owns three groups.
        await _create_group(db_session, email="g1@pingcap.com", owners=[emp])
        await _create_group(db_session, email="g2@pingcap.com", owners=[emp])
        await _create_group(db_session, email="g3@pingcap.com", owners=[emp])

        resp = await client.get(
            "/api/v1/email-groups/owners-directory",
            headers=employee_headers,
        )
        assert resp.status_code == 200
        data = resp.json()["data"]
        ids = [row["id"] for row in data]
        assert ids.count(str(emp.id)) == 1, (
            f"DISTINCT failed; emp.id appeared {ids.count(str(emp.id))} times"
        )

    @pytest.mark.asyncio
    async def test_returns_email_for_picker_search(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        """Each row has display_name + work_email so the picker label
        ``${display_name} · ${work_email}`` resolves and the
        client-side ``filterOption`` matches an email-typed query.
        """
        emp, _ = employee_user
        await _create_group(db_session, email="team@pingcap.com", owners=[emp])

        resp = await client.get(
            "/api/v1/email-groups/owners-directory",
            headers=employee_headers,
        )
        assert resp.status_code == 200
        rows = resp.json()["data"]
        match = next((r for r in rows if r["id"] == str(emp.id)), None)
        assert match is not None
        assert match["display_name"] == emp.display_name
        assert match["work_email"] == emp.work_email

    @pytest.mark.asyncio
    async def test_excludes_leaver_owners(
        self,
        client: AsyncClient,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        """Active filter — a person who left while still listed as an
        owner upstream shouldn't show up in the picker (the picker is
        for actionable filtering; a leaver's groups surface elsewhere
        as the 'orphan owner' / unowned governance lane).

        We create the leaver directly in the ``left`` state rather than
        mutating from active → left so the ``LeaverBlocked`` SQLAlchemy
        listener doesn't intercept (its job is to refuse the flip path,
        not to police existing leaver rows).
        """
        leaver = IdentityPerson(
            id=uuid.uuid4(),
            work_email="leaver@pingcap.com",
            display_name="Past Person",
            employment_status="left",
            source_system="successfactors",
            hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        )
        db_session.add(leaver)
        await db_session.commit()
        await _create_group(
            db_session, email="legacy@pingcap.com", owners=[leaver],
        )

        resp = await client.get(
            "/api/v1/email-groups/owners-directory",
            headers=employee_headers,
        )
        assert resp.status_code == 200
        ids = {r["id"] for r in resp.json()["data"]}
        assert str(leaver.id) not in ids

    @pytest.mark.asyncio
    async def test_empty_when_no_groups_have_owners(
        self,
        client: AsyncClient,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        """Empty response is well-formed; picker degrades to no
        options without crashing the page."""
        await _create_group(db_session, email="orphan@pingcap.com")

        resp = await client.get(
            "/api/v1/email-groups/owners-directory",
            headers=employee_headers,
        )
        assert resp.status_code == 200
        assert resp.json()["data"] == []

    @pytest.mark.asyncio
    async def test_requires_auth(self, client: AsyncClient) -> None:
        """Unauthenticated callers get 401 — same gate as the rest of
        the public directory.
        """
        resp = await client.get("/api/v1/email-groups/owners-directory")
        assert resp.status_code == 401


# ── Transfer-owner endpoint ────────────────────────────────────


class TestTransferOwner:
    @pytest.mark.asyncio
    async def test_current_owner_can_transfer(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        emp, _ = employee_user
        new_owner = await _make_person(
            db_session, email="new-owner@pingcap.com", name="New Owner",
        )
        group = await _create_group(
            db_session, email="team@pingcap.com", owners=[emp],
        )
        # Snapshot identifiers before crossing the request boundary —
        # after the endpoint commits, the instance ``group`` is expired
        # and accessing ``group.id`` would trigger a reload which the
        # test's session can't always satisfy cleanly.
        group_id = group.id
        new_owner_id = new_owner.id
        emp_id = emp.id
        ctx, stub = _patch_driver()
        with ctx:
            resp = await client.post(
                f"/api/v1/email-groups/{group_id}/transfer-owner",
                json={
                    "from_person_id": str(emp_id),
                    "to_person_id": str(new_owner_id),
                    "reason": "Handing off to platform team",
                },
                headers=employee_headers,
            )
        assert resp.status_code == 200, resp.text
        call_names = [c[0] for c in stub.calls]
        # Google ADD must come BEFORE REMOVE — if the order flipped,
        # a single-owner group would temporarily have zero owners and
        # the remove would fail.
        assert call_names == ["set_group_owner", "remove_group_owner"]
        # Local owners table reflects the swap.
        owners = (
            await db_session.execute(
                select(EmailGroupOwner.person_id).where(
                    EmailGroupOwner.email_group_id == group_id,
                )
            )
        ).scalars().all()
        assert set(owners) == {new_owner_id}

    @pytest.mark.asyncio
    async def test_admin_can_transfer_on_any_group(
        self,
        client: AsyncClient,
        admin_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        from_person = await _make_person(
            db_session, email="from@pingcap.com", name="From Person",
        )
        to_person = await _make_person(
            db_session, email="to@pingcap.com", name="To Person",
        )
        group = await _create_group(
            db_session, email="admin-txfer@pingcap.com",
            owners=[from_person],
        )
        ctx, _ = _patch_driver()
        with ctx:
            resp = await client.post(
                f"/api/v1/email-groups/{group.id}/transfer-owner",
                json={
                    "from_person_id": str(from_person.id),
                    "to_person_id": str(to_person.id),
                    "reason": "Admin handoff during leave coverage",
                },
                headers=admin_headers,
            )
        assert resp.status_code == 200, resp.text

    @pytest.mark.asyncio
    async def test_non_owner_is_forbidden(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        emp, _ = employee_user  # NOT an owner of this group
        owner_person = await _make_person(
            db_session, email="owner-x@pingcap.com", name="Owner X",
        )
        to_person = await _make_person(
            db_session, email="to-x@pingcap.com", name="To X",
        )
        group = await _create_group(
            db_session, email="foreign@pingcap.com", owners=[owner_person],
        )
        ctx, _ = _patch_driver()
        with ctx:
            resp = await client.post(
                f"/api/v1/email-groups/{group.id}/transfer-owner",
                json={
                    "from_person_id": str(owner_person.id),
                    "to_person_id": str(to_person.id),
                    "reason": "Trying to hijack someone else's group",
                },
                headers=employee_headers,
            )
        assert resp.status_code == 403

    @pytest.mark.asyncio
    async def test_rejects_same_from_and_to(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        emp, _ = employee_user
        group = await _create_group(
            db_session, email="self@pingcap.com", owners=[emp],
        )
        ctx, _ = _patch_driver()
        with ctx:
            resp = await client.post(
                f"/api/v1/email-groups/{group.id}/transfer-owner",
                json={
                    "from_person_id": str(emp.id),
                    "to_person_id": str(emp.id),
                    "reason": "Trying to transfer to myself",
                },
                headers=employee_headers,
            )
        assert resp.status_code == 400

    @pytest.mark.asyncio
    async def test_rejects_from_not_actually_owner(
        self,
        client: AsyncClient,
        admin_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        # Only an admin may pass a ``from_person_id`` other than their own
        # (a plain owner is restricted to transferring their OWN ownership
        # — see test_owner_cannot_evict_coowner). So the FROM_NOT_OWNER
        # branch is now only reachable by an admin naming a bogus
        # from_person_id; exercise it that way.
        owner = await _make_person(
            db_session, email="realowner4@pingcap.com", name="Real Owner 4",
        )
        stranger = await _make_person(
            db_session, email="stranger4@test.com", name="Stranger 4",
        )
        new_owner = await _make_person(
            db_session, email="new4@pingcap.com", name="New 4",
        )
        group = await _create_group(
            db_session, email="mine4@pingcap.com", owners=[owner],
        )
        ctx, _ = _patch_driver()
        with ctx:
            resp = await client.post(
                f"/api/v1/email-groups/{group.id}/transfer-owner",
                json={
                    # from_person_id isn't actually an owner of this group
                    "from_person_id": str(stranger.id),
                    "to_person_id": str(new_owner.id),
                    "reason": "Bogus from_person_id",
                },
                headers=admin_headers,
            )
        assert resp.status_code == 400

    @pytest.mark.asyncio
    async def test_owner_cannot_evict_coowner(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        # A plain co-owner may only transfer away their OWN ownership.
        # Passing a co-owner's id as ``from_person_id`` (lateral eviction)
        # must be forbidden — only HR / IT admin can transfer on behalf of
        # another owner.
        emp, _ = employee_user
        coowner = await _make_person(
            db_session, email="coowner5@pingcap.com", name="Co Owner 5",
        )
        new_owner = await _make_person(
            db_session, email="new5@pingcap.com", name="New 5",
        )
        group = await _create_group(
            db_session, email="shared5@pingcap.com", owners=[emp, coowner],
        )
        ctx, _ = _patch_driver()
        with ctx:
            resp = await client.post(
                f"/api/v1/email-groups/{group.id}/transfer-owner",
                json={
                    # emp (a co-owner) tries to evict co-owner `coowner`
                    "from_person_id": str(coowner.id),
                    "to_person_id": str(new_owner.id),
                    "reason": "Trying to evict a co-owner",
                },
                headers=employee_headers,
            )
        assert resp.status_code == 403

    @pytest.mark.asyncio
    async def test_short_reason_rejected(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
        db_session: AsyncSession,
    ) -> None:
        emp, _ = employee_user
        new_owner = await _make_person(
            db_session, email="new5@pingcap.com", name="New 5",
        )
        group = await _create_group(
            db_session, email="short5@pingcap.com", owners=[emp],
        )
        resp = await client.post(
            f"/api/v1/email-groups/{group.id}/transfer-owner",
            json={
                "from_person_id": str(emp.id),
                "to_person_id": str(new_owner.id),
                "reason": "x",  # < 4 chars
            },
            headers=employee_headers,
        )
        assert resp.status_code == 422


class TestCreateMailGroupPropagationRetry:
    """Google Directory has eventual consistency on freshly-created
    groups: members().insert() right after groups().insert() can return
    404 "Resource Not Found: groupKey" for a few seconds. The service
    must retry that 404 with backoff before declaring the group a
    failure — otherwise the create looks like it succeeded on Google
    but Helix rolls it back via delete_group, leaving the requester with
    a confusing error like the one observed for Flare Zuo.
    """

    @pytest.mark.asyncio
    async def test_set_group_owner_404_then_ok_succeeds(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
    ) -> None:
        from app.integrations.drivers.google_workspace import WorkspaceOpResult

        emp, _ = employee_user
        ctx, stub = _patch_driver()
        attempts = {"n": 0}

        async def flaky_set_group_owner(**kwargs):
            stub.calls.append(("set_group_owner", kwargs))
            attempts["n"] += 1
            if attempts["n"] == 1:
                return WorkspaceOpResult(
                    ok=False,
                    error_message="Google API error 404: Resource Not Found: groupKey",
                    status=404,
                )
            return WorkspaceOpResult(ok=True, data={"member_id": "m1", "role": "OWNER"})

        stub.set_group_owner = flaky_set_group_owner  # type: ignore[method-assign]

        # Don't actually sleep in the test — we still want to assert the
        # retry happened, just not pay 0.5s wall-clock per attempt.
        with ctx, patch(
            "app.domains.email_group.service.asyncio.sleep",
            new=lambda _delay: _noop(),
        ):
            resp = await client.post(
                "/api/v1/email-groups/requests",
                json={
                    "local_part": "propagation-test",
                    "suffix": "pingcap.com",
                    "name": "Propagation Test",
                    "reason": "Verify Google propagation retry",
                },
                headers=employee_headers,
            )
        assert resp.status_code == 200, resp.text
        # Two set_group_owner attempts: 1st 404, 2nd OK
        owner_calls = [c for c in stub.calls if c[0] == "set_group_owner"]
        assert len(owner_calls) == 2
        # delete_group must NOT have been called — create did succeed
        assert not any(c[0] == "delete_group" for c in stub.calls)

    @pytest.mark.asyncio
    async def test_persistent_404_rolls_back(
        self,
        client: AsyncClient,
        employee_user,
        employee_headers: dict,
    ) -> None:
        from app.integrations.drivers.google_workspace import WorkspaceOpResult

        emp, _ = employee_user
        ctx, stub = _patch_driver()

        async def always_404(**kwargs):
            stub.calls.append(("set_group_owner", kwargs))
            return WorkspaceOpResult(
                ok=False,
                error_message="Google API error 404: Resource Not Found: groupKey",
                status=404,
            )

        stub.set_group_owner = always_404  # type: ignore[method-assign]

        with ctx, patch(
            "app.domains.email_group.service.asyncio.sleep",
            new=lambda _delay: _noop(),
        ):
            resp = await client.post(
                "/api/v1/email-groups/requests",
                json={
                    "local_part": "persistent-404",
                    "suffix": "pingcap.com",
                    "name": "Persistent 404 Test",
                    "reason": "Verify rollback when retries exhausted",
                },
                headers=employee_headers,
            )
        # 502 with our error envelope
        assert resp.status_code == 502, resp.text
        assert "未能给任何申请 owner 成功授权" in resp.text
        # All 5 attempts (0.0, 0.5, 1.0, 2.0, 4.0 backoff slots)
        owner_calls = [c for c in stub.calls if c[0] == "set_group_owner"]
        assert len(owner_calls) == 5
        # Rollback ran
        assert any(c[0] == "delete_group" for c in stub.calls)


async def _noop() -> None:
    """asyncio.sleep replacement for tests — returns immediately."""
    return None