"""Mail-group 申请加入 → in-Helix owner approval flow.

The Google Workspace driver is patched: tests assert RBAC + DB
side-effects (join-request row lifecycle, owner-queue scoping, member
removal guards) without hitting Google. Behaviours covered:

  1. An employee requests to join a group → a ``pending``
     ``email_group_join_requests`` row is created and Google is NOT
     touched yet (nothing added until an owner approves).
  2. A second pending request for the same (group, requester) is
     refused (409) — the ``open_request_key`` single-pending guard.
  3. The group's owner approves → Google ``add_group_member`` is called
     once and the row flips to ``approved``.
  4. A non-owner / non-admin cannot decide (403).
  5. The owner-scoped 待审批 queue lists the request for the owner and
     excludes the requester's own view.
  6. Member removal refuses an OWNER (use transfer) but removes a plain
     MEMBER.
"""
from __future__ import annotations

import uuid
from unittest.mock import AsyncMock, patch

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

from app.core.security import create_access_token
from app.domains.email_group.models import (
    EmailGroup,
    EmailGroupJoinRequest,
    EmailGroupOwner,
)
from app.identity.person_models import IdentityPerson


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.flush()
    return p


async def _make_group(
    db: AsyncSession, *, email: str, owners: list[IdentityPerson],
) -> EmailGroup:
    g = EmailGroup(
        id=uuid.uuid4(), email=email, name=email.split("@", 1)[0], status="active",
    )
    db.add(g)
    await db.flush()
    for p in owners:
        db.add(EmailGroupOwner(
            id=uuid.uuid4(), email_group_id=g.id, person_id=p.id,
        ))
    await db.flush()
    return g


def _ok_result(**data):
    from app.integrations.drivers.google_workspace import WorkspaceOpResult
    return WorkspaceOpResult(ok=True, data=data)


@pytest.fixture
def patch_driver():
    fake = AsyncMock()
    # Requester is not yet a member of anything (join pre-check passes).
    fake.list_groups_for_user.return_value = []
    fake.add_group_member.return_value = _ok_result(member_id="m1", role="MEMBER")
    fake.remove_group_member.return_value = _ok_result(already_absent=False)
    with patch(
        "app.domains.email_group.service.EmailGroupService._get_driver",
        return_value=fake,
    ):
        yield fake


@pytest.mark.asyncio
async def test_request_join_creates_pending_without_google(
    client: AsyncClient, db_session, employee_user, patch_driver,
):
    user, token = employee_user
    me = await db_session.get(IdentityPerson, user.id)
    me.work_email = "shuai.feng@pingcap.com"
    owner = await _make_person(db_session, email="owner@pingcap.com", name="Owner")
    group = await _make_group(db_session, email="eng@pingcap.com", owners=[owner])
    await db_session.commit()

    r = await client.post(
        f"/api/v1/email-groups/{group.id}/join-requests",
        headers={"Authorization": f"Bearer {token}"},
        json={"reason": "need eng list"},
    )
    assert r.status_code == 200, r.text
    assert r.json()["data"]["status"] == "pending"
    # Nothing added to Google until an owner approves.
    assert patch_driver.add_group_member.await_count == 0

    rows = (
        await db_session.execute(
            select(EmailGroupJoinRequest).where(
                EmailGroupJoinRequest.email_group_id == group.id,
            )
        )
    ).scalars().all()
    assert len(rows) == 1
    assert rows[0].status == "pending"
    assert rows[0].requester_id == user.id


@pytest.mark.asyncio
async def test_duplicate_pending_rejected(
    client: AsyncClient, db_session, employee_user, patch_driver,
):
    user, token = employee_user
    me = await db_session.get(IdentityPerson, user.id)
    me.work_email = "shuai.feng@pingcap.com"
    owner = await _make_person(db_session, email="owner2@pingcap.com", name="Owner2")
    group = await _make_group(db_session, email="ops@pingcap.com", owners=[owner])
    await db_session.commit()

    hdr = {"Authorization": f"Bearer {token}"}
    first = await client.post(
        f"/api/v1/email-groups/{group.id}/join-requests", headers=hdr, json={},
    )
    assert first.status_code == 200, first.text
    dup = await client.post(
        f"/api/v1/email-groups/{group.id}/join-requests", headers=hdr, json={},
    )
    assert dup.status_code == 409, dup.text
    assert dup.json()["errors"][0]["code"] == "JOIN_REQUEST_PENDING"


@pytest.mark.asyncio
async def test_owner_approves_adds_member(
    client: AsyncClient, db_session, employee_user, patch_driver,
):
    user, token = employee_user
    me = await db_session.get(IdentityPerson, user.id)
    me.work_email = "shuai.feng@pingcap.com"
    owner = await _make_person(db_session, email="owner3@pingcap.com", name="Owner3")
    group = await _make_group(db_session, email="team@pingcap.com", owners=[owner])
    await db_session.commit()
    owner_token = create_access_token(str(owner.id), ["employee"])

    # Requester submits.
    r = await client.post(
        f"/api/v1/email-groups/{group.id}/join-requests",
        headers={"Authorization": f"Bearer {token}"}, json={"reason": "x"},
    )
    assert r.status_code == 200, r.text
    req_id = r.json()["data"]["request_id"]

    # Owner approves.
    d = await client.post(
        f"/api/v1/email-groups/join-requests/{req_id}/decide",
        headers={"Authorization": f"Bearer {owner_token}"},
        json={"decision": "approve"},
    )
    assert d.status_code == 200, d.text
    assert d.json()["data"]["status"] == "approved"
    assert patch_driver.add_group_member.await_count == 1

    row = await db_session.get(EmailGroupJoinRequest, uuid.UUID(req_id))
    await db_session.refresh(row)
    assert row.status == "approved"
    assert row.decided_by_id == owner.id


@pytest.mark.asyncio
async def test_non_owner_cannot_decide(
    client: AsyncClient, db_session, employee_user, patch_driver,
):
    user, token = employee_user
    me = await db_session.get(IdentityPerson, user.id)
    me.work_email = "shuai.feng@pingcap.com"
    owner = await _make_person(db_session, email="owner4@pingcap.com", name="Owner4")
    stranger = await _make_person(db_session, email="stranger@pingcap.com", name="S")
    group = await _make_group(db_session, email="sec@pingcap.com", owners=[owner])
    await db_session.commit()
    stranger_token = create_access_token(str(stranger.id), ["employee"])

    r = await client.post(
        f"/api/v1/email-groups/{group.id}/join-requests",
        headers={"Authorization": f"Bearer {token}"}, json={},
    )
    req_id = r.json()["data"]["request_id"]

    d = await client.post(
        f"/api/v1/email-groups/join-requests/{req_id}/decide",
        headers={"Authorization": f"Bearer {stranger_token}"},
        json={"decision": "approve"},
    )
    assert d.status_code == 403, d.text
    assert patch_driver.add_group_member.await_count == 0


@pytest.mark.asyncio
async def test_pending_queue_owner_scoped(
    client: AsyncClient, db_session, employee_user, patch_driver,
):
    user, token = employee_user
    me = await db_session.get(IdentityPerson, user.id)
    me.work_email = "shuai.feng@pingcap.com"
    owner = await _make_person(db_session, email="owner5@pingcap.com", name="Owner5")
    group = await _make_group(db_session, email="list@pingcap.com", owners=[owner])
    await db_session.commit()
    owner_token = create_access_token(str(owner.id), ["employee"])

    await client.post(
        f"/api/v1/email-groups/{group.id}/join-requests",
        headers={"Authorization": f"Bearer {token}"}, json={},
    )

    # Owner sees it.
    owner_view = await client.get(
        "/api/v1/email-groups/join-requests/pending",
        headers={"Authorization": f"Bearer {owner_token}"},
    )
    assert owner_view.status_code == 200, owner_view.text
    assert len(owner_view.json()["data"]) == 1

    # Requester (not an owner of anything) sees an empty queue.
    self_view = await client.get(
        "/api/v1/email-groups/join-requests/pending",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert self_view.status_code == 200, self_view.text
    assert self_view.json()["data"] == []


@pytest.mark.asyncio
async def test_no_owner_group_routes_to_admin(
    client: AsyncClient, db_session, employee_user, admin_user, patch_driver,
):
    """A group with NO owner: the request is accepted (has_owner=False),
    nobody is pushed to, but an admin sees it in the pending queue and can
    approve it — so it doesn't rot."""
    user, token = employee_user
    me = await db_session.get(IdentityPerson, user.id)
    me.work_email = "shuai.feng@pingcap.com"
    _admin, admin_token = admin_user
    group = await _make_group(db_session, email="orphan@pingcap.com", owners=[])
    await db_session.commit()

    r = await client.post(
        f"/api/v1/email-groups/{group.id}/join-requests",
        headers={"Authorization": f"Bearer {token}"}, json={"reason": "x"},
    )
    assert r.status_code == 200, r.text
    assert r.json()["data"]["has_owner"] is False
    req_id = r.json()["data"]["request_id"]

    # Admin sees the orphaned-group request in the org-wide pending queue.
    q = await client.get(
        "/api/v1/email-groups/join-requests/pending",
        headers={"Authorization": f"Bearer {admin_token}"},
    )
    assert q.status_code == 200, q.text
    assert any(row["id"] == req_id for row in q.json()["data"])

    # Admin can approve it.
    d = await client.post(
        f"/api/v1/email-groups/join-requests/{req_id}/decide",
        headers={"Authorization": f"Bearer {admin_token}"},
        json={"decision": "approve"},
    )
    assert d.status_code == 200, d.text
    assert patch_driver.add_group_member.await_count == 1


@pytest.mark.asyncio
async def test_requester_can_cancel_pending(
    client: AsyncClient, db_session, employee_user, patch_driver,
):
    """The requester can withdraw their own pending request; a second
    cancel (already terminal) is refused."""
    user, token = employee_user
    me = await db_session.get(IdentityPerson, user.id)
    me.work_email = "shuai.feng@pingcap.com"
    owner = await _make_person(db_session, email="owner6@pingcap.com", name="Owner6")
    group = await _make_group(db_session, email="cancelme@pingcap.com", owners=[owner])
    await db_session.commit()

    r = await client.post(
        f"/api/v1/email-groups/{group.id}/join-requests",
        headers={"Authorization": f"Bearer {token}"}, json={},
    )
    req_id = r.json()["data"]["request_id"]

    c = await client.post(
        f"/api/v1/email-groups/join-requests/{req_id}/cancel",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert c.status_code == 200, c.text
    assert c.json()["data"]["status"] == "cancelled"

    row = await db_session.get(EmailGroupJoinRequest, uuid.UUID(req_id))
    await db_session.refresh(row)
    assert row.status == "cancelled"

    # Re-cancel a terminal request → 409.
    again = await client.post(
        f"/api/v1/email-groups/join-requests/{req_id}/cancel",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert again.status_code == 409, again.text


@pytest.mark.asyncio
async def test_remove_member_refuses_owner_allows_member(
    client: AsyncClient, db_session, employee_user, patch_driver,
):
    user, token = employee_user
    me = await db_session.get(IdentityPerson, user.id)
    me.work_email = "shuai.feng@pingcap.com"
    # The caller owns the group.
    group = await _make_group(db_session, email="mine@pingcap.com", owners=[me])
    await db_session.commit()

    patch_driver.list_group_members.return_value = [
        {"email": "shuai.feng@pingcap.com", "role": "OWNER"},
        {"email": "member@pingcap.com", "role": "MEMBER"},
    ]

    hdr = {"Authorization": f"Bearer {token}"}
    # Refuse removing an OWNER.
    bad = await client.post(
        f"/api/v1/email-groups/{group.id}/members/remove",
        headers=hdr, json={"member_email": "shuai.feng@pingcap.com"},
    )
    assert bad.status_code == 409, bad.text
    assert bad.json()["errors"][0]["code"] == "CANNOT_REMOVE_OWNER"
    assert patch_driver.remove_group_member.await_count == 0

    # Remove a plain MEMBER.
    ok = await client.post(
        f"/api/v1/email-groups/{group.id}/members/remove",
        headers=hdr, json={"member_email": "member@pingcap.com"},
    )
    assert ok.status_code == 200, ok.text
    assert patch_driver.remove_group_member.await_count == 1