werkzeug engine implementation

.github/workflows/tests.yaml

@@ -31,6 +31,8 @@ jobs:
           enable-cache: true
       - name: Install Python ${{ matrix.python }}
         run: uv python install ${{ matrix.python }}
+      - name: Install dependencies
+        run: uv sync --all-extras
       - name: Run tests
         run: uv run pytest --showlocals
 
@@ -44,7 +46,7 @@ jobs:
         with:
           enable-cache: true
       - name: Install minimum dependencies
-        run: uv sync --resolution=lowest-direct
+        run: uv sync --resolution=lowest-direct --all-extras
       - name: Run tests
         run: uv run pytest --showlocals
 

doc/changelog.rst

@@ -7,12 +7,14 @@ Changelog
 .. warning::
 
     This version comes with breaking changes:
+
     - `httpx` is no longer a direct dependency, it is shipped in the `httpx` packaging extra.
     - Use ``scim2_client.engines.httpx.SyncSCIMClient`` instead of ``scim2_client.SCIMClient``.
 
 Added
 ^^^^^
 - The `Unknown resource type` request error keeps a reference to the faulty payload.
+- New `werkzeug` request engine for application development purpose.
 
 Changed
 ^^^^^^^

doc/conf.py

@@ -42,6 +42,7 @@
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3", None),
     "scim2_models": ("https://scim2-models.readthedocs.io/en/latest/", None),
+    "werkzeug": ("https://werkzeug.palletsprojects.com", None),
 }
 
 # -- Options for HTML output ----------------------------------------------

doc/reference.rst

@@ -4,3 +4,11 @@ Reference
 .. automodule:: scim2_client
    :members:
    :member-order: bysource
+
+.. automodule:: scim2_client.engines.httpx
+   :members:
+   :member-order: bysource
+
+.. automodule:: scim2_client.engines.werkzeug
+   :members:
+   :member-order: bysource

doc/tutorial.rst

@@ -83,6 +83,19 @@ To achieve this, all the methods provide the following parameters, all are :data
    which value will excluded from the request payload, and which values are
    expected in the response payload.
 
+Engines
+=======
+
+scim2-client comes with a light abstraction layers that allows for different requests engines.
+Currently those engines are shipped:
+
+- :class:`~scim2_client.engines.httpx.SyncSCIMClient`: A synchronous engine using `httpx <https://github.com/encode/httpx>`_ to perform the HTTP requests.
+- :class:`~scim2_client.engines.werkzeug.TestSCIMClient`: A test engine for development purposes.
+  It takes a WSGI app and directly execute the server code instead of performing real HTTP requests.
+  This is faster in unit test suites, and helpful to catch the server exceptions.
+
+You can easily implement your own engine by inheriting from :class:`~scim2_client.BaseSCIMClient`.
+
 Additional request parameters
 =============================
 

pyproject.toml

@@ -35,6 +35,10 @@ httpx = [
     "httpx>=0.24.0",
 ]
 
+werkzeug = [
+    "werkzeug>=3.1.3",
+]
+
 [project.urls]
 documentation = "https://scim2-client.readthedocs.io"
 repository = "https://github.com/python-scim/scim2-client"
@@ -48,7 +52,9 @@ dev = [
     "pytest>=8.2.1",
     "pytest-coverage>=0.0",
     "pytest-httpserver>=1.0.10",
+    "scim2-server >= 0.1.2; python_version>='3.10'",
     "tox-uv>=1.16.0",
+    "werkzeug>=3.1.3",
 ]
 doc = [
     "autodoc-pydantic>=2.2.0",

scim2_client/engines/werkzeug.py

@@ -0,0 +1,242 @@
+from contextlib import contextmanager
+from typing import Optional
+from typing import Union
+from urllib.parse import urlencode
+
+from scim2_models import AnyResource
+from scim2_models import Context
+from scim2_models import Error
+from scim2_models import ListResponse
+from scim2_models import Resource
+from scim2_models import SearchRequest
+from werkzeug.test import Client
+
+from scim2_client.client import BaseSCIMClient
+from scim2_client.errors import SCIMClientError
+
+
+@contextmanager
+def handle_response_error(response):
+    try:
+        yield
+
+    except SCIMClientError as exc:
+        exc.source = response
+        raise exc
+
+
+class TestSCIMClient(BaseSCIMClient):
+    """A client based on :class:`Werkzeug test Client <werkzeug.test.Client>` for application development purposes.
+
+    This is helpful for developers of SCIM servers.
+    This client avoids to perform real HTTP requests and directly execute the server code instead.
+    This allows to dynamically catch the exceptions if something gets wrong.
+
+    :param client: A WSGI application instance that will be used to send requests.
+    :param scim_prefix: The scim root endpoint in the application.
+    :param resource_types: The client resource types.
+
+    .. code-block:: python
+
+        from scim2_client.engines.werkzeug import TestSCIMClient
+        from scim2_models import User, Group
+
+        testclient = TestSCIMClient(app=scim_provider, resource_types=(User, Group))
+
+        request_user = User(user_name="foo", display_name="bar")
+        response_user = scim_client.create(request_user)
+        assert response_user.user_name == "foo"
+    """
+
+    def __init__(
+        self,
+        app,
+        scim_prefix: str = "",
+        resource_types: Optional[tuple[type[Resource]]] = None,
+    ):
+        super().__init__(resource_types=resource_types)
+        self.client = Client(app)
+        self.scim_prefix = scim_prefix
+
+    def make_url(self, url: str) -> str:
+        prefix = (
+            self.scim_prefix[:-1]
+            if self.scim_prefix.endswith("/")
+            else self.scim_prefix
+        )
+        return f"{prefix}{url}"
+
+    def create(
+        self,
+        resource: Union[AnyResource, dict],
+        check_request_payload: bool = True,
+        check_response_payload: bool = True,
+        expected_status_codes: Optional[
+            list[int]
+        ] = BaseSCIMClient.CREATION_RESPONSE_STATUS_CODES,
+        raise_scim_errors: bool = True,
+        **kwargs,
+    ) -> Union[AnyResource, Error, dict]:
+        url, payload, expected_types, request_kwargs = self.prepare_create_request(
+            resource=resource,
+            check_request_payload=check_request_payload,
+            check_response_payload=check_response_payload,
+            expected_status_codes=expected_status_codes,
+            raise_scim_errors=raise_scim_errors,
+            **kwargs,
+        )
+
+        response = self.client.post(self.make_url(url), json=payload, **request_kwargs)
+
+        with handle_response_error(payload):
+            return self.check_response(
+                payload=response.json if response.text else None,
+                status_code=response.status_code,
+                headers=response.headers,
+                expected_status_codes=expected_status_codes,
+                expected_types=expected_types,
+                check_response_payload=check_response_payload,
+                raise_scim_errors=raise_scim_errors,
+                scim_ctx=Context.RESOURCE_CREATION_RESPONSE,
+            )
+
+    def query(
+        self,
+        resource_type: Optional[type[Resource]] = None,
+        id: Optional[str] = None,
+        search_request: Optional[Union[SearchRequest, dict]] = None,
+        check_request_payload: bool = True,
+        check_response_payload: bool = True,
+        expected_status_codes: Optional[
+            list[int]
+        ] = BaseSCIMClient.QUERY_RESPONSE_STATUS_CODES,
+        raise_scim_errors: bool = True,
+        **kwargs,
+    ):
+        url, payload, expected_types, request_kwargs = self.prepare_query_request(
+            resource_type=resource_type,
+            id=id,
+            search_request=search_request,
+            check_request_payload=check_request_payload,
+            check_response_payload=check_response_payload,
+            expected_status_codes=expected_status_codes,
+            raise_scim_errors=raise_scim_errors,
+            **kwargs,
+        )
+
+        query_string = urlencode(payload, doseq=False) if payload else None
+        response = self.client.get(
+            self.make_url(url), query_string=query_string, **request_kwargs
+        )
+
+        with handle_response_error(payload):
+            return self.check_response(
+                payload=response.json if response.text else None,
+                status_code=response.status_code,
+                headers=response.headers,
+                expected_status_codes=expected_status_codes,
+                expected_types=expected_types,
+                check_response_payload=check_response_payload,
+                raise_scim_errors=raise_scim_errors,
+                scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
+            )
+
+    def search(
+        self,
+        search_request: Optional[SearchRequest] = None,
+        check_request_payload: bool = True,
+        check_response_payload: bool = True,
+        expected_status_codes: Optional[
+            list[int]
+        ] = BaseSCIMClient.SEARCH_RESPONSE_STATUS_CODES,
+        raise_scim_errors: bool = True,
+        **kwargs,
+    ) -> Union[AnyResource, ListResponse[AnyResource], Error, dict]:
+        url, payload, expected_types, request_kwargs = self.prepare_search_request(
+            search_request=search_request,
+            check_request_payload=check_request_payload,
+            check_response_payload=check_response_payload,
+            expected_status_codes=expected_status_codes,
+            raise_scim_errors=raise_scim_errors,
+            **kwargs,
+        )
+
+        response = self.client.post(self.make_url(url), json=payload, **request_kwargs)
+
+        with handle_response_error(response):
+            return self.check_response(
+                payload=response.json if response.text else None,
+                status_code=response.status_code,
+                headers=response.headers,
+                expected_status_codes=expected_status_codes,
+                expected_types=expected_types,
+                check_response_payload=check_response_payload,
+                raise_scim_errors=raise_scim_errors,
+                scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
+            )
+
+    def delete(
+        self,
+        resource_type: type,
+        id: str,
+        check_response_payload: bool = True,
+        expected_status_codes: Optional[
+            list[int]
+        ] = BaseSCIMClient.DELETION_RESPONSE_STATUS_CODES,
+        raise_scim_errors: bool = True,
+        **kwargs,
+    ) -> Optional[Union[Error, dict]]:
+        url, request_kwargs = self.prepare_delete_request(
+            resource_type=resource_type,
+            id=id,
+            check_response_payload=check_response_payload,
+            expected_status_codes=expected_status_codes,
+            raise_scim_errors=raise_scim_errors,
+            **kwargs,
+        )
+
+        response = self.client.delete(self.make_url(url), **request_kwargs)
+
+        with handle_response_error(response):
+            return self.check_response(
+                payload=response.json if response.text else None,
+                status_code=response.status_code,
+                headers=response.headers,
+                expected_status_codes=expected_status_codes,
+                check_response_payload=check_response_payload,
+                raise_scim_errors=raise_scim_errors,
+            )
+
+    def replace(
+        self,
+        resource: Union[AnyResource, dict],
+        check_request_payload: bool = True,
+        check_response_payload: bool = True,
+        expected_status_codes: Optional[
+            list[int]
+        ] = BaseSCIMClient.REPLACEMENT_RESPONSE_STATUS_CODES,
+        raise_scim_errors: bool = True,
+        **kwargs,
+    ) -> Union[AnyResource, Error, dict]:
+        url, payload, expected_types, request_kwargs = self.prepare_replace_request(
+            resource=resource,
+            check_request_payload=check_request_payload,
+            check_response_payload=check_response_payload,
+            expected_status_codes=expected_status_codes,
+            raise_scim_errors=raise_scim_errors,
+            **kwargs,
+        )
+
+        response = self.client.put(self.make_url(url), json=payload, **request_kwargs)
+
+        with handle_response_error(response):
+            return self.check_response(
+                payload=response.json if response.text else None,
+                status_code=response.status_code,
+                headers=response.headers,
+                expected_status_codes=expected_status_codes,
+                expected_types=expected_types,
+                check_response_payload=check_response_payload,
+                raise_scim_errors=raise_scim_errors,
+                scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE,
+            )

tests/engines/test_werkzeug.py

@@ -0,0 +1,60 @@
+import pytest
+from scim2_models import ResourceType
+from scim2_models import SearchRequest
+from scim2_models import User
+
+from scim2_client.engines.werkzeug import TestSCIMClient
+from scim2_client.errors import SCIMResponseErrorObject
+
+scim2_server = pytest.importorskip("scim2_server")
+from scim2_server.backend import InMemoryBackend  # noqa: E402
+from scim2_server.provider import SCIMProvider  # noqa: E402
+
+
+@pytest.fixture
+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",
+        )
+    )
+    return provider
+
+
+@pytest.fixture
+def scim_client(scim_provider):
+    return TestSCIMClient(app=scim_provider, resource_types=(User,))
+
+
+def test_werkzeug_engine(scim_client):
+    request_user = User(user_name="foo", display_name="bar")
+    response_user = scim_client.create(request_user)
+    assert response_user.user_name == "foo"
+    assert response_user.display_name == "bar"
+
+    response_user = scim_client.query(User, response_user.id)
+    assert response_user.user_name == "foo"
+    assert response_user.display_name == "bar"
+
+    req = SearchRequest()
+    response_users = scim_client.search(req)
+    assert response_users.resources[0].user_name == "foo"
+    assert response_users.resources[0].display_name == "bar"
+
+    request_user = User(id=response_user.id, user_name="foo", display_name="baz")
+    response_user = scim_client.replace(request_user)
+    assert response_user.user_name == "foo"
+    assert response_user.display_name == "baz"
+
+    response_user = scim_client.query(User, response_user.id)
+    assert response_user.user_name == "foo"
+    assert response_user.display_name == "baz"
+
+    scim_client.delete(User, response_user.id)
+    with pytest.raises(SCIMResponseErrorObject):
+        scim_client.query(User, response_user.id)