diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index de71b42..0b890c4 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -17,6 +17,87 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Changed + +- **Async control-event HTTP client** โ€” `account_service.send_event`, `Account` event helpers, and case event-reporting methods now use `httpx.AsyncClient` and must be awaited in async agents. Explicit `_sync` shims remain for CLI, startup, and synchronous controller methods. See [RFC: async-control-event-http-client](/docs/rfc/async-control-event-http-client.md) for details. + +### Tests + +`just test` + +| Status | Count | +| ----------- | ----- | +| โœ… Passed | 559 | +| ๐Ÿค” Skipped | 0 | +| ๐Ÿ”ด Failed | 0 | +| โฑ๏ธ in | | +| โฑ๏ธ with cov | 102s | + +### Coverage + +| Name | Stmts | Miss | Cover | +| ------------------------------------------------- | ----- | ---- | ----- | +| src/supervaizer/**init**.py | 23 | 3 | 87% | +| src/supervaizer/**version**.py | 3 | 0 | 100% | +| src/supervaizer/access/**init**.py | 4 | 0 | 100% | +| src/supervaizer/access/api_auth.py | 38 | 1 | 97% | +| src/supervaizer/access/client_ip.py | 37 | 7 | 81% | +| src/supervaizer/access/tailscale.py | 23 | 2 | 91% | +| src/supervaizer/account.py | 122 | 15 | 88% | +| src/supervaizer/account_service.py | 80 | 2 | 98% | +| src/supervaizer/admin/routes.py | 549 | 190 | 65% | +| src/supervaizer/admin/workbench_routes.py | 380 | 256 | 33% | +| src/supervaizer/agent.py | 354 | 80 | 77% | +| src/supervaizer/case.py | 267 | 64 | 76% | +| src/supervaizer/cli.py | 210 | 75 | 64% | +| src/supervaizer/common.py | 142 | 4 | 97% | +| src/supervaizer/contracts.py | 168 | 3 | 98% | +| src/supervaizer/data_resource.py | 70 | 0 | 100% | +| src/supervaizer/data_routes.py | 98 | 21 | 79% | +| src/supervaizer/deploy/**init**.py | 2 | 0 | 100% | +| src/supervaizer/deploy/cli.py | 104 | 46 | 56% | +| src/supervaizer/deploy/commands/**init**.py | 2 | 0 | 100% | +| src/supervaizer/deploy/commands/clean.py | 158 | 17 | 89% | +| src/supervaizer/deploy/commands/down.py | 61 | 12 | 80% | +| src/supervaizer/deploy/commands/local.py | 214 | 15 | 93% | +| src/supervaizer/deploy/commands/plan.py | 75 | 7 | 91% | +| src/supervaizer/deploy/commands/status.py | 100 | 41 | 59% | +| src/supervaizer/deploy/commands/up.py | 113 | 8 | 93% | +| src/supervaizer/deploy/docker.py | 186 | 23 | 88% | +| src/supervaizer/deploy/driver_factory.py | 20 | 0 | 100% | +| src/supervaizer/deploy/drivers/**init**.py | 13 | 4 | 69% | +| src/supervaizer/deploy/drivers/aws_app_runner.py | 221 | 157 | 29% | +| src/supervaizer/deploy/drivers/base.py | 82 | 8 | 90% | +| src/supervaizer/deploy/drivers/cloud_run.py | 205 | 125 | 39% | +| src/supervaizer/deploy/drivers/do_app_platform.py | 164 | 125 | 24% | +| src/supervaizer/deploy/health.py | 161 | 12 | 93% | +| src/supervaizer/deploy/state.py | 115 | 9 | 92% | +| src/supervaizer/deploy/utils.py | 24 | 0 | 100% | +| src/supervaizer/event.py | 40 | 1 | 98% | +| src/supervaizer/examples/local_server.py | 17 | 3 | 82% | +| src/supervaizer/instructions.py | 56 | 1 | 98% | +| src/supervaizer/job.py | 166 | 28 | 83% | +| src/supervaizer/job_service.py | 34 | 0 | 100% | +| src/supervaizer/lifecycle.py | 153 | 3 | 98% | +| src/supervaizer/parameter.py | 79 | 7 | 91% | +| src/supervaizer/protocol/**init**.py | 2 | 0 | 100% | +| src/supervaizer/protocol/a2a/**init**.py | 3 | 0 | 100% | +| src/supervaizer/protocol/a2a/model.py | 37 | 5 | 86% | +| src/supervaizer/protocol/a2a/routes.py | 35 | 3 | 91% | +| src/supervaizer/routers/**init**.py | 4 | 0 | 100% | +| src/supervaizer/routers/api.py | 18 | 1 | 94% | +| src/supervaizer/routers/private.py | 10 | 0 | 100% | +| src/supervaizer/routers/public.py | 22 | 5 | 77% | +| src/supervaizer/routes.py | 333 | 60 | 82% | +| src/supervaizer/server.py | 282 | 84 | 70% | +| src/supervaizer/server_utils.py | 25 | 0 | 100% | +| src/supervaizer/storage.py | 160 | 12 | 92% | +| src/supervaizer/telemetry.py | 40 | 0 | 100% | +| src/supervaizer/utils/**init**.py | 2 | 0 | 100% | +| src/supervaizer/utils/version_check.py | 27 | 6 | 78% | +| ------------------------------------------------- | ----- | ---- | ----- | +| TOTAL | 6133 | 1551 | 75% | + ## [0.16.0] - 2026-04-25 ### Added diff --git a/docs/rfc/002-async-http-client.md b/docs/rfc/002-async-http-client.md new file mode 100644 index 0000000..9f0f8cd --- /dev/null +++ b/docs/rfc/002-async-http-client.md @@ -0,0 +1,137 @@ +# RFC-002: Async HTTP Client for `account_service.send_event` + +**Status:** Proposed +**Date:** 2026-04-27 +**Affects:** `supervaizer` SDK (published PyPI package) + +--- + +## Problem + +Every call from an agent to the Supervaize Control API โ€” reporting a step, starting or closing a case โ€” goes through `account_service.send_event()`, which issues a synchronous `httpx.Client.post()`. This is a blocking network call. + +For agents built on Python's asyncio (FastAPI, Pipecat, etc.), any synchronous blocking call made from within the event loop freezes the entire server for the duration of that call. All concurrent tasks are suspended until it returns. + +The SDK's `async def` hooks give the appearance of being async-safe, but they contain no real `await` points. Any agent that calls them from an asyncio context โ€” directly or as a background task โ€” blocks the event loop. + +### Call chain (current) + +``` +Account.send_update_case() # sync + account_service.send_event() # sync + httpx.Client.post() # โ† BLOCKS the event loop +``` + +`Case.update()`, `Account.send_start_case()`, and `Account.send_register_agent()` all follow the same path. + +### Consequences for async agents + +An agent that calls `report_step()` or `close_case()` from an asyncio background task โ€” even fire-and-forget โ€” blocks the event loop for the full duration of the HTTP round-trip. Observed latencies range from 2s (fast network) to 34s (slow or retried requests). During that window the agent's audio pipeline, WebSocket transport, and all other async work is frozen. + +The only workaround available to consumers is to wrap every SDK call in `asyncio.to_thread(...)`, which is error-prone, requires duplicating logic, and must be applied at every call site across the agent's lifecycle. + +--- + +## Proposed Change + +Make `account_service.send_event()` async and replace `httpx.Client` with `httpx.AsyncClient`. Cascade the `async`/`await` change up through all SDK methods that call it. + +This is a **minor breaking change** for callers that invoke SDK methods from synchronous code (CLI tools, scripts). A sync shim covers those cases โ€” see Backward Compatibility. + +### `account_service.py` + +**Before:** +```python +_httpx_transport = httpx.HTTPTransport(retries=int(os.getenv("SUPERVAIZE_HTTP_MAX_RETRIES", 2))) +_httpx_client = httpx.Client(transport=_httpx_transport) + +def send_event(account, sender, event) -> ApiResult: + ... + response = _httpx_client.post(url_event, headers=headers, json=payload) + response.raise_for_status() + ... +``` + +**After:** +```python +_httpx_transport = httpx.AsyncHTTPTransport(retries=int(os.getenv("SUPERVAIZE_HTTP_MAX_RETRIES", 2))) +_httpx_client = httpx.AsyncClient(transport=_httpx_transport) + +async def send_event(account, sender, event) -> ApiResult: + ... + response = await _httpx_client.post(url_event, headers=headers, json=payload) + response.raise_for_status() + ... +``` + +`httpx.AsyncClient` must remain a module-level singleton to reuse the connection pool. It must be closed gracefully on server shutdown (via a lifespan handler or `atexit`). + +### `account.py` โ€” cascade + +```python +async def send_update_case(self, case, update) -> ApiResult: + from supervaizer.event import CaseUpdateEvent + event = CaseUpdateEvent(case=case, update=update, account=self) + return await account_service.send_event(update, event) + +async def send_start_case(self, case) -> ApiResult: + from supervaizer.event import CaseStartEvent + event = CaseStartEvent(case=case, account=self) + return await account_service.send_event(case, event) + +async def send_register_agent(self, agent, polling) -> ApiResult: + from supervaizer.event import AgentRegisterEvent + event = AgentRegisterEvent(agent=agent, account=self, polling=polling) + return await account_service.send_event(agent, event) +``` + +### `case.py` + +`Case.update()` currently calls `self.account.send_update_case(self, ...)` synchronously: + +```python +async def update(self, update: CaseNodeUpdate) -> ApiResult: + return await self.account.send_update_case(self, update) +``` + +--- + +## Backward Compatibility + +Callers that invoke SDK methods from synchronous code (CLI registration, telemetry scripts) will break if they call the async methods directly. Provide a sync shim in `account_service.py`: + +```python +def send_event_sync(account, sender, event) -> ApiResult: + """Sync entry point for environments without a running event loop (CLI, scripts).""" + import asyncio + try: + asyncio.get_running_loop() + # Already inside a loop โ€” run in a thread to avoid deadlock + import concurrent.futures + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: + return pool.submit(asyncio.run, send_event(account, sender, event)).result() + except RuntimeError: + return asyncio.run(send_event(account, sender, event)) +``` + +All existing sync callers (CLI, `send_telemetry`) switch to `send_event_sync`. Async agent callers use `await send_event(...)`. + +--- + +## What Does NOT Change + +- The public API contract to the Supervaize Control API (URL, headers, payload shape) is unchanged. +- The outbox retry path runs from a scheduled background job, not from the agent hot path, and can remain sync. +- Error handling, retry configuration (`SUPERVAIZE_HTTP_MAX_RETRIES`), and local-mode short-circuit logic are unchanged. + +--- + +## Acceptance Criteria + +1. `account_service.send_event` is `async def` and uses `httpx.AsyncClient`. +2. `Account.send_update_case`, `send_start_case`, `send_register_agent` are `async def`. +3. `Case.update()` is `async def`. +4. `account_service.send_event_sync` provides a safe sync entry point for CLI and script callers. +5. All existing SDK tests pass without modification. +6. New tests verify that `send_event` can be `await`-ed from an async test without blocking. +7. The SDK example agents (FastAPI, Pipecat) register and report steps correctly after the change. diff --git a/pyproject.toml b/pyproject.toml index 3b64858..b3add80 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -126,7 +126,7 @@ mypy_path = "src" disallow_any_expr = false [tool.bumpversion] -current_version = "0.16.0" +current_version = "0.16.0.dev0" commit = true tag = true tag_name = "v{new_version}" diff --git a/src/supervaizer/__version__.py b/src/supervaizer/__version__.py index a16a7ad..19a8f29 100644 --- a/src/supervaizer/__version__.py +++ b/src/supervaizer/__version__.py @@ -11,6 +11,6 @@ # https://mozilla.org/MPL/2.0/. -VERSION = "0.16.0" +VERSION = "0.16.0.dev0" API_VERSION = "v1" TELEMETRY_VERSION = "v1" diff --git a/src/supervaizer/account.py b/src/supervaizer/account.py index a849e18..0d49078 100644 --- a/src/supervaizer/account.py +++ b/src/supervaizer/account.py @@ -170,7 +170,7 @@ def get_url(self, pattern_name: str, **kwargs: Any) -> str: return pattern.format(**url_params) - def send_event( + async def send_event( self, sender: Union["Agent", "Job", "Server", "Case", "CaseNodeUpdate"], event: "Event", @@ -189,9 +189,19 @@ def send_event( # Import here to avoid circular imports from supervaizer.account_service import send_event as service_send_event + return await service_send_event(self, sender, event) + + def send_event_sync( + self, + sender: Union["Agent", "Job", "Server", "Case", "CaseNodeUpdate"], + event: "Event", + ) -> ApiResult: + """Send an event from sync-only contexts.""" + from supervaizer.account_service import send_event_sync as service_send_event + return service_send_event(self, sender, event) - def register_server(self, server: "Server") -> ApiResult: + async def register_server(self, server: "Server") -> ApiResult: """Register a server with the Supervaize Control API. Args: @@ -208,7 +218,18 @@ def register_server(self, server: "Server") -> ApiResult: from supervaizer.event import ServerRegisterEvent event = ServerRegisterEvent(server=server, account=self) - result = self.send_event(sender=server, event=event) + result = await self.send_event(sender=server, event=event) + return self._log_registration_result(result) + + def register_server_sync(self, server: "Server") -> ApiResult: + """Register a server from sync-only startup code.""" + from supervaizer.event import ServerRegisterEvent + + event = ServerRegisterEvent(server=server, account=self) + result = self.send_event_sync(sender=server, event=event) + return self._log_registration_result(result) + + def _log_registration_result(self, result: ApiResult) -> ApiResult: if isinstance(result, ApiSuccess): log.success(result.message) # TODO: Update server with the server ID from the response. store this ID in env variable. @@ -266,7 +287,7 @@ def get_agent_by( exception=e, ) - def register_agent(self, agent: "Agent", polling: bool = True) -> ApiResult: + async def register_agent(self, agent: "Agent", polling: bool = True) -> ApiResult: """Send a registration event to the Supervaize Control API. This will be used for polling, when the agent is registered without a server. Args: @@ -281,16 +302,31 @@ def register_agent(self, agent: "Agent", polling: bool = True) -> ApiResult: from supervaizer.event import AgentRegisterEvent event = AgentRegisterEvent(agent=agent, account=self, polling=polling) - return self.send_event(agent, event) + return await self.send_event(agent, event) + + def register_agent_sync(self, agent: "Agent", polling: bool = True) -> ApiResult: + """Send a registration event from sync-only contexts.""" + from supervaizer.event import AgentRegisterEvent + + event = AgentRegisterEvent(agent=agent, account=self, polling=polling) + return self.send_event_sync(agent, event) - def send_start_case(self, case: "Case") -> ApiResult: + async def send_start_case(self, case: "Case") -> ApiResult: # Import here to avoid circular imports from supervaizer.event import CaseStartEvent event = CaseStartEvent(case=case, account=self) - return self.send_event(case, event) + return await self.send_event(case, event) + + def send_start_case_sync(self, case: "Case") -> ApiResult: + from supervaizer.event import CaseStartEvent - def send_update_case(self, case: "Case", update: "CaseNodeUpdate") -> ApiResult: + event = CaseStartEvent(case=case, account=self) + return self.send_event_sync(case, event) + + async def send_update_case( + self, case: "Case", update: "CaseNodeUpdate" + ) -> ApiResult: # Import here to avoid circular imports log.debug(f"[send_update_case] CaseRef {case} with update {update}") log.debug(f"[send_update_case] {type(case)}") @@ -298,7 +334,18 @@ def send_update_case(self, case: "Case", update: "CaseNodeUpdate") -> ApiResult: from supervaizer.event import CaseUpdateEvent event = CaseUpdateEvent(case=case, update=update, account=self) - return self.send_event(update, event) + return await self.send_event(update, event) + + def send_update_case_sync( + self, case: "Case", update: "CaseNodeUpdate" + ) -> ApiResult: + log.debug(f"[send_update_case] CaseRef {case} with update {update}") + log.debug(f"[send_update_case] {type(case)}") + log.debug(f"[send_update_case] {type(update)}") + from supervaizer.event import CaseUpdateEvent + + event = CaseUpdateEvent(case=case, update=update, account=self) + return self.send_event_sync(update, event) def send_telemetry(self, telemetry: Telemetry) -> ApiResult: """Send telemetry data to the Supervaize Control API. diff --git a/src/supervaizer/account_service.py b/src/supervaizer/account_service.py index b7e0ade..9e4b287 100644 --- a/src/supervaizer/account_service.py +++ b/src/supervaizer/account_service.py @@ -14,9 +14,11 @@ # If a copy of the MPL was not distributed with this file, you can obtain one at # https://mozilla.org/MPL/2.0/. +import asyncio +import atexit import logging import os -from typing import TYPE_CHECKING, Union +from typing import TYPE_CHECKING, Any, NoReturn, Union import httpx @@ -33,10 +35,14 @@ # Enable httpx debug logging (optional - uncomment for transport-level debugging) logger.setLevel(logging.DEBUG) -_httpx_transport = httpx.HTTPTransport( +_httpx_transport = httpx.AsyncHTTPTransport( retries=int(os.getenv("SUPERVAIZE_HTTP_MAX_RETRIES", 2)) ) -_httpx_client = httpx.Client(transport=_httpx_transport) +_httpx_client = httpx.AsyncClient(transport=_httpx_transport) +_sync_httpx_transport = httpx.HTTPTransport( + retries=int(os.getenv("SUPERVAIZE_HTTP_MAX_RETRIES", 2)) +) +_sync_httpx_client = httpx.Client(transport=_sync_httpx_transport) if TYPE_CHECKING: from supervaizer.account import Account @@ -47,7 +53,60 @@ from supervaizer.server import Server -def send_event( +def _event_request( + account: "Account", + sender: Union["Agent", "Server", "Job", "Case", "CaseNodeUpdate"], + event: "Event", +) -> tuple[str, dict[str, str], Any]: + headers = account.api_headers + payload = SvBaseModel.serialize_value(event.payload) + url_event = account.url_event.strip() + return url_event, headers, payload + + +def _event_curl(url_event: str, headers: dict[str, str]) -> str: + curl_headers = " ".join([f'-H "{key}: {value}"' for key, value in headers.items()]) + return f"curl -X 'POST' '{url_event}' {curl_headers}" + + +def _event_success(event: "Event", response: httpx.Response) -> ApiSuccess: + result = ApiSuccess( + message=f"POST Event {event.type.name} sent", detail=response.text + ) + log.success(result.log_message) + return result + + +def _handle_event_http_error( + event: "Event", + url_event: str, + curl_cmd: str, + error: httpx.HTTPError, +) -> NoReturn: + log.error("[Send event] HTTP Error occurred") + log.warning(f"โš ๏ธ Try to connect via curl:\n{curl_cmd}") + + error_result = ApiError( + message=f"Error sending event {event.type.name}", + url=url_event, + payload=event.payload, + exception=error, + ) + log.error(f"[Send event] Error details: {error_result.dict}") + log.error(error_result.log_message) + raise error + + +def _local_mode_result(event: "Event") -> ApiSuccess | None: + if not is_local_mode(): + return None + log.debug(f"[Send event] Local mode โ€” skipping {event.type.name}") + return ApiSuccess( + message=f"Event {event.type.name} skipped (local mode)", detail=None + ) + + +async def send_event( account: "Account", sender: Union["Agent", "Server", "Job", "Case", "CaseNodeUpdate"], event: "Event", @@ -70,35 +129,45 @@ def send_event( Tested in tests/test_account_service.py """ - # In local mode, skip sending events to the SaaS API entirely. - if is_local_mode(): - log.debug(f"[Send event] Local mode โ€” skipping {event.type.name}") - return ApiSuccess( - message=f"Event {event.type.name} skipped (local mode)", detail=None - ) + local_result = _local_mode_result(event) + if local_result: + return local_result - headers = account.api_headers - payload = SvBaseModel.serialize_value(event.payload) - url_event = ( - account.url_event.strip() - ) # defensive: env vars often have trailing newline + url_event, headers, payload = _event_request(account, sender, event) + curl_cmd = _event_curl(url_event, headers) - # Generate curl equivalent for debugging + try: + response = await _httpx_client.post(url_event, headers=headers, json=payload) + response.raise_for_status() + return _event_success(event, response) + except (httpx.ConnectError, httpx.ConnectTimeout) as e: + log.error( + f"Supervaize controller is not available at {url_event}. " + "Connection refused or timed out. Is the controller server running?" + ) + log.error(f"โŒ Error sending event {event.type.name}: {e!s}") + raise e + except httpx.HTTPError as e: + _handle_event_http_error(event, url_event, curl_cmd, e) - curl_headers = " ".join([f'-H "{k}: {v}"' for k, v in headers.items()]) - curl_cmd = f"curl -X 'GET' '{url_event}' {curl_headers}" - try: - response = _httpx_client.post(url_event, headers=headers, json=payload) +def send_event_sync( + account: "Account", + sender: Union["Agent", "Server", "Job", "Case", "CaseNodeUpdate"], + event: "Event", +) -> ApiResult: + """Sync entry point for CLI, startup, and other non-async callers.""" + local_result = _local_mode_result(event) + if local_result: + return local_result - # Log response details before raising for status + url_event, headers, payload = _event_request(account, sender, event) + curl_cmd = _event_curl(url_event, headers) + try: + response = _sync_httpx_client.post(url_event, headers=headers, json=payload) response.raise_for_status() - result = ApiSuccess( - message=f"POST Event {event.type.name} sent", detail=response.text - ) - - log.success(result.log_message) + return _event_success(event, response) except (httpx.ConnectError, httpx.ConnectTimeout) as e: log.error( f"Supervaize controller is not available at {url_event}. " @@ -107,17 +176,23 @@ def send_event( log.error(f"โŒ Error sending event {event.type.name}: {e!s}") raise e except httpx.HTTPError as e: - # Enhanced error logging - log.error("[Send event] HTTP Error occurred") - log.warning(f"โš ๏ธ Try to connect via curl:\n{curl_cmd}") - - error_result = ApiError( - message=f"Error sending event {event.type.name}", - url=url_event, - payload=event.payload, - exception=e, - ) - log.error(f"[Send event] Error details: {error_result.dict}") - log.error(error_result.log_message) - raise e - return result + _handle_event_http_error(event, url_event, curl_cmd, e) + + +async def close_httpx_client() -> None: + """Close the shared async event client.""" + if not _httpx_client.is_closed: + await _httpx_client.aclose() + + +def close_httpx_client_sync() -> None: + """Close shared HTTP clients from sync shutdown hooks.""" + _sync_httpx_client.close() + if not _httpx_client.is_closed: + try: + asyncio.run(_httpx_client.aclose()) + except RuntimeError: + log.warning("[Send event] Could not close async HTTP client at exit") + + +atexit.register(close_httpx_client_sync) diff --git a/src/supervaizer/admin/routes.py b/src/supervaizer/admin/routes.py index f8b76b4..4c31694 100644 --- a/src/supervaizer/admin/routes.py +++ b/src/supervaizer/admin/routes.py @@ -456,7 +456,7 @@ async def register_server_with_supervisor(request: Request) -> JSONResponse: status_code=503, detail="No supervisor account configured", ) - result = server.supervisor_account.register_server(server=server) + result = await server.supervisor_account.register_server(server=server) if isinstance(result, ApiSuccess): return JSONResponse( status_code=200, diff --git a/src/supervaizer/admin/workbench_routes.py b/src/supervaizer/admin/workbench_routes.py index cd693e8..968186b 100644 --- a/src/supervaizer/admin/workbench_routes.py +++ b/src/supervaizer/admin/workbench_routes.py @@ -515,7 +515,7 @@ async def workbench_answer_hitl( if isinstance(answer_with_label, dict): answer_with_label["_hitl_label"] = hitl_label update = CaseNodeUpdate(name="HITL Answer", payload=answer_with_label) - case.receive_human_input(update) + await case.receive_human_input(update) # Step 2: Invoke agent's human_answer method if defined if agent.methods: diff --git a/src/supervaizer/agent.py b/src/supervaizer/agent.py index f0226c7..ba1b8d2 100644 --- a/src/supervaizer/agent.py +++ b/src/supervaizer/agent.py @@ -936,7 +936,7 @@ def job_start( job=job, account=server.supervisor_account, ) - server.supervisor_account.send_event(sender=job, event=event) + server.supervisor_account.send_event_sync(sender=job, event=event) else: log.warning( f"[Agent job_start] No supervisor account defined for server, skipping event send for job {job.id}" diff --git a/src/supervaizer/case.py b/src/supervaizer/case.py index d26e781..25b67d7 100644 --- a/src/supervaizer/case.py +++ b/src/supervaizer/case.py @@ -17,7 +17,7 @@ import shortuuid from pydantic import ConfigDict, Field -from supervaizer.common import SvBaseModel, log, singleton +from supervaizer.common import ApiResult, SvBaseModel, log, singleton from supervaizer.lifecycle import EntityEvents, EntityStatus from supervaizer.storage import PersistentEntityLifecycle, StorageManager @@ -277,23 +277,34 @@ def case_ref(self) -> str: def calculated_cost(self) -> float: return sum(update.cost or 0.0 for update in self.updates) - def update(self, updateCaseNode: CaseNodeUpdate, **kwargs: Any) -> None: - updateCaseNode.index = len(self.updates) + 1 - if updateCaseNode.error: + def _prepare_update(self, update: CaseNodeUpdate) -> None: + update.index = len(self.updates) + 1 + if update.error: success, error = PersistentEntityLifecycle.handle_event( self, EntityEvents.ERROR_ENCOUNTERED ) log.warning( - f"[Case update] CaseRef {self.case_ref} has error {updateCaseNode.error}" + f"[Case update] CaseRef {self.case_ref} has error {update.error}" ) assert self.status == EntityStatus.FAILED # Just to be sure - self.account.send_update_case(self, updateCaseNode) - self.updates.append(updateCaseNode) + def _persist(self) -> None: storage = StorageManager() storage.save_object("Case", self.to_dict) - def patch_step(self, index: int, updateCaseNode: CaseNodeUpdate) -> None: + async def update(self, updateCaseNode: CaseNodeUpdate, **kwargs: Any) -> None: + self._prepare_update(updateCaseNode) + await self.account.send_update_case(self, updateCaseNode) + self.updates.append(updateCaseNode) + self._persist() + + def update_sync(self, updateCaseNode: CaseNodeUpdate, **kwargs: Any) -> None: + self._prepare_update(updateCaseNode) + self.account.send_update_case_sync(self, updateCaseNode) + self.updates.append(updateCaseNode) + self._persist() + + async def patch_step(self, index: int, updateCaseNode: CaseNodeUpdate) -> None: """Update an existing step at the given index instead of appending a new one. Sets upsert=True so Studio performs an update_or_create on the step at that index. @@ -302,52 +313,76 @@ def patch_step(self, index: int, updateCaseNode: CaseNodeUpdate) -> None: """ updateCaseNode.index = index updateCaseNode.upsert = True - self.account.send_update_case(self, updateCaseNode) + await self.account.send_update_case(self, updateCaseNode) # Update the matching entry in the in-memory registry for i, existing in enumerate(self.updates): if existing.index == index: self.updates[i] = updateCaseNode break - storage = StorageManager() - storage.save_object("Case", self.to_dict) + self._persist() - def request_human_input( + def patch_step_sync(self, index: int, updateCaseNode: CaseNodeUpdate) -> None: + """Sync entry point for updating an existing step.""" + updateCaseNode.index = index + updateCaseNode.upsert = True + self.account.send_update_case_sync(self, updateCaseNode) + for i, existing in enumerate(self.updates): + if existing.index == index: + self.updates[i] = updateCaseNode + break + self._persist() + + async def request_human_input( self, updateCaseNode: CaseNodeUpdate, message: str, **kwargs: Any ) -> None: updateCaseNode.index = len(self.updates) + 1 log.info( f"[Update case human_input] CaseRef {self.case_ref} with update {updateCaseNode}" ) - self.account.send_update_case(self, updateCaseNode) + await self.account.send_update_case(self, updateCaseNode) from supervaizer.storage import PersistentEntityLifecycle PersistentEntityLifecycle.handle_event(self, EntityEvents.AWAITING_ON_INPUT) self.updates.append(updateCaseNode) + self._persist() - # Persist updated case to storage (for the updates list change) + def request_human_input_sync( + self, updateCaseNode: CaseNodeUpdate, message: str, **kwargs: Any + ) -> None: + updateCaseNode.index = len(self.updates) + 1 + log.info( + f"[Update case human_input] CaseRef {self.case_ref} with update {updateCaseNode}" + ) + self.account.send_update_case_sync(self, updateCaseNode) + from supervaizer.storage import PersistentEntityLifecycle - storage = StorageManager() - storage.save_object("Case", self.to_dict) + PersistentEntityLifecycle.handle_event(self, EntityEvents.AWAITING_ON_INPUT) + self.updates.append(updateCaseNode) + self._persist() - def receive_human_input( + async def receive_human_input( self, updateCaseNode: CaseNodeUpdate, **kwargs: Any ) -> None: # Add the update to the case (this handles index, send_update_case, and persistence) - self.update(updateCaseNode) + await self.update(updateCaseNode) # Transition from AWAITING to IN_PROGRESS from supervaizer.storage import PersistentEntityLifecycle PersistentEntityLifecycle.handle_event(self, EntityEvents.INPUT_RECEIVED) - def close( + def receive_human_input_sync( + self, updateCaseNode: CaseNodeUpdate, **kwargs: Any + ) -> None: + self.update_sync(updateCaseNode) + from supervaizer.storage import PersistentEntityLifecycle + + PersistentEntityLifecycle.handle_event(self, EntityEvents.INPUT_RECEIVED) + + def _prepare_close( self, case_result: Dict[str, Any], final_cost: Optional[float] = None, - **kwargs: Any, - ) -> None: - """ - Close the case and send the final update to the account. - """ + ) -> CaseNodeUpdate: if final_cost: self.total_cost = final_cost else: @@ -368,13 +403,30 @@ def close( self.final_delivery = case_result self.finished_at = datetime.now() - self.account.send_update_case(self, update) + return update - # Persist updated case to storage - from supervaizer.storage import StorageManager + async def close( + self, + case_result: Dict[str, Any], + final_cost: Optional[float] = None, + **kwargs: Any, + ) -> None: + """ + Close the case and send the final update to the account. + """ + update = self._prepare_close(case_result, final_cost) + await self.account.send_update_case(self, update) + self._persist() - storage = StorageManager() - storage.save_object("Case", self.to_dict) + def close_sync( + self, + case_result: Dict[str, Any], + final_cost: Optional[float] = None, + **kwargs: Any, + ) -> None: + update = self._prepare_close(case_result, final_cost) + self.account.send_update_case_sync(self, update) + self._persist() def cancel_scheduled_steps(self) -> None: """Cancel all pending scheduled steps for this case.""" @@ -402,7 +454,7 @@ def registration_info(self) -> Dict[str, Any]: } @classmethod - def start( + def _create_started_case( cls, job_id: str, name: str, @@ -411,20 +463,6 @@ def start( case_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> "Case": - """ - Start a new case - - Args: - case_id (str): The id of the case - should be unique for the job. If not provided, a shortuuid will be generated. - job_id (str): The id of the job - name (str): The name of the case - account (Account): The account - description (str): The description of the case - - Returns: - Case: The case - """ - case = cls( id=case_id or shortuuid.uuid(), job_id=job_id, @@ -446,9 +484,10 @@ def start( # Transition from STOPPED to IN_PROGRESS PersistentEntityLifecycle.handle_event(case, EntityEvents.START_WORK) + return case - # Send case start event to Supervaize SaaS. - result = account.send_start_case(case=case) + @staticmethod + def _log_start_result(case: "Case", result: ApiResult | None) -> None: if result: log.debug( f"[Case start] Case {case.id} send to Supervaize with result {result}" @@ -458,6 +497,54 @@ def start( f"[Case start] ยงSCCS01 Case {case.id} failed to send to Supervaize" ) + @classmethod + async def start( + cls, + job_id: str, + name: str, + account: "Account", + description: str, + case_id: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> "Case": + """ + Start a new case. + """ + case = cls._create_started_case( + job_id=job_id, + name=name, + account=account, + description=description, + case_id=case_id, + metadata=metadata, + ) + # Send case start event to Supervaize SaaS. + result = await account.send_start_case(case=case) + cls._log_start_result(case, result) + + return case + + @classmethod + def start_sync( + cls, + job_id: str, + name: str, + account: "Account", + description: str, + case_id: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> "Case": + """Start a new case from sync-only contexts.""" + case = cls._create_started_case( + job_id=job_id, + name=name, + account=account, + description=description, + case_id=case_id, + metadata=metadata, + ) + result = account.send_start_case_sync(case=case) + cls._log_start_result(case, result) return case diff --git a/src/supervaizer/examples/hello_world_agent.py b/src/supervaizer/examples/hello_world_agent.py index 463ca60..b506461 100644 --- a/src/supervaizer/examples/hello_world_agent.py +++ b/src/supervaizer/examples/hello_world_agent.py @@ -36,7 +36,10 @@ class _LocalAccount(Account): """No-op account for local mode โ€” skips all Studio HTTP calls.""" - def send_event(self, sender: Any, event: Any) -> ApiSuccess: + async def send_event(self, sender: Any, event: Any) -> ApiSuccess: + return ApiSuccess(message="local-noop", detail=None) + + def send_event_sync(self, sender: Any, event: Any) -> ApiSuccess: return ApiSuccess(message="local-noop", detail=None) @@ -150,7 +153,9 @@ def _is_job_stopped() -> bool: "message": f"Waiting for human review on case {case_idx}", }, ) - case.request_human_input(hitl_update, message=f"Review case {case_idx}") + case.request_human_input_sync( + hitl_update, message=f"Review case {case_idx}" + ) # Poll until human answers or job is stopped while case.status == EntityStatus.AWAITING: @@ -169,7 +174,7 @@ def _is_job_stopped() -> bool: time.sleep(duration) msg = f"The step {step_idx} ({step_name}) in case {case_idx} lasted {duration} s" log.info(msg) - case.update( + case.update_sync( CaseNodeUpdate( name=step_name, payload={"message": msg, "duration": duration}, @@ -181,7 +186,7 @@ def _is_job_stopped() -> bool: stopped = True break - case.close({"result": f"Case {case_idx} completed"}) + case.close_sync({"result": f"Case {case_idx} completed"}) cases_done += 1 if stopped: diff --git a/src/supervaizer/job_service.py b/src/supervaizer/job_service.py index b6bb8c4..4a3b7a5 100644 --- a/src/supervaizer/job_service.py +++ b/src/supervaizer/job_service.py @@ -101,7 +101,7 @@ def service_job_finished(job: Job, server: "Server") -> None: job=job, account=account, ) - account.send_event(sender=job, event=event) + account.send_event_sync(sender=job, event=event) async def service_job_custom( diff --git a/src/supervaizer/routes.py b/src/supervaizer/routes.py index cd5404f..486709b 100644 --- a/src/supervaizer/routes.py +++ b/src/supervaizer/routes.py @@ -274,13 +274,13 @@ async def update_case_with_answer( ) casestep_index = request.answer.get("casestep_index") if casestep_index is not None: - case.patch_step(int(casestep_index), update) + await case.patch_step(int(casestep_index), update) from supervaizer.lifecycle import EntityEvents from supervaizer.storage import PersistentEntityLifecycle PersistentEntityLifecycle.handle_event(case, EntityEvents.INPUT_RECEIVED) else: - case.receive_human_input(update) + await case.receive_human_input(update) case_status = case.status.value owning_agent = server.get_agent_by_name(job.agent_name) diff --git a/src/supervaizer/server.py b/src/supervaizer/server.py index c0652a5..2e5f98b 100644 --- a/src/supervaizer/server.py +++ b/src/supervaizer/server.py @@ -700,7 +700,7 @@ def log_queue_handler(message: Any) -> None: if self.supervisor_account: # Register the server with the supervisor account server_registration_result: ApiResult = ( - self.supervisor_account.register_server(server=self) + self.supervisor_account.register_server_sync(server=self) ) # log.debug(f"[Server launch] Server registration result: {server_registration_result}") # inspect(server_registration_result) diff --git a/tests/test_account.py b/tests/test_account.py index 2187f6f..5ce355a 100644 --- a/tests/test_account.py +++ b/tests/test_account.py @@ -191,16 +191,21 @@ def test_get_agent_by_no_params(account_fixture: Account) -> None: account_fixture.get_agent_by() -def test_register_agent(account_fixture: Account, mocker: MockerFixture) -> None: +@pytest.mark.asyncio +async def test_register_agent(account_fixture: Account, mocker: MockerFixture) -> None: """Test register_agent method.""" # Mock the agent mock_agent = mocker.Mock() mock_agent.name = "test-agent" # Mock the account_service.send_event function instead of patching the instance method - mock_send_event = mocker.patch("supervaizer.account_service.send_event") - mock_send_event.return_value = ApiSuccess( - message="Agent registered", detail={"status": "registered"} + mock_send_event = mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock( + return_value=ApiSuccess( + message="Agent registered", detail={"status": "registered"} + ) + ), ) # Mock the AgentRegisterEvent @@ -209,7 +214,7 @@ def test_register_agent(account_fixture: Account, mocker: MockerFixture) -> None mock_event_class.return_value = mock_event # Test with default polling=True - result = account_fixture.register_agent(mock_agent) + result = await account_fixture.register_agent(mock_agent) assert isinstance(result, ApiSuccess) mock_event_class.assert_called_once_with( @@ -221,21 +226,26 @@ def test_register_agent(account_fixture: Account, mocker: MockerFixture) -> None mock_send_event.reset_mock() mock_event_class.reset_mock() - account_fixture.register_agent(mock_agent, polling=False) + await account_fixture.register_agent(mock_agent, polling=False) mock_event_class.assert_called_once_with( agent=mock_agent, account=account_fixture, polling=False ) -def test_send_start_case(account_fixture: Account, mocker: MockerFixture) -> None: +@pytest.mark.asyncio +async def test_send_start_case(account_fixture: Account, mocker: MockerFixture) -> None: """Test send_start_case method.""" # Mock the case mock_case = mocker.Mock() # Mock the account_service.send_event function instead of patching the instance method - mock_send_event = mocker.patch("supervaizer.account_service.send_event") - mock_send_event.return_value = ApiSuccess( - message="Case started", detail={"status": "started"} + mock_send_event = mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock( + return_value=ApiSuccess( + message="Case started", detail={"status": "started"} + ) + ), ) # Mock the CaseStartEvent @@ -243,23 +253,30 @@ def test_send_start_case(account_fixture: Account, mocker: MockerFixture) -> Non mock_event = mocker.Mock() mock_event_class.return_value = mock_event - result = account_fixture.send_start_case(mock_case) + result = await account_fixture.send_start_case(mock_case) assert isinstance(result, ApiSuccess) mock_event_class.assert_called_once_with(case=mock_case, account=account_fixture) mock_send_event.assert_called_once_with(account_fixture, mock_case, mock_event) -def test_send_update_case(account_fixture: Account, mocker: MockerFixture) -> None: +@pytest.mark.asyncio +async def test_send_update_case( + account_fixture: Account, mocker: MockerFixture +) -> None: """Test send_update_case method.""" # Mock the case and update mock_case = mocker.Mock() mock_update = mocker.Mock() # Mock the account_service.send_event function instead of patching the instance method - mock_send_event = mocker.patch("supervaizer.account_service.send_event") - mock_send_event.return_value = ApiSuccess( - message="Case updated", detail={"status": "updated"} + mock_send_event = mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock( + return_value=ApiSuccess( + message="Case updated", detail={"status": "updated"} + ) + ), ) # Mock the CaseUpdateEvent @@ -267,7 +284,7 @@ def test_send_update_case(account_fixture: Account, mocker: MockerFixture) -> No mock_event = mocker.Mock() mock_event_class.return_value = mock_event - result = account_fixture.send_update_case(mock_case, mock_update) + result = await account_fixture.send_update_case(mock_case, mock_update) assert isinstance(result, ApiSuccess) mock_event_class.assert_called_once_with( @@ -343,42 +360,52 @@ def test_send_telemetry_http_error( ) -def test_account_send_event_delegation( +@pytest.mark.asyncio +async def test_account_send_event_delegation( account_fixture: Account, event_fixture: Event, server_fixture: Server, mocker: MockerFixture, ) -> None: # Mock the account_service.send_event function - mock_service_send_event = mocker.patch("supervaizer.account_service.send_event") - mock_service_send_event.return_value = ApiSuccess( + mock_success = ApiSuccess( message="Event sent", detail={"id": "01JPZ7414FX3JHPNA8N1JXDADX", "response": "success"}, ) + mock_service_send_event = mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=mock_success), + ) # Call the account.send_event method - result = account_fixture.send_event(sender=server_fixture, event=event_fixture) + result = await account_fixture.send_event( + sender=server_fixture, event=event_fixture + ) # Verify that the account_service.send_event was called with correct parameters mock_service_send_event.assert_called_once_with( account_fixture, server_fixture, event_fixture ) - assert result == mock_service_send_event.return_value + assert result == mock_success -def test_account_register_server_success( +@pytest.mark.asyncio +async def test_account_register_server_success( account_fixture: Account, server_fixture: Server, mocker: MockerFixture ) -> None: # Mock the send_event method - mock_send_event = mocker.patch("supervaizer.account_service.send_event") - # Use a dictionary instead of SERVER_REGISTER_RESPONSE to avoid JSON decoding issues detail = {"id": "01JPZ7414FX3JHPNA8N1JXDADX", "response": "success"} - mock_send_event.return_value = ApiSuccess( - message="Event SERVER_REGISTER sent", - detail=detail, + mock_send_event = mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock( + return_value=ApiSuccess( + message="Event SERVER_REGISTER sent", + detail=detail, + ) + ), ) - result = account_fixture.register_server(server_fixture) + result = await account_fixture.register_server(server_fixture) assert isinstance(result, ApiSuccess) assert result.message == "Event SERVER_REGISTER sent" # Verify that send_event was called diff --git a/tests/test_account_service.py b/tests/test_account_service.py index 49d91e9..b4f9260 100644 --- a/tests/test_account_service.py +++ b/tests/test_account_service.py @@ -16,7 +16,8 @@ from pytest_mock import MockerFixture from supervaizer import Account, ApiSuccess -from supervaizer.account_service import send_event +import supervaizer.account_service as account_service +from supervaizer.account_service import send_event, send_event_sync from supervaizer.common import SvBaseModel from supervaizer.event import Event from supervaizer.server import Server @@ -24,19 +25,23 @@ from . import AUTH_ERROR_RESPONSE, WAKEUP_EVENT_RESPONSE -def test_send_event_success( +@pytest.mark.asyncio +async def test_send_event_success( account_fixture: Account, event_fixture: Event, server_fixture: Server, mocker: MockerFixture, ) -> None: # Patch the method on the client instance used in account_service - mock_post = mocker.patch("supervaizer.account_service._httpx_client.post") + mock_post = mocker.patch( + "supervaizer.account_service._httpx_client.post", + new=mocker.AsyncMock(), + ) mock_post.return_value.status_code = 200 mock_post.return_value.text = str(WAKEUP_EVENT_RESPONSE) mock_post.return_value.raise_for_status = mocker.Mock() - result = send_event( + result = await send_event( account=account_fixture, sender=server_fixture, event=event_fixture ) @@ -50,13 +55,17 @@ def test_send_event_success( assert result.detail == {"object": WAKEUP_EVENT_RESPONSE} -def test_send_event_auth_error( +@pytest.mark.asyncio +async def test_send_event_auth_error( account_fixture: Account, event_fixture: Event, server_fixture: Server, mocker: MockerFixture, ) -> None: - mock_post = mocker.patch("supervaizer.account_service._httpx_client.post") + mock_post = mocker.patch( + "supervaizer.account_service._httpx_client.post", + new=mocker.AsyncMock(), + ) # Create a mock response that raises HTTPStatusError when raise_for_status is called mock_response = mocker.Mock() @@ -72,17 +81,186 @@ def test_send_event_auth_error( mock_post.return_value = mock_response with pytest.raises(HTTPStatusError, match="403 Client Error: Forbidden for url"): - send_event(account=account_fixture, sender=server_fixture, event=event_fixture) + await send_event( + account=account_fixture, sender=server_fixture, event=event_fixture + ) -def test_send_event_url_error( +@pytest.mark.asyncio +async def test_send_event_url_error( account_fixture: Account, event_fixture: Event, server_fixture: Server, mocker: MockerFixture, ) -> None: - mock_post = mocker.patch("supervaizer.account_service._httpx_client.post") + mock_post = mocker.patch( + "supervaizer.account_service._httpx_client.post", + new=mocker.AsyncMock(), + ) mock_post.side_effect = ConnectError("HTTPSConnectionPool(host='...") with pytest.raises(ConnectError, match="HTTPSConnectionPool"): - send_event(account=account_fixture, sender=server_fixture, event=event_fixture) + await send_event( + account=account_fixture, sender=server_fixture, event=event_fixture + ) + + +def test_send_event_sync_success( + account_fixture: Account, + event_fixture: Event, + server_fixture: Server, + mocker: MockerFixture, +) -> None: + mock_post = mocker.patch("supervaizer.account_service._sync_httpx_client.post") + mock_post.return_value.status_code = 200 + mock_post.return_value.text = str(WAKEUP_EVENT_RESPONSE) + mock_post.return_value.raise_for_status = mocker.Mock() + + result = send_event_sync( + account=account_fixture, sender=server_fixture, event=event_fixture + ) + + mock_post.assert_called_once_with( + account_fixture.url_event, + headers=account_fixture.api_headers, + json=SvBaseModel.serialize_value(event_fixture.payload), + ) + assert isinstance(result, ApiSuccess) + assert result.message == f"POST Event {event_fixture.type.name} sent" + + +def test_send_event_sync_auth_error( + account_fixture: Account, + event_fixture: Event, + server_fixture: Server, + mocker: MockerFixture, +) -> None: + mock_post = mocker.patch("supervaizer.account_service._sync_httpx_client.post") + mock_response = mocker.Mock() + mock_response.status_code = 403 + mock_response.text = str(AUTH_ERROR_RESPONSE) + + error = HTTPStatusError( + "403 Client Error: Forbidden for url", + request=mocker.Mock(), + response=mock_response, + ) + mock_response.raise_for_status.side_effect = error + mock_post.return_value = mock_response + + with pytest.raises(HTTPStatusError, match="403 Client Error: Forbidden for url"): + send_event_sync( + account=account_fixture, sender=server_fixture, event=event_fixture + ) + + +def test_send_event_sync_url_error( + account_fixture: Account, + event_fixture: Event, + server_fixture: Server, + mocker: MockerFixture, +) -> None: + mock_post = mocker.patch("supervaizer.account_service._sync_httpx_client.post") + mock_post.side_effect = ConnectError("HTTPSConnectionPool(host='...") + + with pytest.raises(ConnectError, match="HTTPSConnectionPool"): + send_event_sync( + account=account_fixture, sender=server_fixture, event=event_fixture + ) + + +def test_send_event_sync_local_mode_skips_http( + account_fixture: Account, + event_fixture: Event, + server_fixture: Server, + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("SUPERVAIZER_LOCAL_MODE", "true") + mock_post = mocker.patch("supervaizer.account_service._sync_httpx_client.post") + + result = send_event_sync( + account=account_fixture, sender=server_fixture, event=event_fixture + ) + + assert isinstance(result, ApiSuccess) + assert result.message == f"Event {event_fixture.type.name} skipped (local mode)" + mock_post.assert_not_called() + + +@pytest.mark.asyncio +async def test_send_event_local_mode_skips_http( + account_fixture: Account, + event_fixture: Event, + server_fixture: Server, + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("SUPERVAIZER_LOCAL_MODE", "true") + mock_post = mocker.patch( + "supervaizer.account_service._httpx_client.post", + new=mocker.AsyncMock(), + ) + + result = await send_event( + account=account_fixture, sender=server_fixture, event=event_fixture + ) + + assert isinstance(result, ApiSuccess) + assert result.message == f"Event {event_fixture.type.name} skipped (local mode)" + mock_post.assert_not_called() + + +@pytest.mark.asyncio +async def test_close_httpx_client_is_idempotent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class FakeAsyncClient: + is_closed = False + + def __init__(self) -> None: + self.close_calls = 0 + + async def aclose(self) -> None: + self.close_calls += 1 + self.is_closed = True + + fake_async_client = FakeAsyncClient() + monkeypatch.setattr(account_service, "_httpx_client", fake_async_client) + + await account_service.close_httpx_client() + await account_service.close_httpx_client() + + assert fake_async_client.close_calls == 1 + + +def test_close_httpx_client_sync_is_idempotent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class FakeAsyncClient: + is_closed = False + + def __init__(self) -> None: + self.close_calls = 0 + + async def aclose(self) -> None: + self.close_calls += 1 + self.is_closed = True + + class FakeSyncClient: + def __init__(self) -> None: + self.close_calls = 0 + + def close(self) -> None: + self.close_calls += 1 + + fake_async_client = FakeAsyncClient() + fake_sync_client = FakeSyncClient() + monkeypatch.setattr(account_service, "_httpx_client", fake_async_client) + monkeypatch.setattr(account_service, "_sync_httpx_client", fake_sync_client) + + account_service.close_httpx_client_sync() + account_service.close_httpx_client_sync() + + assert fake_async_client.close_calls == 1 + assert fake_sync_client.close_calls == 2 diff --git a/tests/test_admin_routes.py b/tests/test_admin_routes.py index c3a83f8..55bed56 100644 --- a/tests/test_admin_routes.py +++ b/tests/test_admin_routes.py @@ -20,6 +20,7 @@ from fastapi import HTTPException from fastapi.testclient import TestClient +from supervaizer.common import ApiError, ApiSuccess from supervaizer.admin.routes import ( # <-- MODIFIED: removed generate_console_token, validate_console_token, verify_admin_access AdminStats, ServerConfiguration, @@ -527,6 +528,78 @@ def test_api_server_status(self, client: TestClient) -> None: assert response.status_code == 200 assert "text/html" in response.headers["content-type"] + def test_api_server_register_success(self, mocker: "MockerFixture") -> None: + """Server register endpoint awaits supervisor registration and returns success.""" + from fastapi import FastAPI + + supervisor_account = Mock() + supervisor_account.register_server = mocker.AsyncMock( + return_value=ApiSuccess( + message="Event SERVER_REGISTER sent", + detail={"id": "evt-1"}, + ) + ) + server = Mock() + server.supervisor_account = supervisor_account + + app = FastAPI() + app.state.server = server + app.include_router(create_admin_routes(), prefix="/manage") + client = TestClient(app) + + response = client.post("/manage/api/server/register") + + assert response.status_code == 200 + assert response.json() == { + "success": True, + "message": "Event SERVER_REGISTER sent", + "detail": {"id": "evt-1"}, + } + supervisor_account.register_server.assert_awaited_once_with(server=server) + + def test_api_server_register_requires_supervisor_account(self) -> None: + """Registration endpoint returns 503 when no supervisor account is configured.""" + from fastapi import FastAPI + + server = Mock() + server.supervisor_account = None + app = FastAPI() + app.state.server = server + app.include_router(create_admin_routes(), prefix="/manage") + client = TestClient(app) + + response = client.post("/manage/api/server/register") + + assert response.status_code == 503 + assert response.json()["detail"] == "No supervisor account configured" + + def test_api_server_register_returns_502_for_api_error( + self, mocker: "MockerFixture" + ) -> None: + """Registration endpoint surfaces non-success API results as 502.""" + from fastapi import FastAPI + + supervisor_account = Mock() + supervisor_account.register_server = mocker.AsyncMock( + return_value=ApiError(message="register failed", detail={"error": "bad"}) + ) + server = Mock() + server.supervisor_account = supervisor_account + + app = FastAPI() + app.state.server = server + app.include_router(create_admin_routes(), prefix="/manage") + client = TestClient(app) + + response = client.post("/manage/api/server/register") + + assert response.status_code == 502 + assert response.json() == { + "success": False, + "message": "register failed", + "detail": {"error": "bad"}, + } + def test_api_agents(self, client: TestClient, mocker: "MockerFixture") -> None: """Test API agents endpoint.""" mock_agent = Mock() diff --git a/tests/test_case.py b/tests/test_case.py index 7d9d6b5..ce2d6c7 100644 --- a/tests/test_case.py +++ b/tests/test_case.py @@ -20,11 +20,13 @@ from typing import Any import pytest +from httpx import ConnectError from pytest_mock import MockerFixture from supervaizer import Account, CaseNode, CaseNodeType, CaseNodes from supervaizer.case import Case, CaseNodeUpdate from supervaizer.lifecycle import EntityStatus +from supervaizer.storage import StorageManager def test_case( @@ -36,7 +38,8 @@ def test_case( assert case_fixture.description == "Test Case Description" -def test_case_start( +@pytest.mark.asyncio +async def test_case_start( account_fixture: Account, respx_mock: Any, mocker: MockerFixture, @@ -74,10 +77,11 @@ def test_case_start( # Mock the account service's send_event method and verify it was called mock_send_event = mocker.patch( - "supervaizer.account_service.send_event", return_value=None + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), ) - new_case = Case.start( + new_case = await Case.start( job_id="job123", account=account_fixture, name="New Case", @@ -88,7 +92,8 @@ def test_case_start( assert mock_send_event.call_count == 1 -def test_case_close( +@pytest.mark.asyncio +async def test_case_close( account_fixture: Account, respx_mock: Any, case_fixture: Case, @@ -111,10 +116,11 @@ def test_case_close( # Mock the account's send_event method to prevent actual API calls mock_send_event = mocker.patch( - "supervaizer.account_service.send_event", return_value=None + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), ) # Execute - case.close(case_result=case_result, final_cost=final_cost) + await case.close(case_result=case_result, final_cost=final_cost) # Assert assert case.status == EntityStatus.COMPLETED @@ -153,11 +159,12 @@ async def test_case_close_without_final_cost( # Mock the account's send_event method to prevent actual API calls mock_send_event = mocker.patch( - "supervaizer.account_service.send_event", return_value=None + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), ) # Execute - case.close(case_result=case_result, final_cost=None) + await case.close(case_result=case_result, final_cost=None) # Assert assert case.status == EntityStatus.COMPLETED @@ -167,6 +174,36 @@ async def test_case_close_without_final_cost( assert mock_send_event.call_count == 1 +@pytest.mark.asyncio +async def test_case_update_failure_does_not_append_or_persist( + storage_manager: StorageManager, + account_fixture: Account, + mocker: MockerFixture, +) -> None: + case = Case( + id="case-update-failure", + job_id="job-update-failure", + account=account_fixture, + status=EntityStatus.IN_PROGRESS, + name="Failure Test Case", + description="Test failed update persistence", + ) + original_update_count = len(case.updates) + update = CaseNodeUpdate(name="Will fail", payload={"status": "pending"}) + mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(side_effect=ConnectError("network down")), + ) + + with pytest.raises(ConnectError, match="network down"): + await case.update(update) + + stored_case = storage_manager.get_object_by_id("Case", case.id) + assert len(case.updates) == original_update_count + assert stored_case is not None + assert stored_case["updates"] == [] + + def test_case_node_instantiation( case_node_confirm_call: CaseNode, ) -> None: @@ -379,11 +416,15 @@ def test_case_node_update_registration_info_includes_upsert() -> None: assert u.registration_info["upsert"] is True -def test_case_patch_step_replaces_in_memory_step_and_sets_upsert( +@pytest.mark.asyncio +async def test_case_patch_step_replaces_in_memory_step_and_sets_upsert( case_fixture: Case, mocker: MockerFixture, ) -> None: - mocker.patch("supervaizer.account_service.send_event", return_value=None) + mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), + ) prior = CaseNodeUpdate(name="Prior", payload={"original": True}) prior.index = 2 case_fixture.updates = [prior] @@ -392,7 +433,7 @@ def test_case_patch_step_replaces_in_memory_step_and_sets_upsert( name="Human Input Response", payload={"answer": {"x": 1}, "message": "m", "response_type": "human_input"}, ) - case_fixture.patch_step(2, new_u) + await case_fixture.patch_step(2, new_u) assert len(case_fixture.updates) == 1 assert case_fixture.updates[0] is new_u @@ -400,20 +441,22 @@ def test_case_patch_step_replaces_in_memory_step_and_sets_upsert( assert new_u.upsert is True -def test_case_patch_step_no_index_match_still_sends_but_keeps_registry_unchanged( +@pytest.mark.asyncio +async def test_case_patch_step_no_index_match_still_sends_but_keeps_registry_unchanged( case_fixture: Case, mocker: MockerFixture, ) -> None: """Studio still receives the upsert; only the local updates list is left as-is if no index matches.""" mock_send = mocker.patch( - "supervaizer.account_service.send_event", return_value=None + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), ) prior = CaseNodeUpdate(name="Only", payload={}) prior.index = 1 case_fixture.updates = [prior] orphan = CaseNodeUpdate(name="Patch", payload={"p": 1}) - case_fixture.patch_step(99, orphan) + await case_fixture.patch_step(99, orphan) assert mock_send.call_count == 1 assert len(case_fixture.updates) == 1 @@ -444,14 +487,18 @@ def test_case_registration_info_includes_metadata(account_fixture: Account) -> N assert info["metadata"] == {"contact_email": "a@b.com", "language": "en"} -def test_case_start_accepts_metadata( +@pytest.mark.asyncio +async def test_case_start_accepts_metadata( account_fixture: Account, mocker: MockerFixture, ) -> None: """Case.start() passes metadata through to the Case instance.""" - mocker.patch("supervaizer.account_service.send_event", return_value=None) + mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), + ) meta = {"contact_email": "test@example.com", "language": "fr"} - case = Case.start( + case = await Case.start( job_id="job-start-meta", name="Test Contact", account=account_fixture, diff --git a/tests/test_deploy_clean.py b/tests/test_deploy_clean.py new file mode 100644 index 0000000..060ab25 --- /dev/null +++ b/tests/test_deploy_clean.py @@ -0,0 +1,205 @@ +# Copyright (c) 2024-2026 Alain Prasquier - Supervaize.com. All rights reserved. +# +# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0. +# If a copy of the MPL was not distributed with this file, you can obtain one at +# https://mozilla.org/MPL/2.0/. + +from pathlib import Path + +import pytest +from pytest_mock import MockerFixture + +from supervaizer.deploy.commands.clean import ( + clean_deployment, + clean_docker_artifacts, + clean_state_only, +) + + +def create_deployment_artifacts(deployment_dir: Path) -> None: + deployment_dir.mkdir() + (deployment_dir / "Dockerfile").write_text("FROM python\n") + (deployment_dir / ".dockerignore").write_text("__pycache__/\n") + (deployment_dir / "state.json").write_text("{}\n") + (deployment_dir / "config.yaml").write_text("name: service\n") + logs_dir = deployment_dir / "logs" + logs_dir.mkdir() + (logs_dir / "deploy.log").write_text("ok\n") + + +class TestCleanDeployment: + def test_clean_deployment_returns_when_directory_is_missing( + self, + tmp_path: Path, + ) -> None: + missing_dir = tmp_path / ".deployment" + + clean_deployment(missing_dir) + + assert not missing_dir.exists() + + def test_clean_deployment_deletes_directory_when_forced( + self, + tmp_path: Path, + ) -> None: + deployment_dir = tmp_path / ".deployment" + create_deployment_artifacts(deployment_dir) + + clean_deployment(deployment_dir, force=True, verbose=True) + + assert not deployment_dir.exists() + + def test_clean_deployment_keeps_directory_when_cancelled( + self, + tmp_path: Path, + mocker: MockerFixture, + ) -> None: + deployment_dir = tmp_path / ".deployment" + create_deployment_artifacts(deployment_dir) + confirm = mocker.patch( + "supervaizer.deploy.commands.clean.Confirm.ask", + return_value=False, + ) + + clean_deployment(deployment_dir) + + assert deployment_dir.exists() + confirm.assert_called_once() + + def test_clean_deployment_wraps_permission_errors( + self, + tmp_path: Path, + mocker: MockerFixture, + ) -> None: + deployment_dir = tmp_path / ".deployment" + create_deployment_artifacts(deployment_dir) + mocker.patch( + "supervaizer.deploy.commands.clean.shutil.rmtree", + side_effect=PermissionError("locked"), + ) + + with pytest.raises(RuntimeError, match="Failed to clean deployment directory"): + clean_deployment(deployment_dir, force=True) + + +class TestCleanDockerArtifacts: + def test_clean_docker_artifacts_returns_when_deployment_dir_is_missing( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.chdir(tmp_path) + + clean_docker_artifacts() + + assert not (tmp_path / ".deployment").exists() + + def test_clean_docker_artifacts_deletes_only_docker_files( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.chdir(tmp_path) + deployment_dir = tmp_path / ".deployment" + create_deployment_artifacts(deployment_dir) + + clean_docker_artifacts(force=True, verbose=True) + + assert not (deployment_dir / "Dockerfile").exists() + assert not (deployment_dir / ".dockerignore").exists() + assert not (deployment_dir / "logs").exists() + assert (deployment_dir / "state.json").exists() + assert (deployment_dir / "config.yaml").exists() + + def test_clean_docker_artifacts_keeps_files_when_cancelled( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mocker: MockerFixture, + ) -> None: + monkeypatch.chdir(tmp_path) + deployment_dir = tmp_path / ".deployment" + create_deployment_artifacts(deployment_dir) + mocker.patch( + "supervaizer.deploy.commands.clean.Confirm.ask", + return_value=False, + ) + + clean_docker_artifacts() + + assert (deployment_dir / "Dockerfile").exists() + assert (deployment_dir / "logs").exists() + + def test_clean_docker_artifacts_returns_when_no_artifacts_exist( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.chdir(tmp_path) + (tmp_path / ".deployment").mkdir() + + clean_docker_artifacts() + + assert (tmp_path / ".deployment").exists() + + +class TestCleanStateOnly: + def test_clean_state_only_returns_when_deployment_dir_is_missing( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.chdir(tmp_path) + + clean_state_only() + + assert not (tmp_path / ".deployment").exists() + + def test_clean_state_only_deletes_state_files( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.chdir(tmp_path) + deployment_dir = tmp_path / ".deployment" + create_deployment_artifacts(deployment_dir) + + clean_state_only(force=True, verbose=True) + + assert not (deployment_dir / "state.json").exists() + assert not (deployment_dir / "config.yaml").exists() + assert (deployment_dir / "Dockerfile").exists() + assert (deployment_dir / "logs").exists() + + def test_clean_state_only_keeps_files_when_cancelled( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mocker: MockerFixture, + ) -> None: + monkeypatch.chdir(tmp_path) + deployment_dir = tmp_path / ".deployment" + create_deployment_artifacts(deployment_dir) + mocker.patch( + "supervaizer.deploy.commands.clean.Confirm.ask", + return_value=False, + ) + + clean_state_only() + + assert (deployment_dir / "state.json").exists() + assert (deployment_dir / "config.yaml").exists() + + def test_clean_state_only_returns_when_no_state_files_exist( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.chdir(tmp_path) + deployment_dir = tmp_path / ".deployment" + deployment_dir.mkdir() + (deployment_dir / "Dockerfile").write_text("FROM python\n") + + clean_state_only() + + assert (deployment_dir / "Dockerfile").exists() diff --git a/tests/test_deploy_health.py b/tests/test_deploy_health.py new file mode 100644 index 0000000..ba01acb --- /dev/null +++ b/tests/test_deploy_health.py @@ -0,0 +1,242 @@ +# Copyright (c) 2024-2026 Alain Prasquier - Supervaize.com. All rights reserved. +# +# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0. +# If a copy of the MPL was not distributed with this file, you can obtain one at +# https://mozilla.org/MPL/2.0/. + +import httpx +import pytest +from pytest_mock import MockerFixture +from unittest.mock import AsyncMock, MagicMock + +from supervaizer.deploy.health import ( + HealthCheckConfig, + HealthCheckResult, + HealthStatus, + HealthVerifier, + display_health_results, + verify_service_health, +) + + +def mock_sync_health_client( + mocker: MockerFixture, + *responses: httpx.Response | Exception, +) -> tuple[MagicMock, MagicMock]: + client_context = mocker.MagicMock() + client = mocker.MagicMock() + client.get.side_effect = responses + client_context.__enter__.return_value = client + + client_class = mocker.patch( + "supervaizer.deploy.health.httpx.Client", + return_value=client_context, + ) + return client_class, client + + +def mock_async_health_client( + mocker: MockerFixture, + *responses: httpx.Response | Exception, +) -> tuple[MagicMock, AsyncMock]: + client_context = mocker.MagicMock() + client = mocker.AsyncMock() + client.get.side_effect = responses + client_context.__aenter__ = mocker.AsyncMock(return_value=client) + client_context.__aexit__ = mocker.AsyncMock(return_value=None) + + client_class = mocker.patch( + "supervaizer.deploy.health.httpx.AsyncClient", + return_value=client_context, + ) + return client_class, client + + +class TestHealthCheckConfig: + def test_default_endpoint_is_created(self) -> None: + config = HealthCheckConfig() + + assert config.endpoints == ["/.well-known/health"] + + def test_result_timestamp_is_initialized(self) -> None: + result = HealthCheckResult(status=HealthStatus.UNKNOWN, response_time=0.0) + + assert result.timestamp > 0 + + +class TestHealthVerifier: + def test_verify_health_success(self, mocker: MockerFixture) -> None: + client_class, client = mock_sync_health_client(mocker, httpx.Response(200)) + + result = HealthVerifier().verify_health("https://service.test/", "api-key") + + assert result.status == HealthStatus.HEALTHY + assert result.status_code == 200 + assert result.endpoint == "/.well-known/health" + client_class.assert_called_once_with(timeout=60) + client.get.assert_called_once_with( + "https://service.test/.well-known/health", + headers={"X-API-Key": "api-key"}, + ) + + def test_verify_health_retries_until_success(self, mocker: MockerFixture) -> None: + mocker.patch("supervaizer.deploy.health.time.sleep") + _, client = mock_sync_health_client( + mocker, + httpx.Response(503), + httpx.Response(200), + ) + config = HealthCheckConfig(max_retries=2, base_delay=0) + + result = HealthVerifier(config).verify_health("https://service.test") + + assert result.status == HealthStatus.HEALTHY + assert client.get.call_count == 2 + + def test_verify_health_returns_unhealthy_for_bad_status( + self, + mocker: MockerFixture, + ) -> None: + mocker.patch("supervaizer.deploy.health.time.sleep") + mock_sync_health_client(mocker, httpx.Response(500)) + config = HealthCheckConfig(max_retries=1) + + result = HealthVerifier(config).verify_health("https://service.test") + + assert result.status == HealthStatus.UNHEALTHY + assert result.error_message == "Endpoint /.well-known/health returned 500" + + def test_verify_health_handles_no_endpoints(self, mocker: MockerFixture) -> None: + mocker.patch("supervaizer.deploy.health.time.sleep") + config = HealthCheckConfig(max_retries=1, endpoints=[]) + + result = HealthVerifier(config).verify_health("https://service.test") + + assert result.status == HealthStatus.UNHEALTHY + assert result.error_message == "No endpoints configured" + assert result.endpoint is None + + def test_verify_health_handles_request_errors( + self, + mocker: MockerFixture, + ) -> None: + mocker.patch("supervaizer.deploy.health.time.sleep") + mock_sync_health_client(mocker, httpx.RequestError("connection failed")) + config = HealthCheckConfig(max_retries=1) + + result = HealthVerifier(config).verify_health("https://service.test") + + assert result.status == HealthStatus.UNHEALTHY + assert result.error_message == "Request error: connection failed" + + def test_verify_health_handles_timeouts(self, mocker: MockerFixture) -> None: + mocker.patch("supervaizer.deploy.health.time.sleep") + mock_sync_health_client(mocker, httpx.TimeoutException("timeout")) + config = HealthCheckConfig(timeout=3, max_retries=1) + + result = HealthVerifier(config).verify_health("https://service.test") + + assert result.status == HealthStatus.UNHEALTHY + assert result.error_message == "Request timeout after 3s" + + def test_verify_multiple_endpoints_returns_each_endpoint_result( + self, + mocker: MockerFixture, + ) -> None: + mock_sync_health_client(mocker, httpx.Response(200), httpx.Response(500)) + config = HealthCheckConfig(max_retries=1) + + results = HealthVerifier(config).verify_multiple_endpoints( + "https://service.test", + ["/health", "/ready"], + ) + + assert results["/health"].status == HealthStatus.HEALTHY + assert results["/ready"].status == HealthStatus.UNHEALTHY + + def test_get_health_summary_counts_endpoint_statuses(self) -> None: + results = { + "/health": HealthCheckResult( + status=HealthStatus.HEALTHY, + response_time=0.2, + ), + "/ready": HealthCheckResult( + status=HealthStatus.UNHEALTHY, + response_time=0.0, + ), + } + + summary = HealthVerifier().get_health_summary(results) + + assert summary["overall_status"] == HealthStatus.UNHEALTHY + assert summary["total_endpoints"] == 2 + assert summary["healthy_endpoints"] == 1 + assert summary["unhealthy_endpoints"] == 1 + assert summary["success_rate"] == 0.5 + assert summary["average_response_time"] == 0.2 + + def test_get_health_summary_handles_empty_results(self) -> None: + summary = HealthVerifier().get_health_summary({}) + + assert summary["overall_status"] == HealthStatus.HEALTHY + assert summary["success_rate"] == 0.0 + assert summary["average_response_time"] == 0.0 + + @pytest.mark.asyncio + async def test_verify_health_async_success(self, mocker: MockerFixture) -> None: + client_class, client = mock_async_health_client(mocker, httpx.Response(200)) + config = HealthCheckConfig(timeout=7) + + result = await HealthVerifier(config)._verify_health_async( + "https://service.test", + "api-key", + ) + + assert result.status == HealthStatus.HEALTHY + client_class.assert_called_once_with(timeout=7) + client.get.assert_awaited_once_with( + "https://service.test/.well-known/health", + headers={"X-API-Key": "api-key"}, + ) + + @pytest.mark.asyncio + async def test_verify_health_async_retries_after_failure( + self, + mocker: MockerFixture, + ) -> None: + sleep = mocker.patch("supervaizer.deploy.health.asyncio.sleep") + _, client = mock_async_health_client( + mocker, + httpx.Response(503), + httpx.Response(200), + ) + config = HealthCheckConfig(max_retries=2, base_delay=0) + + result = await HealthVerifier(config)._verify_health_async( + "https://service.test" + ) + + assert result.status == HealthStatus.HEALTHY + assert client.get.await_count == 2 + sleep.assert_awaited_once_with(0) + + +def test_verify_service_health_returns_boolean(mocker: MockerFixture) -> None: + mock_sync_health_client(mocker, httpx.Response(200)) + + assert verify_service_health("https://service.test") is True + + +def test_display_health_results_prints_table(mocker: MockerFixture) -> None: + console = mocker.patch("supervaizer.deploy.health.console") + results = { + "/health": HealthCheckResult( + status=HealthStatus.HEALTHY, + response_time=0.1, + status_code=200, + ) + } + + display_health_results(results) + + console.print.assert_called_once() diff --git a/tests/test_job_service.py b/tests/test_job_service.py index 608c043..3caceca 100644 --- a/tests/test_job_service.py +++ b/tests/test_job_service.py @@ -224,7 +224,7 @@ def test_service_job_finished(server_fixture: "Server", mocker: MockerFixture) - # Mock account send_event to avoid actual API calls mock_send_event = mocker.patch( - "supervaizer.account_service.send_event", return_value=None + "supervaizer.account_service.send_event_sync", return_value=None ) # Call the function @@ -249,7 +249,7 @@ def test_service_job_finished_without_account( mock_job.id = str(uuid.uuid4()) mock_event_class = mocker.patch("supervaizer.job_service.JobFinishedEvent") mock_send_event = mocker.patch( - "supervaizer.account_service.send_event", return_value=None + "supervaizer.account_service.send_event_sync", return_value=None ) # Remove supervisor account diff --git a/tests/test_routes.py b/tests/test_routes.py index 17137d4..18f2cc9 100644 --- a/tests/test_routes.py +++ b/tests/test_routes.py @@ -423,3 +423,140 @@ def on_list(*, context: DataResourceContext) -> list[dict[str, Any]]: assert context.workspace_slug == "team-slug" assert context.mission_id == "mission-1" assert context.request_id == "request-1" + + +def _make_data_resource_server( + account_fixture: Account, + agent_method_fixture: AgentMethod, + parameters_setup_fixture: ParametersSetup, + resource: DataResource, +) -> tuple[Server, Agent]: + methods = AgentMethods(job_start=agent_method_fixture) + agent = Agent( + name="Data Routes Agent", + author="a", + developer="d", + version="1.0.0", + description="d", + methods=methods, + parameters_setup=parameters_setup_fixture, + data_resources=[resource], + ) + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + server = Server( + scheme="http", + host="localhost", + port=8001, + environment="test", + mac_addr="E2-AC-ED-22-BF-B2", + debug=True, + agent_timeout=10, + private_key=private_key, + a2a_endpoints=False, + supervisor_account=account_fixture, + agents=[agent], + api_key="test-api-key", + ) + return server, agent + + +def test_data_resource_create_requires_id_in_callback_result( + account_fixture: Account, + agent_method_fixture: AgentMethod, + parameters_setup_fixture: ParametersSetup, +) -> None: + resource = DataResource( + name="items", + fields=[], + on_list=lambda: [], + on_create=lambda data: {"name": data["name"]}, + ) + server, agent = _make_data_resource_server( + account_fixture, agent_method_fixture, parameters_setup_fixture, resource + ) + client = TestClient(server.app) + + response = client.post( + f"/api/agents/{agent.slug}/data/items/", + headers={"X-API-Key": "test-api-key"}, + json={"name": "No ID"}, + ) + + assert response.status_code == 500 + assert "must return a dict with 'id'" in response.json()["detail"] + + +def test_data_resource_update_returns_404_when_callback_returns_none( + account_fixture: Account, + agent_method_fixture: AgentMethod, + parameters_setup_fixture: ParametersSetup, +) -> None: + resource = DataResource( + name="items", + fields=[], + on_list=lambda: [], + on_create=lambda data: {**data, "id": "1"}, + on_update=lambda item_id, data: None, + ) + server, agent = _make_data_resource_server( + account_fixture, agent_method_fixture, parameters_setup_fixture, resource + ) + client = TestClient(server.app) + + response = client.put( + f"/api/agents/{agent.slug}/data/items/missing", + headers={"X-API-Key": "test-api-key"}, + json={"name": "Missing"}, + ) + + assert response.status_code == 404 + assert "missing" in response.json()["detail"] + + +def test_data_resource_delete_returns_404_when_callback_is_false( + account_fixture: Account, + agent_method_fixture: AgentMethod, + parameters_setup_fixture: ParametersSetup, +) -> None: + resource = DataResource( + name="items", + fields=[], + on_list=lambda: [], + on_create=lambda data: {**data, "id": "1"}, + on_delete=lambda item_id: False, + ) + server, agent = _make_data_resource_server( + account_fixture, agent_method_fixture, parameters_setup_fixture, resource + ) + client = TestClient(server.app) + + response = client.delete( + f"/api/agents/{agent.slug}/data/items/missing", + headers={"X-API-Key": "test-api-key"}, + ) + + assert response.status_code == 404 + assert "missing" in response.json()["detail"] + + +def test_data_resource_permission_error_becomes_403( + account_fixture: Account, + agent_method_fixture: AgentMethod, + parameters_setup_fixture: ParametersSetup, +) -> None: + def on_list() -> list[dict[str, Any]]: + raise PermissionError("workspace denied") + + resource = DataResource(name="items", fields=[], on_list=on_list, read_only=True) + server, agent = _make_data_resource_server( + account_fixture, agent_method_fixture, parameters_setup_fixture, resource + ) + client = TestClient(server.app) + + response = client.get( + f"/api/agents/{agent.slug}/data/items/", + headers={"X-API-Key": "test-api-key"}, + ) + + assert response.status_code == 403 + assert response.json()["detail"] == "workspace denied" diff --git a/tests/test_routes_case_update.py b/tests/test_routes_case_update.py index 75e1d5a..9ff6381 100644 --- a/tests/test_routes_case_update.py +++ b/tests/test_routes_case_update.py @@ -80,7 +80,8 @@ def test_update_case_with_answer_success( # Mock the account service's send_event method to prevent actual API calls mock_send_event = mocker.patch( - "supervaizer.account_service.send_event", return_value=None + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), ) # Test data @@ -135,7 +136,8 @@ def test_update_case_with_casestep_index_patches_step( headers = {"X-API-Key": server_fixture.api_key} mock_send_event = mocker.patch( - "supervaizer.account_service.send_event", return_value=None + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), ) request_data = { @@ -303,7 +305,10 @@ def test_human_answer_uses_workbench_style_params_and_strips_casestep_index( name="Test Case", description="Test", ) - mocker.patch("supervaizer.account_service.send_event", return_value=None) + mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), + ) client = TestClient(server_fixture.app) headers = {"X-API-Key": server_fixture.api_key} @@ -406,7 +411,10 @@ def test_human_answer_only_owning_agent_executed( name="c", description="c", ) - mocker.patch("supervaizer.account_service.send_event", return_value=None) + mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), + ) mock_owner = mocker.patch.object( owner, @@ -458,7 +466,10 @@ def test_human_answer_skipped_when_job_agent_not_on_server( name="c", description="c", ) - mocker.patch("supervaizer.account_service.send_event", return_value=None) + mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), + ) spy = mocker.patch.object( server_fixture.agents[0], diff --git a/tests/test_storage.py b/tests/test_storage.py index 71c13b6..61cc78b 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -460,10 +460,10 @@ def test_case_persistence( self._clear_singletons() # Mock the send_event service to avoid HTTP calls - mock_send_event = mocker.patch("supervaizer.account_service.send_event") + mock_send_event = mocker.patch("supervaizer.account_service.send_event_sync") mock_send_event.return_value = mocker.MagicMock() - case = Case.start( + case = Case.start_sync( job_id="test-job-123", name="Test Case", account=account_fixture, @@ -492,7 +492,7 @@ def test_foreign_key_relationships( self._clear_singletons() # Mock the send_event service to avoid HTTP calls - mock_send_event = mocker.patch("supervaizer.account_service.send_event") + mock_send_event = mocker.patch("supervaizer.account_service.send_event_sync") mock_send_event.return_value = mocker.MagicMock() # Create job @@ -505,7 +505,7 @@ def test_foreign_key_relationships( ) # Create case (should automatically add to job's case_ids) - case = Case.start( + case = Case.start_sync( job_id="test-job-123", name="Test Case", account=account_fixture, diff --git a/tests/test_workbench_routes.py b/tests/test_workbench_routes.py index df21d6a..2cee60c 100644 --- a/tests/test_workbench_routes.py +++ b/tests/test_workbench_routes.py @@ -17,15 +17,21 @@ import pytest from fastapi import HTTPException from fastapi.testclient import TestClient +from pytest_mock import MockerFixture from supervaizer import ( + Account, Agent, AgentMethod, AgentMethods, + Case, + CaseNodeUpdate, + EntityStatus, Parameter, ParametersSetup, ) from supervaizer.agent import AgentMethodField, FieldTypeEnum +from supervaizer.case import Cases @pytest.fixture @@ -109,6 +115,92 @@ def test_workbench_page_404_for_unknown_agent(self, test_client_with_agent): assert response.status_code == 404 +class TestWorkbenchHitlAnswer: + """Test HITL answer submission through the workbench route.""" + + def setup_method(self) -> None: + Cases().reset() + + def teardown_method(self) -> None: + Cases().reset() + + def test_answer_hitl_updates_awaiting_case( + self, + test_client_with_agent: tuple[TestClient, str], + account_fixture: Account, + mocker: MockerFixture, + ) -> None: + client, agent_slug = test_client_with_agent + case = Case( + id="workbench-hitl-case", + job_id="workbench-job", + account=account_fixture, + status=EntityStatus.AWAITING, + name="Workbench HITL", + description="Awaiting answer", + ) + case.updates = [ + CaseNodeUpdate( + name="Approval Step", + payload={"supervaizer_form": {"question": "Approve?"}}, + ) + ] + mocker.patch( + "supervaizer.account_service.send_event", + new=mocker.AsyncMock(return_value=None), + ) + + response = client.post( + f"/manage/agents/{agent_slug}/workbench/jobs/{case.job_id}/cases/{case.id}/answer", + json={"answer": {"approved": True}}, + ) + + assert response.status_code == 200 + assert response.json()["status"] == "answered" + assert case.status == EntityStatus.IN_PROGRESS + assert case.updates[-1].payload == { + "approved": True, + "_hitl_label": "Approval Step", + } + + def test_answer_hitl_returns_409_when_case_not_awaiting( + self, + test_client_with_agent: tuple[TestClient, str], + account_fixture: Account, + ) -> None: + client, agent_slug = test_client_with_agent + case = Case( + id="workbench-not-awaiting-case", + job_id="workbench-job", + account=account_fixture, + status=EntityStatus.IN_PROGRESS, + name="Workbench Case", + description="Not awaiting", + ) + + response = client.post( + f"/manage/agents/{agent_slug}/workbench/jobs/{case.job_id}/cases/{case.id}/answer", + json={"answer": {"approved": True}}, + ) + + assert response.status_code == 409 + assert "not awaiting input" in response.json()["detail"] + + def test_answer_hitl_returns_404_for_missing_case( + self, + test_client_with_agent: tuple[TestClient, str], + ) -> None: + client, agent_slug = test_client_with_agent + + response = client.post( + f"/manage/agents/{agent_slug}/workbench/jobs/missing-job/cases/missing-case/answer", + json={"answer": {"approved": True}}, + ) + + assert response.status_code == 404 + assert "missing-case" in response.json()["detail"] + + class TestGetAgentBySlug: """Tests for the get_agent_by_slug helper."""