"""Unit tests for the Jira (Atlassian Cloud) software-provisioning driver.

All HTTP is mocked via respx. The driver uses Basic auth on
``<site>.atlassian.net/rest/api/3/`` for site-level user + group ops,
and Bearer auth on ``api.atlassian.com/admin/v1/`` for the optional
``action_on_offboard='delete'`` lifecycle disable.

Covers:
  - metadata + driver registration
  - test_connection: 200 / 401 / 403 / 5xx
  - find_user_by_email: exact-match filter + privacy-hidden email
  - grant_seat: existing user → add, user not found → invite + add,
    already-a-member (text + JSON variants), missing group_name,
    group 404
  - revoke_seat: deactivate happy-path, user not found = already_absent,
    delete with org_id (admin lifecycle disable), delete missing org_id
  - list_group_members: pagination + 404
"""
from __future__ import annotations

import httpx
import pytest
import respx

from app.integrations.driver_base import ProvisioningContext
from app.integrations.drivers.jira_software import (
    _ADMIN_API_BASE,
    JIRA_TAG_ORG_ADMIN_AUTH_FAILED,
    JIRA_TAG_ORG_ADMIN_PERMISSION_DENIED,
    JiraSoftwareDriver,
)

_SITE = "tidb.atlassian.net"
_BASE = f"https://{_SITE}/rest/api/3"
_CONFIG = {
    "site_url": _SITE,
    "admin_email": "it-admin@example.com",
    "api_token": "ATCTT-test-token",
}
_CONFIG_WITH_ORG = {
    **_CONFIG,
    "org_id": "org-uuid-1234",
    "org_admin_api_key": "ATCTT-org-admin-key",
}
_CONFIG_WITH_UNSCOPED = {
    **_CONFIG,
    "org_id": "org-uuid-1234",
    "org_unscoped_api_key": "ATCTT-unscoped-key",
}
_CONFIG_WITH_ORG_ID_NO_KEY = {**_CONFIG, "org_id": "org-uuid-1234"}

# Org remove-access endpoint the delete path now calls (replaces the old
# — and never-valid — manage/lifecycle/disable URL).
_REMOVE_ACCESS_URL = (
    f"{_ADMIN_API_BASE}/orgs/org-uuid-1234/directory/users/acct-1"
)


def _driver(cfg: dict | None = None) -> JiraSoftwareDriver:
    return JiraSoftwareDriver(
        integration_id="jira-test", config=cfg or _CONFIG,
    )


def _ctx(
    *,
    email: str = "alice@example.com",
    display: str = "Alice Doe",
    group_name: str = "jira-software-users",
    action: str = "deactivate",
) -> ProvisioningContext:
    return ProvisioningContext(
        user_email=email,
        user_display_name=display,
        extra={"group_name": group_name, "action_on_offboard": action},
    )


# ── metadata ────────────────────────────────────────────────────


def test_metadata_registered() -> None:
    meta = JiraSoftwareDriver.metadata()
    assert meta.kind == "software_provisioning"
    assert meta.provider == "jira"
    keys = {f.key for f in meta.fields}
    assert {"site_url", "admin_email", "api_token", "org_id",
            "org_admin_api_key"} <= keys
    bkeys = {f.key for f in meta.binding_fields}
    # group_name + action_on_offboard drive grant/revoke; product_key +
    # site_urls (added with the product_access seat-count model) scope the
    # daily reconcile to a specific product × paid-production site;
    # seat_revocable opts a shared-billing add-on plan out of idle-reclaim.
    assert {
        "group_name", "action_on_offboard", "product_key", "site_urls",
        "seat_revocable",
    } == bkeys
    # product_key / site_urls / seat_revocable are optional — counting
    # falls back to group-derived family + the default production-site
    # allowlist, and idle-reclaim treats an absent seat_revocable as True.
    optional = {f.key for f in meta.binding_fields if not f.required}
    assert {"product_key", "site_urls", "seat_revocable"} <= optional
    api_token_field = next(f for f in meta.fields if f.key == "api_token")
    assert api_token_field.is_secret is True
    assert api_token_field.required is True
    # The hint must steer admins toward id.atlassian.com (user token) and
    # warn against the org-admin-keys page — that mistake costs hours.
    hint = api_token_field.hint or ""
    assert hint.count(
        "id.atlassian.com/manage-profile/security/api-tokens",
    ) == 1
    assert hint.count("admin.atlassian.com/o/{orgId}/api-keys") == 1
    org_id_field = next(f for f in meta.fields if f.key == "org_id")
    assert org_id_field.required is False
    org_key_field = next(
        f for f in meta.fields if f.key == "org_admin_api_key"
    )
    assert org_key_field.is_secret is True
    assert org_key_field.required is False


@pytest.mark.asyncio
async def test_site_url_strips_scheme_and_slash() -> None:
    drv = _driver({**_CONFIG, "site_url": "https://tidb.atlassian.net/"})
    # Internal helper — but we reach the prod path via _site_base
    assert await drv._site_base() == _BASE


@pytest.mark.asyncio
async def test_site_url_strips_path_and_query() -> None:
    """Users sometimes paste the full Jira browser URL into the form;
    the driver must reduce it to just the host."""
    pasted = "https://tidb.atlassian.net/jira/projects?page=1&sortKey=name"
    drv = _driver({**_CONFIG, "site_url": pasted})
    assert await drv._site_base() == _BASE


@pytest.mark.asyncio
async def test_site_url_blank_raises() -> None:
    drv = _driver({**_CONFIG, "site_url": "   "})
    with pytest.raises(ValueError, match="missing site_url"):
        await drv._site_base()


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


@pytest.mark.asyncio
@respx.mock
async def test_connection_ok() -> None:
    respx.mock.get(f"{_BASE}/myself").mock(
        return_value=httpx.Response(200, json={
            "accountId": "557058:abcdef-1234-5678-9012-aaaa",
            "displayName": "IT Admin",
            "emailAddress": "it-admin@example.com",
        }),
    )
    result = await _driver().test_connection()
    assert result.ok is True
    assert "IT Admin" in result.message
    assert result.detail["site_url"] == _SITE


@pytest.mark.asyncio
@respx.mock
async def test_connection_401_message_steers_to_user_token() -> None:
    """The 401 message is the only hint the operator gets when they
    paste an org admin key into api_token. It MUST point them at
    id.atlassian.com explicitly — a generic 'check your token' message
    sends them back to the same wrong page."""
    respx.mock.get(f"{_BASE}/myself").mock(
        return_value=httpx.Response(401, text="Unauthorized"),
    )
    result = await _driver().test_connection()
    assert result.ok is False
    assert "401" in result.message
    assert result.message.count(
        "https://id.atlassian.com/manage-profile/security/api-tokens",
    ) == 1
    assert "org admin" in result.message.lower()


@pytest.mark.asyncio
@respx.mock
async def test_connection_403() -> None:
    respx.mock.get(f"{_BASE}/myself").mock(
        return_value=httpx.Response(403, text="Forbidden"),
    )
    result = await _driver().test_connection()
    assert result.ok is False
    assert "403" in result.message


# ── find_user_by_email ──────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_find_user_filters_to_exact_email() -> None:
    """Atlassian's user/search returns near-matches by display name —
    the driver must filter to an exact email match."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "wrong", "emailAddress": "alice.smith@example.com"},
            {"accountId": "right", "emailAddress": "alice@example.com"},
        ]),
    )
    user = await _driver().find_user_by_email("alice@example.com")
    assert user is not None
    assert user["accountId"] == "right"


@pytest.mark.asyncio
@respx.mock
async def test_find_user_no_match() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),
    )
    user = await _driver().find_user_by_email("ghost@example.com")
    assert user is None


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


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_existing_user_added() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    add_route = respx.mock.post(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(201, json={"name": "jira-software-users"}),
    )
    result = await _driver().grant_seat(_ctx())
    assert result.ok is True
    assert result.external_ref == "acct-1"
    assert add_route.called
    # Group name went on the query string, accountId went in body.
    called = add_route.calls.last.request
    assert "groupname=jira-software-users" in str(called.url)
    assert b'"accountId"' in called.content
    assert b'"acct-1"' in called.content


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_invites_when_user_missing() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),
    )
    respx.mock.post(f"{_BASE}/user").mock(
        return_value=httpx.Response(201, json={
            "accountId": "acct-new", "emailAddress": "alice@example.com",
        }),
    )
    respx.mock.post(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(201, json={}),
    )
    result = await _driver().grant_seat(_ctx())
    assert result.ok is True
    assert result.external_ref == "acct-new"


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_already_member_text() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.post(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(
            400, text="user is already a member of the group",
        ),
    )
    result = await _driver().grant_seat(_ctx())
    assert result.ok is True
    assert result.raw_response["already_member"] is True


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_already_member_json() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.post(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(
            400,
            json={"errorMessages": ["The user is already a member of jira-software-users."]},
        ),
    )
    result = await _driver().grant_seat(_ctx())
    assert result.ok is True
    assert result.raw_response["already_member"] is True


@pytest.mark.asyncio
async def test_grant_seat_missing_group_name() -> None:
    ctx = ProvisioningContext(
        user_email="alice@example.com",
        extra={},  # no group_name
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is False
    assert "group_name" in (result.error_message or "")


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_group_404() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.post(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(404, text="Group not found"),
    )
    result = await _driver().grant_seat(_ctx())
    assert result.ok is False
    assert "not found" in (result.error_message or "").lower()


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


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_deactivate_ok() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    rm_route = respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    result = await _driver().revoke_seat(_ctx())
    assert result.ok is True
    assert result.external_ref == "acct-1"
    assert rm_route.called


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_user_already_gone() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),
    )
    result = await _driver().revoke_seat(_ctx())
    assert result.ok is True
    assert result.raw_response.get("already_absent") is True


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_falls_back_to_audit_account_id() -> None:
    """#1651: site user-search is GDPR-blind — a managed/deactivated
    leaver whose email the site omits resolves to None. Pre-fix
    ``revoke_seat`` then returned already_absent (a no-op "success")
    and the leaver kept occupying the paid seat. Post-fix it falls back
    to the authoritative ``account_id`` from ``ctx.extra`` (resolved by
    the offboarding audit via the Org Admin directory) and DELETEs the
    bound group against that account."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),  # email invisible
    )
    seen_accounts: list[str] = []

    def _del(request: httpx.Request) -> httpx.Response:
        seen_accounts.append(request.url.params.get("accountId", ""))
        return httpx.Response(204)

    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=_del)
    ctx = ProvisioningContext(
        user_email="ghost@example.com",
        user_display_name="Ghost",
        extra={
            "group_name": "jira-software-users",
            "action_on_offboard": "deactivate",
            "account_id": "acct-hidden-1",
        },
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is True
    # Real removal ran (not the already_absent no-op path).
    assert result.raw_response.get("already_absent") is not True
    assert del_route.called
    assert seen_accounts == ["acct-hidden-1"]


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_404_treated_idempotent() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(404, text="not in group"),
    )
    result = await _driver().revoke_seat(_ctx())
    assert result.ok is True
    assert result.raw_response.get("already_absent") is True


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_400_not_a_member_json_treated_idempotent() -> None:
    """Atlassian returns HTTP 400 with errorMessages=['Cannot remove user
    from group X since user is not a member of X'] when DELETE targets a
    non-member. Driver must treat this as already-absent so offboarding
    blockers close — observed live in P01458 (Jira eazyBI) revoke.
    """
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(
            400,
            json={"errorMessages": [
                "Cannot remove user from group 'jira-software-users' since "
                "user is not a member of 'jira-software-users'",
            ], "errors": {}},
        ),
    )
    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("removed") is False


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_400_not_a_member_text_treated_idempotent() -> None:
    """Same idempotent treatment when the body comes back as plain text
    instead of JSON (older Atlassian product variants)."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(
            400, text="user is not a member of jira-software-users",
        ),
    )
    result = await _driver().revoke_seat(_ctx())
    assert result.ok is True
    assert result.raw_response.get("already_absent") is True


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_400_unrelated_still_fails() -> None:
    """A 400 that is NOT the not-a-member message must still surface as
    a real failure — we only swallow the specific idempotent variant."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(
            400,
            json={"errorMessages": ["Group name is invalid"]},
        ),
    )
    result = await _driver().revoke_seat(_ctx())
    assert result.ok is False
    assert "HTTP 400" in (result.error_message or "")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_400_not_a_member_still_removes_access() -> None:
    """For action_on_offboard='delete', a 400 not-a-member on the group
    DELETE must NOT abort the flow — the driver still has to call org
    remove-access so the account's product access is stripped."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(
            400,
            json={"errorMessages": [
                "Cannot remove user from group 'jira-software-users' "
                "since user is not a member of 'jira-software-users'",
            ]},
        ),
    )
    remove_route = respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(204),
    )
    result = await _driver(_CONFIG_WITH_UNSCOPED).revoke_seat(_ctx(action="delete"))
    assert result.ok is True
    assert result.raw_response.get("already_absent_from_group") is True
    assert result.raw_response.get("removed_from_group") is False
    assert result.raw_response.get("org_access_removed") is True
    assert remove_route.called


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_calls_remove_access() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    remove_route = respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(204),
    )
    result = await _driver(_CONFIG_WITH_UNSCOPED).revoke_seat(_ctx(action="delete"))
    assert result.ok is True
    assert result.raw_response.get("org_access_removed") is True
    assert remove_route.called


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_remove_access_404_already_absent() -> None:
    """A 404 from remove-access means the account already has no managed
    access — treated as success (already_absent), NOT a false 'removed'."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(404, json={"error": "not found"}),
    )
    result = await _driver(_CONFIG_WITH_UNSCOPED).revoke_seat(_ctx(action="delete"))
    assert result.ok is True
    assert result.raw_response.get("org_access_already_absent") is True
    assert result.raw_response.get("org_access_removed") is False


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_missing_org_id() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    result = await _driver().revoke_seat(_ctx(action="delete"))
    assert result.ok is False
    assert "org_id" in (result.error_message or "")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_missing_unscoped_key() -> None:
    """org_id alone isn't enough for the delete path — the driver must
    NOT fall back to api_token (a different credential type that would
    401). Surface a precise error pointing at org_unscoped_api_key."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    result = await _driver(_CONFIG_WITH_ORG_ID_NO_KEY).revoke_seat(
        _ctx(action="delete"),
    )
    assert result.ok is False
    assert "org_unscoped_api_key" in (result.error_message or "")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_uses_unscoped_key_not_api_token() -> None:
    """The remove-access call must Bearer-auth with the unscoped org key,
    not the user api_token. Inspect the actual request header."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    remove_route = respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(204),
    )
    result = await _driver(_CONFIG_WITH_UNSCOPED).revoke_seat(_ctx(action="delete"))
    assert result.ok is True
    assert remove_route.called
    sent_auth = remove_route.calls.last.request.headers.get("authorization", "")
    assert sent_auth == "Bearer ATCTT-unscoped-key"
    assert "ATCTT-test-token" not in sent_auth  # api_token must not leak here


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_scoped_admin_key_alone_is_rejected() -> None:
    """The scoped org_admin_api_key must NOT be accepted as a write
    credential — no fallback. A config with only org_admin_api_key set
    (the documented read-only key) and action=delete fails the
    precondition loudly instead of firing remove-access with a scoped
    key that would 401. This is the footgun the two-key split prevents."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    remove_route = respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(204),
    )
    result = await _driver(_CONFIG_WITH_ORG).revoke_seat(_ctx(action="delete"))
    assert result.ok is False
    assert "org_unscoped_api_key" in (result.error_message or "")
    # Never even attempted the write with the scoped key.
    assert not remove_route.called


@pytest.mark.asyncio
async def test_revoke_seat_invalid_action() -> None:
    result = await _driver().revoke_seat(_ctx(action="purge"))
    assert result.ok is False
    assert "action_on_offboard" in (result.error_message or "")


# ── Offboarding hardening: remove-access 403/401 partial success ─────


@pytest.mark.asyncio
@respx.mock
async def test_revoke_delete_remove_access_403_surfaces_partial_success() -> None:
    """remove-access 403 (key still scoped / lacks org admin) MUST
    surface as ok=True with the group-removal already counted as success
    and a JIRA_ORG_ADMIN_PERMISSION_DENIED tag — otherwise the
    offboarding blocker hangs forever even though the user IS off the
    licensed group seat."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),  # group removed OK
    )
    respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(403, text="insufficient scope"),
    )
    result = await _driver(_CONFIG_WITH_UNSCOPED).revoke_seat(_ctx(action="delete"))
    assert result.ok is True
    assert result.raw_response["removed_from_group"] is True
    assert result.raw_response["org_access_removed"] is False
    assert result.raw_response["org_access_remove_deferred"] is True
    assert (
        result.raw_response["partial_success_tag"]
        == JIRA_TAG_ORG_ADMIN_PERMISSION_DENIED
    )
    assert result.raw_response["remove_access_status"] == 403


@pytest.mark.asyncio
@respx.mock
async def test_revoke_delete_remove_access_401_surfaces_partial_success() -> None:
    """401 means the key was rotated / revoked. Same partial-success
    semantics, different tag so failure-DM gives a different hint."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(401, text="auth"),
    )
    result = await _driver(_CONFIG_WITH_UNSCOPED).revoke_seat(_ctx(action="delete"))
    assert result.ok is True
    assert (
        result.raw_response["partial_success_tag"]
        == JIRA_TAG_ORG_ADMIN_AUTH_FAILED
    )


@pytest.mark.asyncio
@respx.mock
async def test_revoke_delete_remove_access_5xx_preserves_group_removal_state() -> None:
    """A 500 is NOT a partial-success — it's a true failure (could be
    transient, deserves a retry → ok=False). Group-removal state must
    still be in raw_response for audit forensic queries."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(200),
    )
    respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(500, text="upstream"),
    )
    result = await _driver(_CONFIG_WITH_UNSCOPED).revoke_seat(_ctx(action="delete"))
    assert result.ok is False
    assert result.raw_response["removed_from_group"] is True
    assert result.raw_response["remove_access_status"] == 500


@pytest.mark.asyncio
@respx.mock
async def test_remove_org_user_access_204_returns_removed() -> None:
    route = respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(204),
    )
    res = await _driver(_CONFIG_WITH_UNSCOPED).remove_org_user_access("acct-1")
    assert res == {
        "ok": True, "status": 204, "removed": True,
        "already_absent": False, "error": None,
    }
    assert route.called


@pytest.mark.asyncio
@respx.mock
async def test_remove_org_user_access_404_already_absent() -> None:
    respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(404),
    )
    res = await _driver(_CONFIG_WITH_UNSCOPED).remove_org_user_access("acct-1")
    assert res["ok"] is True
    assert res["already_absent"] is True
    assert res["removed"] is False


@pytest.mark.asyncio
@respx.mock
async def test_remove_org_user_access_route_404_is_hard_error() -> None:
    """A 404 whose body says the request matched no route means a wrong
    endpoint / org_id — must be a HARD error, NOT 'already absent'. This
    is the silent-false-positive class the PR killed for lifecycle."""
    respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(
            404, json={"message": "Request failed to match any route"},
        ),
    )
    res = await _driver(_CONFIG_WITH_UNSCOPED).remove_org_user_access("acct-1")
    assert res["ok"] is False
    assert res["already_absent"] is False
    assert res["status"] == 404


@pytest.mark.asyncio
@respx.mock
async def test_remove_org_user_access_403_not_ok() -> None:
    respx.mock.delete(_REMOVE_ACCESS_URL).mock(
        return_value=httpx.Response(403, text="insufficient scope"),
    )
    res = await _driver(_CONFIG_WITH_UNSCOPED).remove_org_user_access("acct-1")
    assert res["ok"] is False
    assert res["status"] == 403


@pytest.mark.asyncio
async def test_remove_org_user_access_missing_org_id_raises() -> None:
    with pytest.raises(ValueError, match="org_id"):
        await _driver({**_CONFIG, "org_unscoped_api_key": "k"}).remove_org_user_access(
            "acct-1",
        )


@pytest.mark.asyncio
async def test_remove_org_user_access_missing_key_raises() -> None:
    with pytest.raises(ValueError, match="org_unscoped_api_key"):
        await _driver(_CONFIG_WITH_ORG_ID_NO_KEY).remove_org_user_access("acct-1")


@pytest.mark.asyncio
async def test_revoke_seat_missing_action_no_silent_default() -> None:
    """No silent fallback to 'deactivate' — missing action_on_offboard
    must surface as an explicit error so the operator notices and fixes
    the binding rather than getting a behaviour they didn't choose."""
    ctx = ProvisioningContext(
        user_email="alice@example.com",
        extra={"group_name": "jira-software-users"},  # no action_on_offboard
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "missing action_on_offboard" in (result.error_message or "")


@pytest.mark.asyncio
async def test_revoke_seat_blank_action_no_silent_default() -> None:
    ctx = ProvisioningContext(
        user_email="alice@example.com",
        extra={"group_name": "jira-software-users", "action_on_offboard": "  "},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "missing action_on_offboard" in (result.error_message or "")


@pytest.mark.asyncio
async def test_grant_seat_empty_email_rejected() -> None:
    ctx = ProvisioningContext(
        user_email="",
        extra={"group_name": "jira-software-users",
               "action_on_offboard": "deactivate"},
    )
    result = await _driver().grant_seat(ctx)
    assert result.ok is False
    assert "empty user_email" in (result.error_message or "")


@pytest.mark.asyncio
async def test_revoke_seat_empty_email_rejected() -> None:
    """Empty email must NOT silently succeed as 'already absent' —
    that would close an offboarding blocker without ever talking to
    Jira and leave the leaver with active access."""
    ctx = ProvisioningContext(
        user_email="   ",
        extra={"group_name": "jira-software-users",
               "action_on_offboard": "deactivate"},
    )
    result = await _driver().revoke_seat(ctx)
    assert result.ok is False
    assert "empty user_email" in (result.error_message or "")


# ── list_group_members ──────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_list_group_members_paginates() -> None:
    page1 = {"isLast": False, "values": [
        {"accountId": "u1", "emailAddress": "u1@example.com"} for _ in range(50)
    ]}
    page2 = {"isLast": True, "values": [
        {"accountId": "u51", "emailAddress": "u51@example.com"},
    ]}
    seen: list[int] = []

    def _handler(request: httpx.Request) -> httpx.Response:
        start = int(request.url.params.get("startAt", "0"))
        seen.append(start)
        return httpx.Response(200, json=page1 if start == 0 else page2)

    respx.mock.get(f"{_BASE}/group/member").mock(side_effect=_handler)
    out = await _driver().list_group_members("jira-software-users")
    assert len(out) == 51
    assert seen == [0, 50]


@pytest.mark.asyncio
@respx.mock
async def test_list_group_members_404_raises() -> None:
    respx.mock.get(f"{_BASE}/group/member").mock(
        return_value=httpx.Response(404, text="No group named foo"),
    )
    with pytest.raises(RuntimeError, match="not found"):
        await _driver().list_group_members("ghost-group")


# ── revoke_all_access (offboarding: remove from ALL groups) ──────────
#
# The correct offboarding action per PR #532's policy: never deactivate
# the account — strip the leaver from EVERY group they belong to (covers
# access from untracked groups like jira-administrators / SOC2 that a
# single bound-group revoke would miss). Uses get_user_groups (site REST,
# Basic) + DELETE /group/user per group.


def _acct_ctx(email: str = "alice@example.com") -> ProvisioningContext:
    return ProvisioningContext(user_email=email, user_display_name="Alice")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_all_access_removes_every_group() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    # The leaver is in three groups across products incl. an admin group
    # that no single bound-group revoke would ever cover.
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-software-users", "groupId": "g1"},
            {"name": "jira-administrators", "groupId": "g2"},
            {"name": "SOC2", "groupId": "g3"},
        ]),
    )
    seen_groups: list[str] = []

    def _del(request: httpx.Request) -> httpx.Response:
        seen_groups.append(request.url.params.get("groupname", ""))
        return httpx.Response(204)

    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=_del)
    result = await _driver().revoke_all_access(_acct_ctx())
    assert result.ok is True
    assert result.external_ref == "acct-1"
    assert del_route.call_count == 3
    assert set(seen_groups) == {
        "jira-software-users", "jira-administrators", "SOC2",
    }
    assert set(result.raw_response["removed"]) == {
        "jira-software-users", "jira-administrators", "SOC2",
    }


@pytest.mark.asyncio
@respx.mock
async def test_revoke_all_access_no_account_is_idempotent_success() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),
    )
    result = await _driver().revoke_all_access(_acct_ctx())
    assert result.ok is True
    assert result.raw_response.get("no_account") is True


@pytest.mark.asyncio
@respx.mock
async def test_revoke_all_access_falls_back_to_audit_account_id() -> None:
    """GDPR-blind site search returns no match for a managed / deactivated
    leaver, but the offboarding audit supplied the authoritative
    account_id via ``ctx.extra`` — group removal must run by account_id
    instead of silently no-op'ing as ``no_account`` (the bug behind the
    perpetual false 'persistent bounce' on yudingyi / lifu.wu / wangdi)."""
    # Site search can't see the account by email (emailAddress hidden).
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-software-users", "groupId": "g1"},
            {"name": "SOC2", "groupId": "g2"},
        ]),
    )
    seen_groups: list[str] = []
    seen_accounts: list[str] = []

    def _del(request: httpx.Request) -> httpx.Response:
        seen_groups.append(request.url.params.get("groupname", ""))
        seen_accounts.append(request.url.params.get("accountId", ""))
        return httpx.Response(204)

    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=_del)
    ctx = ProvisioningContext(
        user_email="ghost@example.com",
        user_display_name="Ghost",
        extra={"account_id": "acct-hidden-1"},
    )
    result = await _driver().revoke_all_access(ctx)
    assert result.ok is True
    assert result.external_ref == "acct-hidden-1"
    assert del_route.call_count == 2
    assert set(seen_groups) == {"jira-software-users", "SOC2"}
    # Every DELETE targets the audit-supplied account_id, not the
    # (unresolvable) email.
    assert set(seen_accounts) == {"acct-hidden-1"}


@pytest.mark.asyncio
@respx.mock
async def test_revoke_all_access_no_account_and_no_hint_is_noop() -> None:
    """Email truly has no account AND no account_id hint → still the
    idempotent no_account success (don't regress the genuine-absent path)."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),
    )
    result = await _driver().revoke_all_access(_acct_ctx())
    assert result.ok is True
    assert result.raw_response.get("no_account") is True


@pytest.mark.asyncio
@respx.mock
async def test_revoke_all_access_already_absent_is_idempotent() -> None:
    """Re-run after a prior revoke: get_user_groups still lists a group
    (eventual consistency) but the DELETE comes back not-a-member / 404 —
    must count as success, not failure, so the blocker closes."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[{"name": "jira-software-users"}]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(
            400, json={"errorMessages": [
                "Cannot remove user from group 'jira-software-users' since "
                "user is not a member of 'jira-software-users'",
            ]},
        ),
    )
    result = await _driver().revoke_all_access(_acct_ctx())
    assert result.ok is True
    assert result.raw_response["already_absent"] == ["jira-software-users"]
    assert result.raw_response["removed"] == []


@pytest.mark.asyncio
@respx.mock
async def test_revoke_all_access_no_groups_is_clean_success() -> None:
    """Leaver with an account but already in zero groups → trivially
    done, no DELETE calls."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[]),
    )
    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(204),
    )
    result = await _driver().revoke_all_access(_acct_ctx())
    assert result.ok is True
    assert result.raw_response["groups_total"] == 0
    assert not del_route.called


@pytest.mark.asyncio
@respx.mock
async def test_revoke_all_access_partial_failure_is_ok_false() -> None:
    """If a real removal fails (5xx), the result is ok=False (blocker
    stays open / leaver stays visible) but still reports which groups
    WERE removed for the forensic trail."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-software-users"},
            {"name": "locked-group"},
        ]),
    )

    def _del(request: httpx.Request) -> httpx.Response:
        if request.url.params.get("groupname") == "locked-group":
            return httpx.Response(500, text="upstream error")
        return httpx.Response(204)

    respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=_del)
    result = await _driver().revoke_all_access(_acct_ctx())
    assert result.ok is False
    assert result.raw_response["removed"] == ["jira-software-users"]
    assert any("locked-group" in f for f in result.raw_response["failed"])


# ──────────────────────────────────────────────────────────────────
# revoke_product_access — active-employee single-product cancel
# ──────────────────────────────────────────────────────────────────
#
# This is the path that fixed the shengbo.ma / sunxiaoguang bounce:
# their Jira access came from team / role conferring groups
# (``jira-dev-engineers`` / ``RD Leader``), not the connector-bound
# ``jira-software-users``. ``revoke_seat`` removed them from
# ``jira-software-users`` (a no-op for them since they weren't in it,
# returning a 400 "not a member" which is treated as idempotent
# success) and the next ``jira.seat_sync`` re-imported the seat
# because ``product_access`` still showed them as a jira-software
# holder. ``revoke_product_access`` enumerates Atlassian's actual
# conferring groups via ``/rest/api/3/applicationrole``, intersects
# with the user's current groups, and DELETEs that set. Identity
# groups (``employee_in_china`` / ``All users from G Suite``) and
# unrelated product accesses (Confluence, JSM) stay untouched.


def _pa_ctx(
    *,
    email: str = "alice@example.com",
    product_key: str = "jira-software",
    sites: list[str] | None = None,
) -> ProvisioningContext:
    return ProvisioningContext(
        user_email=email,
        user_display_name="Alice",
        extra={"product_key": product_key, "sites": sites or ["tidb.atlassian.net"]},
    )


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_intersects_user_with_conferring() -> None:
    """The shengbo.ma scenario: user holds Jira via team groups,
    not via the default ``jira-software-users``. ``revoke_product_access``
    removes them only from the conferring groups they're actually in
    — leaves identity groups alone.
    """
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-shengbo", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            {
                "key": "jira-software",
                "groups": ["jira-software-users", "jira-dev-engineers", "RD Leader"],
                "defaultGroups": ["jira-software-users"],
            },
            {
                "key": "confluence",
                "groups": ["confluence-users"],
                "defaultGroups": ["confluence-users"],
            },
        ]),
    )
    # User's real groups: 2 conferring (jira-dev-engineers, RD Leader)
    # + 2 identity (employee_in_china, All users from G Suite) — the
    # default jira-software-users isn't among them.
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-dev-engineers"},
            {"name": "RD Leader"},
            {"name": "employee_in_china"},
            {"name": "All users from G Suite"},
        ]),
    )

    deleted: list[str] = []

    def _del(request: httpx.Request) -> httpx.Response:
        deleted.append(request.url.params.get("groupname", ""))
        return httpx.Response(204)

    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=_del)
    result = await _driver().revoke_product_access(_pa_ctx())

    assert result.ok is True
    assert result.external_ref == "acct-shengbo"
    # Only the 2 conferring groups the user is actually in get deleted.
    assert del_route.call_count == 2
    assert set(deleted) == {"jira-dev-engineers", "RD Leader"}
    # Identity groups untouched — verifies the surgical precision.
    assert "employee_in_china" not in deleted
    assert "All users from G Suite" not in deleted
    raw = result.raw_response
    assert raw["product_key"] == "jira-software"
    assert raw["conferring_groups_total"] == 3
    assert raw["user_groups_total"] == 4
    assert sorted(raw["targets"]) == ["RD Leader", "jira-dev-engineers"]
    assert sorted(raw["removed"]) == ["RD Leader", "jira-dev-engineers"]


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_no_intersection_is_clean_success() -> None:
    """User has an Atlassian account but isn't in ANY conferring group
    for the target product (already revoked by hand, or never had
    access). Result is ok=True with empty removed — no API calls, no
    failure surface to the seat-cancel task."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[{
            "key": "jira-software",
            "groups": ["jira-software-users"],
            "defaultGroups": ["jira-software-users"],
        }]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "employee_in_china"},
            {"name": "confluence-users"},
        ]),
    )
    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(204),
    )
    result = await _driver().revoke_product_access(_pa_ctx())

    assert result.ok is True
    assert not del_route.called
    assert result.raw_response["removed"] == []
    assert result.raw_response["targets"] == []


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_no_account_is_idempotent_success() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),
    )
    result = await _driver().revoke_product_access(_pa_ctx())
    assert result.ok is True
    assert result.raw_response.get("no_account") is True
    assert result.raw_response["product_key"] == "jira-software"


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_falls_back_to_audit_account_id() -> None:
    """#1651: same GDPR-blind fallback for ``revoke_product_access``.
    Site search can't see the leaver by email, but ``ctx.extra`` carries
    the audit-resolved ``account_id`` — the product-group DELETEs must
    run against that hint account instead of silently no-op'ing as
    ``no_account`` while the seat stays occupied."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[]),  # email invisible
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[{
            "key": "jira-software",
            "groups": ["jira-software-users", "jira-dev-engineers"],
            "defaultGroups": ["jira-software-users"],
        }]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-dev-engineers"},
        ]),
    )
    seen_accounts: list[str] = []
    seen_groups: list[str] = []

    def _del(request: httpx.Request) -> httpx.Response:
        seen_accounts.append(request.url.params.get("accountId", ""))
        seen_groups.append(request.url.params.get("groupname", ""))
        return httpx.Response(204)

    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=_del)
    ctx = ProvisioningContext(
        user_email="ghost@example.com",
        user_display_name="Ghost",
        extra={"product_key": "jira-software", "account_id": "acct-hidden-1"},
    )
    result = await _driver().revoke_product_access(ctx)
    assert result.ok is True
    assert result.raw_response.get("no_account") is not True
    assert del_route.called
    assert seen_accounts == ["acct-hidden-1"]
    assert seen_groups == ["jira-dev-engineers"]


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_unknown_product_refuses() -> None:
    """``applicationrole`` returns no role matching the binding's
    ``product_key`` → refuse loudly. Silently revoking 0 groups would
    be the same no-op the bug-fix was written to prevent."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            # No "jira-software" entry — site has only Confluence.
            {"key": "confluence", "groups": ["confluence-users"],
             "defaultGroups": ["confluence-users"]},
        ]),
    )
    result = await _driver().revoke_product_access(_pa_ctx())
    assert result.ok is False
    assert "no conferring groups" in (result.error_message or "")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_empty_email_rejected() -> None:
    result = await _driver().revoke_product_access(
        ProvisioningContext(
            user_email="",
            extra={"product_key": "jira-software"},
        ),
    )
    assert result.ok is False
    assert "empty user_email" in (result.error_message or "")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_missing_product_key_rejected() -> None:
    """Dispatcher MUST set product_key in ctx.extra. Without it,
    refuse rather than infer a default (would route the revoke at a
    wrong product)."""
    result = await _driver().revoke_product_access(
        ProvisioningContext(user_email="alice@example.com", extra={}),
    )
    assert result.ok is False
    assert "product_key" in (result.error_message or "")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_partial_failure_is_ok_false() -> None:
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[{
            "key": "jira-software",
            "groups": ["jira-software-users", "jira-dev-engineers"],
            "defaultGroups": ["jira-software-users"],
        }]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-software-users"},
            {"name": "jira-dev-engineers"},
        ]),
    )

    def _del(request: httpx.Request) -> httpx.Response:
        if request.url.params.get("groupname") == "jira-dev-engineers":
            return httpx.Response(500, text="upstream error")
        return httpx.Response(204)

    respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=_del)
    result = await _driver().revoke_product_access(_pa_ctx())

    assert result.ok is False
    assert result.raw_response["removed"] == ["jira-software-users"]
    assert "failed" in result.raw_response
    assert any("jira-dev-engineers" in f for f in result.raw_response["failed"])


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_idempotent_not_a_member_400() -> None:
    """Atlassian's "user is not a member" 400 is treated as
    already_absent, same as 404 — keeps the operation idempotent so
    re-run after a successful first revoke counts as success.
    """
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[{
            "key": "jira-software",
            "groups": ["jira-software-users"],
            "defaultGroups": ["jira-software-users"],
        }]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-software-users"},
        ]),
    )
    respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(400, json={"errorMessages": [
            "Cannot remove user from group since user is not a member of "
            "'jira-software-users'",
        ]}),
    )
    result = await _driver().revoke_product_access(_pa_ctx())
    assert result.ok is True
    assert result.raw_response["already_absent"] == ["jira-software-users"]
    assert result.raw_response["removed"] == []


# ──────────────────────────────────────────────────────────────────
# list_conferring_groups — raw applicationrole helper
# ──────────────────────────────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_list_conferring_groups_unions_groups_and_default_groups() -> None:
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            {
                "key": "jira-software",
                "groups": ["jira-software-users", "jira-dev-engineers"],
                "defaultGroups": ["jira-software-users"],
            },
        ]),
    )
    got = await _driver().list_conferring_groups("jira-software")
    assert got == frozenset({"jira-software-users", "jira-dev-engineers"})


@pytest.mark.asyncio
@respx.mock
async def test_list_conferring_groups_supports_details_shape() -> None:
    """Atlassian sometimes returns ``defaultGroupsDetails`` with
    ``[{groupId, name, self}]`` — the helper should accept both string
    and object shapes."""
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            {
                "key": "confluence",
                "groups": [{"groupId": "g1", "name": "confluence-users"}],
                "defaultGroups": [{"groupId": "g1", "name": "confluence-users"}],
            },
        ]),
    )
    got = await _driver().list_conferring_groups("confluence")
    assert got == frozenset({"confluence-users"})


@pytest.mark.asyncio
@respx.mock
async def test_list_conferring_groups_unknown_key_returns_empty() -> None:
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            {"key": "confluence", "groups": ["confluence-users"], "defaultGroups": []},
        ]),
    )
    got = await _driver().list_conferring_groups("jira-software")
    assert got == frozenset()


@pytest.mark.asyncio
@respx.mock
async def test_list_conferring_groups_empty_key_raises() -> None:
    with pytest.raises(RuntimeError, match="without a product_key"):
        await _driver().list_conferring_groups("")


@pytest.mark.asyncio
@respx.mock
async def test_list_conferring_groups_5xx_raises() -> None:
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(500, text="upstream error"),
    )
    with pytest.raises(RuntimeError, match="applicationrole HTTP 500"):
        await _driver().list_conferring_groups("jira-software")


@pytest.mark.asyncio
@respx.mock
async def test_list_conferring_groups_case_normalisation() -> None:
    """``product_key`` matching is case-insensitive (Atlassian returns
    lower-cased ``key``, but binding_config may be stored with mixed
    case). Don't false-miss because of casing."""
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            {"key": "jira-software", "groups": ["jira-software-users"], "defaultGroups": []},
        ]),
    )
    got = await _driver().list_conferring_groups("JIRA-SOFTWARE")
    assert got == frozenset({"jira-software-users"})


# ──────────────────────────────────────────────────────────────────
# revoke_product_access — admin-declared fallback (Confluence path)
# ──────────────────────────────────────────────────────────────────
#
# applicationrole is Jira-only — Confluence has no equivalent endpoint
# on the same site. Without a fallback, the 2026-05-28 PR refused
# Confluence revokes loudly (sunxiaoguang's Confluence task hit this).
# The 2026-05-29 follow-up adds an admin-declared fallback path: the
# dispatcher loads
# ``software_seat.product_access_revoke.admin_declared_conferring_groups``
# from platform_settings, picks the entry for the task's product_key,
# and passes the list as ``ctx.extra['admin_declared_conferring_groups']``.
# These tests exercise the driver-side behavior; the dispatcher's
# setting-loading is covered by an integration test in
# test_seat_revoke_product_access_2026_05_28.py.


def _pa_ctx_with_fallback(
    *,
    email: str = "alice@example.com",
    product_key: str = "confluence",
    fallback: list[str],
) -> ProvisioningContext:
    return ProvisioningContext(
        user_email=email,
        user_display_name="Alice",
        extra={
            "product_key": product_key,
            "admin_declared_conferring_groups": fallback,
        },
    )


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_fallback_used_when_applicationrole_empty() -> None:
    """Confluence isn't in applicationrole's response. Admin populated
    the platform_setting with the actual conferring groups; driver
    intersects with user's groups and DELETEs as usual."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-confluence", "emailAddress": "alice@example.com"},
        ]),
    )
    # applicationrole has NO confluence entry on this Jira-only site.
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            {"key": "jira-software", "groups": ["jira-software-users"], "defaultGroups": []},
        ]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "confluence-users"},
            {"name": "rd-leader"},
            {"name": "employee_in_china"},
        ]),
    )

    deleted: list[str] = []

    def _del(request: httpx.Request) -> httpx.Response:
        deleted.append(request.url.params.get("groupname", ""))
        return httpx.Response(204)

    respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=_del)

    result = await _driver().revoke_product_access(_pa_ctx_with_fallback(
        fallback=["confluence-users", "rd-leader", "ignored-not-a-member"],
    ))

    assert result.ok is True
    # Only groups the user is actually in get hit.
    assert set(deleted) == {"confluence-users", "rd-leader"}
    # Identity group untouched.
    assert "employee_in_china" not in deleted
    raw = result.raw_response
    assert raw["product_key"] == "confluence"
    # Intersection size reflects the user's group membership, not the
    # raw fallback set.
    assert set(raw["targets"]) == {"confluence-users", "rd-leader"}


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_fallback_merged_with_applicationrole() -> None:
    """When BOTH applicationrole AND the admin fallback have entries,
    the driver unions them (an admin who added an extra custom-group
    that applicationrole misses must not cause it to be dropped).
    """
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-merge", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            {
                "key": "jira-software",
                "groups": ["jira-software-users"],
                "defaultGroups": ["jira-software-users"],
            },
        ]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-software-users"},
            {"name": "custom-jira-grandfather"},  # not in applicationrole
        ]),
    )
    deleted: list[str] = []
    respx.mock.delete(f"{_BASE}/group/user").mock(
        side_effect=lambda r: (deleted.append(r.url.params["groupname"]) or httpx.Response(204))
    )

    result = await _driver().revoke_product_access(_pa_ctx_with_fallback(
        product_key="jira-software",
        fallback=["custom-jira-grandfather"],
    ))
    assert result.ok is True
    # Union: applicationrole + admin extras, both intersected with user's groups.
    assert set(deleted) == {"jira-software-users", "custom-jira-grandfather"}


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_refuses_with_no_applicationrole_and_no_fallback() -> None:
    """The original 2026-05-28 PR's safety: refuse loudly when no
    target set can be computed. With the 2026-05-29 fallback path,
    refusal now covers ``applicationrole empty AND no admin fallback``
    — error message points the admin at the platform_setting key."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[
            {"key": "jira-software", "groups": ["jira-software-users"], "defaultGroups": []},
        ]),
    )
    result = await _driver().revoke_product_access(
        ProvisioningContext(
            user_email="alice@example.com",
            extra={"product_key": "confluence"},  # no fallback in extra
        ),
    )
    assert result.ok is False
    assert "admin_declared_conferring_groups" in (result.error_message or "")


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_uses_fallback_when_applicationrole_raises() -> None:
    """applicationrole 5xx must not block the revoke when the admin
    fallback is configured — the fallback exists precisely for
    Atlassian API instability + endpoint coverage gaps."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(500, text="upstream"),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "confluence-users"},
        ]),
    )
    deleted: list[str] = []
    respx.mock.delete(f"{_BASE}/group/user").mock(
        side_effect=lambda r: (deleted.append(r.url.params["groupname"]) or httpx.Response(204))
    )
    result = await _driver().revoke_product_access(_pa_ctx_with_fallback(
        fallback=["confluence-users"],
    ))
    assert result.ok is True
    assert set(deleted) == {"confluence-users"}


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_fallback_malformed_entries_skipped() -> None:
    """Defensive: non-string entries in the fallback list (admin typo'd
    JSON / null) are silently dropped. An entry of "" / "   " is dropped
    via strip-then-truthy. Stays robust against admin JSON mistakes
    without surfacing a low-value 500."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "confluence-users"},
        ]),
    )
    deleted: list[str] = []
    respx.mock.delete(f"{_BASE}/group/user").mock(
        side_effect=lambda r: (deleted.append(r.url.params["groupname"]) or httpx.Response(204))
    )
    result = await _driver().revoke_product_access(_pa_ctx_with_fallback(
        fallback=["confluence-users", "", "   ", 42, None],  # type: ignore[list-item]
    ))
    assert result.ok is True
    assert set(deleted) == {"confluence-users"}


# ── Org-wide-entitlement guard (2026-06-15) ──────────────────────
#
# Confluence at PingCAP is conferred org-wide via identity groups
# (``employee_in_china`` / ``All users from G Suite``) that SCIM keeps
# re-asserting. ``revoke_product_access`` must NEVER strip those groups
# to reclaim a seat — it would rip an active employee out of a core
# directory group and the next sync bounces the product right back.
# Such a seat is reported ``not_individually_revocable`` so the
# reclaim / audit pipeline stops chasing it (the 13-employee weekly
# bounce-page incident).


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_org_wide_via_default_protected_group() -> None:
    """The prod incident: admin-declared confluence conferring list
    includes ``employee_in_china`` (a built-in protected identity
    group). The user holds confluence via it → refuse to remove ANY
    group (even the genuine ``confluence-users``), return
    ``not_individually_revocable``. Nothing is deleted."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-orgwide", "emailAddress": "alice@example.com"},
        ]),
    )
    # Confluence not in applicationrole (Jira-only endpoint).
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[]),
    )
    # User is in the org-wide identity group AND the real confluence
    # group — mirrors prod (employee_in_china confers confluence).
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "confluence-users"},
            {"name": "employee_in_china"},
        ]),
    )
    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(204),
    )
    # Admin-declared list mirrors prod exactly.
    result = await _driver().revoke_product_access(_pa_ctx_with_fallback(
        fallback=["confluence-users", "All users from G Suite", "employee_in_china"],
    ))

    assert result.ok is True
    assert result.external_ref == "acct-orgwide"
    # CRITICAL: nothing removed — not even the genuine confluence-users,
    # because the org-wide group keeps conferring it anyway.
    assert not del_route.called
    raw = result.raw_response
    assert raw["not_individually_revocable"] is True
    assert raw["org_wide_entitlement"] is True
    assert raw["protected_conferring_groups"] == ["employee_in_china"]
    assert raw["removed"] == []
    assert raw["targets"] == []


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_protected_group_from_platform_setting() -> None:
    """A site-specific "everyone" group not in the built-in defaults is
    protected when supplied via ``ctx.extra['protected_identity_groups']``
    (the platform_setting the dispatcher injects). Case-insensitive."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[]),
    )
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "confluence-users"},
            {"name": "All-Staff"},
        ]),
    )
    del_route = respx.mock.delete(f"{_BASE}/group/user").mock(
        return_value=httpx.Response(204),
    )
    ctx = ProvisioningContext(
        user_email="alice@example.com",
        user_display_name="Alice",
        extra={
            "product_key": "confluence",
            "admin_declared_conferring_groups": ["confluence-users", "All-Staff"],
            "protected_identity_groups": ["all-staff"],  # lowercase → matches
        },
    )
    result = await _driver().revoke_product_access(ctx)

    assert result.ok is True
    assert not del_route.called
    assert result.raw_response["not_individually_revocable"] is True
    assert result.raw_response["protected_conferring_groups"] == ["All-Staff"]


@pytest.mark.asyncio
@respx.mock
async def test_revoke_product_access_protected_group_not_conferring_is_ignored() -> None:
    """Regression guard for the surgical-precision case: a protected
    identity group the user is in but which does NOT confer the target
    product must NOT trigger the org-wide refusal — the genuine
    conferring groups are still removed normally."""
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    respx.mock.get(f"{_BASE}/applicationrole").mock(
        return_value=httpx.Response(200, json=[{
            "key": "jira-software",
            "groups": ["jira-software-users", "jira-dev-engineers"],
            "defaultGroups": ["jira-software-users"],
        }]),
    )
    # User is in employee_in_china (protected) but jira-software is NOT
    # conferred by it here — only jira-dev-engineers confers it.
    respx.mock.get(f"{_BASE}/user/groups").mock(
        return_value=httpx.Response(200, json=[
            {"name": "jira-dev-engineers"},
            {"name": "employee_in_china"},
        ]),
    )
    deleted: list[str] = []
    respx.mock.delete(f"{_BASE}/group/user").mock(
        side_effect=lambda r: (deleted.append(r.url.params["groupname"]) or httpx.Response(204))
    )
    result = await _driver().revoke_product_access(_pa_ctx())  # jira-software

    assert result.ok is True
    assert deleted == ["jira-dev-engineers"]
    assert "not_individually_revocable" not in result.raw_response
    assert "employee_in_china" not in deleted


# ── Org Admin /users transient-error retry (504 gateway timeout) ──
#
# The org-wide roster pull is the heaviest Admin API call and Atlassian's
# gateway sporadically 502/503/504s on large orgs (and 429s under rate
# limit). Those are transient and must be retried, not fail the whole
# seat-sync / idle-reclaim cron. 4xx (other than 429) stay terminal.

_ORG_USERS_URL = f"{_ADMIN_API_BASE}/orgs/org-uuid-1234/users"


@pytest.mark.asyncio
@respx.mock
async def test_list_org_users_retries_transient_504(monkeypatch) -> None:
    slept: list[float] = []

    async def _fake_sleep(secs: float) -> None:
        slept.append(secs)

    monkeypatch.setattr(
        "app.integrations.drivers.jira_software.asyncio.sleep", _fake_sleep,
    )

    responses = [
        httpx.Response(504, text="Gateway Timeout"),
        httpx.Response(502, text="Bad Gateway"),
        httpx.Response(200, json={"data": [{"account_id": "acct-1"}], "links": {}}),
    ]
    respx.mock.get(_ORG_USERS_URL).mock(side_effect=responses)

    users = await _driver(_CONFIG_WITH_ORG).list_org_users_with_activity()

    assert [u["account_id"] for u in users] == ["acct-1"]
    # Two transient hits → two backoff sleeps, exponential: 1s then 2s.
    assert slept == [1.0, 2.0]


@pytest.mark.asyncio
@respx.mock
async def test_list_org_users_honours_retry_after_on_429(monkeypatch) -> None:
    slept: list[float] = []

    async def _fake_sleep(secs: float) -> None:
        slept.append(secs)

    monkeypatch.setattr(
        "app.integrations.drivers.jira_software.asyncio.sleep", _fake_sleep,
    )

    respx.mock.get(_ORG_USERS_URL).mock(side_effect=[
        httpx.Response(429, headers={"Retry-After": "7"}, text="slow down"),
        httpx.Response(200, json={"data": [], "links": {}}),
    ])

    await _driver(_CONFIG_WITH_ORG).list_org_users_with_activity()

    assert slept == [7.0]


@pytest.mark.asyncio
@respx.mock
async def test_list_org_users_gives_up_after_budget(monkeypatch) -> None:
    monkeypatch.setattr(
        "app.integrations.drivers.jira_software.asyncio.sleep",
        _noop_sleep,
    )
    # Every attempt 504s — after the retry budget the transparent helper
    # hands the last 504 back to list_org_users_with_activity, whose
    # existing non-200 handling raises naming the status.
    respx.mock.get(_ORG_USERS_URL).mock(
        return_value=httpx.Response(504, text="Gateway Timeout"),
    )

    with pytest.raises(RuntimeError, match=r"Jira admin /users HTTP 504"):
        await _driver(_CONFIG_WITH_ORG).list_org_users_with_activity()


@pytest.mark.asyncio
@respx.mock
async def test_list_org_users_401_is_terminal_no_retry(monkeypatch) -> None:
    slept: list[float] = []

    async def _fake_sleep(secs: float) -> None:  # pragma: no cover - must not run
        slept.append(secs)

    monkeypatch.setattr(
        "app.integrations.drivers.jira_software.asyncio.sleep", _fake_sleep,
    )
    route = respx.mock.get(_ORG_USERS_URL).mock(
        return_value=httpx.Response(401, text="unauthorized"),
    )

    with pytest.raises(RuntimeError, match="HTTP 401"):
        await _driver(_CONFIG_WITH_ORG).list_org_users_with_activity()

    # No retries on a terminal auth failure.
    assert slept == []
    assert route.call_count == 1


async def _noop_sleep(secs: float) -> None:
    return None


# ── Transient retry on the site-level mutation / read paths ───────
#
# The transparent _request_with_retry layer wraps EVERY Atlassian call,
# not just the org /users pull. Group add/remove are idempotent and the
# add-to-group / invite POSTs already treat "already done" as success, so
# retrying a transient 429/502/503/504 on them is safe and converges.

@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_retries_transient_on_add_to_group(monkeypatch) -> None:
    monkeypatch.setattr(
        "app.integrations.drivers.jira_software.asyncio.sleep", _noop_sleep,
    )
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    # First add-to-group 503s, retry succeeds.
    respx.mock.post(f"{_BASE}/group/user").mock(side_effect=[
        httpx.Response(503, text="Service Unavailable"),
        httpx.Response(201, json={"name": "jira-software-users"}),
    ])
    result = await _driver().grant_seat(_ctx())
    assert result.ok is True
    assert result.external_ref == "acct-1"


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_retries_transient_on_delete(monkeypatch) -> None:
    monkeypatch.setattr(
        "app.integrations.drivers.jira_software.asyncio.sleep", _noop_sleep,
    )
    respx.mock.get(f"{_BASE}/user/search").mock(
        return_value=httpx.Response(200, json=[
            {"accountId": "acct-1", "emailAddress": "alice@example.com"},
        ]),
    )
    # First DELETE 502s (transient), retry returns 204 (removed).
    respx.mock.delete(f"{_BASE}/group/user").mock(side_effect=[
        httpx.Response(502, text="Bad Gateway"),
        httpx.Response(204),
    ])
    result = await _driver().revoke_seat(_ctx())
    assert result.ok is True


@pytest.mark.asyncio
@respx.mock
async def test_transient_transport_error_reraises_original(monkeypatch) -> None:
    """A transport error that persists past the budget is re-raised as
    the original httpx exception — the caller's existing
    ``except httpx.HTTPError`` clause then wraps it (here into the
    user/search RuntimeError), so behaviour is unchanged aside from the
    retries."""
    monkeypatch.setattr(
        "app.integrations.drivers.jira_software.asyncio.sleep", _noop_sleep,
    )
    respx.mock.get(f"{_BASE}/user/search").mock(
        side_effect=httpx.ConnectError("connection reset"),
    )
    with pytest.raises(RuntimeError, match="Jira user/search failed"):
        await _driver().find_user_by_email("alice@example.com")


@pytest.mark.asyncio
@respx.mock
async def test_non_transient_transport_error_not_retried(monkeypatch) -> None:
    """A non-transient httpx.HTTPError (e.g. TooManyRedirects) is
    propagated on the first attempt without burning the retry budget."""
    slept: list[float] = []

    async def _fake_sleep(secs: float) -> None:  # pragma: no cover
        slept.append(secs)

    monkeypatch.setattr(
        "app.integrations.drivers.jira_software.asyncio.sleep", _fake_sleep,
    )
    respx.mock.get(f"{_BASE}/user/search").mock(
        side_effect=httpx.TooManyRedirects("too many redirects"),
    )
    with pytest.raises(RuntimeError, match="Jira user/search failed"):
        await _driver().find_user_by_email("alice@example.com")
    assert slept == []