diff --git a/pyproject.toml b/pyproject.toml index e3a131e5c..aed5101bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,7 @@ dependencies = [ "PyYAML>=6.0.2", "click>=8.2.0", "fastapi>=0.112.0", - "uvicorn", + "uvicorn==0.49", # Latest version that works with websockets https://github.com/Kludex/uvicorn/issues/3040 "requests", "GitPython", "event-model==1.23.1", # https://github.com/DiamondLightSource/blueapi/issues/684 diff --git a/src/blueapi/cli/cli.py b/src/blueapi/cli/cli.py index 8c746fe92..6368da0ac 100644 --- a/src/blueapi/cli/cli.py +++ b/src/blueapi/cli/cli.py @@ -321,6 +321,7 @@ def on_event( @controller.command(name="run") @click.argument("name", type=str) @click.argument("parameters", type=ParametersType(), default={}, required=False) +@click.option("--ws", type=bool, is_flag=True, default=False) @click.option( "--foreground/--background", "--fg/--bg", type=bool, is_flag=True, default=True ) @@ -348,6 +349,7 @@ def run_plan( name: str, timeout: float | None, foreground: bool, + ws: bool, instrument_session: str, parameters: TaskParameters, ) -> None: @@ -374,7 +376,13 @@ def on_event(event: AnyEvent) -> None: elif isinstance(event, DataEvent): callback(event.name, event.doc) - resp = client.run_task(task, on_event=on_event) + client.add_callback(on_event) + + if ws: + resp = client.run_blocking(task) + else: + resp = client.run_task(task) + match resp.result: case TaskResult(result=None, type="NoneType"): print("Plan succeeded") diff --git a/src/blueapi/client/client.py b/src/blueapi/client/client.py index 20de15892..b71da8aa6 100644 --- a/src/blueapi/client/client.py +++ b/src/blueapi/client/client.py @@ -459,6 +459,27 @@ def get_active_task(self) -> WorkerTask: return self.active_task + @start_as_current_span(TRACER, "request") + def run_blocking( + self, request: TaskRequest, on_event: OnAnyEvent | None = None + ) -> TaskStatus: + for event in self._rest.run_blocking(request): + if on_event is not None: + on_event(event) + for cb in self._callbacks.values(): + try: + cb(event) + except Exception as e: + log.error(f"Callback ({cb}) failed for event: {event}", exc_info=e) + if isinstance(event, WorkerEvent) and event.is_complete(): + # task_status will always be present if event is complete + if event.task_status is None: # pragma: no cover + raise BlueskyRemoteControlError( + "Server completed without task status" + ) + return event.task_status + raise BlueskyRemoteControlError("Connection closed before plan completed.") + @start_as_current_span(TRACER, "task", "timeout") def run_task( self, diff --git a/src/blueapi/client/rest.py b/src/blueapi/client/rest.py index 0bddb5c87..2ae87df65 100644 --- a/src/blueapi/client/rest.py +++ b/src/blueapi/client/rest.py @@ -1,6 +1,6 @@ import json import logging -from collections.abc import Callable, Mapping +from collections.abc import Callable, Iterable, Mapping from typing import Any, Literal, TypeVar import requests @@ -12,10 +12,13 @@ ) from pydantic import BaseModel, TypeAdapter, ValidationError from pydantic_core import PydanticSerializationError +from websockets.exceptions import InvalidStatus +from websockets.sync.client import connect from blueapi import __version__ from blueapi.client import client from blueapi.config import RestConfig +from blueapi.core.bluesky_types import DataEvent from blueapi.service.authentication import JWTAuth, SessionManager from blueapi.service.model import ( DeviceModel, @@ -31,7 +34,17 @@ TasksListResponse, WorkerTask, ) +from blueapi.service.protocol import ( + ControlResponse, + InvalidArgs, + PlanNotFound, + ServerBusy, + Submit, + Unauthorized, + Update, +) from blueapi.worker import TrackableTask, WorkerState +from blueapi.worker.event import ProgressEvent, WorkerEvent T = TypeVar("T") @@ -39,6 +52,8 @@ LOGGER = logging.getLogger(__name__) +USER_AGENT = f"blueapi cli {__version__}" + class BlueskyRequestError(Exception): """An error response from the blueapi server.""" @@ -86,8 +101,8 @@ def __init__(self, target_type: type) -> None: class ParameterError(BaseModel): loc: list[str | int] - msg: str - type: str + msg: str | None + type: str | None input: Any def field(self): @@ -307,14 +322,15 @@ def _request_and_deserialize( ) -> T: url = self._config.url.unicode_string().removesuffix("/") + suffix # Get the trace context to propagate to the REST API - carr = get_context_propagator() + headers = get_context_propagator() + headers["User-Agent"] = USER_AGENT try: response = self._pool.request( method, url, json=data, params=params, - headers=carr, + headers=headers, auth=JWTAuth(self._session_manager), ) except requests.exceptions.ConnectionError as ce: @@ -340,6 +356,52 @@ def _request_and_deserialize( ) return deserialized + def run_blocking( + self, req: TaskRequest + ) -> Iterable[DataEvent | WorkerEvent | ProgressEvent]: + url = self._config.ws_address.unicode_string().rstrip("/") + "/api/v2/run_plan" + headers = get_context_propagator() + if self._session_manager: + auth = self._session_manager.get_valid_access_token() + headers["Authorization"] = f"Bearer {auth}" + try: + with connect( + url, + additional_headers=headers, + user_agent_header=USER_AGENT, + ) as ws: + ws.send(Submit(task=req).model_dump_json()) + for message in ws: + event = ControlResponse.validate_json(message) + match event: + case Update(data=data): + yield data + case InvalidArgs(errors=errors): + raise InvalidParametersError( + [ + ParameterError( + loc=e.loc, msg=e.msg, type=e.type, input=e.input + ) + for e in errors + ] + ) + case PlanNotFound(plan_name=name): + raise UnknownPlanError(message=name) + case ServerBusy(): + raise BlueskyRemoteControlError(409, "Server is busy") + case Unauthorized(): + raise UnauthorisedAccessError( + 403, "Not authorized to submit task" + ) + except InvalidStatus as istat: + match istat.response.status_code: + case 401 | 403: + raise UnauthorisedAccessError() from None + case _: + raise BlueskyRemoteControlError() from istat + except ConnectionRefusedError as cre: + raise ServiceUnavailableError() from cre + # https://github.com/DiamondLightSource/blueapi/issues/1256 - remove before 2.0 def __getattr__(name: str): diff --git a/src/blueapi/config.py b/src/blueapi/config.py index a181f4c34..e1b27b435 100644 --- a/src/blueapi/config.py +++ b/src/blueapi/config.py @@ -22,6 +22,7 @@ TypeAdapter, UrlConstraints, ValidationError, + WebsocketUrl, field_validator, model_validator, ) @@ -170,6 +171,22 @@ class RestConfig(BlueapiBaseModel): url: HttpUrl = HttpUrl("http://localhost:8000") cors: CORSConfig | None = None + @property + def ws_address(self) -> WebsocketUrl: + api = self.url + if api.host is None: + # type hints say it could be None but not possible to construct + # HttpUrl without host + raise ValueError("No host configured") # pragma: no cover + scheme = "ws" if api.scheme == "http" else "wss" + + # HttpUrl adds "/" to the start of paths, even if none was specified so + # remove existing leading '/' to prevent duplication + path = (api.path or "").removeprefix("/") + return WebsocketUrl.build( + scheme=scheme, host=api.host, port=api.port, path=path + ) + class ScratchRepository(BlueapiBaseModel): name: str = Field( diff --git a/src/blueapi/service/authentication.py b/src/blueapi/service/authentication.py index 6761256de..9177ad47f 100644 --- a/src/blueapi/service/authentication.py +++ b/src/blueapi/service/authentication.py @@ -15,7 +15,8 @@ import httpx import jwt import requests -from fastapi import Depends, HTTPException, Request +from fastapi import Depends, HTTPException +from fastapi.requests import HTTPConnection from fastapi.security.utils import get_authorization_scheme_param from pydantic import TypeAdapter from requests.auth import AuthBase @@ -278,14 +279,17 @@ def sync_auth_flow(self, request): yield request -def unchecked_bearer_token(req: Request) -> str | None: +def unchecked_bearer_token(req: HTTPConnection) -> str | None: """Get bearer token value from authorization header""" + + auth_header = req.headers.get("Authorization") + auth_cookie = req.cookies.get("Authorization") + # This is an abridged version of the same feature of # OAuth2AuthorizationCodeBearer from fastapi. Replicating here prevents # passing unused configuration and means the schema does not include auth # details for servers that do not support it. - auth = req.headers.get("Authorization") - scheme, param = get_authorization_scheme_param(auth) + scheme, param = get_authorization_scheme_param(auth_header or auth_cookie) if scheme.casefold() != "bearer": return None return param.strip() @@ -303,7 +307,7 @@ def build_access_token_check(config: OIDCConfig): """ jwkclient = jwt.PyJWKClient(config.jwks_uri) - def validate_bearer_token(request: Request, token: UncheckedBearerToken): + def validate_bearer_token(request: HTTPConnection, token: UncheckedBearerToken): """Check that a bearer token is valid and inject into request state""" if not token: raise HTTPException( @@ -326,7 +330,7 @@ def validate_bearer_token(request: Request, token: UncheckedBearerToken): return validate_bearer_token -def access_token(request: Request) -> Mapping[str, Any] | None: +def access_token(request: HTTPConnection) -> Mapping[str, Any] | None: """Get the decoded and verified access token of the user making the request""" return getattr(request.state, "decoded_access_token", None) diff --git a/src/blueapi/service/authorization.py b/src/blueapi/service/authorization.py index f9008138a..c324015a4 100644 --- a/src/blueapi/service/authorization.py +++ b/src/blueapi/service/authorization.py @@ -4,7 +4,8 @@ from typing import Annotated, Any, Self, cast from aiohttp import ClientSession -from fastapi import Depends, HTTPException, Request +from fastapi import Depends, HTTPException +from fastapi.requests import HTTPConnection from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN from blueapi.config import OIDCConfig, OpaConfig, ServiceAccount @@ -114,7 +115,7 @@ async def validate_tiled_config( async def opa( - request: Request, token: str | None = Depends(unchecked_bearer_token) + request: HTTPConnection, token: str | None = Depends(unchecked_bearer_token) ) -> OpaUserClient | None: if opa := cast(OpaClient | None, getattr(request.app.state, "authz", None)): diff --git a/src/blueapi/service/interface.py b/src/blueapi/service/interface.py index da262d33a..0944af59c 100644 --- a/src/blueapi/service/interface.py +++ b/src/blueapi/service/interface.py @@ -1,5 +1,8 @@ +import logging from collections.abc import Mapping +from dataclasses import dataclass from functools import cache +from multiprocessing.connection import Connection from typing import Any from bluesky.callbacks.tiled_writer import TiledWriter @@ -9,6 +12,7 @@ from blueapi.cli.scratch import get_python_environment from blueapi.config import ApplicationConfig, OIDCConfig, ServiceAccount, StompConfig +from blueapi.core.bluesky_types import DataEvent from blueapi.core.context import BlueskyContext from blueapi.core.event import EventStream from blueapi.log import set_up_logging @@ -22,14 +26,14 @@ WorkerTask, ) from blueapi.utils.serialization import access_blob -from blueapi.worker.event import TaskStatusEnum, WorkerEvent, WorkerState +from blueapi.worker.event import ProgressEvent, TaskStatusEnum, WorkerEvent, WorkerState from blueapi.worker.task import Task from blueapi.worker.task_worker import TaskWorker, TrackableTask """This module provides interface between web application and underlying Bluesky context and worker""" - +LOGGER = logging.getLogger(__name__) _CONFIG: ApplicationConfig = ApplicationConfig() @@ -226,7 +230,7 @@ def remove_callback_when_task_finished( if task.task_id is not None: try: active_worker.begin_task(task.task_id) - except KeyError: + except: for channel, token in subscribers: channel.unsubscribe(token) raise @@ -281,3 +285,37 @@ def get_python_env( """Retrieve information about the Python environment""" scratch = config().scratch return get_python_environment(config=scratch, name=name, source=source) + + +@dataclass +class SubHandles: + worker: int + progress: int + data: int + + +def pipe_events(tx: Connection) -> SubHandles: + tw = worker() + + def handler( + worker_event: WorkerEvent | DataEvent | ProgressEvent, + _cor_id: str | None, + ) -> None: + + try: + tx.send(worker_event) + except BrokenPipeError: + LOGGER.warning("Sending event to broken pipe") + pass + + w = tw.worker_events.subscribe(handler) + d = tw.data_events.subscribe(handler) + p = tw.progress_events.subscribe(handler) + return SubHandles(worker=w, data=d, progress=p) + + +def unpipe_events(hnd: SubHandles) -> None: + tw = worker() + tw.worker_events.unsubscribe(hnd.worker) + tw.data_events.unsubscribe(hnd.data) + tw.progress_events.unsubscribe(hnd.progress) diff --git a/src/blueapi/service/main.py b/src/blueapi/service/main.py index 55b929b31..c3e3137b0 100644 --- a/src/blueapi/service/main.py +++ b/src/blueapi/service/main.py @@ -15,6 +15,8 @@ HTTPException, Request, Response, + WebSocket, + WebSocketDisconnect, status, ) from fastapi.datastructures import Address @@ -29,16 +31,31 @@ from opentelemetry.trace import get_tracer_provider from pydantic import ValidationError from starlette.responses import JSONResponse +from starlette.status import WS_1007_INVALID_FRAME_PAYLOAD_DATA, WS_1013_TRY_AGAIN_LATER from blueapi.config import ApplicationConfig, OIDCConfig, Tag -from blueapi.service import interface -from blueapi.service.authentication import Fedid, build_access_token_check +from blueapi.core.bluesky_types import DataEvent +from blueapi.service import interface, protocol +from blueapi.service.authentication import ( + Fedid, + build_access_token_check, +) from blueapi.service.middleware import ( ObservabilityContextPropagator, VersionHeaders, + WebsocketTracing, +) +from blueapi.service.protocol import ( + InvalidArgs, + PlanNotFound, + ServerBusy, + Submit, + Unauthorized, + Update, ) from blueapi.worker import TrackableTask, WorkerState -from blueapi.worker.event import TaskStatusEnum +from blueapi.worker.event import ProgressEvent, TaskStatusEnum, WorkerEvent +from blueapi.worker.worker_errors import WorkerBusyError from .authorization import ( OpaClient, @@ -71,6 +88,9 @@ TRACER = get_tracer("interface") +AnyEvent = WorkerEvent | DataEvent | ProgressEvent + + def _runner() -> WorkerDispatcher: """Intended to be used only with FastAPI Depends""" if RUNNER is None: @@ -114,6 +134,7 @@ async def inner(app: FastAPI): open_router = APIRouter() secure_router = APIRouter(deprecated=True) secure_router_v1 = APIRouter(prefix="/api/v1") +secure_router_v2 = APIRouter(prefix="/api/v2") def get_app(config: ApplicationConfig): @@ -135,12 +156,14 @@ def get_app(config: ApplicationConfig): } app.include_router(open_router) app.include_router(secure_router_v1, dependencies=dependencies) + app.include_router(secure_router_v2, dependencies=dependencies) app.include_router(secure_router, dependencies=dependencies) app.add_exception_handler(KeyError, on_key_error_404) app.add_exception_handler(jwt.PyJWTError, on_token_error_401) app.add_middleware(ObservabilityContextPropagator) app.add_middleware(VersionHeaders) + app.add_middleware(WebsocketTracing) app.middleware("http")(log_request_details) if config.api.cors: app.add_middleware( @@ -604,6 +627,81 @@ def logout(runner: Annotated[WorkerDispatcher, Depends(_runner)]) -> Response: ) +@secure_router_v2.websocket("/run_plan") +async def run_plan( + ws: WebSocket, + runner: Annotated[WorkerDispatcher, Depends(_runner)], + user: Fedid, + opa: Annotated[OpaUserClient | None, Depends(opa)], +): + LOGGER.info("Starting WS plan as %s", user) + await ws.accept() + rq = await ws.receive_text() + try: + task_request = Submit.model_validate_json(rq) + except ValidationError: + LOGGER.info("Failed to deserialize request: %r", rq, exc_info=True) + await ws.close( + code=WS_1007_INVALID_FRAME_PAYLOAD_DATA, reason="Invalid Request" + ) + return + LOGGER.info("Plan request: %s", task_request) + + if opa: + try: + await opa.can_submit_task(task_request.task) + except Exception as e: + LOGGER.info( + "User %s does not have permission to run task", user, exc_info=e + ) + await ws.send_text(Unauthorized().model_dump_json()) + await ws.close(code=protocol.AUTHZ_ERROR, reason="Unauthorized") + return + + try: + task_id: str = runner.run( + 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()) + await ws.close(code=protocol.INVALID_ARGS, reason="Invalid Args") + return + except KeyError as ke: + LOGGER.info("Plan %r not recognised", ke.args[0]) + await ws.send_text(PlanNotFound(plan_name=ke.args[0]).model_dump_json()) + await ws.close(code=protocol.UNKNOWN_PLAN, reason="Unknown Plan") + return + + try: + with runner.event_pipe() as events: + active_task = runner.run(interface.get_active_task) + if active_task is not None and not active_task.is_complete: + raise WorkerBusyError("Task already running") + runner.run(interface.begin_task, task=WorkerTask(task_id=task_id)) + async for evt in events: + if evt.task_id != task_id: + continue + LOGGER.debug("Event: %s", evt) + await ws.send_text(Update(data=evt).model_dump_json()) + if isinstance(evt, WorkerEvent) and evt.is_complete(): + LOGGER.debug("End of stream") + break + except WorkerBusyError: + LOGGER.error("Worker was busy") + await ws.send_text(ServerBusy().model_dump_json()) + await ws.close(code=WS_1013_TRY_AGAIN_LATER, reason="Worker busy") + except WebSocketDisconnect: + LOGGER.info("Client disconnected") + runner.run( + interface.cancel_active_task, failure=True, reason="Client disconnected" + ) + else: + LOGGER.info("Plan complete") + await ws.close() + + @start_as_current_span(TRACER, "config") def start(config: ApplicationConfig): import uvicorn diff --git a/src/blueapi/service/middleware.py b/src/blueapi/service/middleware.py index b31fe0fb9..5adffc368 100644 --- a/src/blueapi/service/middleware.py +++ b/src/blueapi/service/middleware.py @@ -1,4 +1,6 @@ import logging +import uuid +from collections.abc import Iterable from opentelemetry.context import attach from opentelemetry.propagate import get_global_textmap @@ -8,6 +10,7 @@ from blueapi.config import ApplicationConfig OBS_LOGGER = logging.getLogger("blueapi.service.middleware.observability") +WS_LOGGER = logging.getLogger("blueapi.service.middleware.websocket") CONTEXT_HEADER = ApplicationConfig.CONTEXT_HEADER.encode() VENDOR_CONTEXT_HEADER = ApplicationConfig.VENDOR_CONTEXT_HEADER.encode() @@ -56,3 +59,81 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send): attach(get_global_textmap().extract(carrier)) return await self.app(scope, receive, send) + + +Header = tuple[bytes, bytes] + + +def _redact_headers(headers: list[Header] | None) -> Iterable[Header]: + for key, value in headers or []: + if key == b"authorization": + if (space := value.find(b" ")) >= 0: + value = value[:space] + b" [REDACTED]" + yield (key, value) + + +class WebsocketTracing: + def __init__(self, app: ASGIApp): + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send): + active = WS_LOGGER.isEnabledFor(logging.DEBUG) + + if scope.get("type") != "websocket" or not active: + return await self.app(scope, receive, send) + + conn_id = uuid.uuid4() + client: tuple[str, int] = scope.get("client", ("unknown", 0)) + extra = {"conn": conn_id, "client": client} + + WS_LOGGER.debug( + "New Connection from %r", + {**scope, "headers": list(_redact_headers(scope.get("headers")))}, + extra=extra, + ) + + async def local_send(msg: Message): + match msg.get("type"): + case "websocket.send": + WS_LOGGER.debug("Sending: %r", msg.get("text"), extra=extra) + case "websocket.accept": + WS_LOGGER.debug( + "Accepting websocket - sending headers: %r", + msg.get("headers"), + extra=extra, + ) + case "websocket.close": + WS_LOGGER.debug( + "Closing with code: %r, reason: %r", + msg.get("code"), + msg.get("reason"), + extra=extra, + ) + case "websocket.http.response.start": + WS_LOGGER.debug( + "HTTP Response: status=%r, headers=%r", + msg.get("status"), + msg.get("headers"), + extra=extra, + ) + case "websocket.http.response.body": + WS_LOGGER.debug( + "HTTP Response Content: %r", msg.get("body"), extra=extra + ) + case _: + WS_LOGGER.debug("Sending other: %r", msg, extra=extra) + + await send(msg) + + async def local_receive() -> Message: + message = await receive() + match message.get("type"): + case "websocket.receive": + WS_LOGGER.debug("Received: %r", message.get("text"), extra=extra) + case "websocket.connect": + WS_LOGGER.debug("New connection from %s:%d", *client, extra=extra) + case _: + WS_LOGGER.debug("Received other: %r", message, extra=extra) + return message + + return await self.app(scope, local_receive, local_send) diff --git a/src/blueapi/service/protocol.py b/src/blueapi/service/protocol.py new file mode 100644 index 000000000..8af581913 --- /dev/null +++ b/src/blueapi/service/protocol.py @@ -0,0 +1,103 @@ +""" +The application level sub-protocol used to communicate between the server and +client when running plans via websockets +""" + +# Client to server +# * Submit task +# * Pause +# * Resume +# * Abort +# +# Server to client +# * Plan not found +# * Args not valid +# * Server busy +# * Event update + +from typing import Annotated, Any, Literal, Self + +from pydantic import BaseModel, Field, TypeAdapter, ValidationError + +from blueapi.core.bluesky_types import DataEvent +from blueapi.service.model import TaskRequest +from blueapi.worker.event import ProgressEvent, WorkerEvent + +UNKNOWN_PLAN = 4001 +INVALID_ARGS = 4002 +AUTHZ_ERROR = 4003 + + +class ArgumentError(BaseModel): + loc: list[str | int] + msg: str | None + type: str | None + input: Any + + +class Submit(BaseModel): + kind: Literal["submit"] = "submit" + task: TaskRequest + + +class Pause(BaseModel): + kind: Literal["pause"] = "pause" + + +class Resume(BaseModel): + kind: Literal["resume"] = "resume" + + +class Abort(BaseModel): + kind: Literal["abort"] = "abort" + reason: str | None = None + + +ControlRequest = TypeAdapter( + Annotated[Submit | Pause | Resume | Abort, Field(discriminator="kind")] +) + + +class PlanNotFound(BaseModel): + kind: Literal["plan_not_found"] = "plan_not_found" + plan_name: str + + +class InvalidArgs(BaseModel): + kind: Literal["invalid_args"] = "invalid_args" + errors: list[ArgumentError] + + @classmethod + def from_validation_error(cls, e: ValidationError) -> Self: + errors = [ + ArgumentError( + loc=["body", "params", *err.get("loc", [])], + msg=err.get("msg", None), + type=err.get("type", None), + # Input is not listed as required but is useful to have if available + input=err.get("input", None), + ) + for err in e.errors() + ] + return cls(errors=errors) + + +class ServerBusy(BaseModel): + kind: Literal["busy"] = "busy" + + +class Unauthorized(BaseModel): + kind: Literal["unauthorized"] = "unauthorized" + + +class Update(BaseModel): + kind: Literal["update"] = "update" + data: WorkerEvent | DataEvent | ProgressEvent + + +ControlResponse = TypeAdapter( + Annotated[ + PlanNotFound | InvalidArgs | ServerBusy | Unauthorized | Update, + Field(discriminator="kind"), + ] +) diff --git a/src/blueapi/service/runner.py b/src/blueapi/service/runner.py index 2b5a5f37f..ab17c159d 100644 --- a/src/blueapi/service/runner.py +++ b/src/blueapi/service/runner.py @@ -1,10 +1,12 @@ +import asyncio import inspect import logging import signal import uuid -from collections.abc import Callable +from collections.abc import AsyncIterator, Callable from importlib import import_module from multiprocessing import Pool, set_start_method +from multiprocessing.connection import Connection, Pipe from multiprocessing.pool import Pool as PoolClass from typing import Any, ParamSpec, TypeVar @@ -18,8 +20,11 @@ from pydantic import TypeAdapter from blueapi.config import ApplicationConfig -from blueapi.service.interface import setup, teardown +from blueapi.core.bluesky_types import DataEvent +from blueapi.service import interface +from blueapi.service.interface import SubHandles, setup, teardown from blueapi.service.model import EnvironmentResponse +from blueapi.worker.event import ProgressEvent, WorkerEvent # The default multiprocessing start method is fork set_start_method("spawn", force=True) @@ -145,11 +150,57 @@ def run( kwargs, ) + def event_pipe(self): + return EventPipe(self) + @property def state(self) -> EnvironmentResponse: return self._state +class EventStream: + def __init__(self, rx: Connection): + self._rx = rx + + def __aiter__(self) -> AsyncIterator[WorkerEvent | DataEvent | ProgressEvent]: + return self + + async def __anext__(self) -> WorkerEvent | DataEvent | ProgressEvent: + data_available = asyncio.Event() + asyncio.get_event_loop().add_reader(self._rx.fileno(), data_available.set) + try: + while not self._rx.poll(): + await data_available.wait() + data_available.clear() + return self._rx.recv() + except EOFError: + raise StopAsyncIteration() from None + finally: + asyncio.get_event_loop().remove_reader(self._rx.fileno()) + + +class EventPipe: + runner: WorkerDispatcher + handles: list[tuple[SubHandles, Connection]] + + def __init__(self, runner: WorkerDispatcher): + self.runner = runner + self.handles = [] + + def __enter__(self) -> EventStream: + tx, rx = Pipe() + hnd = self.runner.run(interface.pipe_events, tx) + LOGGER.debug("Subscribing new event pipe: %s", hnd) + self.handles.append((hnd, tx)) + return EventStream(rx) + + def __exit__(self, *exc): + hnd, conn = self.handles.pop() + LOGGER.debug("Unsubscribing event pipe: %s", hnd) + conn.close() + self.runner.run(interface.unpipe_events, hnd) + + class InvalidRunnerStateError(Exception): def __init__(self, message): super().__init__(message) diff --git a/src/blueapi/worker/event.py b/src/blueapi/worker/event.py index 880c3da21..25aae20f9 100644 --- a/src/blueapi/worker/event.py +++ b/src/blueapi/worker/event.py @@ -173,3 +173,9 @@ def is_error(self) -> bool: def is_complete(self) -> bool: return self.task_status is not None and self.task_status.task_complete + + @property + def task_id(self) -> str | None: + if task := self.task_status: + return task.task_id + return None diff --git a/tests/conftest.py b/tests/conftest.py index 78af27578..6f517b3bf 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -66,8 +66,13 @@ def oidc_config() -> OIDCConfig: @pytest.fixture -def config_with_auth(tmp_path: Path) -> str: - config = ApplicationConfig(auth_token_path=tmp_path / CACHE_FILE) +def token_cache_file(tmp_path: Path) -> Path: + return tmp_path / CACHE_FILE + + +@pytest.fixture +def config_with_auth(token_cache_file: Path, tmp_path: Path) -> str: + config = ApplicationConfig(auth_token_path=token_cache_file) config_path = tmp_path / "auth_config.yaml" with open(config_path, mode="w") as valid_auth_config_file: valid_auth_config_file.write(yaml.dump(config.model_dump())) @@ -134,7 +139,7 @@ def _make_token( @pytest.fixture def cached_valid_refresh( - tmp_path: Path, expired_token: dict[str, Any], oidc_config: OIDCConfig + token_cache_file: Path, expired_token: dict[str, Any], oidc_config: OIDCConfig ) -> Path: cache = Cache( oidc_config=oidc_config, @@ -142,14 +147,14 @@ def cached_valid_refresh( refresh_token=expired_token["refresh_token"], id_token=expired_token["id_token"], ) - with open(cache_path := tmp_path / CACHE_FILE, "xb") as cache_file: + with open(token_cache_file, "xb") as cache_file: cache_file.write(base64.b64encode(cache.model_dump_json().encode("utf-8"))) - return cache_path + return token_cache_file @pytest.fixture def cached_expired_refresh( - tmp_path: Path, expired_token: dict[str, Any], oidc_config: OIDCConfig + token_cache_file: Path, expired_token: dict[str, Any], oidc_config: OIDCConfig ) -> Path: cache = Cache( oidc_config=oidc_config, @@ -157,14 +162,16 @@ def cached_expired_refresh( refresh_token="expired_refresh", id_token=expired_token["id_token"], ) - with open(cache_path := tmp_path / CACHE_FILE, "xb") as cache_file: + with open(token_cache_file, "xb") as cache_file: cache_file.write(base64.b64encode(cache.model_dump_json().encode("utf-8"))) - return cache_path + return token_cache_file @pytest.fixture def cached_valid_token( - tmp_path: Path, valid_token_with_jwt: dict[str, Any], oidc_config: OIDCConfig + token_cache_file: Path, + valid_token_with_jwt: dict[str, Any], + oidc_config: OIDCConfig, ) -> Path: cache = Cache( oidc_config=oidc_config, @@ -172,14 +179,21 @@ def cached_valid_token( refresh_token=valid_token_with_jwt["refresh_token"], id_token=valid_token_with_jwt["id_token"], ) - with open(cache_path := tmp_path / CACHE_FILE, "xb") as cache_file: + with open(token_cache_file, "xb") as cache_file: cache_file.write(base64.b64encode(cache.model_dump_json().encode("utf-8"))) - return cache_path + return token_cache_file + + +@pytest.fixture +def cached_valid_token_value( + valid_token_with_jwt: dict[str, Any], +) -> str: + return valid_token_with_jwt["access_token"] @pytest.fixture def cache_with_invalid_audience( - tmp_path: Path, + token_cache_file: Path, oidc_config: OIDCConfig, valid_token_with_jwt_invalid_audience: dict[str, Any], ) -> Path: @@ -189,9 +203,9 @@ def cache_with_invalid_audience( refresh_token=valid_token_with_jwt_invalid_audience["refresh_token"], id_token=valid_token_with_jwt_invalid_audience["id_token"], ) - with open(cache_path := tmp_path / CACHE_FILE, "xb") as cache_file: + with open(token_cache_file, "xb") as cache_file: cache_file.write(base64.b64encode(cache.model_dump_json().encode("utf-8"))) - return cache_path + return token_cache_file @pytest.fixture diff --git a/tests/system_tests/test_blueapi_system.py b/tests/system_tests/test_blueapi_system.py index 8e2f5e528..80f708651 100644 --- a/tests/system_tests/test_blueapi_system.py +++ b/tests/system_tests/test_blueapi_system.py @@ -2,8 +2,8 @@ import time from asyncio import Queue from collections.abc import Generator -from contextlib import nullcontext from enum import StrEnum, auto +from itertools import count from pathlib import Path from unittest.mock import MagicMock, patch @@ -117,6 +117,27 @@ def instrument_session(request) -> str: return getattr(request, "param", VALID_INSTRUMENT_SESSION[User.alice]) +@pytest.fixture(scope="module") +def local_numtracker(): + """ + Local version of what we expect the numtracker to provide + + Allows different combinations of tests to be run without having to hard + code the expected scan numbers. + """ + return count(CURRENT_NUMTRACKER_NUM + 1) + + +@pytest.fixture +def scan_id(local_numtracker) -> int: + """ + The next value we expect to use for a scan_id + + Should be used in any test that calls out to numtracker. + """ + return next(local_numtracker) + + def task_factory( user: ValidUser, instrument_session: str | None, time: float = 0.0 ) -> TaskRequest: @@ -232,7 +253,7 @@ def blueapi_rest_client_get_methods() -> list[str]: return [ name for name, method in BlueapiRestClient.__dict__.items() - if not name.startswith("__") + if not name.startswith("_") and callable(method) and len(params := inspect.signature(method).parameters) == 1 and "self" in params @@ -532,20 +553,17 @@ def test_delete_current_environment(client: BlueapiClient): @pytest.mark.parametrize( - "task,scan_id,user", + "task,user", [ ( TaskRequest( name="count", params={ - "detectors": [ - "det", - ], + "detectors": ["det"], "num": 5, }, instrument_session=VALID_INSTRUMENT_SESSION[User.alice], ), - CURRENT_NUMTRACKER_NUM + 1, User.alice, ), ( @@ -574,13 +592,17 @@ def test_delete_current_environment(client: BlueapiClient): }, instrument_session=VALID_INSTRUMENT_SESSION[User.bob], ), - CURRENT_NUMTRACKER_NUM + 2, User.bob, ), ], ) +@pytest.mark.parametrize("run_method", ["run_task", "run_blocking"]) def test_plan_runs( - client_with_stomp: BlueapiClient, task: TaskRequest, scan_id: int, user: ValidUser + client_with_stomp: BlueapiClient, + task: TaskRequest, + scan_id: int, + user: ValidUser, + run_method: str, ): resource = Queue(maxsize=1) start = Queue(maxsize=1) @@ -592,7 +614,8 @@ def on_event(event: AnyEvent) -> None: if event.name == "stream_resource": resource.put_nowait(event.doc) - final_event = client_with_stomp.run_task(task, on_event) + runner = getattr(client_with_stomp, run_method) + final_event = runner(task, on_event) assert isinstance(final_event.result, TaskResult) assert final_event.task_complete assert not final_event.task_failed @@ -669,56 +692,40 @@ def test_task_submission_after_invalid_task(client_with_stomp: BlueapiClient): @pytest.mark.parametrize( - "instrument_session,user,expectation", + "instrument_session,user", [ - ( - # bob cannot submit a task that alice is on - VALID_INSTRUMENT_SESSION[User.alice], - User.bob, - pytest.raises( - UnauthorisedAccessError, match="Not authorized to submit task" - ), - ), - ( - # alice cannot submit a task that bob is on - VALID_INSTRUMENT_SESSION[User.bob], - User.alice, - pytest.raises( - UnauthorisedAccessError, match="Not authorized to submit task" - ), - ), - ( - # alice can submit a task that alice is on - VALID_INSTRUMENT_SESSION[User.alice], - User.alice, - nullcontext(), - ), - ( - # bob can submit a task that bob is on - VALID_INSTRUMENT_SESSION[User.bob], - User.bob, - nullcontext(), - ), - ( - # admin can submit a task that bob is on - VALID_INSTRUMENT_SESSION[User.bob], - User.admin, - nullcontext(), - ), - ( - # admin can submit a task that alice is on - VALID_INSTRUMENT_SESSION[User.alice], - User.admin, - nullcontext(), - ), - ( - # admin still needs to put a valid instrument_session - INVALID_INSTRUMENT_SESSION, - User.admin, - pytest.raises( - UnauthorisedAccessError, match="Not authorized to submit task" - ), - ), + # bob cannot submit a task that alice is on + (VALID_INSTRUMENT_SESSION[User.alice], User.bob), + # alice cannot submit a task that bob is on + (VALID_INSTRUMENT_SESSION[User.bob], User.alice), + # admin still needs to put a valid instrument_session + (INVALID_INSTRUMENT_SESSION, User.admin), + ], +) +@pytest.mark.parametrize("method_name", ["create_task", "run_blocking"]) +def test_run_task_without_authz( + client: BlueapiClient, + small_task: TaskRequest, + user: ValidUser, + instrument_session: str, + method_name: str, +): + method = getattr(client, method_name) + with pytest.raises(UnauthorisedAccessError, match="Not authorized to submit task"): + method(small_task) + + +@pytest.mark.parametrize( + "instrument_session,user", + [ + # alice can submit a task that alice is on + (VALID_INSTRUMENT_SESSION[User.alice], User.alice), + # bob can submit a task that bob is on + (VALID_INSTRUMENT_SESSION[User.bob], User.bob), + # admin can submit a task that bob is on + (VALID_INSTRUMENT_SESSION[User.bob], User.admin), + # admin can submit a task that alice is on + (VALID_INSTRUMENT_SESSION[User.alice], User.admin), ], ) def test_create_task_authorization( @@ -726,10 +733,8 @@ def test_create_task_authorization( small_task: TaskRequest, user: ValidUser, instrument_session: str, - expectation, ): - with expectation: - client.create_task(small_task) + client.create_task(small_task) def test_non_admin_can_only_get_own_tasks( @@ -890,3 +895,10 @@ def test_admin_can_abort_any_task( task_factory(user, VALID_INSTRUMENT_SESSION[user], time=1) ) client_factory[AdminUser.admin].abort() + + +def test_run_blocking_requires_auth( + client_without_auth: BlueapiClient, small_task: TaskRequest +): + with pytest.raises(UnauthorisedAccessError): + client_without_auth.run_blocking(small_task) diff --git a/tests/unit_tests/cli/test_cli.py b/tests/unit_tests/cli/test_cli.py index 3e27cfc98..632f28d65 100644 --- a/tests/unit_tests/cli/test_cli.py +++ b/tests/unit_tests/cli/test_cli.py @@ -7,7 +7,6 @@ from pathlib import Path from textwrap import dedent from typing import Any, TypeVar -from unittest import mock from unittest.mock import Mock, patch import pytest @@ -385,9 +384,9 @@ def test_run_plan_feedback( main, ["controller", "run", "-i", "cm12345-1", "name"], ) + bc.add_callback.assert_called_once() bc.run_task.assert_called_once_with( TaskRequest(name="name", params={}, instrument_session="cm12345-1"), - on_event=mock.ANY, ) assert res.exit_code == 0 assert res.stdout == message @@ -1478,3 +1477,18 @@ def test_host_overrides_config(runner: CliRunner): ) assert response.call_count == 1 assert res.exit_code == 0 + + +@patch("blueapi.cli.cli.BlueapiClient") +def test_run_ws_runs_blocking_plan(mock_client: Mock, runner: CliRunner): + bc = mock_client.from_config() + res = runner.invoke( + main, + ["controller", "run", "-i", "cm12345-1", "--ws", "name"], + ) + bc.add_callback.assert_called_once() + bc.run_task.assert_not_called() + bc.run_blocking.assert_called_once_with( + TaskRequest(name="name", params={}, instrument_session="cm12345-1"), + ) + assert res.exit_code == 0 diff --git a/tests/unit_tests/client/test_client.py b/tests/unit_tests/client/test_client.py index eaf96a3b2..7c3608348 100644 --- a/tests/unit_tests/client/test_client.py +++ b/tests/unit_tests/client/test_client.py @@ -1000,3 +1000,52 @@ def test_client_login_no_oidc( client.login() mock_session_manager.assert_not_called() + + +@pytest.mark.parametrize("event", [COMPLETE_EVENT, FAILED_EVENT]) +def test_run_blocking(event: WorkerEvent, client: BlueapiClient, mock_rest: Mock): + mock_rest.run_blocking.side_effect = lambda req: [event] + res = client.run_blocking(TaskRequest(name="foo", instrument_session="cm12345-1")) + assert res == event.task_status + + +@pytest.mark.parametrize("event", [COMPLETE_EVENT, FAILED_EVENT]) +def test_run_blocking_callbacks( + event: WorkerEvent, client: BlueapiClient, mock_rest: Mock +): + callback = Mock() + mock_rest.run_blocking.side_effect = lambda req: [event] + client.run_blocking(Mock(), on_event=callback) + + callback.assert_called_once_with(event) + + +@pytest.mark.parametrize("event", [COMPLETE_EVENT, FAILED_EVENT]) +def test_run_blocking_client_callbacks( + event: WorkerEvent, client: BlueapiClient, mock_rest: Mock +): + callback = Mock() + mock_rest.run_blocking.side_effect = lambda req: [event] + client.add_callback(callback) + client.run_blocking(Mock()) + + callback.assert_called_once_with(event) + + +def test_run_blocking_error_if_cut_short(client: BlueapiClient, mock_rest: Mock): + mock_rest.run_blocking.side_effect = lambda req: [] + with pytest.raises( + BlueskyRemoteControlError, match="Connection closed before plan completed" + ): + client.run_blocking(Mock()) + + +def test_run_blocking_ignores_callback_error(client: BlueapiClient, mock_rest: Mock): + mock_rest.run_blocking.side_effect = lambda req: [COMPLETE_EVENT] + + def broken_callback(_: AnyEvent): + raise Exception("This callback is broken") + + client.add_callback(broken_callback) + res = client.run_blocking(Mock()) + assert res == COMPLETE_EVENT.task_status diff --git a/tests/unit_tests/client/test_rest.py b/tests/unit_tests/client/test_rest.py index c3386ad55..0fd3bd0cd 100644 --- a/tests/unit_tests/client/test_rest.py +++ b/tests/unit_tests/client/test_rest.py @@ -10,10 +10,12 @@ from packaging.version import Version from pydantic_core import PydanticSerializationError from responses import DELETE, GET, PUT, matchers +from websockets import Headers, InvalidStatus, Response from blueapi import __version__ from blueapi.client.client import DeviceRef from blueapi.client.rest import ( + USER_AGENT, BlueapiRestClient, BlueskyRemoteControlError, BlueskyRequestError, @@ -47,11 +49,13 @@ def rest() -> BlueapiRestClient: @pytest.fixture -def rest_with_auth(oidc_config: OIDCConfig, tmp_path) -> BlueapiRestClient: +def rest_with_auth( + oidc_config: OIDCConfig, token_cache_file: Path +) -> BlueapiRestClient: return BlueapiRestClient( session_manager=SessionManager( server_config=oidc_config, - cache_manager=SessionCacheManager(tmp_path / "blueapi_cache"), + cache_manager=SessionCacheManager(token_cache_file), ) ) @@ -431,3 +435,120 @@ def test_get_missing_plan(rest: BlueapiRestClient): responses.add(GET, "http://localhost:8000/plans/foo", status=404) with pytest.raises(UnknownPlanError): rest.get_plan("foo") + + +@patch("blueapi.client.rest.connect") +def test_run_blocking(mock_connect: Mock, rest: BlueapiRestClient): + ws = MagicMock() + ws.__enter__.return_value.__iter__.return_value = iter( + ['{"kind": "update", "data": {"name": "start", "doc":{}, "task_id":"t_uid"}}'] + ) + mock_connect.return_value = ws + conn = rest.run_blocking( + TaskRequest(name="foo", params={"one": "two"}, instrument_session="cm12345-1") + ) + next(iter(conn)) + mock_connect.assert_called_once_with( + "ws://localhost:8000/api/v2/run_plan", + additional_headers={}, + user_agent_header=USER_AGENT, + ) + + +@patch("blueapi.client.rest.connect") +def test_run_blocking_auth( + mock_connect: Mock, + rest_with_auth: BlueapiRestClient, + mock_authn_server: responses.RequestsMock, + cached_valid_token: Path, # creates the cache file + cached_valid_token_value: str, # the value from the file +): + ws = MagicMock() + ws.__enter__.return_value.__iter__.return_value = iter( + ['{"kind": "update", "data": {"name": "start", "doc":{}, "task_id":"t_uid"}}'] + ) + mock_connect.return_value = ws + conn = rest_with_auth.run_blocking( + TaskRequest(name="foo", params={"one": "two"}, instrument_session="cm12345-1") + ) + next(iter(conn)) + mock_connect.assert_called_once_with( + "ws://localhost:8000/api/v2/run_plan", + additional_headers={"Authorization": f"Bearer {cached_valid_token_value}"}, + user_agent_header=USER_AGENT, + ) + + +@pytest.mark.parametrize( + "event,error,message", + [ + ( + """{ + "kind":"invalid_args", + "errors": [{ + "loc": ["bar"], + "msg": "Field required", + "type": "missing", + "input": {} + }] + }""", + InvalidParametersError, + "ParameterError", + ), + ('{"kind": "busy"}', BlueskyRemoteControlError, "Server is busy"), + ('{"kind": "plan_not_found", "plan_name": "foo"}', UnknownPlanError, "foo"), + ], +) +@patch("blueapi.client.rest.connect") +def test_run_blocking_errors( + mock_connect: Mock, + rest: BlueapiRestClient, + event: str, + error: type[Exception], + message: str, +): + ws = MagicMock() + ws.__enter__.return_value.__iter__.return_value = iter([event]) + mock_connect.return_value = ws + conn = rest.run_blocking( + TaskRequest(name="foo", params={"one": "two"}, instrument_session="cm12345-1") + ) + with pytest.raises(error, match=message): + next(iter(conn)) + mock_connect.assert_called_once_with( + "ws://localhost:8000/api/v2/run_plan", + additional_headers={}, + user_agent_header=USER_AGENT, + ) + + +@pytest.mark.parametrize("status_code", [401, 403]) +@patch("blueapi.client.rest.connect") +def test_run_blocking_ws_failures( + mock_connect: Mock, rest: BlueapiRestClient, status_code: int +): + mock_connect.side_effect = InvalidStatus( + response=Response( + status_code=status_code, reason_phrase="test_error", headers=Headers() + ) + ) + conn = rest.run_blocking( + TaskRequest(name="foo", params={"one": "two"}, instrument_session="cm12345-1") + ) + with pytest.raises(UnauthorisedAccessError): + next(iter(conn)) + + +@patch("blueapi.client.rest.connect") +def test_run_blocking_unknown_error(mock_connect: Mock, rest: BlueapiRestClient): + mock_connect.side_effect = InvalidStatus( + response=Response( + status_code=1234, reason_phrase="test_error", headers=Headers() + ) + ) + + conn = rest.run_blocking( + TaskRequest(name="foo", params={"one": "two"}, instrument_session="cm12345-1") + ) + with pytest.raises(BlueskyRemoteControlError): + next(iter(conn)) diff --git a/tests/unit_tests/service/test_authentication.py b/tests/unit_tests/service/test_authentication.py index 01bc426e2..a76375812 100644 --- a/tests/unit_tests/service/test_authentication.py +++ b/tests/unit_tests/service/test_authentication.py @@ -189,19 +189,23 @@ def test_tiled_auth_sync_auth_flow(): @pytest.mark.parametrize( - "header,token", + "header,cookie,token", [ - (None, None), - ("ApiKey foobar", None), - ("Bearer foobar", "foobar"), - ("Bearer with_whitespace ", "with_whitespace"), - ("Bearerfoobar", None), + (None, None, None), + ("", None, None), + ("ApiKey foobar", None, None), + ("Bearer foobar", None, "foobar"), + ("Bearer with_whitespace ", None, "with_whitespace"), + ("Bearerfoobar", None, None), + (None, "Bearer foobar", "foobar"), + ("", "Bearer foo", "foo"), + ("Bearer foo", "bearer bar", "foo"), ], ) -def test_unchecked_bearer_token(header: str | None, token: str | None): - req = Mock() - req.headers.get.side_effect = lambda key: header if key == "Authorization" else None - +def test_unchecked_bearer_token( + header: str | None, cookie: str | None, token: str | None +): + req = Mock(headers={"Authorization": header}, cookies={"Authorization": cookie}) assert unchecked_bearer_token(req) == token diff --git a/tests/unit_tests/service/test_interface.py b/tests/unit_tests/service/test_interface.py index ac756fe85..d8c0de24b 100644 --- a/tests/unit_tests/service/test_interface.py +++ b/tests/unit_tests/service/test_interface.py @@ -2,6 +2,7 @@ import uuid from dataclasses import dataclass from inspect import isawaitable +from multiprocessing.connection import Connection as PipeConnection from typing import Any from unittest.mock import ANY, MagicMock, Mock, patch @@ -683,3 +684,47 @@ async def test_update_scan_num_side_effect_sets_scan_file_in_re_md( assert isawaitable(scan_id) and await scan_id assert ctx.run_engine.md["scan_file"] == "p46-11" + + +@patch("blueapi.service.interface.worker") +def test_pipe_events(mock_worker: Mock): + worker = mock_worker() + tx = Mock(spec=PipeConnection) + + interface.pipe_events(tx) + + worker.worker_events.subscribe.assert_called_once() + worker.data_events.subscribe.assert_called_once() + worker.progress_events.subscribe.assert_called_once() + + handler = worker.worker_events.subscribe.call_args[0][0] + + evt = Mock() + handler(evt, "ignored correlation id") + tx.send.assert_called_once_with(evt) + + +@patch("blueapi.service.interface.worker") +def test_pipe_events_ignores_broken_pipe(mock_worker: Mock): + worker = mock_worker() + tx = Mock(spec=PipeConnection) + + interface.pipe_events(tx) + + worker.worker_events.subscribe.assert_called_once() + handler = worker.worker_events.subscribe.call_args[0][0] + + tx.send.side_effect = BrokenPipeError() + # ensure that exceptions are not raised + handler(Mock(), "ignored correlation id") + + +@patch("blueapi.service.interface.worker") +def test_unpipe_events(mock_worker: Mock): + worker = mock_worker() + handles = interface.SubHandles(worker=1, progress=2, data=3) + interface.unpipe_events(handles) + + worker.worker_events.unsubscribe.assert_called_once_with(1) + worker.progress_events.unsubscribe.assert_called_once_with(2) + worker.data_events.unsubscribe.assert_called_once_with(3) diff --git a/tests/unit_tests/service/test_middleware.py b/tests/unit_tests/service/test_middleware.py index 5f6dbeac6..1e0a96a93 100644 --- a/tests/unit_tests/service/test_middleware.py +++ b/tests/unit_tests/service/test_middleware.py @@ -1,4 +1,5 @@ -from unittest.mock import AsyncMock, MagicMock, Mock, patch +from typing import Any +from unittest.mock import ANY, AsyncMock, MagicMock, Mock, patch import pytest from starlette.types import ASGIApp @@ -11,6 +12,8 @@ VERSION, ObservabilityContextPropagator, VersionHeaders, + WebsocketTracing, + _redact_headers, ) @@ -106,3 +109,193 @@ async def test_obs_context_passes_vendor_context(app: Mock, protocol: str): ApplicationConfig.VENDOR_CONTEXT_HEADER: "vendor_context", } ) + + +def test_redact_headers(): + assert list(_redact_headers([(b"authorization", b"Bearer foobar")])) == [ + (b"authorization", b"Bearer [REDACTED]") + ] + assert list(_redact_headers([(b"other-header", b"Not affected")])) == [ + (b"other-header", b"Not affected") + ] + + +@pytest.fixture +def asgi() -> AsyncMock: + return AsyncMock(name="asgi-app", spec=ASGIApp) + + +@pytest.fixture +def ws_tracer(asgi: Mock) -> WebsocketTracing: + return WebsocketTracing(asgi) + + +@pytest.fixture +def send() -> AsyncMock: + return AsyncMock() + + +@pytest.fixture +def receive() -> AsyncMock: + return AsyncMock() + + +# logger patch that defaults to enabled for all levels +def patch_ws_logger(func): + return patch( + "blueapi.service.middleware.WS_LOGGER", + isEnabledFor=Mock(name="custom", return_value=True), + )(func) + + +@patch_ws_logger +async def test_websocket_tracing_does_nothing_when_not_debug( + log: Mock, + asgi: AsyncMock, + ws_tracer: WebsocketTracing, + send: AsyncMock, + receive: AsyncMock, +): + scope = {"type": "websocket"} + log.isEnabledFor.return_value = False + await ws_tracer(scope, receive, send) + + asgi.assert_called_once_with(scope, receive, send) + + +@patch_ws_logger +async def test_websocket_tracing_does_nothing_for_http( + log: Mock, + asgi: AsyncMock, + ws_tracer: WebsocketTracing, + send: AsyncMock, + receive: AsyncMock, +): + scope = {"type": "http"} + await ws_tracer(scope, receive, send) + + asgi.assert_called_once_with(scope, receive, send) + + +@patch_ws_logger +async def test_websocket_tracing_logs_new_connection( + log: Mock, ws_tracer: WebsocketTracing, send: AsyncMock, receive: AsyncMock +): + scope = {"type": "websocket", "headers": [(b"authorization", b"bearer foobar")]} + await ws_tracer(scope, receive, send) + log.debug.assert_called_once_with( + "New Connection from %r", + {"type": "websocket", "headers": [(b"authorization", b"bearer [REDACTED]")]}, + extra=ANY, + ) + + +@pytest.mark.parametrize( + "type,other,log_args", + [ + ( + "websocket.send", + {"text": "demo"}, + ("Sending: %r", "demo"), + ), + ( + "websocket.accept", + {"headers": [(b"bapi-version", b"1.2.3")]}, + ( + "Accepting websocket - sending headers: %r", + [(b"bapi-version", b"1.2.3")], + ), + ), + ( + "websocket.close", + {"code": 1234, "reason": "error_code"}, + ("Closing with code: %r, reason: %r", 1234, "error_code"), + ), + ( + "websocket.http.response.start", + {"status": "ws-status", "headers": [(b"bapi-version", b"1.2.3")]}, + ( + "HTTP Response: status=%r, headers=%r", + "ws-status", + [(b"bapi-version", b"1.2.3")], + ), + ), + ( + "websocket.http.response.body", + {"body": "response content"}, + ( + "HTTP Response Content: %r", + "response content", + ), + ), + ( + "unknown.msg.type", + {"other": "data"}, + ("Sending other: %r", {"type": "unknown.msg.type", "other": "data"}), + ), + ], +) +@patch_ws_logger +async def test_websocket_tracing_local_send( + log: Mock, + asgi: AsyncMock, + ws_tracer: WebsocketTracing, + send: AsyncMock, + type: str, + other: dict[str, Any], + log_args: tuple[tuple[str, ...], dict[str, Any]], +): + await ws_tracer({"type": "websocket"}, AsyncMock(), send) + + _, _, local_send = asgi.call_args[0] + + message = {"type": type, **other} + await local_send(message) + log.debug.assert_called_with(*log_args, extra=ANY) + + # Original send method should be called with original message + send.assert_called_once_with(message) + + +@pytest.mark.parametrize( + "type,other,log_args", + [ + ( + "websocket.receive", + {"text": "demo"}, + ("Received: %r", "demo"), + ), + ( + "websocket.connect", + {}, + ("New connection from %s:%d", "unknown", 0), + ), + ("unknown.msg", {}, ("Received other: %r", {"type": "unknown.msg"})), + ], +) +@patch_ws_logger +async def test_websocket_tracing_local_receive( + log: Mock, + asgi: AsyncMock, + ws_tracer: WebsocketTracing, + receive: AsyncMock, + type: str, + other: dict[str, Any], + log_args: tuple[tuple[str, ...], dict[str, Any]], +): + await ws_tracer({"type": "websocket"}, receive, AsyncMock()) + + _, local_recv, _ = asgi.call_args[0] + + message = {"type": type, **other} + receive.return_value = message + + received = await local_recv() + + # original receive called to get message + receive.assert_called_once_with() + + log.debug.assert_called_with(*log_args, extra=ANY) + + # We should not be modifying anything + assert received == message diff --git a/tests/unit_tests/service/test_protocol.py b/tests/unit_tests/service/test_protocol.py new file mode 100644 index 000000000..9178162e1 --- /dev/null +++ b/tests/unit_tests/service/test_protocol.py @@ -0,0 +1,104 @@ +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 ( + Abort, + ArgumentError, + ControlRequest, + ControlResponse, + InvalidArgs, + Pause, + Resume, + Submit, +) + + +@pytest.mark.parametrize( + "src,res", + [ + ( + """{ + "kind": "submit", + "task": { + "name": "foo", + "instrument_session": "cm12345-1" + } + }""", + Submit( + task=TaskRequest(name="foo", params={}, instrument_session="cm12345-1") + ), + ), + ('{"kind": "pause"}', Pause()), + ('{"kind": "resume"}', Resume()), + ('{"kind": "abort"}', Abort()), + ], +) +def test_request_deserialization(src: str, res: Any): + req = ControlRequest.validate_json(src) + assert req == res + + +@pytest.mark.parametrize( + "src,res", + [ + ( + """{ + "kind": "invalid_args", + "errors":[{ + "loc":["body","params","spec"], + "msg":"error_message", + "type":"error_type", + "input":"original input" + }]}""", + InvalidArgs( + errors=[ + ArgumentError( + loc=["body", "params", "spec"], + msg="error_message", + type="error_type", + input="original input", + ) + ] + ), + ), + ], +) +def test_response_deserialization(src: str, res: Any): + req = ControlResponse.validate_json(src) + assert req == res + + +def test_from_empty_validation_error(): + err = InvalidArgs.from_validation_error( + ValidationError("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( + loc=("foo", "bar"), + type="missing", + input={"foo": {"no": "bar"}}, + ) + ], + ), + ) + assert err == InvalidArgs( + errors=[ + ArgumentError( + loc=["body", "params", "foo", "bar"], + msg="Field required", + 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 dedd9d7ff..a10d5ab3c 100644 --- a/tests/unit_tests/service/test_rest_api.py +++ b/tests/unit_tests/service/test_rest_api.py @@ -1,18 +1,21 @@ +import contextlib import uuid -from collections.abc import Iterator +from collections.abc import AsyncIterator, Iterator from dataclasses import dataclass -from typing import Any +from typing import Any, cast from unittest.mock import MagicMock, Mock, patch import jwt import pytest from bluesky._vendor.super_state_machine.errors import TransitionError from bluesky.protocols import Stoppable -from fastapi import HTTPException, status +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 starlette.testclient import WebSocketDenialResponse +from starlette.types import Message, Receive, Scope, Send from blueapi.config import ( ApplicationConfig, @@ -43,7 +46,7 @@ WorkerTask, ) from blueapi.service.runner import WorkerDispatcher -from blueapi.worker.event import WorkerState +from blueapi.worker.event import TaskStatus, WorkerEvent, WorkerState from blueapi.worker.task import Task from blueapi.worker.task_worker import TrackableTask @@ -55,9 +58,19 @@ class MockCountModel(BaseModel): ... FAKE_INSTRUMENT_SESSION = "cm12345-1" +SUBMIT_REQUEST = { + "kind": "submit", + "task": { + "name": "foo", + "params": {"one": "two"}, + "instrument_session": "cm12345-1", + }, +} + + @pytest.fixture def mock_runner() -> Mock: - return Mock(spec=WorkerDispatcher) + return MagicMock(spec=WorkerDispatcher) @pytest.fixture @@ -995,3 +1008,263 @@ def test_logout_when_oidc_config_invalid( response = client_with_auth.get("/logout") assert response.status_code == status.HTTP_205_RESET_CONTENT + + +async def test_websocket_run_plan(mock_runner: Mock, client: TestClient): + mock_runner.run.side_effect = lambda req, *a, **kw: { + interface.get_active_task: None, + interface.submit_task: "task_id", + interface.begin_task: None, + }.get(req) + mock_runner.event_pipe.return_value = contextlib.nullcontext( + _aiter( + WorkerEvent( + state=WorkerState.RUNNING, + task_status=TaskStatus( + task_id="task_id", + result=None, + task_complete=False, + task_failed=False, + ), + ), + WorkerEvent( + state=WorkerState.IDLE, + task_status=TaskStatus( + task_id="task_id", + result=None, + task_complete=True, + task_failed=False, + ), + ), + ) + ) + + with client.websocket_connect("/api/v2/run_plan") as ws: + ws.send_json(SUBMIT_REQUEST) + assert ws.receive_json() == { + "kind": "update", + "data": { + "state": "RUNNING", + "task_status": { + "task_id": "task_id", + "result": None, + "task_complete": False, + "task_failed": False, + }, + "errors": [], + "warnings": [], + }, + } + assert ws.receive_json() == { + "kind": "update", + "data": { + "state": "IDLE", + "task_status": { + "task_id": "task_id", + "result": None, + "task_complete": True, + "task_failed": False, + }, + "errors": [], + "warnings": [], + }, + } + with pytest.raises(WebSocketDisconnect) as discon: + ws.receive_text() + + # Check it's a 'normal' end of stream disconnect + assert discon.value.code == 1000 + assert discon.value.reason == "" + + +@pytest.mark.parametrize("req", ["not a json object", "[]", '{"invalid": "keys"}']) +def test_websocket_run_plan_invalid_request( + req: str, mock_runner: Mock, client: TestClient +): + with client.websocket_connect("/api/v2/run_plan") as ws: + ws.send_text(req) + with pytest.raises(WebSocketDisconnect) as disco: + ws.receive_text() + assert disco.value.code == 1007 + assert disco.value.reason == "Invalid Request" + + +@pytest.mark.parametrize( + "exc,err_message,code,reason", + [ + ( + KeyError("foo"), + {"kind": "plan_not_found", "plan_name": "foo"}, + 4001, + "Unknown Plan", + ), + ( + ValidationError("Not valid", []), + {"kind": "invalid_args", "errors": []}, + 4002, + "Invalid Args", + ), + ], +) +def test_websocket_run_plan_submit_error( + exc: Exception, + err_message: str, + code: int, + reason, + mock_runner: Mock, + client: TestClient, +): + mock_runner.run.side_effect = exc + with client.websocket_connect("/api/v2/run_plan") as ws: + ws.send_json(SUBMIT_REQUEST) + assert ws.receive_json() == err_message + with pytest.raises(WebSocketDisconnect) as disco: + ws.receive_text() + assert disco.value.code == code + assert disco.value.reason == reason + + +def test_websocket_run_plan_server_busy(mock_runner: Mock, client: TestClient): + mock_runner.run.side_effect = [ + "task_id", # submit_task + Mock(name="active_task", is_complete=False), + ] + with client.websocket_connect("/api/v2/run_plan") as ws: + ws.send_json(SUBMIT_REQUEST) + assert ws.receive_json() == {"kind": "busy"} + with pytest.raises(WebSocketDisconnect) as disco: + ws.receive_text() + assert disco.value.code == 1013 + assert disco.value.reason == "Worker busy" + pass + + +def test_websocket_run_plan_unrelated_events(mock_runner: Mock, client: TestClient): + mock_runner.run.side_effect = lambda req, *a, **kw: { + interface.get_active_task: None, + interface.submit_task: "task_id", + interface.begin_task: None, + }.get(req) + mock_runner.event_pipe.return_value = contextlib.nullcontext( + _aiter( + WorkerEvent( + state=WorkerState.RUNNING, + task_status=TaskStatus( + task_id="other_task_id", + result=None, + task_complete=False, + task_failed=False, + ), + ), + WorkerEvent( + state=WorkerState.IDLE, + task_status=TaskStatus( + task_id="task_id", + result=None, + task_complete=True, + task_failed=False, + ), + ), + ) + ) + + with client.websocket_connect("/api/v2/run_plan") as ws: + ws.send_json(SUBMIT_REQUEST) + # first event is not sent + assert ws.receive_json() == { + "kind": "update", + "data": { + "state": "IDLE", + "task_status": { + "task_id": "task_id", + "result": None, + "task_complete": True, + "task_failed": False, + }, + "errors": [], + "warnings": [], + }, + } + with pytest.raises(WebSocketDisconnect) as discon: + ws.receive_text() + + # Check it's a 'normal' end of stream disconnect + assert discon.value.code == 1000 + assert discon.value.reason == "" + pass + + +def test_websocket_run_plan_client_disconnect_cancels( + mock_runner: Mock, client: TestClient +): + mock_runner.run.side_effect = ["task_id", None, None, None] + mock_runner.event_pipe.return_value = contextlib.nullcontext( + _aiter( + WorkerEvent( + state=WorkerState.IDLE, + task_status=TaskStatus( + task_id="task_id", + result=None, + task_complete=False, + task_failed=False, + ), + ), + ) + ) + + class Disconnector: + def __init__(self, app): + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send): + async def local_send(message: Message): + if message.get("type") == "websocket.send": + # Simulate the connection being closed + raise OSError() + await send(message) + + return await self.app(scope, receive, local_send) + + cast(FastAPI, client.app).add_middleware(Disconnector) + with client.websocket_connect("/api/v2/run_plan") as ws: + ws.send_json(SUBMIT_REQUEST) + mock_runner.run.assert_called_with( + interface.cancel_active_task, failure=True, reason="Client disconnected" + ) + + +@pytest.mark.parametrize("token", ["Bearer invalid", None]) +def test_websocket_run_plan_needs_auth_token( + client_with_auth: TestClient, token: str | None +): + del client_with_auth.headers["Authorization"] + if token: + client_with_auth.headers["Authorization"] = token + with pytest.raises(WebSocketDenialResponse) as wsdr: + with client_with_auth.websocket_connect("/api/v2/run_plan"): + pass + assert wsdr.value.status_code == 401 + + +def test_websocket_run_plan_needs_permission( + mock_runner: Mock, + client_with_opa: TestClient, + mock_opa_client: Mock, + access_token: str, +): + client_with_opa.headers["Authorization"] = f"Bearer {access_token}" + mock_opa_client.can_submit_task.side_effect = HTTPException(status_code=403) + mock_runner.run.side_effect = RuntimeError("Task should not be submitted") + with client_with_opa.websocket_connect("/api/v2/run_plan") as ws: + ws.send_json(SUBMIT_REQUEST) + assert ws.receive_json() == {"kind": "unauthorized"} + with pytest.raises(WebSocketDisconnect) as discon: + ws.receive_text() + + assert discon.value.code == 4003 + assert discon.value.reason == "Unauthorized" + + +async def _aiter(*values: Any) -> AsyncIterator: + for value in values: + yield value diff --git a/tests/unit_tests/service/test_runner.py b/tests/unit_tests/service/test_runner.py index b41b3fd4c..8380784e0 100644 --- a/tests/unit_tests/service/test_runner.py +++ b/tests/unit_tests/service/test_runner.py @@ -1,5 +1,7 @@ import uuid from collections.abc import Callable +from multiprocessing import Pipe +from multiprocessing.connection import Connection from multiprocessing.pool import Pool as PoolClass from typing import Any, Generic, TypeVar from unittest.mock import MagicMock, Mock, NonCallableMock, patch @@ -14,12 +16,14 @@ from blueapi.service import interface from blueapi.service.model import EnvironmentResponse from blueapi.service.runner import ( + EventStream, InvalidRunnerStateError, RpcError, WorkerDispatcher, _safe_exception_message, import_and_run_function, ) +from blueapi.worker.event import TaskStatus, WorkerEvent, WorkerState @pytest.fixture @@ -300,3 +304,81 @@ def test_run_span_ok( ): with asserting_span_exporter(exporter, "run", "function", "args", "kwargs"): started_runner.run(interface.get_plans) + + +@patch("blueapi.service.runner.Pipe") +def test_event_pipe(mock_pipe: Mock): + tx = Mock() + rx = Mock() + + mock_pipe.return_value = (tx, rx) + + dispatcher = Mock() + dispatcher.run.side_effect = lambda mth, *a: { + interface.pipe_events: 42, + interface.unpipe_events: None, + }[mth] + evt_pipe = WorkerDispatcher.event_pipe(dispatcher) + + with evt_pipe: + dispatcher.run.assert_called_with(interface.pipe_events, tx) + + dispatcher.run.assert_called_with(interface.unpipe_events, 42) + + assert len(dispatcher.run.mock_calls) == 2 + + +async def test_event_stream(): + tx, rx = Pipe() + stream = EventStream(rx) + + evt = WorkerEvent( + state=WorkerState.RUNNING, + task_status=TaskStatus( + task_id="foo", task_complete=False, task_failed=False, result=None + ), + ) + + tx.send(evt) + + assert (await anext(stream)) == evt + + +async def test_end_of_event_stream(): + tx, rx = Pipe() + stream = EventStream(rx) + + tx.close() + + with pytest.raises(StopAsyncIteration): + await anext(stream) + + +async def test_aiter_event_stream_is_self(): + stream = EventStream(Mock()) + assert aiter(stream) is stream + + +async def test_event_stream_wait_for_event(): + tx, rx = Pipe() + + mrx = Mock(spec=Connection) + mrx.poll.side_effect = [False, False, True] + mrx.fileno.side_effect = rx.fileno + mrx.recv.side_effect = rx.recv + + stream = EventStream(mrx) + + evt = WorkerEvent( + state=WorkerState.RUNNING, + task_status=TaskStatus( + task_id="foo", task_complete=False, task_failed=False, result=None + ), + ) + + tx.send(evt) + + read = await anext(stream) + assert read == evt + + assert len(mrx.poll.mock_calls) == 3 diff --git a/tests/unit_tests/test_config.py b/tests/unit_tests/test_config.py index ed00587a1..8ccdbb369 100644 --- a/tests/unit_tests/test_config.py +++ b/tests/unit_tests/test_config.py @@ -12,13 +12,14 @@ import responses import yaml from bluesky_stomp.models import BasicAuthentication -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, WebsocketUrl from blueapi.config import ( CONFIG_SCHEMA_LOCATION, ApplicationConfig, ConfigLoader, OIDCConfig, + RestConfig, generate_config_schema, ) from blueapi.utils import InvalidConfigError @@ -571,3 +572,15 @@ def test_issuer_url_preferred_over_well_known_url(caplog): oidc._well_known_url == "issuer_url" + "/.well-known/openid-configuration" ) assert "well_known_url and issuer are both set" in caplog.text + + +@pytest.mark.parametrize( + "url,ws_url", + [ + ("http://localhost:8000", "ws://localhost:8000"), + ("https://localhost:8000", "wss://localhost:8000"), + ], +) +def test_ws_address(url: str, ws_url: str): + conf = RestConfig.model_validate({"url": url}) + assert conf.ws_address == WebsocketUrl(ws_url) diff --git a/tests/unit_tests/worker/test_task_worker.py b/tests/unit_tests/worker/test_task_worker.py index 588c37d8e..5e6553d2f 100644 --- a/tests/unit_tests/worker/test_task_worker.py +++ b/tests/unit_tests/worker/test_task_worker.py @@ -922,3 +922,43 @@ def test_task_result_serialization(plan_result, task_result, type_name): res = TaskResult.from_result(plan_result) assert res.result == task_result assert res.type == type_name + + +@pytest.mark.parametrize( + "status,complete", + [ + (None, False), + ( + TaskStatus( + task_id="foo", result=None, task_complete=False, task_failed=False + ), + False, + ), + ( + TaskStatus( + task_id="foo", result=None, task_complete=True, task_failed=False + ), + True, + ), + ], +) +def test_worker_event_complete(status: TaskStatus | None, complete: bool): + event = WorkerEvent( + state=WorkerState.IDLE, task_status=status + ) # state is not used in check + assert event.is_complete() == complete + + +def test_worker_event_task_id(): + event = WorkerEvent( + state=WorkerState.RUNNING, + task_status=TaskStatus( + task_id="foo", result=None, task_complete=False, task_failed=False + ), + ) + assert event.task_id == "foo" + + +def test_worker_event_no_task_id(): + event = WorkerEvent(state=WorkerState.IDLE, task_status=None) + assert event.task_id is None diff --git a/uv.lock b/uv.lock index 5cda625b2..e044a86f6 100644 --- a/uv.lock +++ b/uv.lock @@ -531,7 +531,7 @@ requires-dist = [ { name = "stomp-py" }, { name = "tiled", extras = ["client"], specifier = ">=0.2.4" }, { name = "tomlkit" }, - { name = "uvicorn" }, + { name = "uvicorn", specifier = "==0.49" }, ] [package.metadata.requires-dev] @@ -6097,19 +6097,20 @@ wheels = [ [[package]] name = "uvicorn" -version = "0.51.0" +version = "0.49.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "click" }, { name = "h11" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/a2/65/b7c6c443ccc58678c91e1e973bbe2a878591538655d6e1d47f24ba1c51f3/uvicorn-0.51.0.tar.gz", hash = "sha256:f6f4b69b657c312f516dd2d268ab9ae6f254b11e4bac504f37b2ab58b24dd0b0", size = 94412, upload-time = "2026-07-08T10:59:05.962Z" } +sdist = { url = "https://files.pythonhosted.org/packages/c4/1f/fa18009dea8469069cca78a4e877a008ab78f08b064bfc9ab891579077ff/uvicorn-0.49.0.tar.gz", hash = "sha256:ebf4271aa580d9de97f93192d4595176df6e91f9aae919ca73e4fc07df1e66a3", size = 91284, upload-time = "2026-06-03T22:01:30.448Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/45/ec/dbb7e5a6b91f86bfb9eb7d2988a2730907b6a729875b949c7f022e8b88fa/uvicorn-0.51.0-py3-none-any.whl", hash = "sha256:5d38af6cd620f2ae3849fb44fd4879e0890aa1febe8d47eb355fb45d93fe6a5b", size = 73219, upload-time = "2026-07-08T10:59:04.44Z" }, + { url = "https://files.pythonhosted.org/packages/88/fa/e1388bbcf24ef3274f45c0c1c7b501fd14971037c1b6ee23610553307497/uvicorn-0.49.0-py3-none-any.whl", hash = "sha256:ba3d14c3ee7e41c6c654c46c9eb489d33213cdd30aa1696eab1374337c13f68f", size = 71376, upload-time = "2026-06-03T22:01:29.037Z" }, ] [package.optional-dependencies] standard = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, { name = "httptools" }, { name = "python-dotenv" }, { name = "pyyaml" },