"""Unit tests for the GitHub organization SCIM driver. respx-mocked. Covers test_connection success/403/404, grant_seat create + reactivation + already-active + 500, revoke_seat happy path + already-absent + 404. """ from __future__ import annotations import httpx import pytest import respx from app.integrations.driver_base import ProvisioningContext from app.integrations.drivers.github_software import GitHubSoftwareDriver _CONFIG = { "org_slug": "acme-inc", "token": "ghp_fake", } _USERS_URL = "https://api.github.com/scim/v2/organizations/acme-inc/Users" def _driver() -> GitHubSoftwareDriver: return GitHubSoftwareDriver(integration_id="github-test", config=_CONFIG) def test_metadata() -> None: meta = GitHubSoftwareDriver.metadata() assert meta.kind == "software_provisioning" assert meta.provider == "github" assert meta.binding_fields == [] # org-level, no per-binding params assert any(f.key == "token" and f.is_secret for f in meta.fields) # ── test_connection ────────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_connection_success() -> None: respx.mock.get(_USERS_URL).mock( return_value=httpx.Response( 200, json={"totalResults": 42, "Resources": []}, ), ) result = await _driver().test_connection() assert result.ok is True assert "42 members" in result.message assert "acme-inc" in result.message @pytest.mark.asyncio @respx.mock async def test_connection_404_means_not_enterprise_with_saml() -> None: respx.mock.get(_USERS_URL).mock( return_value=httpx.Response(404, text="Not Found"), ) result = await _driver().test_connection() assert result.ok is False assert "Enterprise Cloud with SAML" in result.message @pytest.mark.asyncio @respx.mock async def test_connection_403_forbidden() -> None: respx.mock.get(_USERS_URL).mock( return_value=httpx.Response(403, text="Forbidden"), ) result = await _driver().test_connection() assert result.ok is False assert "403" in result.message # ── grant_seat ─────────────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_grant_seat_creates_user() -> None: respx.mock.post(_USERS_URL).mock( return_value=httpx.Response( 201, json={ "id": "scim-user-1", "userName": "alice@acme.com", "active": True, }, ), ) result = await _driver().grant_seat(ProvisioningContext( user_email="alice@acme.com", user_display_name="Alice Smith", )) assert result.ok is True assert result.external_ref == "scim-user-1" @pytest.mark.asyncio @respx.mock async def test_grant_seat_409_reactivates() -> None: respx.mock.post(_USERS_URL).mock( return_value=httpx.Response(409, text="already exists"), ) respx.mock.get(_USERS_URL).mock( return_value=httpx.Response( 200, json={ "totalResults": 1, "Resources": [ {"id": "scim-x", "userName": "rehire@acme.com", "active": False}, ], }, ), ) patch_route = respx.mock.patch(f"{_USERS_URL}/scim-x").mock( return_value=httpx.Response(204), ) result = await _driver().grant_seat(ProvisioningContext( user_email="rehire@acme.com", )) assert result.ok is True assert result.raw_response == {"reactivated": True} assert patch_route.called @pytest.mark.asyncio @respx.mock async def test_grant_seat_409_already_active_is_noop() -> None: respx.mock.post(_USERS_URL).mock( return_value=httpx.Response(409, text="exists"), ) respx.mock.get(_USERS_URL).mock( return_value=httpx.Response( 200, json={ "totalResults": 1, "Resources": [ {"id": "scim-y", "userName": "old@acme.com", "active": True}, ], }, ), ) result = await _driver().grant_seat(ProvisioningContext( user_email="old@acme.com", )) assert result.ok is True assert result.raw_response == {"already_active": True} @pytest.mark.asyncio @respx.mock async def test_grant_seat_500() -> None: respx.mock.post(_USERS_URL).mock( return_value=httpx.Response(500, text="GitHub is down"), ) result = await _driver().grant_seat(ProvisioningContext( user_email="x@acme.com", )) assert result.ok is False assert "500" in result.error_message # ── revoke_seat ────────────────────────────────────────────────── @pytest.mark.asyncio @respx.mock async def test_revoke_seat_deprovisions() -> None: respx.mock.get(_USERS_URL).mock( return_value=httpx.Response( 200, json={ "totalResults": 1, "Resources": [{"id": "scim-bye", "userName": "bye@acme.com"}], }, ), ) delete_route = respx.mock.delete(f"{_USERS_URL}/scim-bye").mock( return_value=httpx.Response(204), ) result = await _driver().revoke_seat(ProvisioningContext( user_email="bye@acme.com", )) assert result.ok is True assert result.raw_response == {"deprovisioned": True} assert delete_route.called @pytest.mark.asyncio @respx.mock async def test_revoke_seat_already_absent() -> None: respx.mock.get(_USERS_URL).mock( return_value=httpx.Response( 200, json={"totalResults": 0, "Resources": []}, ), ) result = await _driver().revoke_seat(ProvisioningContext( user_email="gone@acme.com", )) assert result.ok is True assert result.raw_response == {"already_absent": True} @pytest.mark.asyncio @respx.mock async def test_revoke_seat_delete_404_is_success() -> None: """Race: lookup finds the user, but by the time DELETE lands they're already gone. Should still return ok (offboarding idempotency).""" respx.mock.get(_USERS_URL).mock( return_value=httpx.Response( 200, json={ "totalResults": 1, "Resources": [{"id": "scim-race", "userName": "race@acme.com"}], }, ), ) respx.mock.delete(f"{_USERS_URL}/scim-race").mock( return_value=httpx.Response(404, text="not found"), ) result = await _driver().revoke_seat(ProvisioningContext( user_email="race@acme.com", )) assert result.ok is True assert result.raw_response == {"already_absent": True} @pytest.mark.asyncio @respx.mock async def test_revoke_seat_lookup_network_error_does_NOT_confirm_absent() -> None: """Round-19 audit regression: a transport-layer failure on the SCIM lookup MUST NOT be silently confirmed as ``already_absent``. Pre-fix the driver collapsed httpx.HTTPError into the same ``(scim_id=None, lookup_resp=None)`` tuple that meant "user not found". ``revoke_seat`` then returned ok=True / already_absent — flipping the local seat row to revoked while the GitHub org membership stayed live (billing leak + access leak). """ respx.mock.get(_USERS_URL).mock( side_effect=httpx.ConnectError("name resolution failed"), ) result = await _driver().revoke_seat(ProvisioningContext( user_email="nettrouble@acme.com", )) assert result.ok is False assert "transport error" in (result.error_message or "") assert "ConnectError" in (result.error_message or "") @pytest.mark.asyncio @respx.mock async def test_reactivate_lookup_network_error_propagates() -> None: """Round-19 audit regression: the 409→reactivate path must surface transport-layer failures on the SCIM lookup, not pretend the user doesn't exist.""" respx.mock.post(_USERS_URL).mock( return_value=httpx.Response(409, text="conflict"), ) respx.mock.get(_USERS_URL).mock( side_effect=httpx.TimeoutException("read timeout"), ) result = await _driver().grant_seat(ProvisioningContext( user_email="conflict@acme.com", user_display_name="C", )) assert result.ok is False assert "transport error" in (result.error_message or "")