"""Unit tests for the Zoom software-provisioning driver.

All HTTP is mocked via respx so tests are hermetic and deterministic.
Covers:
  - test_connection: happy path + 401 + network error
  - token caching: a single token fetch covers N API calls
  - grant_seat: happy path, existing user upgrade, existing user
    already-correct-tier, invalid plan_type
  - revoke_seat: disassociate, delete, 404 idempotent success,
    invalid action

A separate ``@pytest.mark.live`` smoke test at the bottom is skipped
in CI; run manually against a sandbox Zoom account by exporting
``HELIX_ZOOM_*`` env vars and running
``pytest -m live tests/test_integrations/test_zoom_driver.py``.
"""
from __future__ import annotations

import os

import httpx
import pytest
import respx

from app.integrations.driver_base import ProvisioningContext
from app.integrations.drivers.zoom_software import (
    _ZOOM_BASE,
    _ZOOM_TOKEN_URL,
    ZOOM_TAG_BASIC_CAP_EXCEEDED,
    ZOOM_TAG_LICENSED_CAP_EXCEEDED,
    ZOOM_TAG_USER_IN_OTHER_ACCOUNT,
    ZoomSoftwareDriver,
    _classify_zoom_user_error,
    _clear_token_cache_for_tests,
)

_CONFIG = {
    "account_id": "test-account-id",
    "client_id": "test-client-id",
    "client_secret": "test-client-secret",
}


@pytest.fixture(autouse=True)
def _clear_zoom_token_cache() -> None:
    """Wipe the process-global token cache before every test so that
    a cached token from one test doesn't bleed into the next and cause
    unexpected unmocked HTTP calls or wrong token-fetch counts."""
    _clear_token_cache_for_tests()


def _driver() -> ZoomSoftwareDriver:
    return ZoomSoftwareDriver(integration_id="zoom-test", config=_CONFIG)


def _token_route(mock: respx.MockRouter) -> None:
    mock.post(_ZOOM_TOKEN_URL).mock(
        return_value=httpx.Response(
            200, json={"access_token": "fake-token", "expires_in": 3600},
        ),
    )


# ── Metadata ─────────────────────────────────────────────────────


def test_metadata_registered() -> None:
    meta = ZoomSoftwareDriver.metadata()
    assert meta.kind == "software_provisioning"
    assert meta.provider == "zoom"
    assert any(f.key == "plan_type" for f in meta.binding_fields)
    assert any(f.key == "client_secret" and f.is_secret for f in meta.fields)


def test_revoke_requires_successor_only_on_delete() -> None:
    assert ZoomSoftwareDriver.revoke_requires_successor(
        {"action_on_offboard": "delete"},
    ) is True
    assert ZoomSoftwareDriver.revoke_requires_successor(
        {"action_on_offboard": "disassociate"},
    ) is False
    # Missing key falls back to the default (disassociate → no successor)
    assert ZoomSoftwareDriver.revoke_requires_successor({}) is False
    assert ZoomSoftwareDriver.revoke_requires_successor(None) is False


# ── test_connection ──────────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_connection_success() -> None:
    _token_route(respx.mock)
    respx.mock.get(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(200, json={"total_records": 42, "users": []}),
    )
    result = await _driver().test_connection()
    assert result.ok is True
    assert "42 active users" in result.message


@pytest.mark.asyncio
@respx.mock
async def test_connection_invalid_credentials() -> None:
    respx.mock.post(_ZOOM_TOKEN_URL).mock(
        return_value=httpx.Response(401, json={"reason": "Invalid client_id or client_secret"}),
    )
    result = await _driver().test_connection()
    assert result.ok is False
    assert "Token error" in result.message


@pytest.mark.asyncio
@respx.mock
async def test_connection_api_error() -> None:
    _token_route(respx.mock)
    respx.mock.get(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(500, text="Internal Server Error"),
    )
    result = await _driver().test_connection()
    assert result.ok is False
    assert "500" in result.message


# ── Token caching ────────────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_token_cached_across_calls() -> None:
    token_route = respx.mock.post(_ZOOM_TOKEN_URL).mock(
        return_value=httpx.Response(
            200, json={"access_token": "cache-me", "expires_in": 3600},
        ),
    )
    users_route = respx.mock.get(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(200, json={"total_records": 1, "users": []}),
    )
    driver = _driver()
    await driver.test_connection()
    await driver.test_connection()
    await driver.test_connection()
    # Token fetched once, users GET three times
    assert token_route.call_count == 1
    assert users_route.call_count == 3


# ── grant_seat ───────────────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_licensed_success() -> None:
    _token_route(respx.mock)
    create_route = respx.mock.post(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(
            201,
            json={
                "id": "zoom-user-123",
                "email": "alice@example.com",
                "type": 2,
            },
        ),
    )
    ctx = ProvisioningContext(
        user_email="alice@example.com",
        user_display_name="Alice Smith",
        plan_name="Zoom Licensed",
        extra={"plan_type": "licensed"},
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is True
    assert result.external_ref == "zoom-user-123"
    # Verify the payload sent to Zoom (httpx JSON-encodes without
    # spaces after the colon, hence `"type":2` not `"type": 2`).
    req = create_route.calls.last.request
    body = req.content.decode()
    assert '"type":2' in body
    assert '"email":"alice@example.com"' in body
    assert '"first_name":"Alice"' in body
    assert '"last_name":"Smith"' in body


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_basic_tier() -> None:
    _token_route(respx.mock)
    create_route = respx.mock.post(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(
            201, json={"id": "zoom-u", "email": "bob@example.com", "type": 1},
        ),
    )
    ctx = ProvisioningContext(
        user_email="bob@example.com",
        user_display_name=None,
        extra={"plan_type": "basic"},
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is True
    req = create_route.calls.last.request
    assert '"type":1' in req.content.decode()


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_invalid_plan_type() -> None:
    _token_route(respx.mock)
    ctx = ProvisioningContext(
        user_email="c@example.com",
        extra={"plan_type": "enterprise_xl"},  # not in the enum
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is False
    assert "invalid plan_type" in result.error_message


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_existing_user_with_matching_type() -> None:
    """409 conflict + existing user already has the desired type → success."""
    _token_route(respx.mock)
    respx.mock.post(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(
            409, json={"code": 1005, "message": "User already in the account"},
        ),
    )
    respx.mock.get(f"{_ZOOM_BASE}/users/dup@example.com").mock(
        return_value=httpx.Response(
            200, json={"id": "zoom-dup", "email": "dup@example.com", "type": 2},
        ),
    )
    ctx = ProvisioningContext(
        user_email="dup@example.com",
        user_display_name="Dup User",
        extra={"plan_type": "licensed"},
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is True
    assert result.raw_response == {"already_active": True, "type": 2}


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_existing_user_type_upgrade() -> None:
    """409 + existing type differs → PATCH to the desired type."""
    _token_route(respx.mock)
    respx.mock.post(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(
            409, json={"code": 1005, "message": "User already in the account"},
        ),
    )
    respx.mock.get(f"{_ZOOM_BASE}/users/upg@example.com").mock(
        return_value=httpx.Response(
            200, json={"id": "zoom-upg", "email": "upg@example.com", "type": 1},
        ),
    )
    patch_route = respx.mock.patch(f"{_ZOOM_BASE}/users/zoom-upg").mock(
        return_value=httpx.Response(204),
    )
    ctx = ProvisioningContext(
        user_email="upg@example.com",
        extra={"plan_type": "licensed"},
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is True
    assert result.raw_response == {
        "upgraded_from_type": 1, "upgraded_to_type": 2,
    }
    assert patch_route.called


# ── revoke_seat ──────────────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_disassociate() -> None:
    _token_route(respx.mock)
    delete_route = respx.mock.delete(f"{_ZOOM_BASE}/users/alice@example.com").mock(
        return_value=httpx.Response(204),
    )
    ctx = ProvisioningContext(
        user_email="alice@example.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is True
    assert result.external_ref.endswith(":disassociate")
    req = delete_route.calls.last.request
    assert "action=disassociate" in str(req.url)


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_with_successor_passes_transfer_params() -> None:
    """Zoom delete requires handoff. When a successor is in the
    context, the driver must add ``transfer_email`` +
    ``transfer_meeting/webinar/recording`` query params so Zoom moves
    content instead of 400ing."""
    _token_route(respx.mock)
    delete_route = respx.mock.delete(f"{_ZOOM_BASE}/users/bob@example.com").mock(
        return_value=httpx.Response(204),
    )
    ctx = ProvisioningContext(
        user_email="bob@example.com",
        successor_email="carol@example.com",
        extra={"action_on_offboard": "delete"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is True
    url = str(delete_route.calls.last.request.url)
    assert "action=delete" in url
    assert "transfer_email=carol%40example.com" in url
    assert "transfer_meeting=true" in url
    assert "transfer_webinar=true" in url
    assert "transfer_recording=true" in url


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_without_successor_fails() -> None:
    """Zoom delete without a successor must fail fast — Zoom would
    otherwise return 400 for licensed users that host content."""
    _token_route(respx.mock)
    ctx = ProvisioningContext(
        user_email="bob@example.com",
        extra={"action_on_offboard": "delete"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "successor" in result.error_message.lower()


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_idempotent_on_404() -> None:
    """Revoking an already-absent user should succeed (offboarding
    idempotency — we don't care whether they were already gone)."""
    _token_route(respx.mock)
    respx.mock.delete(f"{_ZOOM_BASE}/users/gone@example.com").mock(
        return_value=httpx.Response(404, json={"code": 1001, "message": "User not exist"}),
    )
    ctx = ProvisioningContext(
        user_email="gone@example.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is True
    assert result.raw_response == {"already_absent": True}


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_idempotent_on_429_concurrent_disassociate() -> None:
    """Round-13 (2026-05-08): a leaver holding N add-on seats on the
    same Zoom account creates N revoke tasks against the SAME user.
    Zoom serialises these at the account level — the 1st succeeds,
    the rest get HTTP 429 + ``A request to disassociate this user
    has already been made.`` Treat that as success for offboarding
    idempotency (same semantics as 404 already_absent).

    Caught on P01121 in prod: 1 of 3 Zoom revokes succeeded, the
    other 2 dead-lettered with the 429 and stranded the leaver in
    blocked status. Without this branch, every multi-Zoom-seat
    leaver hits the same trap.
    """
    _token_route(respx.mock)
    respx.mock.delete(f"{_ZOOM_BASE}/users/sara.fox@example.com").mock(
        return_value=httpx.Response(
            429,
            json={
                "code": 429,
                "message": (
                    "Too many concurrent requests. A request to "
                    "disassociate this user has already been made."
                ),
            },
        ),
    )
    ctx = ProvisioningContext(
        user_email="sara.fox@example.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is True
    assert result.raw_response.get("already_in_progress") is True
    assert result.raw_response.get("status") == 429
    assert result.external_ref.endswith(":already_in_progress")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_429_unrelated_message_still_fails() -> None:
    """A 429 *without* the "already been made" marker is a real rate
    limit — must NOT be treated as idempotent success or we'd mask
    legit retry-after signals as completed work.
    """
    _token_route(respx.mock)
    respx.mock.delete(f"{_ZOOM_BASE}/users/x@example.com").mock(
        return_value=httpx.Response(
            429,
            json={"code": 429, "message": "Rate limit exceeded; retry later"},
        ),
    )
    ctx = ProvisioningContext(
        user_email="x@example.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "429" in result.error_message


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_invalid_action() -> None:
    _token_route(respx.mock)
    ctx = ProvisioningContext(
        user_email="x@example.com",
        extra={"action_on_offboard": "nuke"},  # not in the enum
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "invalid action_on_offboard" in result.error_message


# ── Managed-domain 1107 fallback ─────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_managed_domain_1107_falls_back_to_basic_downgrade() -> None:
    """When DELETE returns 400 / code 1107 ("managed domain"), driver
    must PATCH user.type=1 (Basic) and report success — that's the
    only Zoom-permitted way to release a paid license on an SSO-
    bound domain. Caught 2026-05-08 (kevin.lu@pingcap.com prod bug).
    """
    _token_route(respx.mock)
    delete_route = respx.mock.delete(
        f"{_ZOOM_BASE}/users/managed@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={
                "code": 1107,
                "message": "You can not disassociate a user with managed domain.",
            },
        ),
    )
    patch_route = respx.mock.patch(
        f"{_ZOOM_BASE}/users/managed@pingcap.com",
    ).mock(return_value=httpx.Response(204))

    ctx = ProvisioningContext(
        user_email="managed@pingcap.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)

    assert result.ok is True, result.error_message
    assert result.external_ref == "zoom:managed@pingcap.com:downgraded_to_basic"
    assert result.raw_response["fallback"] == "managed_domain_type_downgrade"
    assert result.raw_response["to_type"] == 1
    assert delete_route.called
    assert patch_route.called
    # PATCH body must be {"type": 1}
    import json as _json
    body = _json.loads(patch_route.calls.last.request.content.decode())
    assert body == {"type": 1}


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_managed_domain_1107_404_on_patch_treated_as_absent() -> None:
    """If the fallback PATCH itself 404s (vendor lost the user between
    DELETE and PATCH — race), treat as already-absent so the revoke
    settles successfully rather than dead-lettering."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/orphan@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={"code": 1107, "message": "managed domain"},
        ),
    )
    respx.mock.patch(
        f"{_ZOOM_BASE}/users/orphan@pingcap.com",
    ).mock(return_value=httpx.Response(404))

    ctx = ProvisioningContext(
        user_email="orphan@pingcap.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is True
    assert result.raw_response.get("already_absent") is True
    assert result.raw_response.get("via") == "managed_domain_fallback"


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_managed_domain_1107_patch_failure_surfaces() -> None:
    """If the fallback PATCH itself fails with non-2xx/non-404, the
    whole revoke must fail with a message that captures BOTH legs
    (the original 1107 path + the fallback's status) so the operator
    can see the full timeline in one error_message column."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/locked@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={"code": 1107, "message": "managed domain"},
        ),
    )
    respx.mock.patch(
        f"{_ZOOM_BASE}/users/locked@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            403,
            json={"code": 200, "message": "Account does not have user-update privilege"},
        ),
    )

    ctx = ProvisioningContext(
        user_email="locked@pingcap.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "managed-domain fallback PATCH" in result.error_message
    assert "403" in result.error_message
    assert result.raw_response["fallback"] == "managed_domain_type_downgrade"
    assert result.raw_response["status"] == 403


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_400_non_1107_does_not_trigger_fallback() -> None:
    """Only Zoom code 1107 should fan out to the PATCH fallback;
    other 400s (validation errors, missing fields, etc.) must remain
    as failures so the operator sees the real cause instead of a
    false-success from a fortuitous PATCH."""
    _token_route(respx.mock)
    delete_route = respx.mock.delete(
        f"{_ZOOM_BASE}/users/whatever@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={"code": 300, "message": "Invalid parameter."},
        ),
    )
    patch_route = respx.mock.patch(
        f"{_ZOOM_BASE}/users/whatever@pingcap.com",
    ).mock(return_value=httpx.Response(204))

    ctx = ProvisioningContext(
        user_email="whatever@pingcap.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "300" not in (result.external_ref or "")  # not auto-success
    assert delete_route.called
    assert not patch_route.called  # fallback NOT triggered


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_action_1107_does_not_fallback() -> None:
    """The 1107 fallback is scoped to ``action=disassociate`` only.
    For ``action=delete`` (admin chose hard-delete with successor
    handoff), 1107 should still surface as failure — silently
    converting to a Basic downgrade would skip the meeting/recording
    handoff the admin asked for."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/managed@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={"code": 1107, "message": "managed domain"},
        ),
    )
    patch_route = respx.mock.patch(
        f"{_ZOOM_BASE}/users/managed@pingcap.com",
    ).mock(return_value=httpx.Response(204))

    ctx = ProvisioningContext(
        user_email="managed@pingcap.com",
        successor_email="successor@pingcap.com",
        extra={"action_on_offboard": "delete"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "1107" in result.error_message or "managed domain" in result.error_message.lower()
    assert not patch_route.called


# ── Live smoke (manual) ──────────────────────────────────────────


# ── list_users / get_account_plan / extract_licensed_seats ─────


@pytest.mark.asyncio
@respx.mock
async def test_list_users_paginates_until_empty_token() -> None:
    _token_route(respx.mock)
    page1 = {
        "users": [
            {"id": "u1", "email": "a@x.io", "type": 2, "last_login_time": "2026-04-01T09:00:00Z"},
            {"id": "u2", "email": "b@x.io", "type": 1, "last_login_time": ""},
        ],
        "next_page_token": "tok-2",
    }
    page2 = {
        "users": [
            {"id": "u3", "email": "c@x.io", "type": 2, "last_login_time": "2025-12-01T00:00:00Z"},
        ],
        "next_page_token": "",
    }
    call_log: list[str] = []

    def _responder(request: httpx.Request) -> httpx.Response:
        token = request.url.params.get("next_page_token")
        call_log.append(token or "")
        if token == "tok-2":
            return httpx.Response(200, json=page2)
        return httpx.Response(200, json=page1)

    respx.mock.get(f"{_ZOOM_BASE}/users").mock(side_effect=_responder)
    users = await _driver().list_users(status="active")
    assert [u["id"] for u in users] == ["u1", "u2", "u3"]
    # First call has no token, second passes tok-2
    assert call_log == ["", "tok-2"]


@pytest.mark.asyncio
@respx.mock
async def test_list_users_raises_on_api_error() -> None:
    _token_route(respx.mock)
    respx.mock.get(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(500, text="boom"),
    )
    with pytest.raises(RuntimeError, match="500"):
        await _driver().list_users()


@pytest.mark.asyncio
@respx.mock
async def test_get_account_plan_success_and_missing_scope() -> None:
    _token_route(respx.mock)
    respx.mock.get(f"{_ZOOM_BASE}/accounts/me/plans").mock(
        return_value=httpx.Response(
            200,
            json={"plan_base": {"type": "Pro", "hosts": 77}},
        ),
    )
    payload = await _driver().get_account_plan()
    assert ZoomSoftwareDriver.extract_licensed_seats(payload) == 77

    # Fresh driver + 403 → returns None instead of raising
    _clear_token_cache_for_tests()  # force a fresh token fetch for the next call
    driver = _driver()
    respx.mock.get(f"{_ZOOM_BASE}/accounts/me/plans").mock(
        return_value=httpx.Response(403, text="missing scope"),
    )
    _token_route(respx.mock)
    assert await driver.get_account_plan() is None

    # Zoom's granular-scope migration started returning HTTP 400
    # (code 4711) for missing scopes on this endpoint, not 403 —
    # the swallow list has to cover it or the seat-sync run fails
    # whole-cloth on a best-effort max_seats fetch.
    _clear_token_cache_for_tests()  # force a fresh token fetch for the next call
    driver = _driver()
    respx.mock.get(f"{_ZOOM_BASE}/accounts/me/plans").mock(
        return_value=httpx.Response(
            400,
            json={
                "code": 4711,
                "message": (
                    "Invalid access token, does not contain "
                    "scopes:[billing:read:plan_information:admin]."
                ),
            },
        ),
    )
    _token_route(respx.mock)
    assert await driver.get_account_plan() is None


def test_extract_licensed_seats_handles_missing_fields() -> None:
    assert ZoomSoftwareDriver.extract_licensed_seats(None) is None
    assert ZoomSoftwareDriver.extract_licensed_seats({}) is None
    assert ZoomSoftwareDriver.extract_licensed_seats({"plan_base": {}}) is None
    assert (
        ZoomSoftwareDriver.extract_licensed_seats(
            {"plan_base": {"hosts": 0}}
        )
        is None
    )
    assert (
        ZoomSoftwareDriver.extract_licensed_seats(
            {"plan_base": {"hosts": 12}}
        )
        == 12
    )


@pytest.mark.asyncio
@respx.mock
async def test_get_user_404_returns_none() -> None:
    _token_route(respx.mock)
    respx.mock.get(f"{_ZOOM_BASE}/users/gone@x.io").mock(
        return_value=httpx.Response(404, json={"code": 1001, "message": "user not found"}),
    )
    assert await _driver().get_user("gone@x.io") is None


@pytest.mark.asyncio
@respx.mock
async def test_get_user_refreshes_token_on_401_then_succeeds() -> None:
    """#1652 — directory helpers now route through ``_zoom_request``, so a
    Zoom-side S2S token invalidation (401) transparently force-refreshes
    the token and retries once instead of hard-failing. Previously these
    methods opened a bare ``httpx.AsyncClient`` and a 401 propagated as a
    RuntimeError / None with no refresh."""
    _token_route(respx.mock)
    responses = [
        httpx.Response(401, json={"code": 124, "message": "Invalid access token"}),
        httpx.Response(200, json={"id": "u9", "email": "alice@x.io", "type": 2}),
    ]

    def _responder(request: httpx.Request) -> httpx.Response:
        return responses.pop(0)

    respx.mock.get(f"{_ZOOM_BASE}/users/alice@x.io").mock(side_effect=_responder)
    out = await _driver().get_user("alice@x.io")
    assert out is not None
    assert out["id"] == "u9"
    # Both the 401 and the retried 200 were consumed → refresh fired.
    assert responses == []


# ── Structured error tags (Zoom 3412 + cousins) ──────────────────


def test_classify_zoom_user_error_basic_cap_on_create() -> None:
    """Zoom code 3413 = "user count limit reached" when creating a
    Basic user directly. Same admin remediation as 3412 (clean up
    dormant Basic users on Zoom), so they share the BASIC_CAP_EXCEEDED
    tag."""
    tag, hint = _classify_zoom_user_error(3413, "User count limit reached")
    assert tag == ZOOM_TAG_BASIC_CAP_EXCEEDED
    assert hint is not None
    assert "Basic" in hint


def test_classify_zoom_user_error_user_in_other_account() -> None:
    """Zoom codes 1005 / 1108 / 2030 all mean "email is already on
    another Zoom account" — admin must coordinate cross-tenant.
    """
    for code in (1005, 1108, 2030):
        tag, hint = _classify_zoom_user_error(code, "email exists")
        assert tag == ZOOM_TAG_USER_IN_OTHER_ACCOUNT, f"code={code}"
        assert hint is not None


def test_classify_zoom_user_error_licensed_cap_phrase_match() -> None:
    """Zoom code 200 (generic "user type cannot be modified") + a
    message mentioning maximum/limit + licensed → Licensed-cap tag.
    Bare code 200 with an unrelated message stays untagged so we
    don't false-positive on per-user setting conflicts."""
    tag, _ = _classify_zoom_user_error(
        200, "The maximum number of Licensed users has been reached.",
    )
    assert tag == ZOOM_TAG_LICENSED_CAP_EXCEEDED
    tag, _ = _classify_zoom_user_error(
        200, "Account does not have user-update privilege",
    )
    assert tag is None
    # 200 without ANY message also stays untagged
    tag, _ = _classify_zoom_user_error(200, "")
    assert tag is None


def test_classify_zoom_user_error_unknown_code_returns_none() -> None:
    """Codes we don't have a remediation for must fall through —
    framework adds no false-positive guidance. The bare vendor body
    still appears in error_message."""
    assert _classify_zoom_user_error(9999, "weird thing") == (None, None)
    assert _classify_zoom_user_error(None, "no code at all") == (None, None)


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_managed_domain_3412_basic_cap_tagged() -> None:
    """The PR's primary fix: managed-domain fallback PATCH=Basic that
    hits code 3412 (Basic-user cap reached) must surface a tagged,
    actionable error_message AND enrich raw_response with the current
    Basic user count.

    Bit prod 2026-05-22: P01451 looperth.wong's revoke task died
    with an opaque Zoom 3412 body and admin had to look up the
    code manually."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/looperth.wong@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={"code": 1107, "message": "managed domain"},
        ),
    )
    respx.mock.patch(
        f"{_ZOOM_BASE}/users/looperth.wong@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={
                "code": 3412,
                "message": (
                    "Your request to convert the user type of this "
                    "Licensed user to a basic user was not approved at "
                    "this time because your account has reached the "
                    "permitted maximum of basic users."
                ),
            },
        ),
    )
    # Basic-count enrichment — _count_basic_users_safe paginates
    # active + pending. Stub both with a small fixed roster so the
    # count is deterministic. Three Basic users + one Licensed user
    # → expected count = 3.
    respx.mock.get(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(
            200,
            json={
                "users": [
                    {"id": "b1", "email": "b1@pingcap.com", "type": 1},
                    {"id": "b2", "email": "b2@pingcap.com", "type": 1},
                    {"id": "b3", "email": "b3@pingcap.com", "type": 1},
                    {"id": "l1", "email": "l1@pingcap.com", "type": 2},
                ],
                "next_page_token": "",
            },
        ),
    )

    ctx = ProvisioningContext(
        user_email="looperth.wong@pingcap.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)

    assert result.ok is False
    # Tag prefix is at index 0 so the DM dispatcher's tag regex matches
    assert result.error_message.startswith(f"[{ZOOM_TAG_BASIC_CAP_EXCEEDED}]")
    # Chinese-locale actionable text
    assert "Basic 用户数" in result.error_message
    assert "Zoom Admin" in result.error_message
    # raw_response carries the structured detail for logs / audit /
    # future-PR analytics — count is 6 (3 Basic × 2 status fetches).
    assert result.raw_response["error_tag"] == ZOOM_TAG_BASIC_CAP_EXCEEDED
    assert result.raw_response["zoom_code"] == 3412
    assert result.raw_response["current_basic_users"] == 6


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_managed_domain_3412_count_failure_falls_back_to_none() -> None:
    """If the enrichment list-users call itself fails (transient API
    outage, rate limit, scope removal), the 3412 error_message must
    STILL fire with the tag — the count is best-effort. Without this
    behaviour, a flaky list_users would mask the underlying 3412 as
    a generic 500 with no remediation hint."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/managed@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400, json={"code": 1107, "message": "managed domain"},
        ),
    )
    respx.mock.patch(
        f"{_ZOOM_BASE}/users/managed@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400, json={"code": 3412, "message": "basic users cap"},
        ),
    )
    # list_users 500 → _count_basic_users_safe returns None
    respx.mock.get(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(500, text="kaboom"),
    )

    ctx = ProvisioningContext(
        user_email="managed@pingcap.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)

    assert result.ok is False
    assert result.error_message.startswith(f"[{ZOOM_TAG_BASIC_CAP_EXCEEDED}]")
    assert "Basic count unavailable" in result.error_message
    assert result.raw_response["current_basic_users"] is None


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_400_other_code_no_fallback_carries_zoom_code() -> None:
    """A non-1107 400 must NOT trigger the PATCH fallback (existing
    invariant) but its error_message must now carry the parsed
    ``zoom_code`` so logs / future analytics can act on it without
    re-parsing the vendor body."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/whatever@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400, json={"code": 300, "message": "Invalid parameter."},
        ),
    )
    patch_route = respx.mock.patch(
        f"{_ZOOM_BASE}/users/whatever@pingcap.com",
    ).mock(return_value=httpx.Response(204))

    ctx = ProvisioningContext(
        user_email="whatever@pingcap.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert not patch_route.called
    # zoom_code now lives in both error_message and raw_response
    assert "code=300" in result.error_message
    assert result.raw_response["zoom_code"] == 300


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_user_in_other_account_tagged() -> None:
    """Zoom code 1005 on revoke (rare — usually surfaces on create —
    but possible if the user got moved between accounts mid-flight).
    Must tag USER_IN_OTHER_ACCOUNT so the admin DM renders the
    cross-tenant remediation hint."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/elsewhere@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={"code": 1005, "message": "User belongs to another account"},
        ),
    )
    ctx = ProvisioningContext(
        user_email="elsewhere@pingcap.com",
        extra={"action_on_offboard": "disassociate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert result.error_message.startswith(f"[{ZOOM_TAG_USER_IN_OTHER_ACCOUNT}]")
    assert result.raw_response["error_tag"] == ZOOM_TAG_USER_IN_OTHER_ACCOUNT
    assert result.raw_response["zoom_code"] == 1005


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_user_in_other_account_tagged() -> None:
    """Zoom code 1005 on POST /users (most common path for this code)
    must tag USER_IN_OTHER_ACCOUNT. Critical: 1005 here comes back
    as HTTP 400 NOT 409 — Zoom's response is inconsistent across
    "exists in this account" (409) vs "exists in another account"
    (often 400). Without the tag, the admin sees a raw vendor body
    and has to guess which case they're in."""
    _token_route(respx.mock)
    respx.mock.post(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(
            400,
            json={
                "code": 1005,
                "message": "User email belongs to a different account",
            },
        ),
    )
    ctx = ProvisioningContext(
        user_email="claimed@pingcap.com",
        extra={"plan_type": "licensed"},
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is False
    assert result.error_message.startswith(f"[{ZOOM_TAG_USER_IN_OTHER_ACCOUNT}]")
    assert result.raw_response["error_tag"] == ZOOM_TAG_USER_IN_OTHER_ACCOUNT


@pytest.mark.asyncio
@respx.mock
async def test_existing_user_patch_upgrade_failure_carries_zoom_code() -> None:
    """The upgrade-PATCH branch in _reconcile_existing_user used to
    swallow the vendor code into a plain text message. Verify the new
    structured path: zoom_code lands in both message + raw_response."""
    _token_route(respx.mock)
    respx.mock.post(f"{_ZOOM_BASE}/users").mock(
        return_value=httpx.Response(
            409, json={"code": 1005, "message": "exists"},
        ),
    )
    respx.mock.get(f"{_ZOOM_BASE}/users/p@pingcap.com").mock(
        return_value=httpx.Response(
            200, json={"id": "ZP", "email": "p@pingcap.com", "type": 1},
        ),
    )
    # PATCH refused — Zoom's generic 400 + code 200 + Licensed-cap
    # phrasing → LICENSED_CAP_EXCEEDED tag.
    respx.mock.patch(f"{_ZOOM_BASE}/users/ZP").mock(
        return_value=httpx.Response(
            400,
            json={
                "code": 200,
                "message": (
                    "The maximum number of Licensed users has been "
                    "reached for this account."
                ),
            },
        ),
    )
    ctx = ProvisioningContext(
        user_email="p@pingcap.com",
        extra={"plan_type": "licensed"},
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is False
    assert result.error_message.startswith(f"[{ZOOM_TAG_LICENSED_CAP_EXCEEDED}]")
    assert result.raw_response["error_tag"] == ZOOM_TAG_LICENSED_CAP_EXCEEDED
    assert result.raw_response["zoom_code"] == 200


# ── delete_user_unconditional (Basic-pool eviction path) ───────


@pytest.mark.asyncio
@respx.mock
async def test_delete_user_unconditional_204_success() -> None:
    """Happy path: DELETE returns 204, driver reports ok with the
    canonical external_ref shape."""
    _token_route(respx.mock)
    route = respx.mock.delete(
        f"{_ZOOM_BASE}/users/ghost@pingcap.com",
    ).mock(return_value=httpx.Response(204))
    result = await _driver().delete_user_unconditional("ghost@pingcap.com")
    assert result.ok is True
    assert result.external_ref == "zoom:ghost@pingcap.com:deleted"
    # action=delete is the only managed-domain-permitted path
    assert route.calls[0].request.url.params["action"] == "delete"


@pytest.mark.asyncio
@respx.mock
async def test_delete_user_unconditional_404_idempotent_success() -> None:
    """User already absent → caller treats as success so re-running
    after a partial Basic-eviction round doesn't double-fail."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/already_gone@pingcap.com",
    ).mock(return_value=httpx.Response(404))
    result = await _driver().delete_user_unconditional("already_gone@pingcap.com")
    assert result.ok is True
    assert result.external_ref == "zoom:already_gone@pingcap.com:already_absent"
    assert result.raw_response["already_absent"] is True


@pytest.mark.asyncio
@respx.mock
async def test_delete_user_unconditional_400_returns_structured_failure() -> None:
    """Zoom can refuse DELETE if the user has unmovable content. The
    coordinator's retry loop reads the parsed code to decide whether
    to advance to the next candidate, so the failure must carry
    structured raw_response — not just an opaque error_message."""
    _token_route(respx.mock)
    respx.mock.delete(
        f"{_ZOOM_BASE}/users/heavy_user@pingcap.com",
    ).mock(
        return_value=httpx.Response(
            400,
            json={"code": 1010, "message": "transfer_email required"},
        ),
    )
    result = await _driver().delete_user_unconditional("heavy_user@pingcap.com")
    assert result.ok is False
    assert result.raw_response["zoom_code"] == 1010
    assert "code=1010" in result.error_message


@pytest.mark.live
@pytest.mark.asyncio
async def test_live_connection_against_real_zoom() -> None:
    """Run manually with real sandbox credentials:

        HELIX_ZOOM_ACCOUNT_ID=... HELIX_ZOOM_CLIENT_ID=... \\
        HELIX_ZOOM_CLIENT_SECRET=... pytest -m live \\
            tests/test_integrations/test_zoom_driver.py

    CI skips this marker. The point is to catch real Zoom API quirks
    (rate limits, unexpected error shapes, plan restrictions) that
    mocked tests can't.
    """
    account_id = os.environ.get("HELIX_ZOOM_ACCOUNT_ID")
    client_id = os.environ.get("HELIX_ZOOM_CLIENT_ID")
    client_secret = os.environ.get("HELIX_ZOOM_CLIENT_SECRET")
    if not (account_id and client_id and client_secret):
        pytest.skip(
            "Set HELIX_ZOOM_ACCOUNT_ID / _CLIENT_ID / _CLIENT_SECRET to run",
        )
    driver = ZoomSoftwareDriver(
        integration_id="zoom-live",
        config={
            "account_id": account_id,
            "client_id": client_id,
            "client_secret": client_secret,
        },
    )
    result = await driver.test_connection()
    assert result.ok is True, f"Live Zoom probe failed: {result.message}"


# ── Transfer-recipient refusals on the terminal delete (2026-09-06) ───
#
# Zoom answers this whole family with HTTP 400 + ``code: 200`` — the
# same catch-all code as a refused type PATCH — so the code carries no
# information and the message is the only signal. That makes the driver
# the one correct place to string-match: callers read a tag.
#
# The only wording verified against the real vendor is the webinar one
# (prod 2026-08-20..08-23). Everything else that names the data receiver
# is deliberately classified as "recipient invalid" rather than guessed
# sentence by sentence.


def test_classify_transfer_error_recognises_the_verified_prod_message() -> None:
    from app.integrations.drivers.zoom_software import (
        ZOOM_TAG_TRANSFER_RECIPIENT_NO_WEBINAR,
        _classify_zoom_transfer_error,
    )

    assert _classify_zoom_transfer_error(
        "Unable to transfer data because data receiver does not have Zoom "
        "Webinars licenses."
    ) == ZOOM_TAG_TRANSFER_RECIPIENT_NO_WEBINAR


def test_classify_transfer_error_survives_a_truncated_body() -> None:
    """Ledger rows cap the vendor body at 200 chars and have cut
    "Webinars" mid-word; the classification must not flip to the
    opposite treatment because of where the truncation landed."""
    from app.integrations.drivers.zoom_software import (
        ZOOM_TAG_TRANSFER_RECIPIENT_NO_WEBINAR,
        _classify_zoom_transfer_error,
    )

    assert _classify_zoom_transfer_error(
        "Unable to transfer data because data receiver does not have Zoom Webi"
    ) == ZOOM_TAG_TRANSFER_RECIPIENT_NO_WEBINAR


def test_classify_transfer_error_tags_a_missing_receiver_separately() -> None:
    from app.integrations.drivers.zoom_software import (
        ZOOM_TAG_TRANSFER_RECIPIENT_INVALID,
        _classify_zoom_transfer_error,
    )

    assert _classify_zoom_transfer_error(
        "Unable to transfer data because data receiver does not exist."
    ) == ZOOM_TAG_TRANSFER_RECIPIENT_INVALID


def test_classify_transfer_error_ignores_unrelated_failures() -> None:
    from app.integrations.drivers.zoom_software import (
        _classify_zoom_transfer_error,
    )

    assert _classify_zoom_transfer_error("Internal Server Error") is None
    assert _classify_zoom_transfer_error("") is None
    # A refusal about the LEAVER, not the receiver, must stay untagged —
    # it is a real fault and has to keep failing loudly.
    assert _classify_zoom_transfer_error(
        "User does not exist: alice@x.io",
    ) is None


@pytest.mark.asyncio
@respx.mock
async def test_delete_with_transfer_tags_a_recipient_refusal() -> None:
    from app.integrations.drivers.zoom_software import (
        ZOOM_TAG_TRANSFER_RECIPIENT_INVALID,
        zoom_transfer_refusal_tag,
    )

    _token_route(respx.mock)
    respx.mock.delete(f"{_ZOOM_BASE}/users/leaver@x.io").mock(
        return_value=httpx.Response(400, json={
            "code": 200,
            "message": (
                "Unable to transfer data because data receiver does not exist."
            ),
        }),
    )
    result = await _driver().delete_user_unconditional(
        "leaver@x.io", transfer_email="ghost@x.io",
    )
    assert result.ok is False
    assert result.raw_response["error_tag"] == ZOOM_TAG_TRANSFER_RECIPIENT_INVALID
    assert zoom_transfer_refusal_tag(result) == ZOOM_TAG_TRANSFER_RECIPIENT_INVALID
    # The recipient stays on the failure for the ledger / the alert.
    assert result.raw_response["transfer_email"] == "ghost@x.io"


@pytest.mark.asyncio
@respx.mock
async def test_delete_without_transfer_never_tags_a_recipient_refusal() -> None:
    """No recipient was asked for, so no failure can be about one — the
    Basic-pool eviction path must not be re-routed into the offboarding
    sweep's refusal branch by a coincidental message."""
    from app.integrations.drivers.zoom_software import zoom_transfer_refusal_tag

    _token_route(respx.mock)
    respx.mock.delete(f"{_ZOOM_BASE}/users/idle@x.io").mock(
        return_value=httpx.Response(400, json={
            "code": 200, "message": "data receiver does not exist",
        }),
    )
    result = await _driver().delete_user_unconditional("idle@x.io")
    assert result.ok is False
    assert result.raw_response.get("error_tag") is None
    assert result.raw_response["transfer_email"] is None
    assert zoom_transfer_refusal_tag(result) is None