From ee28c91d46af9aa8f6d6043d62fb9e02be681172 Mon Sep 17 00:00:00 2001 From: Peter Holloway Date: Tue, 20 Oct 2026 15:01:01 +0100 Subject: [PATCH 1/2] Convert ValidationErrors into something picklable --- src/blueapi/service/main.py | 11 +++--- src/blueapi/service/protocol.py | 7 ++-- src/blueapi/worker/task.py | 44 ++++++++++++++++++++--- tests/unit_tests/service/test_protocol.py | 24 ++++++------- tests/unit_tests/service/test_rest_api.py | 26 +++++++------- 5 files changed, 75 insertions(+), 37 deletions(-) diff --git a/src/blueapi/service/main.py b/src/blueapi/service/main.py index 0d8fbf4774..225413304b 100644 --- a/src/blueapi/service/main.py +++ b/src/blueapi/service/main.py @@ -57,6 +57,7 @@ ) from blueapi.worker import TrackableTask, WorkerState from blueapi.worker.event import ProgressEvent, TaskStatusEnum, WorkerEvent +from blueapi.worker.task import TaskParameterError from blueapi.worker.worker_errors import WorkerBusyError from .authorization import ( @@ -347,7 +348,7 @@ def submit_task( task_id: str = runner.run(interface.submit_task, task_request, {"user": fedid}) response.headers["Location"] = f"{request.url}/{task_id}" return TaskResponse(task_id=task_id) - except ValidationError as e: + except TaskParameterError as e: # Add body/params context to location and ensure that all required # fields defined in the generated schema are present errors = [ @@ -358,7 +359,7 @@ def submit_task( # Input is not listed as required but is useful to have if available "input": err.get("input", None), } - for err in e.errors() + for err in e.errors ] LOGGER.info("Error submitting task: %s - %s", task_request, e) @@ -669,9 +670,9 @@ async def run_plan( interface.submit_task, task_request.task, {"user": user} ) LOGGER.info("Task ID: %s", task_id) - except ValidationError as ve: - LOGGER.info("Plan args not valid: %s - %s", task_request, ve) - await ws.send_text(InvalidArgs.from_validation_error(ve).model_dump_json()) + except TaskParameterError as tpe: + LOGGER.info("Plan args not valid: %s - %s", task_request, tpe) + await ws.send_text(InvalidArgs.from_parameter_error(tpe).model_dump_json()) await ws.close(code=protocol.INVALID_ARGS, reason="Invalid Args") return except KeyError as ke: diff --git a/src/blueapi/service/protocol.py b/src/blueapi/service/protocol.py index 8af5819133..e1b923c72d 100644 --- a/src/blueapi/service/protocol.py +++ b/src/blueapi/service/protocol.py @@ -17,11 +17,12 @@ from typing import Annotated, Any, Literal, Self -from pydantic import BaseModel, Field, TypeAdapter, ValidationError +from pydantic import BaseModel, Field, TypeAdapter from blueapi.core.bluesky_types import DataEvent from blueapi.service.model import TaskRequest from blueapi.worker.event import ProgressEvent, WorkerEvent +from blueapi.worker.task import TaskParameterError UNKNOWN_PLAN = 4001 INVALID_ARGS = 4002 @@ -68,7 +69,7 @@ class InvalidArgs(BaseModel): errors: list[ArgumentError] @classmethod - def from_validation_error(cls, e: ValidationError) -> Self: + def from_parameter_error(cls, e: TaskParameterError) -> Self: errors = [ ArgumentError( loc=["body", "params", *err.get("loc", [])], @@ -77,7 +78,7 @@ def from_validation_error(cls, e: ValidationError) -> Self: # Input is not listed as required but is useful to have if available input=err.get("input", None), ) - for err in e.errors() + for err in e.errors ] return cls(errors=errors) diff --git a/src/blueapi/worker/task.py b/src/blueapi/worker/task.py index e2efade0d7..6dbbaa0202 100644 --- a/src/blueapi/worker/task.py +++ b/src/blueapi/worker/task.py @@ -1,8 +1,8 @@ import logging from collections.abc import Mapping -from typing import Any +from typing import Any, Self, TypedDict -from pydantic import BaseModel, Field, TypeAdapter +from pydantic import BaseModel, Field, TypeAdapter, ValidationError from blueapi.core import BlueskyContext from blueapi.utils import BlueapiBaseModel @@ -25,9 +25,12 @@ class Task(BlueapiBaseModel): ) def prepare_params(self, ctx: BlueskyContext) -> Mapping[str, Any]: - model = _lookup_params(ctx, self) - # Re-create dict manually to avoid nesting in model_dump output - return {field: getattr(model, field) for field in model.__pydantic_fields__} + try: + model = _lookup_params(ctx, self) + # Re-create dict manually to avoid nesting in model_dump output + return {field: getattr(model, field) for field in model.__pydantic_fields__} + except ValidationError as ve: + raise TaskParameterError.from_validation_error(ve) from None def do_task(self, ctx: BlueskyContext) -> None: LOGGER.info( @@ -62,3 +65,34 @@ def _lookup_params(ctx: BlueskyContext, task: Task) -> BaseModel: model = plan.model adapter = TypeAdapter(model) return adapter.validate_python(task.params) + + +class InvalidTaskParameter(TypedDict): + loc: tuple[str | int, ...] + msg: str | None + type: str | None + input: Any | None + + +class TaskParameterError(ValueError): + errors: list[InvalidTaskParameter] + + def __init__(self, message: str, errors: list[InvalidTaskParameter]): + super().__init__(message) + self.errors = errors + + @classmethod + def from_validation_error(cls, error: ValidationError) -> Self: + errors = [ + InvalidTaskParameter( + loc=e.get("loc", []), + msg=e.get("msg", None), + type=e.get("type", None), + input=e.get("input", None), + ) + for e in error.errors() + ] + return cls(str(error), errors) + + def __reduce__(self): + return (self.__class__, (*self.args, self.errors)) diff --git a/tests/unit_tests/service/test_protocol.py b/tests/unit_tests/service/test_protocol.py index 9178162e18..be583f956b 100644 --- a/tests/unit_tests/service/test_protocol.py +++ b/tests/unit_tests/service/test_protocol.py @@ -1,8 +1,6 @@ from typing import Any import pytest -from pydantic import ValidationError -from pydantic_core import InitErrorDetails from blueapi.service.model import TaskRequest from blueapi.service.protocol import ( @@ -15,6 +13,7 @@ Resume, Submit, ) +from blueapi.worker.task import InvalidTaskParameter, TaskParameterError @pytest.mark.parametrize( @@ -72,19 +71,20 @@ def test_response_deserialization(src: str, res: Any): assert req == res -def test_from_empty_validation_error(): - err = InvalidArgs.from_validation_error( - ValidationError("Error validating request", []) +def test_from_empty_parameter_error(): + err = InvalidArgs.from_parameter_error( + TaskParameterError("Error validating request", []) ) assert err == InvalidArgs(errors=[]) -def test_from_validation_error(): - err = InvalidArgs.from_validation_error( - ValidationError.from_exception_data( - title="Error validating request", - line_errors=[ - InitErrorDetails( +def test_from_parameter_error(): + err = InvalidArgs.from_parameter_error( + TaskParameterError( + "Error validating request", + [ + InvalidTaskParameter( + msg="foo.bar is missing", loc=("foo", "bar"), type="missing", input={"foo": {"no": "bar"}}, @@ -96,7 +96,7 @@ def test_from_validation_error(): errors=[ ArgumentError( loc=["body", "params", "foo", "bar"], - msg="Field required", + msg="foo.bar is missing", type="missing", input={"foo": {"no": "bar"}}, ) diff --git a/tests/unit_tests/service/test_rest_api.py b/tests/unit_tests/service/test_rest_api.py index a10d5ab3cf..a02b1e3016 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -12,8 +12,7 @@ from fastapi import FastAPI, HTTPException, WebSocketDisconnect, status from fastapi.testclient import TestClient from httpx2 import Headers -from pydantic import BaseModel, ValidationError -from pydantic_core import InitErrorDetails +from pydantic import BaseModel from starlette.testclient import WebSocketDenialResponse from starlette.types import Message, Receive, Scope, Send @@ -47,7 +46,7 @@ ) from blueapi.service.runner import WorkerDispatcher from blueapi.worker.event import TaskStatus, WorkerEvent, WorkerState -from blueapi.worker.task import Task +from blueapi.worker.task import InvalidTaskParameter, Task, TaskParameterError from blueapi.worker.task_worker import TrackableTask @@ -330,21 +329,24 @@ def test_create_task_inserts_auth_metadata( task_id = str(uuid.uuid4()) # mock_runner.run.side_effect = [task_id] - mock_runner.run.return_value = [task_id] + mock_runner.run.return_value = task_id client_with_auth.post("/tasks", json=task.model_dump()) mock_runner.run.assert_called_with(submit_task, task, {"user": "jd1"}) -def test_create_task_validation_error(mock_runner: Mock, client: TestClient) -> None: +def test_create_task_parameter_error(mock_runner: Mock, client: TestClient) -> None: mock_runner.run.side_effect = [ - ValidationError.from_exception_data( - title="ValueError", - line_errors=[ - InitErrorDetails( - type="missing", loc=("id",), msg="value is required for Identifier" - ) # type: ignore + TaskParameterError( + "ValueError", + [ + InvalidTaskParameter( + type="missing", + loc=("id",), + msg="Field required", + input=None, + ) ], ), ] @@ -1099,7 +1101,7 @@ def test_websocket_run_plan_invalid_request( "Unknown Plan", ), ( - ValidationError("Not valid", []), + TaskParameterError("Not valid", []), {"kind": "invalid_args", "errors": []}, 4002, "Invalid Args", From 55756fd9ecd0a65c8b38ad2b32a7b9499a4a39dc Mon Sep 17 00:00:00 2001 From: Peter Holloway Date: Tue, 20 Oct 2026 18:21:02 +0100 Subject: [PATCH 2/2] Add test for running with invalid parameters Regression tests that would have caught the errors earlier. --- tests/system_tests/test_blueapi_system.py | 69 +++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index 138ee85c98..04ba545658 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -18,7 +18,9 @@ BlueapiRestClient, BlueskyRemoteControlError, BlueskyRequestError, + InvalidParametersError, NotFoundError, + ParameterError, ServiceUnavailableError, UnauthorisedAccessError, UnknownPlanError, @@ -661,6 +663,73 @@ def test_stub_runs(client_with_stomp: BlueapiClient, task: TaskRequest): assert not final_event.task_failed +@pytest.mark.parametrize( + "task,param_error,user", + [ + ( + TaskRequest( + name="count", + params={"detectors": "det", "num": 3}, # detectors should be list + instrument_session=VALID_INSTRUMENT_SESSION[User.alice], + ), + ParameterError( + loc=["body", "params", "detectors"], + msg="'str' instances are not allowed as a Sequence value", + type="sequence_str", + input="det", + ), + User.alice, + ), + ( + TaskRequest( + name="spec_scan", + params={ + "detectors": ["det"], + "spec": { + "axis": "missing", # doesn't exist + "start": 0.0, + "stop": 0.4, + "num": 2, + "type": "Linspace", + }, + }, + instrument_session=VALID_INSTRUMENT_SESSION[User.bob], + ), + ParameterError( + loc=["body", "params", "spec", "Linspace", "axis"], + msg="Value error, Device missing cannot be found", + type="value_error", + input="missing", + ), + User.bob, + ), + ], +) +@pytest.mark.parametrize("run_method", ["run_task", "run_blocking"]) +def test_invalid_parameters_fail_cleanly( + client_with_stomp: BlueapiClient, + task: TaskRequest, + user: ValidUser, + param_error: ParameterError, + run_method: str, +): + resource = Queue(maxsize=1) + start = Queue(maxsize=1) + + def on_event(event: AnyEvent) -> None: + if isinstance(event, DataEvent): + if event.name == "start": + start.put_nowait(event.doc) + if event.name == "stream_resource": + resource.put_nowait(event.doc) + + runner = getattr(client_with_stomp, run_method) + with pytest.raises(InvalidParametersError) as ipe: + runner(task, on_event) + + assert ipe.value.errors == [param_error] + + # Regression test for #1480 def test_task_submission_after_invalid_task(client_with_stomp: BlueapiClient): with pytest.raises(NotFoundError):