refactor: describe the server with a ScimProvider built from the configuration files and the discovery

Author: azmeuk

doc/changelog.rst

@@ -7,12 +7,17 @@ Changelog
 Changed
 ^^^^^^^
 - Python 3.11 is now the minimum supported version.
+- scim2-client 0.9.0 and scim2-tester 0.4.0 are now the minimum supported versions.
 
 Fixed
 ^^^^^
-- Without ``--schemas``, the schemas and the resource types are discovered on the server
-  again with scim2-client 0.9. Only ``User`` and ``Group`` were known, without their
-  extensions.
+- What the configuration files leave out is discovered on the server again with
+  scim2-client 0.9. Without ``--schemas``, only ``User`` and ``Group`` were known,
+  without their extensions. With ``--schemas`` only, the resource types were guessed
+  from the schemas.
+- ``--resource-types`` without ``--schemas`` no longer crashes.
+- A server description whose resource types name unknown schemas is reported as a
+  readable error instead of a traceback.
 
 Security
 ^^^^^^^^

pyproject.toml

@@ -27,8 +27,8 @@ classifiers = [
 requires-python = ">= 3.11"
 dependencies = [
     "click>=8.1.7",
-    "scim2-client[httpx2]>=0.8.0",
-    "scim2-tester>=0.3.0",
+    "scim2-client[httpx2]>=0.9.0",
+    "scim2-tester>=0.4.0",
     "sphinx-click-rst-to-ansi-formatter>=0.1.0",
     "pydanclick>=0.5.0",
     "pygments>=2.18.0",

scim2_cli/__init__.py

@@ -5,9 +5,10 @@
 from httpx2 import Client
 from scim2_client.engines.httpx2 import SyncSCIMClient
 from scim2_models import ListResponse
-from scim2_models import Resource
 from scim2_models import ResourceType
 from scim2_models import Schema
+from scim2_models import ScimProvider
+from scim2_models import ScimProviderError
 from scim2_models import ServiceProviderConfig
 from sphinx_click.rst_to_ansi_formatter import make_rst_to_ansi_formatter
 
@@ -25,46 +26,34 @@
 from scim2_cli.utils import split_headers
 
 
-def load_config_files(
-    schemas_fd, resource_types_fd, service_provider_config_fd
-) -> tuple[
-    list[type[Resource]] | None,
-    list[ResourceType] | None,
-    ServiceProviderConfig | None,
-]:
-    if schemas_fd:
-        schemas_payload = json.load(schemas_fd)
-        if isinstance(schemas_payload, dict):
-            schemas_obj = ListResponse[Schema].model_validate(schemas_payload).resources
-        else:
-            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 = None
-
-    if resource_types_fd:
-        resource_types_payload = json.load(resource_types_fd)
-        if isinstance(schemas_payload, dict):
-            resource_types = (
-                ListResponse[ResourceType]
-                .model_validate(resource_types_payload)
-                .resources
-            )
-        else:
-            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
+def load_objects(fd, model):
+    """Read a list of objects, bare or wrapped in a ListResponse, from a JSON file."""
+    payload = json.load(fd)
+    if isinstance(payload, dict):
+        return ListResponse[model].model_validate(payload).resources or []
+    return [model.model_validate(item) for item in payload]
+
+
+def describe_server(
+    scim_client, schemas_fd, resource_types_fd, service_provider_config_fd
+) -> ScimProvider:
+    """Describe the server with the configuration files, and query it for the others."""
+    resource_types = (
+        load_objects(resource_types_fd, ResourceType)
+        if resource_types_fd
+        else scim_client.query(ResourceType).resources or []
+    )
+    schemas = (
+        load_objects(schemas_fd, Schema)
+        if schemas_fd
+        else scim_client.query(Schema).resources or []
+    )
+    config = (
+        ServiceProviderConfig.model_validate(json.load(service_provider_config_fd))
+        if service_provider_config_fd
+        else scim_client.query(ServiceProviderConfig)
+    )
+    return ScimProvider.from_discovery(schemas, resource_types, config)
 
 
 @click.group(cls=make_rst_to_ansi_formatter(DOC_URL, group=True))
@@ -123,31 +112,21 @@ def cli(
     headers_dict = split_headers(header)
     client = Client(base_url=url, headers=headers_dict, verify=not no_verify)
 
-    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,
-    )
+    scim_client = SyncSCIMClient(client)
     try:
-        scim_client.discover(
-            schemas=not bool(schemas),
-            resource_types=not bool(resource_types),
-            service_provider_config=not bool(service_provider_config),
+        scim_client.provider = describe_server(
+            scim_client, schemas, resource_types, service_provider_config
         )
-    except SCIM_EXCEPTIONS as exc:
+    except (*SCIM_EXCEPTIONS, ScimProviderError) as exc:
         raise exception_to_click_error(exc) from exc
 
+    provider = scim_client.provider
     ctx.obj["client"] = scim_client
     ctx.obj["resource_models"] = {
         escape_control_characters(
             re.sub(r"\[.*\]", "", resource_model.__name__.lower())
         ): resource_model
-        for resource_model in ctx.obj["client"].resource_models
+        for resource_model in map(provider.model_for, provider.resource_types)
     }
 
     if not click.get_text_stream("stdin").isatty():  # pragma: no cover

tests/test_cli.py

@@ -275,6 +275,134 @@ def test_custom_configuration_in_list_response(
     assert result.exit_code == 0
 
 
+@pytest.mark.parametrize("in_list_response", [False, True])
+def test_custom_resource_types_only(
+    runner, httpserver, simple_user_payload, tmp_path, resource_types, in_list_response
+):
+    """Test passing only the resource types, the schemas being discovered."""
+    payload = (
+        ListResponse[ResourceType](
+            total_results=len(resource_types),
+            start_index=1,
+            items_per_page=len(resource_types),
+            resources=resource_types,
+        ).model_dump()
+        if in_list_response
+        else [resource_type.model_dump() for resource_type in resource_types]
+    )
+    resource_types_path = tmp_path / "resource_types.json"
+    with open(resource_types_path, "w") as fd:
+        json.dump(payload, fd)
+
+    httpserver.expect_request(
+        "/somewhere-different/foobar",
+        method="GET",
+    ).respond_with_json(
+        simple_user_payload("foobar"),
+        status=200,
+        content_type="application/scim+json",
+    )
+
+    result = runner.invoke(
+        cli,
+        [
+            "--url",
+            httpserver.url_for("/"),
+            "--resource-types",
+            resource_types_path,
+            "query",
+            "user",
+            "foobar",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0, result.output
+
+
+def test_custom_schemas_only(
+    runner,
+    httpserver,
+    simple_user_payload,
+    tmp_path,
+    service_provider_configuration,
+    schemas,
+    resource_types,
+):
+    """Test passing only the schemas, the resource types being discovered."""
+    schemas_path = tmp_path / "schemas.json"
+    with open(schemas_path, "w") as fd:
+        json.dump([schema.model_dump() for schema in schemas], fd)
+
+    httpserver.clear_all_handlers()
+    httpserver.expect_request("/ResourceTypes").respond_with_json(
+        ListResponse[ResourceType](
+            total_results=len(resource_types), resources=resource_types
+        ).model_dump(),
+        content_type="application/scim+json",
+    )
+    httpserver.expect_request("/ServiceProviderConfig").respond_with_json(
+        service_provider_configuration.model_dump(),
+        content_type="application/scim+json",
+    )
+    httpserver.expect_request(
+        "/somewhere-different/foobar",
+        method="GET",
+    ).respond_with_json(
+        simple_user_payload("foobar"),
+        status=200,
+        content_type="application/scim+json",
+    )
+
+    result = runner.invoke(
+        cli,
+        [
+            "--url",
+            httpserver.url_for("/"),
+            "--schemas",
+            schemas_path,
+            "query",
+            "user",
+            "foobar",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0, result.output
+
+
+def test_incoherent_server_description(runner, httpserver, tmp_path, schemas):
+    """Test that a resource type naming an unknown schema displays a readable error."""
+    resource_types_path = tmp_path / "resource_types.json"
+    with open(resource_types_path, "w") as fd:
+        json.dump(
+            [
+                ResourceType(
+                    id="Group",
+                    name="Group",
+                    endpoint="/Groups",
+                    schema_="urn:ietf:params:scim:schemas:core:2.0:Group",
+                ).model_dump()
+            ],
+            fd,
+        )
+
+    result = runner.invoke(
+        cli,
+        [
+            "--url",
+            httpserver.url_for("/"),
+            "--resource-types",
+            resource_types_path,
+            "query",
+        ],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 1, result.output
+    assert (
+        "Error: No resource describes urn:ietf:params:scim:schemas:core:2.0:Group"
+        in result.output
+    )
+
+
 def test_custom_configuration_by_env(
     runner,
     httpserver,