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

All HTTP is mocked via respx. The rewritten driver uses two tokens
(xoxb- bot + optional xoxp- user token) and Slack SCIM v1 (not v2).

Covers:
  - test_connection: bot-token-only + with SCIM probe + plus_teams_only
  - list_full_members: pagination, filters (guest, bot, deleted)
  - grant_seat: create success, duplicate (409) reactivation,
    already-active noop, plus_teams_only plan gate
  - revoke_seat: deactivate, delete, already-absent, invalid action,
    plus_teams_only
  - SCIM not available when no user_token configured
"""
from __future__ import annotations

import httpx
import pytest
import respx

from app.integrations.driver_base import ProvisioningContext
from app.integrations.drivers.slack_software import (
    _DEFAULT_SCIM_BASE,
    _SLACK_WEB_BASE,
    SLACK_TAG_RATE_LIMITED,
    SlackSoftwareDriver,
)

_CONFIG_FULL = {
    "bot_token": "xoxb-test-bot",
    "user_token": "xoxp-test-user",
}
_CONFIG_BOT_ONLY = {
    "bot_token": "xoxb-test-bot",
}


def _driver(cfg: dict | None = None) -> SlackSoftwareDriver:
    return SlackSoftwareDriver(
        integration_id="slack-test",
        config=cfg or _CONFIG_FULL,
    )


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


def test_metadata_registered() -> None:
    meta = SlackSoftwareDriver.metadata()
    assert meta.kind == "software_provisioning"
    assert meta.provider == "slack"
    assert any(f.key == "bot_token" and f.is_secret and f.required for f in meta.fields)
    assert any(f.key == "user_token" and f.is_secret and not f.required for f in meta.fields)
    exec_field = next(f for f in meta.binding_fields if f.key == "execution_mode")
    assert exec_field.default == "manual_notify"
    assert {"value": "auto_api", "label": "SCIM 自动 (Business+ / Enterprise Grid 才可用)"} in (exec_field.options or [])


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


@pytest.mark.asyncio
@respx.mock
async def test_connection_bot_only_ok() -> None:
    """Bot-only config reports SCIM as 'no user_token' without
    making a SCIM network call."""
    respx.mock.post(f"{_SLACK_WEB_BASE}/auth.test").mock(
        return_value=httpx.Response(200, json={
            "ok": True, "team": "PingCAP", "url": "https://pingcap.slack.com/",
            "is_enterprise_install": False,
        }),
    )
    result = await _driver(_CONFIG_BOT_ONLY).test_connection()
    assert result.ok is True
    assert "team=PingCAP" in result.message
    assert "no user_token" in result.message
    assert result.detail["is_enterprise_install"] is False


@pytest.mark.asyncio
@respx.mock
async def test_connection_bot_and_scim_ok() -> None:
    respx.mock.post(f"{_SLACK_WEB_BASE}/auth.test").mock(
        return_value=httpx.Response(200, json={
            "ok": True, "team": "PingCAP", "url": "https://pingcap.slack.com/",
        }),
    )
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={"totalResults": 5, "Resources": []}),
    )
    result = await _driver().test_connection()
    assert result.ok is True
    assert "available" in result.message


@pytest.mark.asyncio
@respx.mock
async def test_connection_plus_teams_only() -> None:
    respx.mock.post(f"{_SLACK_WEB_BASE}/auth.test").mock(
        return_value=httpx.Response(200, json={"ok": True, "team": "PingCAP"}),
    )
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(
            403, json={"Errors": {"description": "plus_teams_only", "code": 403}},
        ),
    )
    result = await _driver().test_connection()
    assert result.ok is True  # bot token works — reports overall OK
    assert "plus_teams_only" in result.message


@pytest.mark.asyncio
@respx.mock
async def test_connection_bot_auth_fail() -> None:
    respx.mock.post(f"{_SLACK_WEB_BASE}/auth.test").mock(
        return_value=httpx.Response(200, json={
            "ok": False, "error": "invalid_auth",
        }),
    )
    result = await _driver().test_connection()
    assert result.ok is False
    assert "invalid_auth" in result.message


@pytest.mark.asyncio
@respx.mock
async def test_connection_http_error_redacts_token_in_message() -> None:
    """Regression: Slack SCIM/auth error bodies occasionally echo back the
    request token. A non-200 auth.test body containing an xoxp- token must be
    routed through _redact_secrets before landing in the admin-visible
    IntegrationTestResult.message / provisioning_tasks.error_message."""
    leaked = "xoxp-9999-abcdefghijklmnop-secrettoken"
    respx.mock.post(f"{_SLACK_WEB_BASE}/auth.test").mock(
        return_value=httpx.Response(
            401, text=f'{{"error":"echoed request token {leaked}"}}',
        ),
    )
    result = await _driver().test_connection()
    assert result.ok is False
    assert leaked not in result.message
    assert "xox*-***" in result.message


# ── list_full_members ───────────────────────────────────────────


@pytest.mark.asyncio
@respx.mock
async def test_list_full_members_filters_out_guests_bots_deleted() -> None:
    respx.mock.get(f"{_SLACK_WEB_BASE}/users.list").mock(
        return_value=httpx.Response(200, json={
            "ok": True,
            "members": [
                {
                    "id": "U1", "deleted": False, "is_bot": False,
                    "is_restricted": False, "is_ultra_restricted": False,
                    "profile": {"email": "alice@pingcap.com"},
                },
                # Deleted
                {"id": "U2", "deleted": True, "profile": {"email": "x@x"}},
                # Bot
                {"id": "U3", "is_bot": True, "profile": {"email": "y@y"}},
                # Slackbot
                {"id": "USLACKBOT", "profile": {"email": "z@z"}},
                # Single-channel guest
                {
                    "id": "U4", "is_restricted": True,
                    "profile": {"email": "guest@ext.com"},
                },
                # Multi-channel guest
                {
                    "id": "U5", "is_ultra_restricted": True,
                    "profile": {"email": "multi@ext.com"},
                },
                {
                    "id": "U6", "profile": {"email": "bob@pingcap.com"},
                },
            ],
            "response_metadata": {"next_cursor": ""},
        }),
    )
    members = await _driver().list_full_members()
    emails = [(m.get("profile") or {}).get("email") for m in members]
    assert emails == ["alice@pingcap.com", "bob@pingcap.com"]


@pytest.mark.asyncio
@respx.mock
async def test_list_full_members_paginates() -> None:
    route = respx.mock.get(f"{_SLACK_WEB_BASE}/users.list")
    route.mock(side_effect=[
        httpx.Response(200, json={
            "ok": True,
            "members": [{"id": "U1", "profile": {"email": "a@p.com"}}],
            "response_metadata": {"next_cursor": "CURSOR2"},
        }),
        httpx.Response(200, json={
            "ok": True,
            "members": [{"id": "U2", "profile": {"email": "b@p.com"}}],
            "response_metadata": {"next_cursor": ""},
        }),
    ])
    members = await _driver().list_full_members()
    assert len(members) == 2
    assert route.call_count == 2


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


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_create_success() -> None:
    respx.mock.post(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(201, json={
            "id": "U12345", "userName": "alice@example.com", "active": True,
        }),
    )
    result = await _driver().grant_seat(ProvisioningContext(
        user_email="alice@example.com", user_display_name="Alice Smith",
    ))
    assert result.ok is True
    assert result.external_ref == "U12345"


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_409_reactivates() -> None:
    respx.mock.post(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(409, text='{"Errors":{"description":"duplicate"}}'),
    )
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [
                {"id": "U99", "userName": "dup@example.com", "active": False},
            ],
        }),
    )
    put_route = respx.mock.put(f"{_DEFAULT_SCIM_BASE}/Users/U99").mock(
        return_value=httpx.Response(200),
    )
    result = await _driver().grant_seat(ProvisioningContext(
        user_email="dup@example.com",
    ))
    assert result.ok is True
    assert result.raw_response == {"reactivated": True}
    assert put_route.called
    # PUT body includes active=true
    body = put_route.calls.last.request.content.decode()
    assert '"active": true' in body or '"active":true' in body


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_409_already_active_noop() -> None:
    respx.mock.post(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(409, text="exists"),
    )
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [
                {"id": "U77", "userName": "x@example.com", "active": True},
            ],
        }),
    )
    result = await _driver().grant_seat(ProvisioningContext(
        user_email="x@example.com",
    ))
    assert result.ok is True
    assert result.raw_response == {"already_active": True}


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_plus_teams_only_bubbles_clear_error() -> None:
    respx.mock.post(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(
            403, json={"Errors": {"description": "plus_teams_only", "code": 403}},
        ),
    )
    result = await _driver().grant_seat(ProvisioningContext(
        user_email="whatever@example.com",
    ))
    assert result.ok is False
    assert "plus_teams_only" in result.error_message
    assert "manual_notify" in result.error_message


@pytest.mark.asyncio
async def test_grant_seat_without_user_token_refuses_cleanly() -> None:
    """Bot-only config: SCIM can't be attempted. Return a clear error
    telling the admin to flip binding to manual_notify."""
    result = await _driver(_CONFIG_BOT_ONLY).grant_seat(ProvisioningContext(
        user_email="a@a.com",
    ))
    assert result.ok is False
    assert "SCIM not usable" in result.error_message


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


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_deactivate_happy_path() -> None:
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [
                {"id": "U42", "userName": "bob@example.com", "active": True},
            ],
        }),
    )
    put_route = respx.mock.put(f"{_DEFAULT_SCIM_BASE}/Users/U42").mock(
        return_value=httpx.Response(200),
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="bob@example.com",
        extra={"action_on_offboard": "deactivate"},
    ))
    assert result.ok is True
    assert result.raw_response == {"deactivated": True}
    body = put_route.calls.last.request.content.decode()
    assert '"active": false' in body or '"active":false' in body


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_delete_happy_path() -> None:
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [{"id": "U42", "userName": "c@example.com"}],
        }),
    )
    delete_route = respx.mock.delete(f"{_DEFAULT_SCIM_BASE}/Users/U42").mock(
        return_value=httpx.Response(204),
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="c@example.com",
        extra={"action_on_offboard": "delete"},
    ))
    assert result.ok is True
    assert result.raw_response == {"deleted": True}
    assert delete_route.called


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_deactivate_404_treated_idempotent() -> None:
    """TOCTOU between SCIM lookup and PUT — admin removed the user
    out-of-band between the find and the deactivate. The desired
    end-state (user not active) is already met, so the offboarding
    blocker must close. Mirrors the delete branch's 404 contract."""
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [
                {"id": "U42", "userName": "bob@example.com", "active": True},
            ],
        }),
    )
    respx.mock.put(f"{_DEFAULT_SCIM_BASE}/Users/U42").mock(
        return_value=httpx.Response(404, text="resource_not_found"),
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="bob@example.com",
        extra={"action_on_offboard": "deactivate"},
    ))
    assert result.ok is True
    assert result.raw_response == {"already_absent": True, "user_id": "U42"}
    # external_ref must match the synthetic form used by the
    # `find.user is None` branch — by the time PUT 404s, U42 may
    # already be reaped on Slack's side and an audit replay keying
    # on it would re-404.
    assert result.external_ref == "slack:bob@example.com:already_absent"


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_user_already_absent() -> None:
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={"totalResults": 0, "Resources": []}),
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="gone@example.com",
        extra={"action_on_offboard": "deactivate"},
    ))
    assert result.ok is True
    assert result.raw_response == {"already_absent": True}


@pytest.mark.asyncio
@respx.mock
async def test_revoke_seat_plus_teams_only() -> None:
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(
            403, json={"Errors": {"description": "plus_teams_only", "code": 403}},
        ),
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="anyone@example.com",
        extra={"action_on_offboard": "deactivate"},
    ))
    assert result.ok is False
    assert "plus_teams_only" in result.error_message
    assert "manual_notify" in result.error_message


@pytest.mark.asyncio
async def test_revoke_seat_invalid_action() -> None:
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="y@example.com",
        extra={"action_on_offboard": "purge"},
    ))
    assert result.ok is False
    assert "invalid action_on_offboard" in result.error_message


# ── security / robustness ───────────────────────────────────────


@pytest.mark.asyncio
async def test_list_full_members_without_bot_token_raises_clearly() -> None:
    """Sync handler expects a clear error when someone enables the
    integration with only a user_token (xoxp-) and no bot_token. Silent
    failure here would make the daily sync attempt appear to succeed
    with zero members."""
    driver = SlackSoftwareDriver(
        integration_id="slack-broken",
        config={"user_token": "xoxp-only"},  # no bot_token
    )
    with pytest.raises(ValueError, match="bot_token"):
        await driver.list_full_members()


def test_scim_filter_value_is_escaped() -> None:
    """Emails can legally contain double-quotes (RFC 3986 local part).
    The driver must escape them so crafted input can't break out of
    the ``userName eq "…"`` SCIM filter. Backslashes get doubled so
    Slack's unescaping recovers the original value."""
    raw = 'foo"bar\\baz@example.com'
    escaped = SlackSoftwareDriver._escape_scim_filter_value(raw)
    # Backslash is doubled, quote is backslashed.
    assert escaped == 'foo\\"bar\\\\baz@example.com'
    # Sanity: every literal " in the escaped string is preceded by a
    # backslash (no naked quotes that would close the filter early).
    assert escaped.count('"') == escaped.count('\\"')


@pytest.mark.asyncio
@respx.mock
async def test_grant_seat_uses_lowercased_email_in_payload() -> None:
    """external_ref stability depends on the email we send matching
    what ``slack.seat_sync`` will later return. Both sides lowercase."""
    route = respx.mock.post(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(201, json={
            "id": "U1", "userName": "alice@pingcap.com", "active": True,
        }),
    )
    result = await _driver().grant_seat(ProvisioningContext(
        user_email="Alice@PingCAP.COM", user_display_name="Alice",
    ))
    assert result.ok is True
    body = route.calls.last.request.content.decode()
    assert "alice@pingcap.com" in body
    assert "Alice@PingCAP.COM" not in body


@pytest.mark.asyncio
@respx.mock
async def test_reactivate_missing_user_id_returns_clear_error() -> None:
    """If Slack's SCIM response for a 409-conflict lookup lacks an
    ``id`` field, don't PUT to ``/Users/None`` with a cryptic 404;
    surface the malformed response to the caller."""
    respx.mock.post(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(409, text="dup"),
    )
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            # No "id" field in the record!
            "Resources": [{"userName": "weird@x.com", "active": False}],
        }),
    )
    result = await _driver().grant_seat(ProvisioningContext(
        user_email="weird@x.com",
    ))
    assert result.ok is False
    assert "without an id field" in result.error_message


# ── Round-1 offboarding hardening: SCIM 429/5xx retry ────────────


@pytest.mark.asyncio
@respx.mock
async def test_scim_revoke_retries_on_429_then_succeeds(monkeypatch) -> None:
    """A single 429 from Slack on the deactivate PUT must NOT fail the
    revoke — the helper retries with the Retry-After delay and then
    completes. Without this fix, transient throttling dead-letters the
    offboarding task immediately.
    """
    # Avoid real sleeps in the retry loop.
    import asyncio as _asyncio
    sleeps: list[float] = []

    async def _fast_sleep(s: float) -> None:
        sleeps.append(s)

    monkeypatch.setattr(_asyncio, "sleep", _fast_sleep)

    # SCIM find succeeds.
    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [{"id": "U123", "userName": "x@y.io", "active": True}],
        }),
    )
    # PUT: first call 429, second call 204.
    put_route = respx.mock.put(f"{_DEFAULT_SCIM_BASE}/Users/U123").mock(
        side_effect=[
            httpx.Response(429, headers={"Retry-After": "2"}, text="rate"),
            httpx.Response(204),
        ],
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="x@y.io",
        extra={"action_on_offboard": "deactivate"},
    ))
    assert result.ok is True
    assert result.raw_response == {"deactivated": True}
    assert put_route.call_count == 2
    assert sleeps and sleeps[0] >= 1.0


@pytest.mark.asyncio
@respx.mock
async def test_scim_revoke_retries_on_5xx_then_succeeds(monkeypatch) -> None:
    """5xx (Slack datacenter cutover / overload) must trigger exponential
    backoff retry, not immediate failure."""
    import asyncio as _asyncio

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

    monkeypatch.setattr(_asyncio, "sleep", _noop_sleep)

    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [{"id": "U456", "userName": "a@b.io", "active": True}],
        }),
    )
    put_route = respx.mock.put(f"{_DEFAULT_SCIM_BASE}/Users/U456").mock(
        side_effect=[
            httpx.Response(503, text="cutover"),
            httpx.Response(502, text="bad gateway"),
            httpx.Response(204),
        ],
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="a@b.io",
        extra={"action_on_offboard": "deactivate"},
    ))
    assert result.ok is True
    assert put_route.call_count == 3


@pytest.mark.asyncio
@respx.mock
async def test_scim_revoke_429_budget_exhausted_tagged(monkeypatch) -> None:
    """When retries are exhausted on persistent 429, the error must be
    tagged [SLACK_RATE_LIMITED] so the failure-DM dispatcher appends
    the throttle remediation hint."""
    import asyncio as _asyncio

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

    monkeypatch.setattr(_asyncio, "sleep", _noop_sleep)

    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [{"id": "U789", "userName": "p@q.io", "active": True}],
        }),
    )
    # Always 429 — budget exhausts.
    respx.mock.put(f"{_DEFAULT_SCIM_BASE}/Users/U789").mock(
        return_value=httpx.Response(429, headers={"Retry-After": "1"}, text="rate"),
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="p@q.io",
        extra={"action_on_offboard": "deactivate"},
    ))
    assert result.ok is False
    assert result.error_message.startswith(f"[{SLACK_TAG_RATE_LIMITED}]")
    assert (result.raw_response or {}).get("error_tag") == SLACK_TAG_RATE_LIMITED


@pytest.mark.asyncio
@respx.mock
async def test_scim_find_user_retries_on_429(monkeypatch) -> None:
    """The lookup step itself must retry — without this, a 429 on the
    pre-revoke SCIM find dead-letters the task without ever attempting
    the PUT/DELETE."""
    import asyncio as _asyncio

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

    monkeypatch.setattr(_asyncio, "sleep", _noop_sleep)

    get_route = respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        side_effect=[
            httpx.Response(429, headers={"Retry-After": "1"}, text="rate"),
            httpx.Response(200, json={
                "totalResults": 1,
                "Resources": [{"id": "U999", "userName": "z@w.io", "active": True}],
            }),
        ],
    )
    respx.mock.put(f"{_DEFAULT_SCIM_BASE}/Users/U999").mock(
        return_value=httpx.Response(204),
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="z@w.io",
        extra={"action_on_offboard": "deactivate"},
    ))
    assert result.ok is True
    assert get_route.call_count == 2


@pytest.mark.asyncio
@respx.mock
async def test_scim_delete_retries_on_429(monkeypatch) -> None:
    """DELETE path also needs retry — symmetric with PUT."""
    import asyncio as _asyncio

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

    monkeypatch.setattr(_asyncio, "sleep", _noop_sleep)

    respx.mock.get(f"{_DEFAULT_SCIM_BASE}/Users").mock(
        return_value=httpx.Response(200, json={
            "totalResults": 1,
            "Resources": [{"id": "U111", "userName": "d@e.io", "active": True}],
        }),
    )
    del_route = respx.mock.delete(f"{_DEFAULT_SCIM_BASE}/Users/U111").mock(
        side_effect=[
            httpx.Response(429, text="rate"),
            httpx.Response(204),
        ],
    )
    result = await _driver().revoke_seat(ProvisioningContext(
        user_email="d@e.io",
        extra={"action_on_offboard": "delete"},
    ))
    assert result.ok is True
    assert del_route.call_count == 2