feat: discover the server configuration endpoints

Author: azmeuk

pyproject.toml

@@ -27,7 +27,7 @@ classifiers = [
 requires-python = ">= 3.10"
 dependencies = [
     "click>=8.1.7",
-    "scim2-client>=0.4.0",
+    "scim2-client>=0.4.2",
     "scim2-tester[httpx]>=0.1.10",
     "sphinx-click-rst-to-ansi-formatter>=0.1.0",
     "pydanclick>=0.3.0",

scim2_cli/__init__.py

@@ -44,6 +44,36 @@ def _is_pydantic_model(model: Any) -> TypeGuard[type[BaseModel]]:
 patch_pydanclick()
 
 
+def load_config_files(
+    schemas_fd, resource_types_fd, service_provider_config_fd
+) -> tuple[
+    list[type[Resource]], list[ResourceType] | None, ServiceProviderConfig | None
+]:
+    if schemas_fd:
+        schemas_payload = json.load(schemas_fd)
+        schemas_obj = [Schema.model_validate(schema) for schema in schemas_payload]
+        resource_models = [Resource.from_schema(schema) for schema in schemas_obj]
+
+    else:
+        resource_models = [User, Group]
+
+    if resource_types_fd:
+        resource_types_payload = json.load(resource_types_fd)
+        resource_types = [
+            ResourceType.model_validate(item) for item in resource_types_payload
+        ]
+    else:
+        resource_types = None
+
+    if service_provider_config_fd:
+        spc_payload = json.load(service_provider_config_fd)
+        service_provider_config = ServiceProviderConfig.model_validate(spc_payload)
+    else:
+        service_provider_config = None
+
+    return resource_models, resource_types, service_provider_config
+
+
 @click.group(cls=make_rst_to_ansi_formatter(DOC_URL, group=True))
 @click.option("-u", "--url", help="The SCIM server endpoint.", envvar="SCIM_CLI_URL")
 @click.option(
@@ -86,36 +116,21 @@ def cli(
     headers_dict = split_headers(header)
     client = Client(base_url=ctx.obj["URL"], headers=headers_dict)
 
-    if schemas:
-        schemas_payload = json.load(schemas)
-        schemas_obj = [Schema.model_validate(schema) for schema in schemas_payload]
-        resource_models = [Resource.from_schema(schema) for schema in schemas_obj]
-
-    else:
-        resource_models = [User, Group]
-
-    if resource_types:
-        resource_types_payload = json.load(resource_types)
-        resource_types_obj = [
-            ResourceType.model_validate(item) for item in resource_types_payload
-        ]
-    else:
-        resource_types_obj = None
-
-    if service_provider_config:
-        spc_payload = json.load(service_provider_config)
-        spc_obj = ServiceProviderConfig.model_validate(spc_payload)
-    else:
-        spc_obj = None
+    resource_models, resource_types_obj, spc_obj = load_config_files(
+        schemas, resource_types, service_provider_config
+    )
 
     scim_client = SyncSCIMClient(
         client,
         resource_models=resource_models,
         resource_types=resource_types_obj,
         service_provider_config=spc_obj,
     )
-    if not resource_types:
-        scim_client.register_naive_resource_types()
+    scim_client.discover(
+        schemas=not bool(schemas),
+        resource_types=not bool(resource_types),
+        service_provider_config=not bool(service_provider_config),
+    )
 
     ctx.obj["client"] = scim_client
     ctx.obj["resource_models"] = {

tests/conftest.py

@@ -1,12 +1,82 @@
 import pytest
 from click.testing import CliRunner
+from scim2_models import AuthenticationScheme
+from scim2_models import Bulk
+from scim2_models import ChangePassword
+from scim2_models import ETag
+from scim2_models import Filter
+from scim2_models import ListResponse
+from scim2_models import Patch
+from scim2_models import ResourceType
+from scim2_models import Schema
+from scim2_models import ServiceProviderConfig
+from scim2_models import Sort
+from scim2_models import User
 
 
 @pytest.fixture
 def runner():
     return CliRunner()
 
 
+@pytest.fixture
+def httpserver(httpserver):
+    httpserver.expect_request("/ServiceProviderConfig").respond_with_json(
+        ServiceProviderConfig(
+            documentation_uri="https://scim.test",
+            patch=Patch(supported=False),
+            bulk=Bulk(supported=False, max_operations=0, max_payload_size=0),
+            change_password=ChangePassword(supported=True),
+            filter=Filter(supported=False, max_results=0),
+            sort=Sort(supported=False),
+            etag=ETag(supported=False),
+            authentication_schemes=[
+                AuthenticationScheme(
+                    name="OAuth Bearer Token",
+                    description="Authentication scheme using the OAuth Bearer Token Standard",
+                    spec_uri="http://www.rfc-editor.org/info/rfc6750",
+                    documentation_uri="https://scim.test",
+                    type="oauthbearertoken",
+                    primary=True,
+                ),
+            ],
+        ).model_dump(),
+        status=200,
+        content_type="application/scim+json",
+    )
+
+    httpserver.expect_request("/ResourceTypes").respond_with_json(
+        ListResponse[ResourceType](
+            total_results=1,
+            start_index=1,
+            items_per_page=1,
+            resources=[
+                ResourceType(
+                    name="User",
+                    id="User",
+                    endpoint="/Users",
+                    schema_="urn:ietf:params:scim:schemas:core:2.0:User",
+                )
+            ],
+        ).model_dump(),
+        status=200,
+        content_type="application/scim+json",
+    )
+
+    httpserver.expect_request("/Schemas").respond_with_json(
+        ListResponse[Schema](
+            total_results=1,
+            start_index=1,
+            items_per_page=1,
+            resources=[User.to_schema()],
+        ).model_dump(),
+        status=200,
+        content_type="application/scim+json",
+    )
+
+    return httpserver
+
+
 @pytest.fixture
 def simple_user_payload(httpserver):
     def wrapped(id):

tests/test_cli.py

@@ -27,6 +27,7 @@ def test_stdin_bad_json(runner, httpserver):
         cli,
         ["--url", httpserver.url_for("/"), "query", "user"],
         input="invalid",
+        catch_exceptions=False,
     )
     assert result.exit_code == 1
     assert "Invalid JSON input." in result.stdout
@@ -99,8 +100,9 @@ def test_env_vars(runner, httpserver, simple_user_payload):
 
 
 def test_custom_configuration(runner, httpserver, simple_user_payload, tmp_path):
+    """Test passing custom .JSON configuration files to the command."""
     spc = ServiceProviderConfig(
-        documentation_uri="https://canaille.readthedocs.io",
+        documentation_uri="https://scim.test",
         patch=Patch(supported=False),
         bulk=Bulk(supported=False, max_operations=0, max_payload_size=0),
         change_password=ChangePassword(supported=True),
@@ -112,7 +114,7 @@ def test_custom_configuration(runner, httpserver, simple_user_payload, tmp_path)
                 name="OAuth Bearer Token",
                 description="Authentication scheme using the OAuth Bearer Token Standard",
                 spec_uri="http://www.rfc-editor.org/info/rfc6750",
-                documentation_uri="https://canaille.readthedocs.io",
+                documentation_uri="https://scim.test",
                 type="oauthbearertoken",
                 primary=True,
             ),
@@ -141,7 +143,6 @@ def test_custom_configuration(runner, httpserver, simple_user_payload, tmp_path)
     with open(resource_types_path, "w") as fd:
         json.dump(resource_types, fd)
 
-    """Test passing custom .JSON configuration files to the command."""
     httpserver.expect_request(
         "/somewhere-different/foobar",
         method="GET",