"""Unit tests for the Google Workspace workspace-ops methods.

These test the PR-D additions (``create_group``, ``delete_group``,
``add_alias``, ``remove_alias``, ``reset_user_password``,
``set_group_owner``, ``remove_group_owner``). Google's Admin SDK
client is NOT httpx — it wraps ``googleapiclient.discovery.build``
and returns a deeply-nested fluent builder. respx can't intercept it,
so we monkeypatch ``_build_directory_service`` to return a stub that
records the call chain and emits either a success body or a crafted
``HttpError`` failure.

The stub implements just enough of the Google client's fluent API
for our driver methods:

    svc.groups().insert(body=...).execute()
    svc.members().insert(groupKey=..., body=...).execute()
    svc.members().patch(groupKey=..., memberKey=..., body=...).execute()
    svc.members().delete(groupKey=..., memberKey=...).execute()
    svc.users().update(userKey=..., body=...).execute()
    svc.users().aliases().insert(userKey=..., body=...).execute()
    svc.users().aliases().delete(userKey=..., alias=...).execute()
    svc.groups().aliases().insert(groupKey=..., body=...).execute()
    svc.groups().aliases().delete(groupKey=..., alias=...).execute()

Every builder method just records its args on a shared log and
returns a leaf object whose ``.execute()`` returns the preconfigured
response for that call path.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

import httplib2
import pytest
from googleapiclient.errors import HttpError

from app.integrations.drivers import google_workspace as gw
from app.integrations.drivers.google_workspace import (
    GoogleWorkspaceDriver,
)

_CONFIG = {
    "service_account_json": "{\"client_email\":\"sa@example.iam.gserviceaccount.com\"}",
    "impersonate_subject_email": "admin@example.com",
}


def _driver() -> GoogleWorkspaceDriver:
    return GoogleWorkspaceDriver(integration_id="gw-test", config=_CONFIG)


# ─── Stub Google Admin SDK client ──────────────────────────────────


@dataclass
class _CallLog:
    chain: list[tuple[str, tuple, dict]] = field(default_factory=list)


class _Leaf:
    """Terminal fluent node; ``.execute()`` returns a preset response
    or raises HttpError."""

    def __init__(self, response: Any | Exception):
        self._response = response

    def execute(self) -> Any:
        if isinstance(self._response, Exception):
            raise self._response
        return self._response


class _FluentNode:
    """Catch-all recording fluent node.

    A call like ``node.members().insert(...)`` records ``members`` +
    ``insert`` into the log and returns another node. Leaf responses
    are matched against the dotted path built from the chain.
    """

    def __init__(self, log: _CallLog, path: tuple[str, ...], routes: dict[str, Any]):
        self._log = log
        self._path = path
        self._routes = routes

    def __getattr__(self, name: str) -> Any:
        def _call(*args: Any, **kwargs: Any) -> Any:
            self._log.chain.append((name, args, kwargs))
            new_path = self._path + (name,)
            key = ".".join(new_path)
            if key in self._routes:
                return _Leaf(self._routes[key])
            return _FluentNode(self._log, new_path, self._routes)

        return _call


def _make_http_error(status: int, reason: str = "err") -> HttpError:
    resp = httplib2.Response({"status": status, "reason": reason})
    resp.reason = reason
    return HttpError(resp, b"{\"error\": {\"message\": \"boom\"}}")


@pytest.fixture
def patch_service(monkeypatch):
    """Yield a helper that installs a stub ``_build_directory_service``
    returning a fluent node with the given leaf routes.

    Usage::

        with patch_service({"groups.insert": {"email": "x@y"}}):
            await driver.create_group(...)
    """
    routes: dict[str, Any] = {}
    log = _CallLog()

    def _set_routes(new_routes: dict[str, Any]) -> _CallLog:
        routes.clear()
        routes.update(new_routes)
        log.chain.clear()
        return log

    def _fake_build(sa_json: str, subject: str, scopes: list[str]):
        return _FluentNode(log, (), routes)

    monkeypatch.setattr(gw, "_build_directory_service", _fake_build)
    return _set_routes


# ─── create_group ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_create_group_happy_path(patch_service) -> None:
    patch_service({
        "groups.insert": {"id": "grp-1", "email": "team@pingcap.com"},
    })
    result = await _driver().create_group(
        email="team@pingcap.com", name="Team", description="x",
    )
    assert result.ok is True
    assert result.data["email"] == "team@pingcap.com"
    assert result.data["id"] == "grp-1"
    assert result.already_exists is False


@pytest.mark.asyncio
async def test_create_group_already_exists_is_idempotent(patch_service) -> None:
    patch_service({
        "groups.insert": _make_http_error(409, "conflict"),
    })
    result = await _driver().create_group(
        email="team@pingcap.com", name="Team", description=None,
    )
    assert result.ok is True
    assert result.already_exists is True
    assert result.status == 409


@pytest.mark.asyncio
async def test_create_group_other_error_surfaces(patch_service) -> None:
    patch_service({
        "groups.insert": _make_http_error(500, "oops"),
    })
    result = await _driver().create_group(
        email="team@pingcap.com", name="Team", description=None,
    )
    assert result.ok is False
    assert "500" in (result.error_message or "")


# ─── delete_group ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_delete_group_happy_path(patch_service) -> None:
    patch_service({"groups.delete": None})
    result = await _driver().delete_group(email="team@pingcap.com")
    assert result.ok is True


@pytest.mark.asyncio
async def test_delete_group_404_is_idempotent(patch_service) -> None:
    patch_service({"groups.delete": _make_http_error(404, "not found")})
    result = await _driver().delete_group(email="team@pingcap.com")
    assert result.ok is True
    assert result.data.get("already_absent") is True


# ─── add_alias / remove_alias ──────────────────────────────────────


@pytest.mark.asyncio
async def test_add_alias_user_happy_path(patch_service) -> None:
    patch_service({
        "users.aliases.insert": {"alias": "dev@pingcap.com"},
    })
    result = await _driver().add_alias(
        target_email="ada@pingcap.com",
        alias_email="dev@pingcap.com",
        target_kind="user",
    )
    assert result.ok is True
    assert result.data["alias"] == "dev@pingcap.com"


@pytest.mark.asyncio
async def test_add_alias_group_happy_path(patch_service) -> None:
    patch_service({
        "groups.aliases.insert": {"alias": "engineers@pingcap.com"},
    })
    result = await _driver().add_alias(
        target_email="team@pingcap.com",
        alias_email="engineers@pingcap.com",
        target_kind="group",
    )
    assert result.ok is True


@pytest.mark.asyncio
async def test_add_alias_invalid_kind() -> None:
    result = await _driver().add_alias(
        target_email="a@b.com", alias_email="c@b.com", target_kind="other",
    )
    assert result.ok is False
    assert "unknown target_kind" in (result.error_message or "")


@pytest.mark.asyncio
async def test_add_alias_conflict_is_idempotent(patch_service) -> None:
    patch_service({
        "users.aliases.insert": _make_http_error(409, "exists"),
    })
    result = await _driver().add_alias(
        target_email="ada@pingcap.com",
        alias_email="dev@pingcap.com",
        target_kind="user",
    )
    assert result.ok is True
    assert result.already_exists is True


@pytest.mark.asyncio
async def test_remove_alias_404_is_idempotent(patch_service) -> None:
    patch_service({
        "users.aliases.delete": _make_http_error(404, "missing"),
    })
    result = await _driver().remove_alias(
        target_email="ada@pingcap.com",
        alias_email="dev@pingcap.com",
        target_kind="user",
    )
    assert result.ok is True


# ─── reset_user_password ───────────────────────────────────────────


@pytest.mark.asyncio
async def test_reset_user_password_requires_password(patch_service) -> None:
    """Round-5 fix: the caller MUST supply new_password. The old
    'generate internally if None' branch minted a password, pushed it to
    Google, then discarded it → silent account lockout (nobody could sign
    in). Refusing a missing password fails closed instead."""
    log = patch_service({
        "users.update": {"primaryEmail": "ada@pingcap.com"},
    })
    # Empty string must fail closed (would otherwise set a blank password).
    result = await _driver().reset_user_password(
        user_email="ada@pingcap.com", new_password="",
    )
    assert result.ok is False
    assert "new_password" in (result.error_message or "")
    # Nothing was pushed to Google — we bailed before the update call.
    assert not any(call[0] == "update" for call in log.chain)


@pytest.mark.asyncio
async def test_reset_user_password_sends_supplied_password(patch_service) -> None:
    log = patch_service({
        "users.update": {"primaryEmail": "ada@pingcap.com"},
    })
    result = await _driver().reset_user_password(
        user_email="ada@pingcap.com", new_password="Sup3r-Str0ng!pw",
    )
    assert result.ok is True
    assert result.data.get("password_reset") is True
    # Plaintext is never echoed back in result.data (no DB-persist leak).
    assert "new_password" not in result.data
    update_call = next(call for call in log.chain if call[0] == "update")
    body = update_call[2].get("body") or {}
    assert body.get("password") == "Sup3r-Str0ng!pw"
    assert "changePasswordAtNextLogin" not in body


@pytest.mark.asyncio
async def test_reset_user_password_http_error(patch_service) -> None:
    patch_service({"users.update": _make_http_error(403, "forbidden")})
    result = await _driver().reset_user_password(
        user_email="ada@pingcap.com", new_password="Sup3r-Str0ng!pw",
    )
    assert result.ok is False
    assert "403" in (result.error_message or "")


# ─── set_group_owner / remove_group_owner ──────────────────────────


@pytest.mark.asyncio
async def test_set_group_owner_new_member(patch_service) -> None:
    patch_service({
        "members.insert": {"id": "m-1", "role": "OWNER"},
    })
    result = await _driver().set_group_owner(
        group_email="team@pingcap.com", member_email="ada@pingcap.com",
    )
    assert result.ok is True
    assert result.data["role"] == "OWNER"


@pytest.mark.asyncio
async def test_set_group_owner_existing_member_promoted(patch_service) -> None:
    # insert raises 409, fallback to patch
    patch_service({
        "members.insert": _make_http_error(409, "exists"),
        "members.patch": {"id": "m-1", "role": "OWNER"},
    })
    result = await _driver().set_group_owner(
        group_email="team@pingcap.com", member_email="ada@pingcap.com",
    )
    assert result.ok is True
    assert result.data["role"] == "OWNER"


@pytest.mark.asyncio
async def test_remove_group_owner_404_idempotent(patch_service) -> None:
    patch_service({"members.delete": _make_http_error(404, "missing")})
    result = await _driver().remove_group_owner(
        group_email="team@pingcap.com", member_email="ada@pingcap.com",
    )
    assert result.ok is True


# ─── Password generator unit ───────────────────────────────────────


def test_generate_strong_password_composition() -> None:
    for _ in range(50):
        pw = GoogleWorkspaceDriver._generate_strong_password(16)
        assert len(pw) == 16
        assert any(c.islower() for c in pw)
        assert any(c.isupper() for c in pw)
        assert any(c.isdigit() for c in pw)
        assert any(c in "!@#%^&*()-_=+[]{};:,.?" for c in pw)


def test_generate_strong_password_enforces_min_length() -> None:
    pw = GoogleWorkspaceDriver._generate_strong_password(8)
    # Implementation clamps shorter requests up to 12 so composition
    # constraints remain satisfiable deterministically.
    assert len(pw) == 12


# ─── revoke_seat: user_suspend + Data Transfer ──────────────────


def test_revoke_requires_successor_only_for_user_suspend() -> None:
    assert GoogleWorkspaceDriver.revoke_requires_successor(
        {"operation": "user_suspend"},
    ) is True
    assert GoogleWorkspaceDriver.revoke_requires_successor(
        {"operation": "group_member"},
    ) is False
    assert GoogleWorkspaceDriver.revoke_requires_successor({}) is False
    assert GoogleWorkspaceDriver.revoke_requires_successor(None) is False


@pytest.fixture
def patch_services(monkeypatch):
    """Dual-fixture variant of :func:`patch_service` that patches BOTH
    ``_build_directory_service`` and ``_build_datatransfer_service`` so
    we can simulate the full Drive/Calendar transfer flow.

    The fluent log is SHARED across both services so assertions can
    inspect the full call sequence in order (users.get → users.get →
    transfers.insert → users.update).
    """
    routes: dict[str, Any] = {}
    log = _CallLog()

    def _set_routes(new_routes: dict[str, Any]) -> _CallLog:
        routes.clear()
        routes.update(new_routes)
        log.chain.clear()
        return log

    def _fake_dir(sa_json: str, subject: str, scopes: list[str]):
        return _FluentNode(log, (), routes)

    def _fake_dt(sa_json: str, subject: str):
        return _FluentNode(log, (), routes)

    monkeypatch.setattr(gw, "_build_directory_service", _fake_dir)
    monkeypatch.setattr(gw, "_build_datatransfer_service", _fake_dt)
    return _set_routes


@pytest.mark.asyncio
async def test_revoke_seat_user_suspend_transfers_then_suspends(
    patch_services,
) -> None:
    """Happy path: user_suspend with a successor fires the Data
    Transfer insert for Drive+Calendar, then flips suspended=True."""
    from app.integrations.driver_base import ProvisioningContext

    log = patch_services({
        # The directory users.get is called twice — once for the old
        # user, once for the successor. Our fluent stub returns the
        # same leaf for the same key, which is fine here because both
        # callers just need an `id` field. If we needed different
        # bodies per call we'd add a deterministic sequence helper.
        "users.get": {"id": "user-123", "primaryEmail": "x@x"},
        "transfers.insert": {"id": "transfer-xyz", "overallTransferStatusCode": "inProgress"},
        # Pre-launch P1 fix added a ``transfers.get`` poll between
        # insert and the eventual users.update suspend so we don't
        # suspend while the async transfer is still in flight. The
        # mock returns ``completed`` immediately so the poll exits on
        # the first iteration without blocking the test on a real
        # 5-min budget.
        "transfers.get": {"id": "transfer-xyz", "overallTransferStatusCode": "completed"},
        "users.update": {"primaryEmail": "leaver@x", "suspended": True},
    })
    ctx = ProvisioningContext(
        user_email="leaver@example.com",
        successor_email="successor@example.com",
        extra={"operation": "user_suspend"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is True
    assert result.raw_response.get("data_transfer_id") == "transfer-xyz"
    # Sequence: users.get (×2 for recipient lookup + source-user lookup
    # in ``_initiate_data_transfer``) → transfers.insert → transfers.get
    # (poll, returns 'completed' on first call) → users.get (×1 for
    # ``_set_suspended`` already-suspended pre-read, added 2026-06-02)
    # → users.update. Four "get" leafs total: 3 directory users.get
    # plus 1 datatransfer transfers.get. The pre-read returns a dict
    # with no ``suspended`` key, so ``cur.get("suspended") is True``
    # is False and the flow falls through to the actual update.
    methods = [name for name, _, _ in log.chain]
    assert methods.count("get") == 4
    assert "insert" in methods  # transfers.insert
    assert "update" in methods  # users.update (suspend)
    # data_transfer_status surfaces the poll outcome.
    assert result.raw_response.get("data_transfer_status") == "completed"


@pytest.mark.asyncio
async def test_revoke_seat_user_suspend_without_successor_fails(
    patch_services,
) -> None:
    """user_suspend without a successor must fail fast — Drive data
    would otherwise be orphaned when the account goes dark."""
    from app.integrations.driver_base import ProvisioningContext

    patch_services({})
    ctx = ProvisioningContext(
        user_email="leaver@example.com",
        extra={"operation": "user_suspend"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "successor" in result.error_message.lower()


@pytest.mark.asyncio
async def test_revoke_seat_group_member_ignores_successor(
    patch_services,
) -> None:
    """group_member removal doesn't touch user-owned data; the driver
    must skip the data-transfer step even if a successor happens to
    be set in the context."""
    from app.integrations.driver_base import ProvisioningContext

    log = patch_services({
        "members.delete": None,  # delete() returns None
    })
    ctx = ProvisioningContext(
        user_email="leaver@example.com",
        successor_email="successor@example.com",
        extra={"operation": "group_member", "group_email": "eng@x.com"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is True
    methods = [name for name, _, _ in log.chain]
    # No transfers.insert should have been called.
    assert "insert" not in methods