feat: add BaseSCIMClient 'resource_types' parameter

Author: azmeuk

doc/changelog.rst

@@ -1,6 +1,15 @@
 Changelog
 =========
 
+[0.4.0] - Unreleased
+--------------------
+
+.. warning::
+
+    This version comes with breaking changes:
+
+    - :class:`~scim2_client.BaseSCIMClient` takes a mandatory :paramref:`~scim2_client.BaseSCIMClient.resource_types` parameter.
+
 [0.3.3] - 2024-11-29
 --------------------
 

scim2_client/client.py

@@ -47,8 +47,10 @@ class BaseSCIMClient:
 
     This class can be inherited and used as a basis for request engine integration.
 
-    :param resource_models: A tuple of :class:`~scim2_models.Resource` types expected to be handled by the SCIM client.
+    :param resource_models: A collection of :class:`~scim2_models.Resource` models expected to be handled by the SCIM client.
         If a request payload describe a resource that is not in this list, an exception will be raised.
+    :param resource_types: A collection of :class:`~scim2_models.ResourceType` that will be used to guess the
+        server endpoints associated with the resources.
     :param check_request_payload: If :data:`False`,
         :code:`resource` is expected to be a dict that will be passed as-is in the request.
         This value can be overwritten in methods.
@@ -146,13 +148,15 @@ class BaseSCIMClient:
     def __init__(
         self,
         resource_models: Optional[Collection[type[Resource]]] = None,
+        resource_types: Optional[Collection[ResourceType]] = None,
         check_request_payload: bool = True,
         check_response_payload: bool = True,
         raise_scim_errors: bool = True,
     ):
         self.resource_models = tuple(
             set(resource_models or []) | {ResourceType, Schema, ServiceProviderConfig}
         )
+        self.resource_types = resource_types
         self.check_request_payload = check_request_payload
         self.check_response_payload = check_response_payload
         self.raise_scim_errors = raise_scim_errors
@@ -169,16 +173,19 @@ def resource_endpoint(self, resource_model: Optional[type[Resource]]) -> str:
         if resource_model is None:
             return "/"
 
+        if resource_model in (ResourceType, Schema):
+            return f"/{resource_model.__name__}s"
+
         # This one takes no final 's'
         if resource_model is ServiceProviderConfig:
             return "/ServiceProviderConfig"
 
-        try:
-            first_bracket_index = resource_model.__name__.index("[")
-            root_name = resource_model.__name__[:first_bracket_index]
-        except ValueError:
-            root_name = resource_model.__name__
-        return f"/{root_name}s"
+        schema = resource_model.model_fields["schemas"].default[0]
+        for resource_type in self.resource_types or []:
+            if schema == resource_type.schema_:
+                return resource_type.endpoint
+
+        raise SCIMRequestError(f"No ResourceType is matching the schema: {schema}")
 
     def check_response(
         self,

tests/conftest.py

@@ -0,0 +1,21 @@
+import pytest
+from httpx import Client
+from scim2_models import Group
+from scim2_models import ResourceType
+from scim2_models import User
+
+from scim2_client.engines.httpx import SyncSCIMClient
+
+
+@pytest.fixture
+def sync_client(httpserver):
+    client = Client(base_url=f"http://localhost:{httpserver.port}")
+    scim_client = SyncSCIMClient(
+        client,
+        resource_models=[User, Group],
+        resource_types=[
+            ResourceType.from_resource(User),
+            ResourceType.from_resource(Group),
+        ],
+    )
+    return scim_client

tests/engines/test_httpx.py

@@ -23,15 +23,7 @@ def server():
     backend = InMemoryBackend()
     provider = SCIMProvider(backend)
     provider.register_schema(User.to_schema())
-    provider.register_resource_type(
-        ResourceType(
-            id="User",
-            name="User",
-            endpoint="/Users",
-            schema="urn:ietf:params:scim:schemas:core:2.0:User",
-        )
-    )
-
+    provider.register_resource_type(ResourceType.from_resource(User))
     host = "localhost"
     port = portpicker.pick_unused_port()
     httpd = wsgiref.simple_server.make_server(host, port, provider)
@@ -48,7 +40,11 @@ def server():
 def test_sync_engine(server):
     host, port = server
     client = Client(base_url=f"http://{host}:{port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
+    scim_client = SyncSCIMClient(
+        client,
+        resource_models=[User],
+        resource_types=[ResourceType.from_resource(User)],
+    )
 
     request_user = User(user_name="foo", display_name="bar")
     response_user = scim_client.create(request_user)
@@ -81,7 +77,11 @@ def test_sync_engine(server):
 async def test_async_engine(server):
     host, port = server
     client = AsyncClient(base_url=f"http://{host}:{port}")
-    scim_client = AsyncSCIMClient(client, resource_models=(User,))
+    scim_client = AsyncSCIMClient(
+        client,
+        resource_models=(User,),
+        resource_types=[ResourceType.from_resource(User)],
+    )
 
     request_user = User(user_name="foo", display_name="bar")
     response_user = await scim_client.create(request_user)

tests/engines/test_werkzeug.py

@@ -18,20 +18,17 @@
 def scim_provider():
     provider = SCIMProvider(InMemoryBackend())
     provider.register_schema(User.to_schema())
-    provider.register_resource_type(
-        ResourceType(
-            id="User",
-            name="User",
-            endpoint="/Users",
-            schema="urn:ietf:params:scim:schemas:core:2.0:User",
-        )
-    )
+    provider.register_resource_type(ResourceType.from_resource(User))
     return provider
 
 
 @pytest.fixture
 def scim_client(scim_provider):
-    return TestSCIMClient(app=scim_provider, resource_models=(User,))
+    return TestSCIMClient(
+        app=scim_provider,
+        resource_models=(User,),
+        resource_types=[ResourceType.from_resource(User)],
+    )
 
 
 def test_werkzeug_engine(scim_client):

tests/test_create.py

@@ -1,21 +1,20 @@
 import datetime
+from typing import Optional
 
 import pytest
-from httpx import Client
 from scim2_models import Error
-from scim2_models import Group
 from scim2_models import Meta
+from scim2_models import Resource
 from scim2_models import User
 
 from scim2_client import RequestNetworkError
 from scim2_client import RequestPayloadValidationError
 from scim2_client import SCIMClientError
 from scim2_client import SCIMRequestError
 from scim2_client import UnexpectedStatusCode
-from scim2_client.engines.httpx import SyncSCIMClient
 
 
-def test_create_user(httpserver):
+def test_create_user(httpserver, sync_client):
     """Nominal case for a User creation object."""
     httpserver.expect_request("/Users", method="POST").respond_with_json(
         {
@@ -34,10 +33,7 @@ def test_create_user(httpserver):
     )
 
     user_request = User(user_name="bjensen@example.com")
-
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
-    response = scim_client.create(user_request)
+    response = sync_client.create(user_request)
 
     user_created = User(
         id="2819c223-7f76-453a-919d-413861904646",
@@ -57,7 +53,7 @@ def test_create_user(httpserver):
     assert response == user_created
 
 
-def test_create_dict_user(httpserver):
+def test_create_dict_user(httpserver, sync_client):
     """Nominal case for a User creation object, when passing a dict instead of a resource."""
     httpserver.expect_request("/Users", method="POST").respond_with_json(
         {
@@ -80,9 +76,7 @@ def test_create_dict_user(httpserver):
         "userName": "bjensen@example.com",
     }
 
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
-    response = scim_client.create(user_request)
+    response = sync_client.create(user_request)
 
     user_created = User(
         id="2819c223-7f76-453a-919d-413861904646",
@@ -102,7 +96,7 @@ def test_create_dict_user(httpserver):
     assert response == user_created
 
 
-def test_create_dict_user_bad_schema(httpserver):
+def test_create_dict_user_bad_schema(httpserver, sync_client):
     """Test when passing a resource dict with an unknown or invalid schema."""
     httpserver.expect_request("/Users", method="POST").respond_with_json(
         {
@@ -125,29 +119,25 @@ def test_create_dict_user_bad_schema(httpserver):
         "userName": "bjensen@example.com",
     }
 
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
     with pytest.raises(
         SCIMClientError, match="Cannot guess resource type from the payload"
     ):
-        scim_client.create(user_request)
+        sync_client.create(user_request)
 
 
-def test_dont_check_response_payload(httpserver):
+def test_dont_check_response_payload(httpserver, sync_client):
     """Test the check_response_payload_attribute."""
     httpserver.expect_request("/Users", method="POST").respond_with_json(
         {"foo": "bar"}, status=201
     )
 
     user_request = User(user_name="bjensen@example.com")
 
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
-    response = scim_client.create(resource=user_request, check_response_payload=False)
+    response = sync_client.create(resource=user_request, check_response_payload=False)
     assert response == {"foo": "bar"}
 
 
-def test_dont_check_request_payload(httpserver):
+def test_dont_check_request_payload(httpserver, sync_client):
     """Test the check_request_payload_attribute.
 
     TODO: Actually check that the payload is sent through the network
@@ -173,9 +163,7 @@ def test_dont_check_request_payload(httpserver):
         "userName": "bjensen@example.com",
     }
 
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
-    response = scim_client.create(
+    response = sync_client.create(
         resource=user_request, check_request_payload=False, url="/Users"
     )
 
@@ -194,7 +182,7 @@ def test_dont_check_request_payload(httpserver):
     assert response == user_created
 
 
-def test_conflict(httpserver):
+def test_conflict(httpserver, sync_client):
     """Nominal case for a User creation object."""
     httpserver.expect_request("/Users", method="POST").respond_with_json(
         {
@@ -208,9 +196,7 @@ def test_conflict(httpserver):
 
     user_request = User(user_name="bjensen@example.com")
 
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
-    response = scim_client.create(user_request, raise_scim_errors=False)
+    response = sync_client.create(user_request, raise_scim_errors=False)
     assert response == Error(
         schemas=["urn:ietf:params:scim:api:messages:2.0:Error"],
         status=409,
@@ -219,7 +205,7 @@ def test_conflict(httpserver):
     )
 
 
-def test_no_200(httpserver):
+def test_no_200(httpserver, sync_client):
     """User creation object should return 201 codes and no 200."""
     httpserver.expect_request("/Users", method="POST").respond_with_json(
         {
@@ -239,16 +225,14 @@ def test_no_200(httpserver):
 
     user_request = User(user_name="bjensen@example.com")
 
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
     with pytest.raises(UnexpectedStatusCode):
-        scim_client.create(user_request)
-    scim_client.create(user_request, expected_status_codes=None)
-    scim_client.create(user_request, expected_status_codes=[200, 201])
+        sync_client.create(user_request)
+    sync_client.create(user_request, expected_status_codes=None)
+    sync_client.create(user_request, expected_status_codes=[200, 201])
 
 
 @pytest.mark.parametrize("code", [400, 401, 403, 404, 500])
-def test_errors(httpserver, code):
+def test_errors(httpserver, code, sync_client):
     """Test error cases defined in RFC7644."""
     httpserver.expect_request("/Users", method="POST").respond_with_json(
         {
@@ -261,9 +245,7 @@ def test_errors(httpserver, code):
 
     user_request = User(user_name="bjensen@example.com")
 
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
-    response = scim_client.create(user_request, raise_scim_errors=False)
+    response = sync_client.create(user_request, raise_scim_errors=False)
 
     assert response == Error(
         schemas=["urn:ietf:params:scim:api:messages:2.0:Error"],
@@ -272,35 +254,34 @@ def test_errors(httpserver, code):
     )
 
 
-def test_invalid_resource_model(httpserver):
+def test_invalid_resource_model(sync_client):
     """Test that resource_models passed to the method must be part of BaseSCIMClient.resource_models."""
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
+
+    class MyResource(Resource):
+        schemas: list[str] = ["urn:ietf:params:scim:schemas:core:2.0:MyResource"]
+        display_name: Optional[str] = None
+
     with pytest.raises(SCIMRequestError, match=r"Unknown resource type"):
-        scim_client.create(Group(display_name="foobar"))
+        sync_client.create(MyResource(display_name="foobar"))
 
 
-def test_request_validation_error(httpserver):
+def test_request_validation_error(sync_client):
     """Test that incorrect input raise a RequestPayloadValidationError."""
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
     with pytest.raises(
         RequestPayloadValidationError, match="Server request payload validation error"
     ):
-        scim_client.create(
+        sync_client.create(
             {
                 "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
                 "active": "not-a-bool",
             }
         )
 
 
-def test_request_network_error(httpserver):
+def test_request_network_error(sync_client):
     """Test that httpx exceptions are transformed in RequestNetworkError."""
-    client = Client(base_url=f"http://localhost:{httpserver.port}")
-    scim_client = SyncSCIMClient(client, resource_models=(User,))
     user_request = User(user_name="bjensen@example.com")
     with pytest.raises(
         RequestNetworkError, match="Network error happened during request"
     ):
-        scim_client.create(user_request, url="http://invalid.test")
+        sync_client.create(user_request, url="http://invalid.test")