diff --git a/README.md b/README.md index 9492fa8..c21fd59 100644 --- a/README.md +++ b/README.md @@ -13,8 +13,7 @@ they are lost once the process exits. - [x] Unique Constraints - [x] HTTP PATCH (Add/Remove/Replace) - [x] Sorting - -The only optional feature currently missing is support for Bulk operations ([RFC 7644, Section 3.7](https://datatracker.ietf.org/doc/html/rfc7644#section-3.7)). +- [x] Bulk operations ([RFC 7644, Section 3.7](https://datatracker.ietf.org/doc/html/rfc7644#section-3.7)) ## Usage diff --git a/scim2_server/bulk.py b/scim2_server/bulk.py new file mode 100644 index 0000000..8d14689 --- /dev/null +++ b/scim2_server/bulk.py @@ -0,0 +1,164 @@ +from collections.abc import Callable +from typing import Any + +from scim2_models import InvalidValueException +from scim2_models import Resource +from werkzeug.exceptions import Conflict + +BULK_ID_PREFIX = "bulkId:" + +Resolver = Callable[[Any], Any] +"""Replaces the bulkId references of a raw operation.""" + +OperationRunner = Callable[[Any, Resolver], tuple[dict[str, Any], Resource | None]] +"""Applies a raw operation once resolved, and returns its outcome and the resource it acted on.""" + + +def raw_attribute(payload: Any, name: str) -> Any: + """Return an attribute of a raw payload, whose names are case insensitive.""" + if not isinstance(payload, dict): + return None + return next( + (value for key, value in payload.items() if key.casefold() == name.casefold()), + None, + ) + + +def replace_bulk_ids(value: Any, replace: Callable[[str], str]) -> Any: + """Replace every "bulkId:" reference of a raw value.""" + if isinstance(value, str) and value.startswith(BULK_ID_PREFIX): + return replace(value.removeprefix(BULK_ID_PREFIX)) + if isinstance(value, list): + return [replace_bulk_ids(item, replace) for item in value] + if isinstance(value, dict): + return {key: replace_bulk_ids(item, replace) for key, item in value.items()} + return value + + +def resolve_operation(payload: Any, replace: Callable[[str], str]) -> Any: + """Replace the "bulkId:" references of the path and the data of a raw bulk operation.""" + if not isinstance(payload, dict): + return payload + + resolved = {} + for key, value in payload.items(): + if key.casefold() == "path" and isinstance(value, str): + value = "/".join( + replace_bulk_ids(segment, replace) for segment in value.split("/") + ) + elif key.casefold() == "data": + value = replace_bulk_ids(value, replace) + resolved[key] = value + return resolved + + +class BulkJob: + """Run the operations of a bulk request, and resolve their "bulkId:" references. + + RFC 7644 §3.7.2 lets an operation reference a resource that another POST + of the same request creates. The operations run in the order of the + request, except that a POST runs before the first operation that + references it. The results keep the order of the request. + """ + + def __init__( + self, + operations: list[Any], + fail_on_errors: int | None, + run: OperationRunner, + ): + self.run_resolved = run + self.operations = operations + self.fail_on_errors = fail_on_errors + self.results: dict[int, dict[str, Any]] = {} + self.created: dict[str, Resource] = {} + self.running: set[int] = set() + self.errors = 0 + + self.creations: dict[str, int] = {} + for index, payload in enumerate(operations): + bulk_id = raw_attribute(payload, "bulkId") + if raw_attribute(payload, "method") == "POST" and isinstance(bulk_id, str): + self.creations.setdefault(bulk_id, index) + + @property + def stopped(self) -> bool: + """Whether the job reached the number of errors the client accepts. + + RFC 7644 §3.7.3: the job goes on despite failures, unless the client + caps the errors it accepts with "failOnErrors". + """ + return ( + self.errors > 0 + and self.fail_on_errors is not None + and self.errors >= self.fail_on_errors + ) + + def run(self) -> list[dict[str, Any]]: + """Run every operation, and return the results of the operations that ran.""" + for index in range(len(self.operations)): + self.run_operation(index) + return [self.results[index] for index in sorted(self.results)] + + def run_operation(self, index: int) -> None: + """Run an operation, after the creations it references.""" + if index in self.results or index in self.running or self.stopped: + return + + payload = self.operations[index] + self.running.add(index) + for bulk_id in self.references(payload): + if bulk_id in self.creations: + self.run_operation(self.creations[bulk_id]) + + if not self.stopped: + result, resource = self.run_resolved( + payload, lambda payload: self.resolve(index, payload) + ) + self.results[index] = result + if result["status"] >= 400: + self.errors += 1 + elif resource is not None and self.is_creation(index, result["bulk_id"]): + self.created[result["bulk_id"]] = resource + self.running.discard(index) + + def is_creation(self, index: int, bulk_id: str | None) -> bool: + """Whether an operation is the creation a bulkId references.""" + return bulk_id is not None and self.creations.get(bulk_id) == index + + @staticmethod + def references(payload: Any) -> list[str]: + """Return the bulkIds an operation references.""" + bulk_ids: list[str] = [] + + def collect(bulk_id: str) -> str: + bulk_ids.append(bulk_id) + return bulk_id + + resolve_operation(payload, collect) + return bulk_ids + + def resolve(self, index: int, payload: Any) -> Any: + """Replace the bulkId references of an operation with the identifiers of the created resources. + + :raises Conflict: When a referenced resource was not created, as + RFC 7644 §3.7.1 allows for circular references. + """ + bulk_id = raw_attribute(payload, "bulkId") + if ( + raw_attribute(payload, "method") == "POST" + and isinstance(bulk_id, str) + and not self.is_creation(index, bulk_id) + ): + raise InvalidValueException( + detail=f"The bulkId {bulk_id} is not unique in the request" + ) + + def replace(bulk_id: str) -> str: + if bulk_id in self.created: + return str(self.created[bulk_id].id) + if self.creations.get(bulk_id) in self.running: + raise Conflict(f"The bulkId {bulk_id} is part of a circular reference") + raise Conflict(f"No resource was created with the bulkId {bulk_id}") + + return resolve_operation(payload, replace) diff --git a/scim2_server/provider.py b/scim2_server/provider.py index 358734c..110522d 100644 --- a/scim2_server/provider.py +++ b/scim2_server/provider.py @@ -7,9 +7,14 @@ from urllib.parse import urljoin from pydantic import ValidationError +from scim2_models import Bulk +from scim2_models import BulkOperation +from scim2_models import BulkRequest +from scim2_models import BulkResponse from scim2_models import Context from scim2_models import Error from scim2_models import Filter +from scim2_models import InvalidValueException from scim2_models import ListResponse from scim2_models import Meta from scim2_models import Patch @@ -25,18 +30,24 @@ from scim2_models import Sort from werkzeug import Request from werkzeug import Response +from werkzeug.datastructures import ETags from werkzeug.exceptions import Forbidden from werkzeug.exceptions import HTTPException from werkzeug.exceptions import NotFound from werkzeug.exceptions import NotImplemented as WerkzeugNotImplemented from werkzeug.exceptions import PreconditionFailed +from werkzeug.exceptions import RequestEntityTooLarge from werkzeug.exceptions import Unauthorized +from werkzeug.http import parse_etags from werkzeug.http import unquote_etag from werkzeug.routing import Map from werkzeug.routing import Rule from werkzeug.routing.exceptions import RequestRedirect from scim2_server.backend import Backend +from scim2_server.bulk import BulkJob +from scim2_server.bulk import Resolver +from scim2_server.bulk import raw_attribute from scim2_server.utils import load_default_service_provider_config SEARCH_REQUEST_PARAMETERS = ( @@ -49,6 +60,13 @@ "count", ) +BULK_SUCCESS_STATUS = { + BulkOperation.Method.post: 201, + BulkOperation.Method.put: 200, + BulkOperation.Method.patch: 200, + BulkOperation.Method.delete: 204, +} + class SCIMApplication: """A WSGI application implementing a SCIM provider (server).""" @@ -166,8 +184,14 @@ def etag_header(resource: Resource) -> dict[str, str]: """Return the ETag header of a published resource, if it has a version.""" return {"ETag": resource.meta.version} if resource.meta.version else {} - def check_preconditions(self, request: Request, resource: Resource) -> bool: - """Evaluate the "If-Match" and "If-None-Match" headers against a resource. + def check_preconditions( + self, + resource: Resource, + method: str, + if_match: ETags | None = None, + if_none_match: ETags | None = None, + ) -> bool: + """Evaluate the "If-Match" and "If-None-Match" conditions against a resource. RFC 7232 §6 evaluates "If-Match" first: a failed "If-Match" answers 412 whatever the method, a failed "If-None-Match" answers 304 to a GET @@ -183,16 +207,84 @@ def check_preconditions(self, request: Request, resource: Resource) -> bool: ) # RFC 7232 §3.1 compares If-Match strongly, which would never match # the weak ETags RFC 7644 §3.14 recommends and sends in its example. - if request.if_match and not request.if_match.contains_weak(version): + if if_match and not if_match.contains_weak(version): raise PreconditionFailed - if request.if_none_match and request.if_none_match.contains_weak(version): - if request.method == "GET": + if if_none_match and if_none_match.contains_weak(version): + if method == "GET": return False raise PreconditionFailed return True + def get_existing_resource( + self, resource_type: ResourceType, resource_id: str + ) -> Resource: + """Return a stored resource. + + :raises NotFound: When no resource of this type has this identifier. + """ + resource = self.backend.get_resource(resource_type, resource_id) + if resource is None: + raise NotFound + return resource + + def create(self, resource_type: ResourceType, payload: Any) -> Resource: + """Validate a creation payload and store the new resource.""" + resource = self.get_model(resource_type).model_validate( + payload, scim_ctx=Context.RESOURCE_CREATION_REQUEST + ) + return self.backend.create_resource(resource_type, resource) + + def replace( + self, + resource_type: ResourceType, + resource_id: str, + payload: Any, + if_match: ETags | None = None, + if_none_match: ETags | None = None, + ) -> Resource: + """Replace a stored resource with a payload and return the stored result.""" + resource = self.get_existing_resource(resource_type, resource_id) + self.check_preconditions(resource, "PUT", if_match, if_none_match) + + replacement = self.get_model(resource_type).model_validate( + payload, scim_ctx=Context.RESOURCE_REPLACEMENT_REQUEST + ) + replacement.replace(resource) + return self.backend.update_resource(resource_type, replacement) + + def patch( + self, + resource_type: ResourceType, + resource_id: str, + payload: Any, + if_match: ETags | None = None, + if_none_match: ETags | None = None, + ) -> Resource: + """Apply a PATCH payload to a stored resource and return the stored result.""" + self.ensure_supported(self.config.patch, "PATCH") + patch_operation = PatchOp[self.get_model(resource_type)].model_validate(payload) + resource = self.get_existing_resource(resource_type, resource_id) + self.check_preconditions(resource, "PATCH", if_match, if_none_match) + + # A PATCH that changes nothing keeps meta.lastModified and the ETag. + if not patch_operation.patch(resource): + return resource + return self.backend.update_resource(resource_type, resource) + + def delete( + self, + resource_type: ResourceType, + resource_id: str, + if_match: ETags | None = None, + if_none_match: ETags | None = None, + ) -> None: + """Delete a stored resource.""" + resource = self.get_existing_resource(resource_type, resource_id) + self.check_preconditions(resource, "DELETE", if_match, if_none_match) + self.backend.delete_resource(resource_type, resource_id) + def call_single_resource( self, request: Request, resource_endpoint: str, resource_id: str, **kwargs ) -> Response: @@ -202,11 +294,12 @@ def call_single_resource( match request.method: case "GET": - resource = self.backend.get_resource(resource_type, resource_id) - if resource is None: - raise NotFound - resource = self.publish(request, resource) - if not self.check_preconditions(request, resource): + resource = self.publish( + request, self.get_existing_resource(resource_type, resource_id) + ) + if not self.check_preconditions( + resource, "GET", request.if_match, request.if_none_match + ): # RFC 7232 §4.1: a 304 carries the ETag a 200 would have return self.make_response( None, status=304, headers=self.etag_header(resource) @@ -222,69 +315,56 @@ def call_single_resource( ) ) case "DELETE": - resource = self.backend.get_resource(resource_type, resource_id) - if resource is None: - raise NotFound - self.check_preconditions(request, resource) - self.backend.delete_resource(resource_type, resource_id) + self.delete( + resource_type, resource_id, request.if_match, request.if_none_match + ) return self.make_response(None, 204) case "PUT": response_parameters = self.get_response_parameters( request, self.get_model(resource_type) ) - resource = self.backend.get_resource(resource_type, resource_id) - if resource is None: - raise NotFound - self.check_preconditions(request, resource) - - replacement = self.get_model(resource_type).model_validate( - request.json, scim_ctx=Context.RESOURCE_REPLACEMENT_REQUEST + resource = self.replace( + resource_type, + resource_id, + request.json, + request.if_match, + request.if_none_match, ) - replacement.replace(resource) - updated = self.backend.update_resource(resource_type, replacement) - updated = self.publish(request, updated) + resource = self.publish(request, resource) return self.make_response( - updated.model_dump( + resource.model_dump( scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE, response_parameters=response_parameters, ) ) case _: # "PATCH" - self.ensure_supported(self.config.patch, "PATCH") - ResourceModel = self.get_model(resource_type) - patch_operation = PatchOp[ResourceModel].model_validate(request.json) response_parameters = self.get_response_parameters( - request, ResourceModel + request, self.get_model(resource_type) ) - resource = self.backend.get_resource(resource_type, resource_id) - if resource is None: - raise NotFound - self.check_preconditions(request, resource) - - # A PATCH that changes nothing keeps meta.lastModified and the ETag. - if patch_operation.patch(resource): - resource = self.backend.update_resource(resource_type, resource) - + resource = self.patch( + resource_type, + resource_id, + request.json, + request.if_match, + request.if_none_match, + ) + resource = self.publish(request, resource) if ( - response_parameters.attributes - or response_parameters.excluded_attributes + not response_parameters.attributes + and not response_parameters.excluded_attributes ): - resource = self.publish(request, resource) + # RFC 7644 §3.5.2: a PATCH MAY answer 204 when no + # attributes were requested. return self.make_response( - resource.model_dump( - scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE, - response_parameters=response_parameters, - ) + None, 204, headers=self.etag_header(resource) ) - else: - # RFC 7644, section 3.5.2: - # A PATCH operation MAY return a 204 (no content) - # if no attributes were requested - return self.make_response( - None, - 204, - headers=self.etag_header(self.publish(request, resource)), + + return self.make_response( + resource.model_dump( + scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE, + response_parameters=response_parameters, ) + ) @staticmethod def get_response_parameters( @@ -377,12 +457,9 @@ def call_resource( ) ) case _: # "POST" - payload = request.json - resource = self.get_model(resource_type).model_validate( - payload, scim_ctx=Context.RESOURCE_CREATION_REQUEST + created_resource = self.publish( + request, self.create(resource_type, request.json) ) - created_resource = self.backend.create_resource(resource_type, resource) - created_resource = self.publish(request, created_resource) return self.make_response( created_resource.model_dump( scim_ctx=Context.RESOURCE_CREATION_RESPONSE @@ -411,7 +488,9 @@ def call_resource_search( ) @staticmethod - def ensure_supported(capability: Patch | Filter | Sort | None, operation: str): + def ensure_supported( + capability: Patch | Bulk | Filter | Sort | None, operation: str + ): """Refuse with a 501 an operation the configuration does not declare supported. RFC 7644 §3.12 answers 501 when the service provider does not support @@ -420,9 +499,140 @@ def ensure_supported(capability: Patch | Filter | Sort | None, operation: str): if capability is None or not capability.supported: raise WerkzeugNotImplemented(f"{operation} is not supported") - def call_bulk(self, request: Request, **kwargs): - """Implement the /Bulk endpoint, which this server does not support.""" - raise WerkzeugNotImplemented("Bulk operations are not supported") + def call_bulk(self, request: Request, **kwargs) -> Response: + """Implement the /Bulk endpoint (RFC 7644 §3.7).""" + self.ensure_supported(self.config.bulk, "Bulk") + bulk = cast(Bulk, self.config.bulk) + + if ( + bulk.max_payload_size is not None + and len(request.get_data()) > bulk.max_payload_size + ): + raise RequestEntityTooLarge( + f"The payload exceeds the maxPayloadSize ({bulk.max_payload_size} bytes)" + ) + + bulk_request, operations = self.read_bulk_request(request.json) + if bulk.max_operations is not None and len(operations) > bulk.max_operations: + raise RequestEntityTooLarge( + f"The number of operations exceeds the maxOperations ({bulk.max_operations})" + ) + + results = BulkJob( + operations, + bulk_request.fail_on_errors, + lambda payload, resolve: self.run_bulk_operation(request, payload, resolve), + ).run() + return self.make_response( + BulkResponse[Union[tuple(self.get_models())]]( # noqa: UP007 + operations=results + ).model_dump(scim_ctx=Context.BULK_RESPONSE) + ) + + def read_bulk_request(self, payload: Any) -> tuple[BulkRequest, list[Any]]: + """Validate the envelope of a bulk request, and return it with its raw operations. + + Each operation is validated on its own, so an invalid operation only + fails itself (RFC 7644 §3.7.3). + """ + key = ( + next((key for key in payload if key.casefold() == "operations"), None) + if isinstance(payload, dict) + else None + ) + operations = payload[key] if key else None + envelope = {**payload, key: []} if isinstance(operations, list) else payload + bulk_request = BulkRequest[Union[tuple(self.get_models())]].model_validate( # noqa: UP007 + envelope, scim_ctx=Context.BULK_REQUEST + ) + return bulk_request, cast(list[Any], operations) + + def run_bulk_operation( + self, request: Request, payload: Any, resolve: Resolver + ) -> tuple[dict[str, Any], Resource | None]: + """Apply one operation of a bulk job. + + :return: The outcome of the operation, and the resource it created or updated. + """ + method = raw_attribute(payload, "method") + bulk_id = raw_attribute(payload, "bulkId") + result: dict[str, Any] = { + "method": method + if method in [member.value for member in BulkOperation.Method] + else None, + "bulk_id": bulk_id if isinstance(bulk_id, str) else None, + } + + try: + payload = resolve(payload) + resource_type, resource_id = self.get_bulk_target(payload) + if resource_id: + result["location"] = urljoin( + request.url, f"{resource_type.endpoint.strip('/')}/{resource_id}" + ) + operation = BulkOperation[self.get_model(resource_type)].model_validate( + payload, scim_ctx=Context.BULK_REQUEST + ) + resource = self.apply_bulk_operation(resource_type, resource_id, operation) + except Exception as exception: + error = self.error_from(exception) + return {**result, "status": error.status, "response": error}, None + + result["status"] = BULK_SUCCESS_STATUS[operation.method] + if resource is None: + return result, None + + resource = self.publish(request, resource) + result["location"] = resource.meta.location + result["version"] = resource.meta.version + return result, resource + + def get_bulk_target(self, payload: Any) -> tuple[ResourceType, str]: + """Return the resource type and the resource identifier of a bulk operation path. + + :raises NotFound: When the path does not start with a resource type endpoint. + """ + path = raw_attribute(payload, "path") + if not isinstance(path, str): + raise InvalidValueException( + detail="path is required for request operations" + ) + + endpoint, _, resource_id = path.lstrip("/").partition("/") + resource_type = self.get_resource_type_by_endpoint(endpoint) + if resource_type is None: + raise NotFound + return resource_type, resource_id + + def apply_bulk_operation( + self, + resource_type: ResourceType, + resource_id: str, + operation: BulkOperation, + ) -> Resource | None: + """Apply a validated bulk operation, and return the resource it acted on. + + The data of the operation is already validated, and the resource + operations take it as it is. + """ + if (operation.method == BulkOperation.Method.post) == bool(resource_id): + raise InvalidValueException( + detail="A POST path must target a resource type endpoint, other methods a resource" + ) + + if_match = parse_etags(operation.version) if operation.version else None + match operation.method: + case BulkOperation.Method.post: + return self.create(resource_type, operation.data) + case BulkOperation.Method.put: + return self.replace( + resource_type, resource_id, operation.data, if_match + ) + case BulkOperation.Method.patch: + return self.patch(resource_type, resource_id, operation.data, if_match) + case _: # DELETE + self.delete(resource_type, resource_id, if_match) + return None def call_me(self, request: Request, **kwargs): """Implement the /Me endpoint. @@ -468,6 +678,19 @@ def make_response(content, status=200, **kwargs) -> Response: **kwargs, ) + def error_from(self, exception: Exception) -> Error: + """Log an exception raised while serving a request and return its SCIM Error.""" + self.log.exception(exception) + match exception: + case HTTPException(): + return Error(status=exception.code, detail=exception.description) + case SCIMException(): + return exception.to_error() + case ValidationError(): + return Error.from_validation_errors(exception)[0] + case _: + return Error(status=500, detail="Internal server error") + def make_error(self, error: Error): """Construct a werkzeug response from a SCIM Error.""" return self.make_response(error.model_dump(), status=int(error.status)) @@ -553,18 +776,8 @@ def wsgi_app(self, request: Request, environ): # urls.match may cause a redirect, handle it as a special case of HTTPException self.log.exception(e) return e.get_response(environ) - except HTTPException as e: - self.log.exception(e) - return self.make_error(Error(status=e.code, detail=e.description)) - except SCIMException as e: - self.log.exception(e) - return self.make_error(e.to_error()) - except ValidationError as e: - self.log.exception(e) - return self.make_error(Error.from_validation_errors(e)[0]) except Exception as e: - self.log.exception(e) - return self.make_error(Error(status=500, detail="Internal server error")) + return self.make_error(self.error_from(e)) def __call__(self, environ, start_response): """Return the actual WSGI server implementation.""" diff --git a/scim2_server/resources/default-service-provider-config.json b/scim2_server/resources/default-service-provider-config.json index 5a7003e..cc8d9d6 100644 --- a/scim2_server/resources/default-service-provider-config.json +++ b/scim2_server/resources/default-service-provider-config.json @@ -6,7 +6,9 @@ "supported": true }, "bulk": { - "supported": false + "supported": true, + "maxOperations": 1000, + "maxPayloadSize": 1048576 }, "filter": { "supported": true, diff --git a/tests/integration/test_bulk.py b/tests/integration/test_bulk.py new file mode 100644 index 0000000..d9731a7 --- /dev/null +++ b/tests/integration/test_bulk.py @@ -0,0 +1,542 @@ +import pytest +from scim2_models import Bulk +from scim2_models import ETag +from scim2_models import Patch + +from scim2_server.utils import load_default_service_provider_config + +BULK_REQUEST = "urn:ietf:params:scim:api:messages:2.0:BulkRequest" +PATCH_OP = "urn:ietf:params:scim:api:messages:2.0:PatchOp" + + +def bulk(client, operations, **attributes): + return client.post( + "/v2/Bulk", + json={"schemas": [BULK_REQUEST], "Operations": operations, **attributes}, + ) + + +def create_user(user_name, bulk_id="u"): + return { + "method": "POST", + "path": "/Users", + "bulkId": bulk_id, + "data": {"userName": user_name}, + } + + +def configured(wsgi_with, **settings): + config = load_default_service_provider_config() + for name, value in settings.items(): + setattr(config, name, value) + return wsgi_with(config) + + +class TestBulkOperations: + def test_post(self, wsgi): + """RFC 7644 §3.7: a creation answers 201 with the location, the version and the bulkId.""" + r = bulk(wsgi, [create_user("alice", "qwerty")]) + assert r.status_code == 200 + assert r.json()["schemas"] == [ + "urn:ietf:params:scim:api:messages:2.0:BulkResponse" + ] + (result,) = r.json()["Operations"] + assert result["method"] == "POST" + assert result["bulkId"] == "qwerty" + assert result["status"] == "201" + assert "response" not in result + + user = wsgi.get(result["location"]).json() + assert user["userName"] == "alice" + assert result["version"] == user["meta"]["version"] + + def test_put(self, wsgi, first_fake_user): + """A replacement answers 200 and replaces the resource.""" + r = bulk( + wsgi, + [ + { + "method": "PUT", + "path": f"/Users/{first_fake_user}", + "data": {"userName": "bob"}, + } + ], + ) + (result,) = r.json()["Operations"] + assert result["status"] == "200" + assert ( + result["location"] == f"https://scim.example.com/v2/Users/{first_fake_user}" + ) + user = wsgi.get(f"/v2/Users/{first_fake_user}").json() + assert user["userName"] == "bob" + assert result["version"] == user["meta"]["version"] + + def test_patch(self, wsgi, first_fake_user): + """A PATCH answers 200 and edits the resource.""" + r = bulk( + wsgi, + [ + { + "method": "PATCH", + "path": f"/Users/{first_fake_user}", + "data": { + "schemas": [PATCH_OP], + "Operations": [ + {"op": "replace", "path": "userName", "value": "carol"} + ], + }, + } + ], + ) + (result,) = r.json()["Operations"] + assert result["status"] == "200" + user = wsgi.get(f"/v2/Users/{first_fake_user}").json() + assert user["userName"] == "carol" + assert result["version"] == user["meta"]["version"] + + def test_delete(self, wsgi, first_fake_user): + """A deletion answers 204 with the location of the deleted resource.""" + r = bulk(wsgi, [{"method": "DELETE", "path": f"/Users/{first_fake_user}"}]) + (result,) = r.json()["Operations"] + assert result == { + "method": "DELETE", + "location": f"https://scim.example.com/v2/Users/{first_fake_user}", + "status": "204", + } + assert wsgi.get(f"/v2/Users/{first_fake_user}").status_code == 404 + + def test_operations_run_in_order(self, wsgi): + """Each operation sees the changes of the previous ones.""" + r = bulk(wsgi, [create_user("dave"), create_user("dave", "other")]) + first, second = r.json()["Operations"] + assert first["status"] == "201" + assert second["status"] == "409" + assert second["response"]["scimType"] == "uniqueness" + assert "location" not in second + + def test_paths_are_case_insensitive_and_may_omit_the_leading_slash(self, wsgi): + """The endpoint of a path is compared as the HTTP endpoints are.""" + operation = {**create_user("erin"), "path": "users"} + (result,) = bulk(wsgi, [operation]).json()["Operations"] + assert result["status"] == "201" + + +class TestBulkOperationErrors: + def test_invalid_data_only_fails_its_operation(self, wsgi): + """RFC 7644 §3.7.3: an invalid operation answers 400 inside a job that answers 200.""" + r = bulk( + wsgi, + [ + {"method": "POST", "path": "/Users", "bulkId": "a", "data": {}}, + create_user("frank"), + ], + ) + assert r.status_code == 200 + first, second = r.json()["Operations"] + assert first["method"] == "POST" + assert first["bulkId"] == "a" + assert first["status"] == "400" + assert first["response"]["status"] == "400" + assert "location" not in first + assert second["status"] == "201" + + def test_unknown_resource(self, wsgi): + """An operation on a missing resource answers 404 with its location.""" + (result,) = bulk(wsgi, [{"method": "DELETE", "path": "/Users/unknown"}]).json()[ + "Operations" + ] + assert result["status"] == "404" + assert result["location"] == "https://scim.example.com/v2/Users/unknown" + + def test_unknown_endpoint(self, wsgi): + """An operation on an endpoint that serves no resource type answers 404.""" + (result,) = bulk(wsgi, [{"method": "DELETE", "path": "/Unknown/x"}]).json()[ + "Operations" + ] + assert result["status"] == "404" + + @pytest.mark.parametrize( + "operation", + [ + {"method": "POST", "path": "/Users/x", "bulkId": "a", "data": {}}, + {"method": "DELETE", "path": "/Users"}, + ], + ids=["post-to-a-resource", "delete-an-endpoint"], + ) + def test_path_does_not_fit_the_method(self, wsgi, operation): + """RFC 7644 §3.7: a POST targets a resource type endpoint, other methods a resource.""" + (result,) = bulk(wsgi, [operation]).json()["Operations"] + assert result["status"] == "400" + assert result["response"]["scimType"] == "invalidValue" + + @pytest.mark.parametrize( + "operation", + [{"method": "DELETE"}, {"method": "DELETE", "path": 1}, "DELETE"], + ids=["missing", "not-a-string", "operation-not-an-object"], + ) + def test_path_is_required(self, wsgi, operation): + """RFC 7644 §3.7: the path of an operation is required.""" + (result,) = bulk(wsgi, [operation]).json()["Operations"] + assert result["status"] == "400" + assert result["response"]["detail"] == "path is required for request operations" + + @pytest.mark.parametrize("method", ["GET", "post", ["POST"]]) + def test_invalid_method(self, wsgi, method): + """An operation with an invalid method answers 400, and does not repeat the method.""" + (result,) = bulk(wsgi, [{"method": method, "path": "/Users/x"}]).json()[ + "Operations" + ] + assert result["status"] == "400" + assert "method" not in result + + def test_version_mismatch(self, wsgi, first_fake_user): + """RFC 7644 §3.7: an operation carrying another version answers 412.""" + r = bulk( + wsgi, + [ + { + "method": "DELETE", + "path": f"/Users/{first_fake_user}", + "version": 'W/"abc"', + } + ], + ) + (result,) = r.json()["Operations"] + assert result["status"] == "412" + assert wsgi.get(f"/v2/Users/{first_fake_user}").status_code == 200 + + def test_version_match(self, wsgi, first_fake_user): + """An operation carrying the current version is applied.""" + version = wsgi.get(f"/v2/Users/{first_fake_user}").headers["etag"] + r = bulk( + wsgi, + [ + { + "method": "DELETE", + "path": f"/Users/{first_fake_user}", + "version": version, + } + ], + ) + (result,) = r.json()["Operations"] + assert result["status"] == "204" + + def test_patch_not_supported(self, wsgi_with, first_fake_user): + """A PATCH operation answers 501 when the service does not support PATCH.""" + client = configured(wsgi_with, patch=Patch(supported=False)) + r = bulk( + client, + [ + { + "method": "PATCH", + "path": f"/Users/{first_fake_user}", + "data": { + "schemas": [PATCH_OP], + "Operations": [ + {"op": "replace", "path": "userName", "value": "x"} + ], + }, + } + ], + ) + (result,) = r.json()["Operations"] + assert result["status"] == "501" + + +class TestBulkFailOnErrors: + def test_without_fail_on_errors_every_operation_runs(self, wsgi): + """RFC 7644 §3.7.3: the job performs as many changes as possible.""" + operations = [ + {"method": "DELETE", "path": "/Users/a"}, + {"method": "DELETE", "path": "/Users/b"}, + create_user("grace"), + ] + results = bulk(wsgi, operations).json()["Operations"] + assert [result["status"] for result in results] == ["404", "404", "201"] + + @pytest.mark.parametrize("fail_on_errors", [0, 1]) + def test_fail_on_errors_stops_the_job(self, wsgi, fail_on_errors): + """RFC 7644 §3.7.3: the job stops at the first error, and later operations are left out.""" + operations = [ + {"method": "DELETE", "path": "/Users/a"}, + create_user("heidi"), + ] + r = bulk(wsgi, operations, failOnErrors=fail_on_errors) + assert r.status_code == 200 + assert [result["status"] for result in r.json()["Operations"]] == ["404"] + assert wsgi.get("/v2/Users").json()["totalResults"] == 0 + + def test_fail_on_errors_accepts_errors_below_the_limit(self, wsgi): + """The job goes on until the number of errors reaches failOnErrors.""" + operations = [ + {"method": "DELETE", "path": "/Users/a"}, + create_user("ivan"), + {"method": "DELETE", "path": "/Users/b"}, + create_user("judy", "j"), + ] + results = bulk(wsgi, operations, failOnErrors=2).json()["Operations"] + assert [result["status"] for result in results] == ["404", "201", "404"] + + +class TestBulkRequest: + def test_not_supported(self, wsgi_with): + """RFC 7644 §3.7: bulk is optional, and a service that does not support it answers 501.""" + client = configured(wsgi_with, bulk=Bulk(supported=False)) + r = bulk(client, []) + assert r.status_code == 501 + assert r.json()["detail"] == "Bulk is not supported" + + def test_empty(self, wsgi): + """A job without operations answers an empty list of results.""" + r = bulk(wsgi, []) + assert r.status_code == 200 + assert r.json()["Operations"] == [] + + def test_operations_key_is_case_insensitive(self, wsgi): + """The attribute names of the envelope are case insensitive.""" + r = wsgi.post( + "/v2/Bulk", + json={"schemas": [BULK_REQUEST], "operations": [create_user("kim")]}, + ) + assert [result["status"] for result in r.json()["Operations"]] == ["201"] + + @pytest.mark.parametrize( + "payload", + [ + {"schemas": [BULK_REQUEST]}, + {"schemas": [BULK_REQUEST], "Operations": "x"}, + { + "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + "Operations": [], + }, + {"schemas": [BULK_REQUEST], "failOnErrors": "x", "Operations": []}, + [], + ], + ids=[ + "missing-operations", + "operations-not-a-list", + "wrong-schema", + "invalid-fail-on-errors", + "not-an-object", + ], + ) + def test_invalid_envelope(self, wsgi, payload): + """An invalid envelope fails the whole request with a 400.""" + r = wsgi.post("/v2/Bulk", json=payload) + assert r.status_code == 400 + + def test_too_many_operations(self, wsgi_with): + """RFC 7644 §3.7.4: a job beyond maxOperations answers 413 and names the limit.""" + client = configured( + wsgi_with, + bulk=Bulk(supported=True, max_operations=1, max_payload_size=1048576), + ) + r = bulk(client, [create_user("leo"), create_user("mia", "m")]) + assert r.status_code == 413 + assert ( + r.json()["detail"] + == "The number of operations exceeds the maxOperations (1)" + ) + assert client.get("/v2/Users").json()["totalResults"] == 0 + + def test_payload_too_large(self, wsgi_with): + """RFC 7644 §3.7.4: a payload beyond maxPayloadSize answers 413 and names the limit.""" + client = configured( + wsgi_with, + bulk=Bulk(supported=True, max_operations=1000, max_payload_size=10), + ) + r = bulk(client, [create_user("nina")]) + assert r.status_code == 413 + assert r.json()["detail"] == "The payload exceeds the maxPayloadSize (10 bytes)" + + def test_limits_are_optional(self, wsgi_with): + """A configuration without limits accepts any job.""" + client = configured(wsgi_with, bulk=Bulk(supported=True)) + r = bulk(client, [create_user("olga")]) + assert [result["status"] for result in r.json()["Operations"]] == ["201"] + + def test_without_etags_results_carry_no_version(self, wsgi_with): + """RFC 7644 §3.14: a service without ETags sends no version.""" + client = configured(wsgi_with, etag=ETag(supported=False)) + (result,) = bulk(client, [create_user("paul")]).json()["Operations"] + assert result["status"] == "201" + assert "version" not in result + + +def create_group(display_name, bulk_id, *member_bulk_ids): + return { + "method": "POST", + "path": "/Groups", + "bulkId": bulk_id, + "data": { + "displayName": display_name, + "members": [ + {"type": "User", "value": f"bulkId:{member_bulk_id}"} + for member_bulk_id in member_bulk_ids + ], + }, + } + + +class TestBulkIdReferences: + def test_reference_in_data(self, wsgi): + """RFC 7644 §3.7.2: a bulkId reference is replaced with the id of the created resource.""" + r = bulk( + wsgi, [create_user("alice", "qwerty"), create_group("g", "g", "qwerty")] + ) + user, group = r.json()["Operations"] + assert group["status"] == "201" + user_id = wsgi.get(user["location"]).json()["id"] + members = wsgi.get(group["location"]).json()["members"] + assert [member["value"] for member in members] == [user_id] + + def test_forward_reference(self, wsgi): + """A creation runs before the operation that references it, and the results keep the order of the request.""" + r = bulk(wsgi, [create_group("g", "g", "qwerty"), create_user("bob", "qwerty")]) + group, user = r.json()["Operations"] + assert group["bulkId"] == "g" + assert user["bulkId"] == "qwerty" + assert group["status"] == "201" + user_id = wsgi.get(user["location"]).json()["id"] + members = wsgi.get(group["location"]).json()["members"] + assert [member["value"] for member in members] == [user_id] + + def test_reference_in_path(self, wsgi): + """An operation can target a resource the same request creates.""" + r = bulk( + wsgi, + [ + create_user("carol", "qwerty"), + {"method": "DELETE", "path": "/Users/bulkId:qwerty"}, + ], + ) + user, deletion = r.json()["Operations"] + assert deletion["status"] == "204" + assert deletion["location"] == user["location"] + assert wsgi.get(user["location"]).status_code == 404 + + def test_reference_in_patch_value(self, wsgi): + """A bulkId reference is replaced in the values of a PATCH.""" + r = bulk( + wsgi, + [ + create_group("g", "g"), + create_user("dave", "qwerty"), + { + "method": "PATCH", + "path": "/Groups/bulkId:g", + "data": { + "schemas": [PATCH_OP], + "Operations": [ + { + "op": "add", + "path": "members", + "value": [{"value": "bulkId:qwerty"}], + } + ], + }, + }, + ], + ) + group, user, patch = r.json()["Operations"] + assert patch["status"] == "200" + user_id = wsgi.get(user["location"]).json()["id"] + members = wsgi.get(group["location"]).json()["members"] + assert [member["value"] for member in members] == [user_id] + + def test_reference_to_a_manager(self, wsgi): + """A reference to a manager needs no $ref.""" + enterprise = "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User" + r = bulk( + wsgi, + [ + create_user("erin", "boss"), + { + "method": "POST", + "path": "/Users", + "bulkId": "employee", + "data": { + "schemas": [ + "urn:ietf:params:scim:schemas:core:2.0:User", + enterprise, + ], + "userName": "frank", + enterprise: {"manager": {"value": "bulkId:boss"}}, + }, + }, + ], + ) + boss, employee = r.json()["Operations"] + assert employee["status"] == "201" + boss_id = wsgi.get(boss["location"]).json()["id"] + assert ( + wsgi.get(employee["location"]).json()[enterprise]["manager"]["value"] + == boss_id + ) + + def test_circular_reference(self, wsgi): + """RFC 7644 §3.7.1: the operations of a circular reference answer 409.""" + r = bulk(wsgi, [create_group("a", "a", "b"), create_group("b", "b", "a")]) + first, second = r.json()["Operations"] + assert first["status"] == "409" + assert ( + first["response"]["detail"] == "No resource was created with the bulkId b" + ) + assert second["status"] == "409" + assert ( + second["response"]["detail"] + == "The bulkId a is part of a circular reference" + ) + assert wsgi.get("/v2/Groups").json()["totalResults"] == 0 + + def test_self_reference(self, wsgi): + """A creation that references itself answers 409.""" + (result,) = bulk(wsgi, [create_group("a", "a", "a")]).json()["Operations"] + assert result["status"] == "409" + assert ( + result["response"]["detail"] + == "The bulkId a is part of a circular reference" + ) + + def test_unknown_reference(self, wsgi): + """A reference to a bulkId no operation creates answers 409.""" + (result,) = bulk(wsgi, [create_group("a", "a", "unknown")]).json()["Operations"] + assert result["status"] == "409" + assert ( + result["response"]["detail"] + == "No resource was created with the bulkId unknown" + ) + + def test_reference_to_a_failed_creation(self, wsgi): + """A reference to a creation that failed answers 409.""" + failed = {"method": "POST", "path": "/Users", "bulkId": "qwerty", "data": {}} + r = bulk(wsgi, [failed, create_group("g", "g", "qwerty")]) + user, group = r.json()["Operations"] + assert user["status"] == "400" + assert group["status"] == "409" + + def test_duplicate_bulk_id(self, wsgi): + """A bulkId is unique in a request: a second creation with it answers 400, and references go to the first.""" + r = bulk( + wsgi, + [ + create_user("grace", "qwerty"), + create_user("heidi", "qwerty"), + create_group("g", "g", "qwerty"), + ], + ) + first, second, group = r.json()["Operations"] + assert second["status"] == "400" + assert second["response"]["scimType"] == "invalidValue" + user_id = wsgi.get(first["location"]).json()["id"] + members = wsgi.get(group["location"]).json()["members"] + assert [member["value"] for member in members] == [user_id] + + def test_fail_on_errors_counts_the_referenced_creations(self, wsgi): + """A failed creation that runs first can stop the job before the operation that references it.""" + failed = {"method": "POST", "path": "/Users", "bulkId": "qwerty", "data": {}} + r = bulk(wsgi, [create_group("g", "g", "qwerty"), failed], failOnErrors=1) + (result,) = r.json()["Operations"] + assert result["bulkId"] == "qwerty" + assert result["status"] == "400" + assert wsgi.get("/v2/Groups").json()["totalResults"] == 0 diff --git a/tests/integration/test_scim_application.py b/tests/integration/test_scim_application.py index d632cde..c6f3403 100644 --- a/tests/integration/test_scim_application.py +++ b/tests/integration/test_scim_application.py @@ -51,7 +51,9 @@ def test_service_provider_configuration(self, wsgi): assert r.json() == { "authenticationSchemes": [], "bulk": { - "supported": False, + "maxOperations": 1000, + "maxPayloadSize": 1048576, + "supported": True, }, "changePassword": {"supported": True}, "etag": {"supported": True}, @@ -73,7 +75,9 @@ def test_no_version_prefix(self, wsgi): assert r.json() == { "authenticationSchemes": [], "bulk": { - "supported": False, + "maxOperations": 1000, + "maxPayloadSize": 1048576, + "supported": True, }, "changePassword": {"supported": True}, "etag": {"supported": True}, @@ -865,18 +869,6 @@ def test_patch_path_filters_do_not_require_the_filter_capability( ) assert r.status_code == 204 - def test_bulk_is_not_implemented(self, wsgi): - """RFC 7644 §3.7: bulk is optional, and this server answers 501 to it.""" - r = wsgi.post( - "/v2/Bulk", - json={ - "schemas": ["urn:ietf:params:scim:api:messages:2.0:BulkRequest"], - "Operations": [], - }, - ) - assert r.status_code == 501 - assert r.json()["detail"] == "Bulk operations are not supported" - def test_validation_error_carries_scim_type(self, wsgi): """A payload refused by validation answers with a SCIM error keyword.""" r = wsgi.post(