fed8b2eed7
Backend release / release (push) Waiting to run
Bandit Security Scan / bandit_scan (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push multi-arch DocsGPT Docker image / manifest (push) Blocked by required conditions
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Waiting to run
Build and push DocsGPT FE Docker image for development / manifest (push) Blocked by required conditions
Python linting / ruff (push) Waiting to run
Run python tests with pytest / Run tests and count coverage (3.12) (push) Waiting to run
React Widget Build / build (push) Waiting to run
592 lines
27 KiB
Python
592 lines
27 KiB
Python
"""Unit tests for the SCIM 2.0 provisioning endpoints (application/api/scim/)."""
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from application.core.settings import settings
|
|
|
|
TOKEN = "test-scim-token"
|
|
AUTH = {"Authorization": f"Bearer {TOKEN}"}
|
|
USER_PK = "3f1f2bd6-8a87-4f7e-9c2b-2f76c1a4f0d3"
|
|
|
|
ERROR_URN = "urn:ietf:params:scim:api:messages:2.0:Error"
|
|
LIST_RESPONSE_URN = "urn:ietf:params:scim:api:messages:2.0:ListResponse"
|
|
PATCH_OP_URN = "urn:ietf:params:scim:api:messages:2.0:PatchOp"
|
|
USER_URN = "urn:ietf:params:scim:schemas:core:2.0:User"
|
|
|
|
|
|
def _user_row(pk=USER_PK, user_id="alice@example.com", active=True):
|
|
return {
|
|
"id": pk,
|
|
"user_id": user_id,
|
|
"active": active,
|
|
"agent_preferences": {"pinned": [], "shared_with_me": []},
|
|
"tool_preferences": {},
|
|
"created_at": "2026-06-09T10:00:00+00:00",
|
|
"updated_at": "2026-06-09T11:00:00+00:00",
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def app():
|
|
"""Import the Flask app with auth mocked to avoid JWT setup issues."""
|
|
with patch("application.app.handle_auth", return_value={"sub": "test_user"}):
|
|
from application.app import app as flask_app
|
|
|
|
flask_app.config["TESTING"] = True
|
|
yield flask_app
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
return app.test_client()
|
|
|
|
|
|
@pytest.fixture
|
|
def scim_settings(monkeypatch):
|
|
"""Enable SCIM with a known bearer token on the real settings singleton."""
|
|
monkeypatch.setattr(settings, "SCIM_ENABLED", True)
|
|
monkeypatch.setattr(settings, "SCIM_TOKEN", TOKEN)
|
|
|
|
|
|
@pytest.fixture
|
|
def scim_mocks():
|
|
"""Patch DB plumbing, repositories, and denylist functions used by the routes."""
|
|
with patch("application.api.scim.routes.db_session") as db_session_mock, patch(
|
|
"application.api.scim.routes.db_readonly"
|
|
) as db_readonly_mock, patch(
|
|
"application.api.scim.routes.UsersRepository"
|
|
) as users_cls, patch(
|
|
"application.api.scim.routes.AuthEventsRepository"
|
|
) as audit_cls, patch(
|
|
"application.api.scim.routes.deny_user"
|
|
) as deny_user_mock:
|
|
conn = MagicMock(name="conn")
|
|
db_session_mock.return_value.__enter__.return_value = conn
|
|
db_readonly_mock.return_value.__enter__.return_value = conn
|
|
# Don't let the mocked context manager swallow exceptions (a real
|
|
# ``engine.begin()`` __exit__ returns None) — _RevocationUnavailable
|
|
# must propagate to the blueprint error handler.
|
|
db_session_mock.return_value.__exit__.return_value = False
|
|
db_readonly_mock.return_value.__exit__.return_value = False
|
|
yield SimpleNamespace(
|
|
users=users_cls.return_value,
|
|
audit=audit_cls.return_value,
|
|
deny_user=deny_user_mock,
|
|
conn=conn,
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestScimGate:
|
|
|
|
def test_disabled_returns_scim_404_everywhere(self, client, scim_mocks, monkeypatch):
|
|
monkeypatch.setattr(settings, "SCIM_ENABLED", False)
|
|
monkeypatch.setattr(settings, "SCIM_TOKEN", TOKEN)
|
|
for path in ("/scim/v2/Users", "/scim/v2/ServiceProviderConfig", "/scim/v2/Groups"):
|
|
response = client.get(path, headers=AUTH)
|
|
assert response.status_code == 404
|
|
body = response.get_json()
|
|
assert body["schemas"] == [ERROR_URN]
|
|
assert body["status"] == "404"
|
|
scim_mocks.users.list_paginated.assert_not_called()
|
|
|
|
def test_enabled_without_token_returns_503(self, client, scim_mocks, monkeypatch):
|
|
monkeypatch.setattr(settings, "SCIM_ENABLED", True)
|
|
monkeypatch.setattr(settings, "SCIM_TOKEN", None)
|
|
response = client.get("/scim/v2/Users", headers=AUTH)
|
|
assert response.status_code == 503
|
|
body = response.get_json()
|
|
assert body["schemas"] == [ERROR_URN]
|
|
assert body["status"] == "503"
|
|
|
|
def test_enabled_with_empty_token_returns_503(self, client, scim_mocks, monkeypatch):
|
|
monkeypatch.setattr(settings, "SCIM_ENABLED", True)
|
|
monkeypatch.setattr(settings, "SCIM_TOKEN", "")
|
|
response = client.get("/scim/v2/Users", headers=AUTH)
|
|
assert response.status_code == 503
|
|
|
|
def test_wrong_bearer_returns_401(self, client, scim_settings, scim_mocks):
|
|
response = client.get("/scim/v2/Users", headers={"Authorization": "Bearer wrong-token"})
|
|
assert response.status_code == 401
|
|
assert response.get_json()["schemas"] == [ERROR_URN]
|
|
scim_mocks.users.list_paginated.assert_not_called()
|
|
|
|
def test_missing_authorization_returns_401(self, client, scim_settings, scim_mocks):
|
|
response = client.get("/scim/v2/Users")
|
|
assert response.status_code == 401
|
|
assert response.get_json()["status"] == "401"
|
|
|
|
def test_non_bearer_scheme_returns_401(self, client, scim_settings, scim_mocks):
|
|
response = client.get("/scim/v2/Users", headers={"Authorization": f"Basic {TOKEN}"})
|
|
assert response.status_code == 401
|
|
|
|
def test_errors_use_scim_content_type(self, client, scim_settings, scim_mocks):
|
|
response = client.get("/scim/v2/Users", headers={"Authorization": "Bearer nope"})
|
|
assert response.content_type.startswith("application/scim+json")
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestDiscoveryEndpoints:
|
|
|
|
def test_service_provider_config_shape(self, client, scim_settings, scim_mocks):
|
|
response = client.get("/scim/v2/ServiceProviderConfig", headers=AUTH)
|
|
assert response.status_code == 200
|
|
assert response.content_type.startswith("application/scim+json")
|
|
body = response.get_json()
|
|
assert body["schemas"] == ["urn:ietf:params:scim:schemas:core:2.0:ServiceProviderConfig"]
|
|
assert body["patch"] == {"supported": True}
|
|
assert body["bulk"]["supported"] is False
|
|
assert body["filter"] == {"supported": True, "maxResults": 200}
|
|
assert body["changePassword"] == {"supported": False}
|
|
assert body["sort"] == {"supported": False}
|
|
assert body["etag"] == {"supported": False}
|
|
scheme = body["authenticationSchemes"][0]
|
|
assert scheme["type"] == "oauthbearertoken"
|
|
assert scheme["name"] == "Bearer Token"
|
|
|
|
def test_resource_types_advertise_user_only(self, client, scim_settings, scim_mocks):
|
|
response = client.get("/scim/v2/ResourceTypes", headers=AUTH)
|
|
assert response.status_code == 200
|
|
body = response.get_json()
|
|
assert body["schemas"] == [LIST_RESPONSE_URN]
|
|
assert body["totalResults"] == 1
|
|
resource = body["Resources"][0]
|
|
assert resource["name"] == "User"
|
|
assert resource["endpoint"] == "/scim/v2/Users"
|
|
assert resource["schema"] == USER_URN
|
|
|
|
def test_schemas_advertise_user_only(self, client, scim_settings, scim_mocks):
|
|
response = client.get("/scim/v2/Schemas", headers=AUTH)
|
|
assert response.status_code == 200
|
|
body = response.get_json()
|
|
assert body["schemas"] == [LIST_RESPONSE_URN]
|
|
assert body["totalResults"] == 1
|
|
assert body["Resources"][0]["id"] == USER_URN
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestListUsers:
|
|
|
|
def test_filter_username_eq_passes_value_to_repo(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.list_paginated.return_value = (1, [_user_row()])
|
|
response = client.get(
|
|
"/scim/v2/Users",
|
|
headers=AUTH,
|
|
query_string={"filter": 'userName eq "alice smith@example.com"'},
|
|
)
|
|
assert response.status_code == 200
|
|
scim_mocks.users.list_paginated.assert_called_once_with("alice smith@example.com", 0, 100)
|
|
|
|
def test_filter_keywords_are_case_insensitive(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.list_paginated.return_value = (0, [])
|
|
response = client.get(
|
|
"/scim/v2/Users", headers=AUTH, query_string={"filter": 'UserName EQ "bob@example.com"'}
|
|
)
|
|
assert response.status_code == 200
|
|
scim_mocks.users.list_paginated.assert_called_once_with("bob@example.com", 0, 100)
|
|
|
|
@pytest.mark.parametrize(
|
|
"bad_filter",
|
|
[
|
|
'userName co "alice"',
|
|
'displayName eq "alice"',
|
|
'userName eq "a" and active eq true',
|
|
"userName eq alice",
|
|
],
|
|
)
|
|
def test_unsupported_filter_rejected(self, client, scim_settings, scim_mocks, bad_filter):
|
|
response = client.get("/scim/v2/Users", headers=AUTH, query_string={"filter": bad_filter})
|
|
assert response.status_code == 400
|
|
body = response.get_json()
|
|
assert body["schemas"] == [ERROR_URN]
|
|
assert body["scimType"] == "invalidFilter"
|
|
scim_mocks.users.list_paginated.assert_not_called()
|
|
|
|
@pytest.mark.parametrize(
|
|
("query", "expected_offset", "expected_limit"),
|
|
[
|
|
({}, 0, 100),
|
|
({"startIndex": "3", "count": "2"}, 2, 2),
|
|
({"startIndex": "0", "count": "999"}, 0, 200),
|
|
({"startIndex": "-4", "count": "-5"}, 0, 0),
|
|
],
|
|
)
|
|
def test_pagination_offset_limit_math(
|
|
self, client, scim_settings, scim_mocks, query, expected_offset, expected_limit
|
|
):
|
|
scim_mocks.users.list_paginated.return_value = (0, [])
|
|
response = client.get("/scim/v2/Users", headers=AUTH, query_string=query)
|
|
assert response.status_code == 200
|
|
scim_mocks.users.list_paginated.assert_called_once_with(None, expected_offset, expected_limit)
|
|
|
|
def test_list_response_shape(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.list_paginated.return_value = (42, [_user_row()])
|
|
response = client.get(
|
|
"/scim/v2/Users", headers=AUTH, query_string={"startIndex": "5", "count": "1"}
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.content_type.startswith("application/scim+json")
|
|
body = response.get_json()
|
|
assert body["schemas"] == [LIST_RESPONSE_URN]
|
|
assert body["totalResults"] == 42
|
|
assert body["startIndex"] == 5
|
|
assert body["itemsPerPage"] == 1
|
|
resource = body["Resources"][0]
|
|
assert resource["schemas"] == [USER_URN]
|
|
assert resource["id"] == USER_PK
|
|
assert resource["userName"] == "alice@example.com"
|
|
assert resource["active"] is True
|
|
assert resource["emails"] == [{"value": "alice@example.com", "primary": True}]
|
|
assert resource["meta"]["resourceType"] == "User"
|
|
assert resource["meta"]["location"] == f"/scim/v2/Users/{USER_PK}"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestCreateUser:
|
|
|
|
def test_missing_username_returns_400(self, client, scim_settings, scim_mocks):
|
|
response = client.post("/scim/v2/Users", headers=AUTH, json={"active": True})
|
|
assert response.status_code == 400
|
|
assert response.get_json()["scimType"] == "invalidValue"
|
|
scim_mocks.users.create.assert_not_called()
|
|
|
|
def test_blank_username_returns_400(self, client, scim_settings, scim_mocks):
|
|
response = client.post("/scim/v2/Users", headers=AUTH, json={"userName": " "})
|
|
assert response.status_code == 400
|
|
assert response.get_json()["scimType"] == "invalidValue"
|
|
|
|
def test_conflict_returns_409(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.create.return_value = None
|
|
response = client.post("/scim/v2/Users", headers=AUTH, json={"userName": "alice@example.com"})
|
|
assert response.status_code == 409
|
|
body = response.get_json()
|
|
assert body["schemas"] == [ERROR_URN]
|
|
assert body["scimType"] == "uniqueness"
|
|
scim_mocks.audit.insert.assert_not_called()
|
|
|
|
def test_create_happy_path(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.create.return_value = _user_row()
|
|
response = client.post("/scim/v2/Users", headers=AUTH, json={"userName": "alice@example.com"})
|
|
assert response.status_code == 201
|
|
assert response.content_type.startswith("application/scim+json")
|
|
assert response.headers["Location"].endswith(f"/scim/v2/Users/{USER_PK}")
|
|
scim_mocks.users.create.assert_called_once_with("alice@example.com", active=True)
|
|
scim_mocks.audit.insert.assert_called_once_with(
|
|
"alice@example.com", "scim_created", metadata={"via": "scim"}
|
|
)
|
|
body = response.get_json()
|
|
assert body["id"] == USER_PK
|
|
assert body["userName"] == "alice@example.com"
|
|
assert body["active"] is True
|
|
assert body["emails"] == [{"value": "alice@example.com", "primary": True}]
|
|
assert body["meta"]["resourceType"] == "User"
|
|
|
|
def test_create_honors_active_false(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.create.return_value = _user_row(active=False)
|
|
response = client.post(
|
|
"/scim/v2/Users", headers=AUTH, json={"userName": "alice@example.com", "active": False}
|
|
)
|
|
assert response.status_code == 201
|
|
scim_mocks.users.create.assert_called_once_with("alice@example.com", active=False)
|
|
assert response.get_json()["active"] is False
|
|
|
|
def test_create_accepts_scim_content_type(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.create.return_value = _user_row()
|
|
response = client.post(
|
|
"/scim/v2/Users",
|
|
headers=AUTH,
|
|
data=json.dumps({"userName": "alice@example.com"}),
|
|
content_type="application/scim+json",
|
|
)
|
|
assert response.status_code == 201
|
|
|
|
def test_create_without_email_username_has_no_emails(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.create.return_value = _user_row(user_id="ldap-user-1")
|
|
response = client.post("/scim/v2/Users", headers=AUTH, json={"userName": "ldap-user-1"})
|
|
assert response.status_code == 201
|
|
assert "emails" not in response.get_json()
|
|
|
|
def test_audit_failure_does_not_fail_request(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.create.return_value = _user_row()
|
|
scim_mocks.audit.insert.side_effect = RuntimeError("audit down")
|
|
response = client.post("/scim/v2/Users", headers=AUTH, json={"userName": "alice@example.com"})
|
|
assert response.status_code == 201
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetUser:
|
|
|
|
def test_get_user_found(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row()
|
|
response = client.get(f"/scim/v2/Users/{USER_PK}", headers=AUTH)
|
|
assert response.status_code == 200
|
|
scim_mocks.users.get_by_pk.assert_called_once_with(USER_PK)
|
|
body = response.get_json()
|
|
assert body["id"] == USER_PK
|
|
assert body["userName"] == "alice@example.com"
|
|
assert body["meta"]["location"] == f"/scim/v2/Users/{USER_PK}"
|
|
|
|
def test_get_user_missing_returns_404(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = None
|
|
response = client.get(f"/scim/v2/Users/{USER_PK}", headers=AUTH)
|
|
assert response.status_code == 404
|
|
assert response.get_json()["schemas"] == [ERROR_URN]
|
|
|
|
def test_get_user_malformed_uuid_returns_404(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = None
|
|
response = client.get("/scim/v2/Users/not-a-uuid", headers=AUTH)
|
|
assert response.status_code == 404
|
|
assert response.get_json()["status"] == "404"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestReplaceUser:
|
|
|
|
def test_username_change_rejected_as_mutability(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row()
|
|
response = client.put(
|
|
f"/scim/v2/Users/{USER_PK}",
|
|
headers=AUTH,
|
|
json={"userName": "bob@example.com", "active": True},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.get_json()["scimType"] == "mutability"
|
|
scim_mocks.users.set_active.assert_not_called()
|
|
|
|
def test_put_active_true_reactivates(self, client, scim_settings, scim_mocks):
|
|
row = _user_row(active=False)
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": True}
|
|
response = client.put(
|
|
f"/scim/v2/Users/{USER_PK}",
|
|
headers=AUTH,
|
|
json={"userName": "alice@example.com", "active": True},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.get_json()["active"] is True
|
|
scim_mocks.users.set_active.assert_called_once_with(USER_PK, True)
|
|
# Reactivation does not touch the denylist (the deactivation watermark
|
|
# already lets a fresh login through on a newer iat).
|
|
scim_mocks.deny_user.assert_not_called()
|
|
scim_mocks.audit.insert.assert_called_once_with(
|
|
"alice@example.com", "scim_reactivated", metadata={"via": "scim"}
|
|
)
|
|
|
|
def test_put_active_false_triggers_deny(self, client, scim_settings, scim_mocks):
|
|
row = _user_row(active=True)
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": False}
|
|
response = client.put(
|
|
f"/scim/v2/Users/{USER_PK}",
|
|
headers=AUTH,
|
|
json={"userName": "alice@example.com", "active": False},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.get_json()["active"] is False
|
|
scim_mocks.users.set_active.assert_called_once_with(USER_PK, False)
|
|
scim_mocks.deny_user.assert_called_once_with("alice@example.com")
|
|
scim_mocks.audit.insert.assert_called_once_with(
|
|
"alice@example.com", "scim_deactivated", metadata={"via": "scim"}
|
|
)
|
|
|
|
def test_put_differently_cased_username_deprovisions(self, client, scim_settings, scim_mocks):
|
|
# userName is case-insensitive: a PUT echoing the stored name under
|
|
# different casing must deactivate, not 400 "userName is immutable".
|
|
row = _user_row(active=True) # user_id == "alice@example.com"
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": False}
|
|
response = client.put(
|
|
f"/scim/v2/Users/{USER_PK}",
|
|
headers=AUTH,
|
|
json={"userName": "Alice@Example.com", "active": False},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.get_json()["active"] is False
|
|
scim_mocks.users.set_active.assert_called_once_with(USER_PK, False)
|
|
scim_mocks.deny_user.assert_called_once_with("alice@example.com")
|
|
|
|
def test_put_unchanged_active_has_no_side_effects(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row(active=True)
|
|
response = client.put(
|
|
f"/scim/v2/Users/{USER_PK}",
|
|
headers=AUTH,
|
|
json={"userName": "alice@example.com", "active": True},
|
|
)
|
|
assert response.status_code == 200
|
|
scim_mocks.users.set_active.assert_not_called()
|
|
scim_mocks.deny_user.assert_not_called()
|
|
scim_mocks.audit.insert.assert_not_called()
|
|
|
|
def test_put_missing_user_returns_404(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = None
|
|
response = client.put(
|
|
f"/scim/v2/Users/{USER_PK}", headers=AUTH, json={"userName": "x", "active": True}
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestPatchUser:
|
|
|
|
def _patch(self, client, operations):
|
|
return client.patch(
|
|
f"/scim/v2/Users/{USER_PK}",
|
|
headers=AUTH,
|
|
json={"schemas": [PATCH_OP_URN], "Operations": operations},
|
|
)
|
|
|
|
def _expect_deactivated(self, scim_mocks, response):
|
|
assert response.status_code == 200
|
|
assert response.get_json()["active"] is False
|
|
scim_mocks.users.set_active.assert_called_once_with(USER_PK, False)
|
|
scim_mocks.deny_user.assert_called_once_with("alice@example.com")
|
|
scim_mocks.audit.insert.assert_called_once_with(
|
|
"alice@example.com", "scim_deactivated", metadata={"via": "scim"}
|
|
)
|
|
|
|
def test_replace_with_path_deactivates(self, client, scim_settings, scim_mocks):
|
|
row = _user_row(active=True)
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": False}
|
|
response = self._patch(client, [{"op": "replace", "path": "active", "value": False}])
|
|
self._expect_deactivated(scim_mocks, response)
|
|
|
|
def test_replace_without_path_object_value_deactivates(self, client, scim_settings, scim_mocks):
|
|
row = _user_row(active=True)
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": False}
|
|
response = self._patch(client, [{"op": "replace", "value": {"active": False}}])
|
|
self._expect_deactivated(scim_mocks, response)
|
|
|
|
def test_replace_with_string_false_deactivates(self, client, scim_settings, scim_mocks):
|
|
row = _user_row(active=True)
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": False}
|
|
response = self._patch(client, [{"op": "Replace", "path": "active", "value": "False"}])
|
|
self._expect_deactivated(scim_mocks, response)
|
|
|
|
def test_replace_with_string_true_reactivates(self, client, scim_settings, scim_mocks):
|
|
row = _user_row(active=False)
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": True}
|
|
response = self._patch(client, [{"op": "replace", "path": "active", "value": "true"}])
|
|
assert response.status_code == 200
|
|
assert response.get_json()["active"] is True
|
|
scim_mocks.users.set_active.assert_called_once_with(USER_PK, True)
|
|
scim_mocks.deny_user.assert_not_called()
|
|
scim_mocks.audit.insert.assert_called_once_with(
|
|
"alice@example.com", "scim_reactivated", metadata={"via": "scim"}
|
|
)
|
|
|
|
def test_bogus_path_rejected(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row()
|
|
response = self._patch(client, [{"op": "replace", "path": "displayName", "value": "X"}])
|
|
assert response.status_code == 400
|
|
assert response.get_json()["scimType"] == "invalidPath"
|
|
scim_mocks.users.set_active.assert_not_called()
|
|
|
|
def test_bogus_op_rejected(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row()
|
|
response = self._patch(client, [{"op": "add", "path": "active", "value": True}])
|
|
assert response.status_code == 400
|
|
assert response.get_json()["scimType"] == "invalidPath"
|
|
|
|
def test_invalid_active_value_rejected(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row()
|
|
response = self._patch(client, [{"op": "replace", "path": "active", "value": "maybe"}])
|
|
assert response.status_code == 400
|
|
assert response.get_json()["scimType"] == "invalidValue"
|
|
|
|
def test_unchanged_active_is_noop(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row(active=True)
|
|
response = self._patch(client, [{"op": "replace", "path": "active", "value": True}])
|
|
assert response.status_code == 200
|
|
scim_mocks.users.set_active.assert_not_called()
|
|
scim_mocks.deny_user.assert_not_called()
|
|
scim_mocks.audit.insert.assert_not_called()
|
|
|
|
def test_no_path_value_without_active_is_noop(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row(active=True)
|
|
response = self._patch(client, [{"op": "replace", "value": {"displayName": "X"}}])
|
|
assert response.status_code == 200
|
|
scim_mocks.users.set_active.assert_not_called()
|
|
|
|
def test_missing_operations_rejected(self, client, scim_settings, scim_mocks):
|
|
response = client.patch(
|
|
f"/scim/v2/Users/{USER_PK}", headers=AUTH, json={"schemas": [PATCH_OP_URN]}
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
def test_patch_missing_user_returns_404(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = None
|
|
response = self._patch(client, [{"op": "replace", "path": "active", "value": False}])
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestDeleteUser:
|
|
|
|
def test_delete_deactivates_and_returns_204(self, client, scim_settings, scim_mocks):
|
|
row = _user_row(active=True)
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": False}
|
|
response = client.delete(f"/scim/v2/Users/{USER_PK}", headers=AUTH)
|
|
assert response.status_code == 204
|
|
assert response.data == b""
|
|
scim_mocks.users.set_active.assert_called_once_with(USER_PK, False)
|
|
scim_mocks.deny_user.assert_called_once_with("alice@example.com")
|
|
scim_mocks.audit.insert.assert_called_once_with(
|
|
"alice@example.com", "scim_deactivated", metadata={"via": "scim"}
|
|
)
|
|
|
|
def test_delete_already_inactive_has_no_side_effects(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = _user_row(active=False)
|
|
response = client.delete(f"/scim/v2/Users/{USER_PK}", headers=AUTH)
|
|
assert response.status_code == 204
|
|
scim_mocks.users.set_active.assert_not_called()
|
|
scim_mocks.deny_user.assert_not_called()
|
|
|
|
def test_delete_missing_user_returns_404(self, client, scim_settings, scim_mocks):
|
|
scim_mocks.users.get_by_pk.return_value = None
|
|
response = client.delete(f"/scim/v2/Users/{USER_PK}", headers=AUTH)
|
|
assert response.status_code == 404
|
|
|
|
def test_deactivation_revocation_failure_returns_503(self, client, scim_settings, scim_mocks):
|
|
# Redis down: deny_user returns False, so the deactivation rolls back and
|
|
# the IdP gets a 503 to retry instead of a false deprovision success.
|
|
row = _user_row(active=True)
|
|
scim_mocks.users.get_by_pk.return_value = row
|
|
scim_mocks.users.set_active.return_value = {**row, "active": False}
|
|
scim_mocks.deny_user.return_value = False
|
|
response = client.delete(f"/scim/v2/Users/{USER_PK}", headers=AUTH)
|
|
assert response.status_code == 503
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGroups:
|
|
|
|
def test_get_groups_returns_empty_list(self, client, scim_settings, scim_mocks):
|
|
response = client.get("/scim/v2/Groups", headers=AUTH)
|
|
assert response.status_code == 200
|
|
body = response.get_json()
|
|
assert body["schemas"] == [LIST_RESPONSE_URN]
|
|
assert body["totalResults"] == 0
|
|
assert body["Resources"] == []
|
|
|
|
def test_post_groups_not_implemented(self, client, scim_settings, scim_mocks):
|
|
response = client.post("/scim/v2/Groups", headers=AUTH, json={"displayName": "Team"})
|
|
assert response.status_code == 501
|
|
body = response.get_json()
|
|
assert body["schemas"] == [ERROR_URN]
|
|
assert body["detail"] == "Group provisioning is not supported"
|
|
|
|
@pytest.mark.parametrize("method", ["put", "patch", "delete"])
|
|
def test_group_mutations_not_implemented(self, client, scim_settings, scim_mocks, method):
|
|
response = getattr(client, method)("/scim/v2/Groups/some-group", headers=AUTH)
|
|
assert response.status_code == 501
|
|
assert response.get_json()["detail"] == "Group provisioning is not supported"
|