"""Integration tests for the SCIM 2.0 endpoints against a live Postgres. These tests drive the real Flask app through its test client and verify row state in the database configured by ``POSTGRES_URI``. Redis is not required — the denylist functions are stubbed. They are skipped by the default ``pytest`` run (``--ignore=tests/integration`` in ``pytest.ini``) and marked ``@pytest.mark.integration``. Run them locally with:: .venv/bin/python -m pytest tests/integration/test_scim.py -q --no-cov \\ -p no:cacheprovider --override-ini "addopts=" """ from __future__ import annotations import uuid from types import SimpleNamespace from unittest.mock import patch import pytest from sqlalchemy import text from application.core.settings import settings from application.storage.db.repositories.auth_events import AuthEventsRepository from application.storage.db.repositories.users import UsersRepository from application.storage.db.session import db_readonly, db_session SCIM_TOKEN = "scim-test-token" AUTH = {"Authorization": f"Bearer {SCIM_TOKEN}"} ERROR_URN = "urn:ietf:params:scim:api:messages:2.0:Error" PATCH_OP_URN = "urn:ietf:params:scim:api:messages:2.0:PatchOp" pytestmark = [ pytest.mark.integration, pytest.mark.skipif( not settings.POSTGRES_URI, reason="POSTGRES_URI not set — skipping SCIM integration tests", ), ] @pytest.fixture(scope="module") def app(): """Real Flask app; /scim/ paths bypass JWT auth so no handle_auth patching is needed.""" from application.app import app as flask_app flask_app.config["TESTING"] = True return flask_app @pytest.fixture def client(app): return app.test_client() @pytest.fixture def scim_env(monkeypatch): """Enable SCIM on the settings singleton and stub the Redis denylist.""" monkeypatch.setattr(settings, "SCIM_ENABLED", True) monkeypatch.setattr(settings, "SCIM_TOKEN", SCIM_TOKEN) with patch("application.api.scim.routes.deny_user") as deny_user_mock: yield SimpleNamespace(deny_user=deny_user_mock) @pytest.fixture def scim_user_name(): """Unique per-run userName; deletes the user + audit rows afterwards.""" name = f"scim-it-{uuid.uuid4().hex[:10]}@example.com" yield name with db_session() as conn: conn.execute(text("DELETE FROM auth_events WHERE user_id = :user_id"), {"user_id": name}) conn.execute(text("DELETE FROM users WHERE user_id = :user_id"), {"user_id": name}) def _fetch_user(user_name: str): with db_readonly() as conn: return UsersRepository(conn).get(user_name) def _fetch_events(user_name: str): with db_readonly() as conn: return AuthEventsRepository(conn).list_recent(user_name) @pytest.mark.integration class TestScimLifecycle: def test_full_user_lifecycle(self, client, scim_env, scim_user_name): # Create response = client.post("/scim/v2/Users", headers=AUTH, json={"userName": scim_user_name}) assert response.status_code == 201, response.get_data(as_text=True) created = response.get_json() pk = created["id"] assert created["userName"] == scim_user_name assert created["active"] is True assert created["emails"] == [{"value": scim_user_name, "primary": True}] assert response.headers["Location"].endswith(f"/scim/v2/Users/{pk}") row = _fetch_user(scim_user_name) assert row is not None assert row["active"] is True assert [event["event"] for event in _fetch_events(scim_user_name)] == ["scim_created"] # GET by id response = client.get(f"/scim/v2/Users/{pk}", headers=AUTH) assert response.status_code == 200 assert response.get_json()["id"] == pk # List with exact userName filter response = client.get( "/scim/v2/Users", headers=AUTH, query_string={"filter": f'userName eq "{scim_user_name}"'}, ) assert response.status_code == 200 body = response.get_json() assert body["totalResults"] == 1 assert body["itemsPerPage"] == 1 assert body["Resources"][0]["id"] == pk # Deactivate via PATCH (Okta no-path form with string value) response = client.patch( f"/scim/v2/Users/{pk}", headers=AUTH, json={ "schemas": [PATCH_OP_URN], "Operations": [{"op": "replace", "value": {"active": "False"}}], }, ) assert response.status_code == 200 assert response.get_json()["active"] is False assert _fetch_user(scim_user_name)["active"] is False deactivations = [e for e in _fetch_events(scim_user_name) if e["event"] == "scim_deactivated"] assert len(deactivations) == 1 assert deactivations[0]["metadata"] == {"via": "scim"} scim_env.deny_user.assert_called_once_with(scim_user_name) # Reactivate via PUT (full replace; only active is honored) response = client.put( f"/scim/v2/Users/{pk}", headers=AUTH, json={"userName": scim_user_name, "active": True}, ) assert response.status_code == 200 assert response.get_json()["active"] is True assert _fetch_user(scim_user_name)["active"] is True # Reactivation no longer clears the denylist (watermark model). scim_env.deny_user.assert_called_once_with(scim_user_name) assert any(e["event"] == "scim_reactivated" for e in _fetch_events(scim_user_name)) # Soft delete deactivates again response = client.delete(f"/scim/v2/Users/{pk}", headers=AUTH) assert response.status_code == 204 assert response.data == b"" assert _fetch_user(scim_user_name)["active"] is False assert scim_env.deny_user.call_count == 2 # Duplicate create conflicts (row still exists after soft delete) response = client.post("/scim/v2/Users", headers=AUTH, json={"userName": scim_user_name}) assert response.status_code == 409 body = response.get_json() assert body["schemas"] == [ERROR_URN] assert body["scimType"] == "uniqueness" @pytest.mark.integration class TestScimAccess: def test_wrong_bearer_token_rejected(self, client, scim_env): response = client.get("/scim/v2/Users", headers={"Authorization": "Bearer wrong"}) assert response.status_code == 401 assert response.get_json()["schemas"] == [ERROR_URN] def test_get_user_with_malformed_uuid_returns_404(self, client, scim_env): response = client.get("/scim/v2/Users/not-a-uuid", headers=AUTH) assert response.status_code == 404 assert response.get_json()["status"] == "404" def test_service_provider_config_served(self, client, scim_env): response = client.get("/scim/v2/ServiceProviderConfig", headers=AUTH) assert response.status_code == 200 assert response.content_type.startswith("application/scim+json") assert response.get_json()["patch"] == {"supported": True}