diff --git a/README.md b/README.md index e5dff28..82b372c 100644 --- a/README.md +++ b/README.md @@ -157,6 +157,23 @@ When using custom CA certificate bundles, you must configure both: --- +## Transports + +Perfecto MCP runs over **stdio** by default. It can also serve **streamable HTTP**, where credentials are +resolved per request from an `Authorization: Bearer` header and the target cloud from a `Perfecto-Cloud-Name` +header, so one server can serve several users and clouds. + +```bash +perfecto-mcp --mcp http +``` + +Transport resolution precedence: **CLI `--mcp` > `PERFECTO_MCP_TRANSPORT` > stdio**. + +See [docs/hosted-http.md](docs/hosted-http.md) for client configuration, auth behavior, health probes and +environment variables. + +--- + ## OpenTelemetry Perfecto MCP reports traces and metrics for MCP tool calls using [OpenTelemetry](https://opentelemetry.io/). This gives you visibility into which tools are used, how long they take, and when errors occur. diff --git a/build.py b/build.py index 2578413..ae05432 100644 --- a/build.py +++ b/build.py @@ -123,6 +123,9 @@ def run_pyinstaller(name: str, icon: str): '--hidden-import=opentelemetry.propagate', '--collect-submodules=opentelemetry', '--collect-all=grpc', + # Streamable HTTP transport: uvicorn resolves its loop/protocol + # implementations by name at runtime, so PyInstaller cannot see them. + '--collect-submodules=uvicorn', ]) diff --git a/config/auth.py b/config/auth.py new file mode 100644 index 0000000..5ef261d --- /dev/null +++ b/config/auth.py @@ -0,0 +1,168 @@ +"""Per-request authentication for the streamable HTTP transport.""" +from __future__ import annotations + +import os +from typing import Optional, Protocol, runtime_checkable + +from mcp.server.fastmcp import Context, FastMCP +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.types import ASGIApp, Receive, Scope, Send + +from config.perfecto import PERFECTO_CLOUD_NAME_ENV_NAME +from config.token import PerfectoToken, PerfectoTokenError + +PERFECTO_TOKEN_STATE_ATTR = "token" +PERFECTO_USER_CONFIG_STATE_ATTR = "user_config" +PERFECTO_CLOUD_NAME_HEADER = "perfecto-cloud-name" + +# Unauthenticated probe paths for orchestrators / load balancers. +HEALTH_PATHS = frozenset({"/health", "/healthz"}) + + +class AuthError(Exception): + """Raised when Authorization cannot be parsed into credentials.""" + + +@runtime_checkable +class AuthPort(Protocol): + """Resolves the Perfecto security token for the current tool invocation.""" + + def get_token(self, ctx: Context) -> Optional[PerfectoToken]: + ... + + +class StdioAuthProvider: + """Process-lifetime token from env / token file / Docker secrets.""" + + def __init__(self, token: Optional[PerfectoToken]): + self._token = token + + def get_token(self, ctx: Context) -> Optional[PerfectoToken]: + return self._token + + +class HttpAuthProvider: + """Per-request token attached by Bearer auth middleware to request.state.""" + + def get_token(self, ctx: Context) -> Optional[PerfectoToken]: + request = ctx.request_context.request + if request is None: + return None + return getattr(request.state, PERFECTO_TOKEN_STATE_ATTR, None) + + +def resolve_cloud_name(header_value: Optional[str] = None) -> Optional[str]: + """ + Resolve the Perfecto cloud for a request. + + Precedence: ``Perfecto-Cloud-Name`` header > PERFECTO_CLOUD_NAME env var. + """ + candidate = (header_value or "").strip() + if candidate: + return candidate + return os.getenv(PERFECTO_CLOUD_NAME_ENV_NAME, "").strip() or None + + +def parse_authorization_header(value: Optional[str], cloud_name: Optional[str] = None) -> PerfectoToken: + """ + Parse ``Authorization: Bearer `` into a PerfectoToken. + + The cloud name is not carried in the credentials; it comes from the + ``Perfecto-Cloud-Name`` header or PERFECTO_CLOUD_NAME. Does not call the + Perfecto API — parse only. + """ + if not value or not value.strip(): + raise AuthError("Missing Authorization header") + + scheme, _, credentials = value.strip().partition(" ") + if scheme.lower() != "bearer" or not credentials.strip(): + raise AuthError("Authorization header must use Bearer scheme") + + try: + return PerfectoToken.from_bearer_credentials(credentials.strip(), cloud_name) + except PerfectoTokenError as exc: + raise AuthError("Unparseable Bearer credentials") from exc + + +class BearerAuthMiddleware: + """ + HTTP gate: require a parseable Bearer token on every request. + + Attaches PerfectoToken to ``request.state``; does not validate against Perfecto. + A missing cloud name is not rejected here — tools surface it as a + configuration error, the same way stdio does. + """ + + def __init__(self, app: ASGIApp): + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + + if scope.get("method") == "OPTIONS": + await self.app(scope, receive, send) + return + + path = scope.get("path", "") or "" + if path in HEALTH_PATHS: + await self.app(scope, receive, send) + return + + request = Request(scope, receive) + cloud_name = resolve_cloud_name(request.headers.get(PERFECTO_CLOUD_NAME_HEADER)) + try: + token = parse_authorization_header( + request.headers.get("authorization"), + cloud_name, + ) + except AuthError: + response = JSONResponse( + {"error": "Unauthorized"}, + status_code=401, + headers={"WWW-Authenticate": "Bearer"}, + ) + await response(scope, receive, send) + return + + setattr(request.state, PERFECTO_TOKEN_STATE_ATTR, token) + setattr( + request.state, + PERFECTO_USER_CONFIG_STATE_ATTR, + {"token": token, "cloud_name": token.cloud_name}, + ) + await self.app(scope, receive, send) + + +def register_health_routes(mcp: FastMCP) -> None: + """Register unauthenticated health probes on the FastMCP ASGI app.""" + + @mcp.custom_route("/health", methods=["GET"]) + async def health(_request: Request) -> JSONResponse: + return JSONResponse({"status": "ok"}) + + @mcp.custom_route("/healthz", methods=["GET"]) + async def healthz(_request: Request) -> JSONResponse: + return JSONResponse({"status": "ok"}) + + +def run_streamable_http(mcp: FastMCP) -> None: + """Serve FastMCP over streamable HTTP with Bearer auth middleware.""" + import anyio + import uvicorn + + register_health_routes(mcp) + + async def _serve() -> None: + app = BearerAuthMiddleware(mcp.streamable_http_app()) + config = uvicorn.Config( + app, + host=mcp.settings.host, + port=mcp.settings.port, + log_level=mcp.settings.log_level.lower(), + ) + await uvicorn.Server(config).serve() + + anyio.run(_serve) diff --git a/config/context_resolution.py b/config/context_resolution.py new file mode 100644 index 0000000..2f1a2db --- /dev/null +++ b/config/context_resolution.py @@ -0,0 +1,36 @@ +"""Helpers to read the per-session user config carried by the MCP context.""" +from typing import Any + +from config.auth import PERFECTO_TOKEN_STATE_ATTR, PERFECTO_USER_CONFIG_STATE_ATTR + + +def get_request_context(ctx: Any) -> Any: + return getattr(ctx, "request_context", None) + + +def get_request_state(ctx: Any) -> Any: + request_context = get_request_context(ctx) + request = getattr(request_context, "request", None) + return getattr(request, "state", None) + + +def resolve_ctx_user_config(ctx: Any) -> dict[str, Any]: + request_context = get_request_context(ctx) + request_state = get_request_state(ctx) + + request_context_config = getattr(request_context, PERFECTO_USER_CONFIG_STATE_ATTR, None) + if isinstance(request_context_config, dict): + return request_context_config + + request_state_config = getattr(request_state, PERFECTO_USER_CONFIG_STATE_ATTR, None) + if isinstance(request_state_config, dict): + return request_state_config + + return {} + + +def resolve_ctx_token(ctx: Any) -> Any: + user_config = resolve_ctx_user_config(ctx) + request_state = get_request_state(ctx) + request_state_token = getattr(request_state, PERFECTO_TOKEN_STATE_ATTR, None) + return user_config.get("token") or request_state_token diff --git a/config/perfecto.py b/config/perfecto.py index 1001cef..324291a 100644 --- a/config/perfecto.py +++ b/config/perfecto.py @@ -8,8 +8,8 @@ SECURITY_TOKEN_ENV_NAME: str = "PERFECTO_SECURITY_TOKEN" PERFECTO_CLOUD_NAME_ENV_NAME: str = 'PERFECTO_CLOUD_NAME' -SECURITY_TOKEN_NOT_SET_MESSAGE: str = f"Perfecto Security Token not set. Set environment variable {SECURITY_TOKEN_FILE_ENV_NAME} or {SECURITY_TOKEN_ENV_NAME}" -PERFECTO_CLOUD_NAME_NOT_SET_MESSAGE: str = f"Perfecto Environment Cloud Name not set. Set environment variable {PERFECTO_CLOUD_NAME_ENV_NAME}" +SECURITY_TOKEN_NOT_SET_MESSAGE: str = f"Perfecto Security Token not set. Set environment variable {SECURITY_TOKEN_FILE_ENV_NAME} or {SECURITY_TOKEN_ENV_NAME}, or send it as 'Authorization: Bearer ' when connecting over HTTP" +PERFECTO_CLOUD_NAME_NOT_SET_MESSAGE: str = f"Perfecto Environment Cloud Name not set. Set environment variable {PERFECTO_CLOUD_NAME_ENV_NAME}, or send the 'Perfecto-Cloud-Name' header when connecting over HTTP" HELP_TOC_URL = "https://help.perfecto.io/perfecto-help/Data/Tocs/" HELP_INDEX_URL = f"{HELP_TOC_URL}perfecto_help.js" diff --git a/config/runtime.py b/config/runtime.py new file mode 100644 index 0000000..efd7b7d --- /dev/null +++ b/config/runtime.py @@ -0,0 +1,102 @@ +"""Process-level runtime wiring shared by all tool registrations.""" +from dataclasses import dataclass +from typing import Any, Literal, Optional + +from config.auth import ( + AuthPort, + PERFECTO_USER_CONFIG_STATE_ATTR, + HttpAuthProvider, + StdioAuthProvider, +) +from config.token import PerfectoToken + +Transport = Literal["stdio", "streamable-http"] + + +@dataclass(frozen=True) +class AppRuntime: + """Process-level collaborators shared by tool registrations.""" + + transport: Transport + auth: AuthPort + user_config: dict[str, Any] + + def resolve_user_config(self, ctx: Any) -> dict[str, Any]: + user_config = dict(self.user_config) + user_config.update(_read_ctx_user_config(ctx)) + token = self.auth.get_token(ctx) + if token is not None: + user_config["token"] = token + return user_config + + def configure_context(self, ctx: Any) -> dict[str, Any]: + user_config = self.resolve_user_config(ctx) + _hydrate_ctx_user_config(ctx, user_config) + return user_config + + +def _read_ctx_user_config(ctx: Any) -> dict[str, Any]: + if ctx is None: + return {} + + user_config: dict[str, Any] = {} + request_context = getattr(ctx, "request_context", None) + request = getattr(request_context, "request", None) + request_state = getattr(request, "state", None) + + for target, attr_name in ( + (ctx, "user_config"), + (request_context, PERFECTO_USER_CONFIG_STATE_ATTR), + (request_state, PERFECTO_USER_CONFIG_STATE_ATTR), + ): + request_config = getattr(target, attr_name, None) + if isinstance(request_config, dict): + user_config.update(request_config) + + return user_config + + +def _hydrate_ctx_user_config(ctx: Any, user_config: dict[str, Any]) -> None: + if ctx is None: + return + + config_copy = dict(user_config) + request_context = getattr(ctx, "request_context", None) + request = getattr(request_context, "request", None) + request_state = getattr(request, "state", None) + + for target in (request_context, request_state): + if target is not None: + setattr(target, PERFECTO_USER_CONFIG_STATE_ATTR, dict(config_copy)) + + +def build_runtime( + transport: Transport, + startup_token: Optional[PerfectoToken] = None, +) -> AppRuntime: + """ + Compose auth for the selected transport. + + - stdio: process-lifetime ``startup_token``. + - streamable-http: request-scoped Bearer auth. + """ + if transport == "stdio": + stdio_user_config = { + "startup_token": startup_token, + "token": startup_token, + "cloud_name": startup_token.cloud_name if startup_token else None, + } + return AppRuntime( + transport=transport, + auth=StdioAuthProvider(startup_token), + user_config=stdio_user_config, + ) + + if transport == "streamable-http": + return AppRuntime( + transport=transport, + auth=HttpAuthProvider(), + user_config={}, + ) + + raise ValueError(f"Unknown transport: {transport}") diff --git a/config/token.py b/config/token.py index 624bc10..a702506 100644 --- a/config/token.py +++ b/config/token.py @@ -1,6 +1,6 @@ from functools import lru_cache from pathlib import Path -from typing import Union +from typing import Optional, Union from config.perfecto import SECURITY_TOKEN_NOT_SET_MESSAGE, PERFECTO_CLOUD_NAME_NOT_SET_MESSAGE @@ -52,5 +52,22 @@ def from_file(cls, path: Union[str, Path], cloud_name: str) -> "PerfectoToken": return cls(token=token_val, cloud_name=cloud_name) + @classmethod + def from_bearer_credentials(cls, credentials: str, cloud_name: Optional[str] = None) -> "PerfectoToken": + """ + Parse Bearer credential material into a PerfectoToken. + + Perfecto credentials are a single security token, so the cloud name is + not part of them: it is supplied by the caller from the + ``Perfecto-Cloud-Name`` header (falling back to PERFECTO_CLOUD_NAME). + Does not call the Perfecto API. + """ + raw = (credentials or "").strip() + if not raw: + raise PerfectoTokenError("Empty bearer credentials") + + normalized_cloud_name = (cloud_name or "").strip() or None + return cls(token=raw, cloud_name=normalized_cloud_name) + def __repr__(self): return "" diff --git a/docs/hosted-http.md b/docs/hosted-http.md new file mode 100644 index 0000000..5fa4bd7 --- /dev/null +++ b/docs/hosted-http.md @@ -0,0 +1,75 @@ +# Hosted HTTP (streamable-http) + +Operator and advanced client guide for running Perfecto MCP over HTTP. For the standard local install (binary, uvx, Docker stdio), see the [README](../README.md). + +## Client configuration + +Configure the MCP client with the server URL, your Perfecto security token as Bearer credentials, and the Perfecto cloud to target: + +```json +{ + "mcpServers": { + "Perfecto MCP": { + "url": "http://localhost:8000/mcp", + "headers": { + "Authorization": "Bearer ", + "Perfecto-Cloud-Name": "" + } + } + } +} +``` + +### Auth behavior + +- Over HTTP, credentials are resolved **per request** from the `Authorization` header. +- Missing or malformed Bearer credentials return `401` before any tool runs. +- Well-formed but wrong security tokens fail later inside Perfecto API calls (same as stdio). +- Stdio / local Docker transport uses the security token file / env vars instead of Bearer auth. + +### Cloud name header + +- Header: `Perfecto-Cloud-Name` +- Resolution precedence: **`Perfecto-Cloud-Name` header > `PERFECTO_CLOUD_NAME` env var**. +- A missing cloud name is **not** rejected at the auth gate. The request is accepted and tools return the + usual "Perfecto Environment Cloud Name not set" error, exactly as stdio does. This keeps `401` meaning + "bad credentials" only. +- Each request resolves its own cloud, so a single server can serve several Perfecto clouds concurrently. + +### Health probes + +`GET /health` and `GET /healthz` bypass authentication and return `{"status": "ok"}`, for orchestrators and load balancers. + +## Local / operator run + +Transport resolution precedence: **CLI `--mcp` > `PERFECTO_MCP_TRANSPORT` > stdio**. + +```bash +# From source +uv run python main.py --mcp http +# or +PERFECTO_MCP_TRANSPORT=http FASTMCP_HOST=0.0.0.0 FASTMCP_PORT=8000 uv run python main.py --mcp + +# Container image (stdio by default; pass hosted HTTP env vars) +docker run --rm -p 8000:8000 \ + -e PERFECTO_MCP_TRANSPORT=http \ + -e FASTMCP_HOST=0.0.0.0 \ + -e FASTMCP_PORT=8000 \ + -e FASTMCP_STREAMABLE_HTTP_PATH=/mcp \ + ghcr.io/perfectocode/perfecto-mcp:latest +``` + +### Environment variables + +| Variable | Description | Default | +|----------|-------------|---------| +| `PERFECTO_MCP_TRANSPORT` | Logical transport: `stdio`, `http`, or `docker` | `stdio` | +| `FASTMCP_HOST` | Bind address (HTTP only) | `127.0.0.1` | +| `FASTMCP_PORT` | Listen port (HTTP only). Also accepts `PORT` (e.g. Cloud Run) | `8000` | +| `FASTMCP_STREAMABLE_HTTP_PATH` | HTTP path for the MCP endpoint | `/mcp` | +| `PERFECTO_CLOUD_NAME` | Fallback cloud when the `Perfecto-Cloud-Name` header is absent | *(unset)* | + +## Limitations + +- No session storage service: session state is not shared across server instances. +- File upload / local file access is not supported by Perfecto MCP on any transport. diff --git a/main.py b/main.py index 016e47b..4b89cee 100644 --- a/main.py +++ b/main.py @@ -7,8 +7,10 @@ from mcp.server.fastmcp import FastMCP, Icon +from config.auth import run_streamable_http from config.perfecto import SECURITY_TOKEN_FILE_ENV_NAME, SECURITY_TOKEN_ENV_NAME, PERFECTO_CLOUD_NAME_ENV_NAME, \ GITHUB +from config.runtime import build_runtime from config.token import PerfectoToken, PerfectoTokenError from config.version import __version__, __executable__, __bundle__, __uvx__, get_version from server import register_tools @@ -20,6 +22,7 @@ PERFECTO_CLOUD_NAME = os.getenv(PERFECTO_CLOUD_NAME_ENV_NAME) LOG_LEVELS = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] +MCP_TRANSPORTS = ("stdio", "http", "docker") def init_logging(level_name: str) -> None: @@ -63,23 +66,106 @@ def get_token() -> PerfectoToken: return token -def run(log_level: str = "CRITICAL"): - init_telemetry("perfecto-mcp", __version__) - token = get_token() +def resolve_mcp_transport(raw_cli_transport: str) -> str: + """ + Resolve transport with precedence: CLI > PERFECTO_MCP_TRANSPORT > stdio. + + `raw_cli_transport` comes from argparse `--mcp`: + - empty string means `--mcp` was provided without an explicit value + - non-empty string means an explicit CLI transport was provided + """ + raw_cli_transport = raw_cli_transport.strip() + + if raw_cli_transport: + candidate = raw_cli_transport + source = "CLI --mcp" + else: + candidate = os.getenv("PERFECTO_MCP_TRANSPORT", "").strip() + source = "PERFECTO_MCP_TRANSPORT" + + if not candidate: + return "stdio" + + normalized = candidate.lower() + if normalized not in MCP_TRANSPORTS: + allowed = ", ".join(MCP_TRANSPORTS) + raise ValueError( + f"Invalid MCP transport '{candidate}' from {source}. " + f"Valid values: {allowed}." + ) + return normalized + + +def to_wire_transport(logical_transport: str) -> Literal["stdio", "streamable-http"]: + """Map CLI/logical transport (stdio|http|docker) to FastMCP wire transport.""" + return "streamable-http" if logical_transport == "http" else "stdio" + + +def build_mcp_server( + log_level: str = "CRITICAL", + transport: str = "stdio", +) -> tuple[FastMCP, str]: + """ + Build FastMCP + auth wiring for a logical CLI transport (stdio|http|docker). + Returns ``(mcp, wire_transport)`` where ``wire_transport`` is the FastMCP + transport name (``stdio`` or ``streamable-http``). + """ + init_telemetry("perfecto-mcp", __version__) + host = "127.0.0.1" + port = 8000 + streamable_http_path = "/mcp" + if transport == "http": + host = os.getenv("FASTMCP_HOST", "127.0.0.1").strip() or "127.0.0.1" + # Cloud Run injects PORT; prefer FASTMCP_PORT when set, else PORT, else 8000. + port_raw = ( + os.getenv("FASTMCP_PORT") + or os.getenv("PORT") + or "8000" + ).strip() or "8000" + port = int(port_raw) + streamable_http_path = os.getenv("FASTMCP_STREAMABLE_HTTP_PATH", "/mcp").strip() or "/mcp" + + # docker and stdio share process-lifetime credentials; http uses Bearer per request. + wire_transport = to_wire_transport(transport) + app_runtime = build_runtime( + wire_transport, + startup_token=get_token() if wire_transport == "stdio" else None, + ) instructions = """ # Perfecto MCP Server """ + mcp = FastMCP( + "perfecto-mcp", + instructions=instructions, + log_level=cast(LOG_LEVELS, log_level), + host=host, + port=port, + streamable_http_path=streamable_http_path, + stateless_http=False, + ) + register_tools(mcp, app_runtime) + return mcp, wire_transport + - mcp = FastMCP("perfecto-mcp", instructions=instructions, - log_level=cast(LOG_LEVELS, log_level)) - register_tools(mcp, token) - mcp.run(transport="stdio") +def run(log_level: str = "CRITICAL", transport: str = "stdio"): + mcp, runtime_transport = build_mcp_server( + log_level=log_level, + transport=transport, + ) + if runtime_transport == "stdio": + mcp.run(transport=runtime_transport) + else: + # Hosted HTTP requires Bearer auth middleware around the ASGI app. + run_streamable_http(mcp) def main(): - parser = argparse.ArgumentParser(prog="perfecto-mcp") + parser = argparse.ArgumentParser( + prog="perfecto-mcp", + formatter_class=argparse.RawTextHelpFormatter, + ) parser.add_argument( "--version", @@ -89,8 +175,13 @@ def main(): parser.add_argument( "--mcp", - action="store_true", - help="Execute MCP Server" + nargs="?", + const="", + metavar="TRANSPORT", + help=( + "Execute MCP Server. Optional TRANSPORT values: stdio, http, docker.\n" + "Resolution precedence: CLI > PERFECTO_MCP_TRANSPORT > stdio." + ), ) parser.add_argument( @@ -102,9 +193,17 @@ def main(): args = parser.parse_args() - if args.mcp: - init_logging(args.log_level) - run(log_level=args.log_level.upper()) + if args.mcp is not None: + try: + transport = resolve_mcp_transport(args.mcp) + if transport == "docker": + os.environ["MCP_DOCKER"] = "true" + elif transport == "http": + os.environ["MCP_DOCKER"] = "false" + init_logging(args.log_level) + run(log_level=args.log_level.upper(), transport=transport) + except ValueError as e: + parser.error(str(e)) else: logo_ascii = ( diff --git a/models/manager.py b/models/manager.py index 32e7665..9cf5433 100644 --- a/models/manager.py +++ b/models/manager.py @@ -1,11 +1,9 @@ -from typing import Optional - from mcp.server.fastmcp import Context -from config.token import PerfectoToken +from config.context_resolution import resolve_ctx_token class Manager: - def __init__(self, token: Optional[PerfectoToken], ctx: Context): - self.token = token + def __init__(self, ctx: Context): self.ctx = ctx + self.token = resolve_ctx_token(ctx) diff --git a/server.py b/server.py index 1600788..25f7827 100644 --- a/server.py +++ b/server.py @@ -1,6 +1,4 @@ -from typing import Optional - -from config.token import PerfectoToken +from config.runtime import AppRuntime from tools.ai_scriptless_manager import register as register_ai_scriptless_manager from tools.device_manager import register as register_device_manager from tools.execution_manager import register as register_execution_manager @@ -9,17 +7,17 @@ from tools.user_manager import register as register_user_manager -def register_tools(mcp, token: Optional[PerfectoToken]): +def register_tools(mcp, runtime: AppRuntime): """ Register all available tools with the MCP server. Args: mcp: The MCP server instance - token: Optional Perfecto token (can be None if not configured) + runtime: App runtime (transport + auth port and shared collaborators) """ - register_user_manager(mcp, token) - register_device_manager(mcp, token) - register_execution_manager(mcp, token) - register_help_manager(mcp, token) - register_ai_scriptless_manager(mcp, token) - register_tools_manager(mcp, token) + register_user_manager(mcp, runtime) + register_device_manager(mcp, runtime) + register_execution_manager(mcp, runtime) + register_help_manager(mcp, runtime) + register_ai_scriptless_manager(mcp, runtime) + register_tools_manager(mcp, runtime) diff --git a/tests/conftest.py b/tests/conftest.py index f1d39c6..929955d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,12 +13,38 @@ See the License for the specific language governing permissions and limitations under the License. """ +from types import SimpleNamespace +from typing import Any, Optional import pytest +from config.auth import PERFECTO_USER_CONFIG_STATE_ATTR from config.token import PerfectoToken @pytest.fixture def perfecto_token() -> PerfectoToken: return PerfectoToken("test-token", "demo") + + +def make_ctx(token: Optional[PerfectoToken] = None, **user_config: Any) -> SimpleNamespace: + """ + Minimal Context stand-in carrying the per-session user config. + + Mirrors what AppRuntime.configure_context hydrates at runtime, so managers + resolve their token from the context in tests the same way they do in + production. + """ + config: dict[str, Any] = dict(user_config) + if token is not None: + config["token"] = token + config.setdefault("cloud_name", token.cloud_name) + + request_context = SimpleNamespace(request=None) + setattr(request_context, PERFECTO_USER_CONFIG_STATE_ATTR, config) + return SimpleNamespace(request_context=request_context) + + +@pytest.fixture +def perfecto_ctx(perfecto_token: PerfectoToken) -> SimpleNamespace: + return make_ctx(perfecto_token) diff --git a/tests/test_ai_scriptless_manager.py b/tests/test_ai_scriptless_manager.py index d2e19f7..fea88cf 100644 --- a/tests/test_ai_scriptless_manager.py +++ b/tests/test_ai_scriptless_manager.py @@ -30,6 +30,8 @@ build_logical_step, new_empty_script, ) +from config.runtime import build_runtime +from tests.conftest import make_ctx from tools.ai_scriptless_manager import AiScriptlessManager, STEP_PATH_REFRESH_NOTES TEST_ID = "PRIVATE:Folder/Test.xml" @@ -117,7 +119,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.execute_test( "PRIVATE:Folder/Test.xml", "real", @@ -137,7 +139,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.execute_test( "PRIVATE:Folder/Test.xml", "real", @@ -148,7 +150,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): assert captured["json"]["params"]["DUT"] == "DEVICE-456" def test_real_device_requires_device_id(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.execute_test( "PRIVATE:Folder/Test.xml", "real", @@ -165,7 +167,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.execute_test( "PRIVATE:Folder/Test.xml", "virtual", @@ -191,7 +193,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.execute_test( "PRIVATE:Folder/Test.xml", "virtual", @@ -205,7 +207,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): assert dut["model"] is None def test_invalid_device_type_returns_error(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.execute_test( "PRIVATE:Folder/Test.xml", "unknown", @@ -222,7 +224,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.execute_test( TEST_ID, "desktop", @@ -251,7 +253,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.execute_test( TEST_ID, "desktop", @@ -292,7 +294,7 @@ async def fake_api_request( lambda _cloud: "https://demo.app.perfectomobile.com/perfectomobile/ai-scriptless/api", ) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_tests({"page_index": 1, "visibility": "PRIVATE"})) assert result.error is None @@ -307,7 +309,7 @@ async def fake_api_request(*_args, **_kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_tests({})) assert result.error == "tree unavailable" @@ -326,7 +328,7 @@ async def fake_api_request( monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_filter_values(["test_name", "owner_list"])) assert result.error is None @@ -340,7 +342,7 @@ async def fake_api_request(*_args, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_filter_values(["bad_filter"])) assert "invalid filter_names" in result.error @@ -371,7 +373,7 @@ async def fake_api_request( monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_commands()) assert result.error is None @@ -387,7 +389,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_commands(checkpoint=True)) assert result.error is None @@ -410,7 +412,7 @@ async def fake_api_request( monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.view_test_structure(TEST_ID)) assert result.error is None @@ -419,7 +421,7 @@ async def fake_api_request( assert result.info is not None def test_view_test_structure_requires_test_id(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.view_test_structure("")) assert result.error == "test_id is required (itemKey from list_tests)" @@ -438,7 +440,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.get_command_definitions(["wait"])) assert result.error is None @@ -465,7 +467,7 @@ async def fake_api_request( monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.view_snapshot("PRIVATE:Folder/Test.xml@uuid-1")) assert result.error is None @@ -475,37 +477,37 @@ async def fake_api_request( class TestManagerValidation: def test_add_command_requires_test_id(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_command("", "ai_user-action")) assert result.error == "test_id is required" def test_add_command_requires_command_id(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_command("PRIVATE:Folder/Test.xml", "")) assert "command_id is required" in result.error def test_view_snapshot_rejects_current_marker(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.view_snapshot("")) assert "not a historical snapshot" in result.error def test_modify_command_requires_arguments(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.modify_command(TEST_ID, "0", {})) assert result.error == "arguments is required" def test_save_test_requires_test_id(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.save_test("")) assert result.error == "test_id is required" def test_create_test_requires_name(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.create_test("")) assert result.error == "name is required" def test_save_test_as_requires_name(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.save_test_as(TEST_ID, "")) assert result.error == "name is required" @@ -515,7 +517,7 @@ def test_add_command_inserts_and_returns_step_path(self, perfecto_token, monkeyp captured: dict = {} _mock_load_and_mutate(monkeypatch, captured=captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_command( TEST_ID, "wait", @@ -532,7 +534,7 @@ def test_modify_command_updates_arguments(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, _script_with_steps("wait"), captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.modify_command( TEST_ID, "0", @@ -549,7 +551,7 @@ def test_modify_command_updates_arguments(self, perfecto_token, monkeypatch): def test_modify_command_returns_error_for_missing_step_path(self, perfecto_token, monkeypatch): _mock_load_and_mutate(monkeypatch, _script_with_steps("wait")) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.modify_command(TEST_ID, "9", {"duration": "1"})) assert result.error == "step_path not found: 9" @@ -558,7 +560,7 @@ def test_delete_command_removes_step(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, _script_with_steps("wait", "comment"), captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.delete_command(TEST_ID, "0")) _assert_step_path_notes(result) @@ -569,7 +571,7 @@ def test_set_command_enabled(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, _script_with_steps("wait"), captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.set_command_enabled(TEST_ID, "0", False)) _assert_step_path_notes(result) @@ -577,7 +579,7 @@ def test_set_command_enabled(self, perfecto_token, monkeypatch): assert captured["script"]["flowElements"][0]["active"] is False def test_move_command_requires_target_path(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.move_command(TEST_ID, "0")) assert result.error == "after_path or parent_path is required" @@ -585,7 +587,7 @@ def test_move_command_reorders_within_root(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, _script_with_steps("wait", "comment"), captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.move_command(TEST_ID, "0", after_path="0")) _assert_step_path_notes(result) @@ -596,7 +598,7 @@ def test_add_command_after_path(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, _script_with_steps("wait", "comment"), captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_command( TEST_ID, "ai_user-action", @@ -614,7 +616,7 @@ def test_add_command_inside_logical_step(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, _script_with_logical_group(), captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_command( TEST_ID, "comment", @@ -634,7 +636,7 @@ async def fake_load_and_mutate(*_args, **_kwargs): monkeypatch.setattr(ai_scriptless_manager, "load_and_mutate", fake_load_and_mutate) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_command(TEST_ID, "wait")) assert result.error == "persist failed" @@ -646,7 +648,7 @@ def test_move_command_into_logical_step(self, perfecto_token, monkeypatch): script["numOfFlowElements"] = 2 _mock_load_and_mutate(monkeypatch, script, captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.move_command(TEST_ID, "1", parent_path="0")) _assert_step_path_notes(result) @@ -659,7 +661,7 @@ def test_add_logical_step(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, captured=captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_logical_step(TEST_ID, label="Setup")) _assert_step_path_notes(result) @@ -670,7 +672,7 @@ def test_add_logical_step_inside_container(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, _script_with_logical_group(), captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_logical_step(TEST_ID, label="Nested", parent_path="0")) _assert_step_path_notes(result) @@ -680,7 +682,7 @@ def test_add_logical_step_inside_container(self, perfecto_token, monkeypatch): assert result.result["step_path"] == "0.0" def test_add_loop_rejects_invalid_count(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_loop(TEST_ID, count=0)) assert result.error == "count must be at least 1" @@ -688,7 +690,7 @@ def test_add_loop_returns_count(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, captured=captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_loop(TEST_ID, count=3)) _assert_step_path_notes(result) @@ -699,7 +701,7 @@ def test_add_loop_inside_logical_step(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, _script_with_logical_group(), captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_loop(TEST_ID, count=2, parent_path="0")) _assert_step_path_notes(result) @@ -713,7 +715,7 @@ def test_add_condition_with_expression(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, captured=captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_condition(TEST_ID, expression="x == 1", label="Check")) _assert_step_path_notes(result) @@ -726,7 +728,7 @@ def test_add_condition_inside_then_branch(self, perfecto_token, monkeypatch): script["flowElements"] = [build_if_statement("x == 1", "Check")] _mock_load_and_mutate(monkeypatch, script, captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_condition( TEST_ID, expression="y == 2", @@ -745,7 +747,7 @@ def test_set_condition_expression(self, perfecto_token, monkeypatch): script["flowElements"] = [build_if_statement("old", "If")] _mock_load_and_mutate(monkeypatch, script, captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.set_condition_expression(TEST_ID, "0", "new == true")) _assert_step_path_notes(result) @@ -753,7 +755,7 @@ def test_set_condition_expression(self, perfecto_token, monkeypatch): assert captured["script"]["flowElements"][0]["expression"] == "new == true" def test_set_condition_expression_requires_expression(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.set_condition_expression(TEST_ID, "0", "")) assert result.error == "expression is required" @@ -778,7 +780,7 @@ async def fake_fetch(_token, _test_id): monkeypatch.setattr(ai_scriptless_manager, "fetch_script_payload", fake_fetch) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_test_variables(TEST_ID)) assert result.error is None @@ -791,7 +793,7 @@ async def fake_fetch(_token, _test_id): monkeypatch.setattr(ai_scriptless_manager, "fetch_script_payload", fake_fetch) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_test_variables(TEST_ID)) assert result.error == "script missing" @@ -799,7 +801,7 @@ def test_add_test_variable(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, captured=captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.add_test_variable( TEST_ID, "count", "number", 42, set_at_runtime=True, )) @@ -811,7 +813,7 @@ def test_add_test_variable(self, perfecto_token, monkeypatch): assert captured["script"]["variables"][0]["@type"] == "Parameter" def test_modify_test_variable_requires_change(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.modify_test_variable(TEST_ID, "token")) assert "At least one of value, variable_type, or set_at_runtime is required" in result.error @@ -831,7 +833,7 @@ def test_modify_test_variable(self, perfecto_token, monkeypatch): }] _mock_load_and_mutate(monkeypatch, script, captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.modify_test_variable(TEST_ID, "token", value="new")) assert result.error is None @@ -853,7 +855,7 @@ def test_delete_test_variable(self, perfecto_token, monkeypatch): }] _mock_load_and_mutate(monkeypatch, script, captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.delete_test_variable(TEST_ID, "token")) assert result.error is None @@ -871,7 +873,7 @@ async def fake_persist(_token, item_key, script, saved_script=None, snapshot_com monkeypatch.setattr(ai_scriptless_manager, "persist_script", fake_persist) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.create_test("Login", folder="My Folder")) assert result.error is None @@ -883,7 +885,7 @@ def test_save_test_passes_snapshot_comment(self, perfecto_token, monkeypatch): captured: dict = {} _mock_load_and_mutate(monkeypatch, captured=captured) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.save_test(TEST_ID, comment="checkpoint")) assert result.error is None @@ -905,7 +907,7 @@ async def fake_persist(_token, item_key, script, saved_script=None, snapshot_com monkeypatch.setattr(ai_scriptless_manager, "fetch_script_payload", fake_fetch) monkeypatch.setattr(ai_scriptless_manager, "persist_script", fake_persist) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.save_test_as( TEST_ID, "Copy", @@ -925,7 +927,7 @@ async def fake_fetch(_token, _test_id): monkeypatch.setattr(ai_scriptless_manager, "fetch_script_payload", fake_fetch) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.save_test_as(TEST_ID, "Copy")) assert result.error == "source missing" @@ -935,12 +937,12 @@ async def fake_api_request(*_args, **_kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.move_test(TEST_ID, "Archive")) assert result.error == "move denied" def test_list_snapshots_rejects_invalid_item_key(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_snapshots("invalid")) assert "Invalid itemKey format" in result.error @@ -954,7 +956,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.move_test(TEST_ID, "Archive", visibility="PUBLIC")) assert result.error is None @@ -965,7 +967,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): assert result.result["source_item_key"] == TEST_ID def test_move_test_rejects_invalid_item_key(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.move_test("not-an-item-key", "Archive")) assert "Invalid itemKey format" in result.error @@ -979,7 +981,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.delete_test(TEST_ID)) assert result.error is None @@ -992,7 +994,7 @@ async def fake_api_request(*_args, **_kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.delete_test(TEST_ID)) assert result.error == "delete denied" @@ -1006,7 +1008,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_snapshots(TEST_ID)) assert result.error is None @@ -1020,12 +1022,12 @@ async def fake_api_request(*_args, **_kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.list_snapshots(TEST_ID)) assert result.error == "snapshots unavailable" def test_get_command_definitions_requires_ids(self, perfecto_token): - manager = AiScriptlessManager(perfecto_token, ctx=None) + manager = AiScriptlessManager(make_ctx(perfecto_token)) result = asyncio.run(manager.get_command_definitions([])) assert result.error == "command_ids is required and must not be empty" @@ -1156,7 +1158,7 @@ async def fake_load_and_mutate(_token, test_id, mutator, snapshot_comment=None): class TestAiScriptlessDispatcher: def test_unknown_action_returns_error(self, perfecto_token): tool = _register_tool(perfecto_token) - result = asyncio.run(tool(action="not_a_real_action", args={}, ctx=None)) + result = asyncio.run(tool(action="not_a_real_action", args={}, ctx=make_ctx())) assert "not found in AI Scriptless manager tool" in result.error def test_routes_add_command(self, perfecto_token, monkeypatch): @@ -1171,7 +1173,7 @@ def test_routes_add_command(self, perfecto_token, monkeypatch): "command_id": "wait", "arguments": {"duration": "1"}, }, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1201,7 +1203,7 @@ async def fake_fetch(_token, _test_id): result = asyncio.run(tool( action="list_test_variables", args={"test_id": TEST_ID}, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1214,7 +1216,7 @@ async def fake_api_request(*_args, **_kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", fake_api_request) tool = _register_tool(perfecto_token) - result = asyncio.run(tool(action="list_tests", args=None, ctx=None)) + result = asyncio.run(tool(action="list_tests", args=None, ctx=make_ctx())) assert result.error == "tree unavailable" def test_routes_move_command(self, perfecto_token, monkeypatch): @@ -1225,7 +1227,7 @@ def test_routes_move_command(self, perfecto_token, monkeypatch): result = asyncio.run(tool( action="move_command", args={"test_id": TEST_ID, "step_path": "0", "after_path": "0"}, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1240,7 +1242,7 @@ def test_routes_delete_command(self, perfecto_token, monkeypatch): result = asyncio.run(tool( action="delete_command", args={"test_id": TEST_ID, "step_path": "0"}, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1263,7 +1265,7 @@ async def fake_api_request(_token, method, endpoint=None, **kwargs): "device_type": "real", "device_under_test": {"device_id": "DEV-1"}, }, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1282,7 +1284,7 @@ async def fake_persist(_token, item_key, script, saved_script=None, snapshot_com result = asyncio.run(tool( action="create_test", args={"name": "Smoke", "folder": "QA"}, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1295,7 +1297,7 @@ async def fake_persist(_token, item_key, script, saved_script=None, snapshot_com ]) def test_dispatcher_validation_errors(self, perfecto_token, action, args, expected_error): tool = _register_tool(perfecto_token) - result = asyncio.run(tool(action=action, args=args, ctx=None)) + result = asyncio.run(tool(action=action, args=args, ctx=make_ctx())) if expected_error: assert expected_error in result.error @@ -1308,7 +1310,7 @@ async def raise_http_error(*_args, **_kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", raise_http_error) tool = _register_tool(perfecto_token) - result = asyncio.run(tool(action="list_tests", args={}, ctx=None)) + result = asyncio.run(tool(action="list_tests", args={}, ctx=make_ctx())) assert result.error is not None assert result.error.startswith("Error:") @@ -1321,7 +1323,7 @@ async def raise_runtime_error(*_args, **_kwargs): monkeypatch.setattr(ai_scriptless_manager, "api_request", raise_runtime_error) tool = _register_tool(perfecto_token) - result = asyncio.run(tool(action="list_tests", args={}, ctx=None)) + result = asyncio.run(tool(action="list_tests", args={}, ctx=make_ctx())) assert "boom" in result.error assert SUPPORT_MESSAGE in result.error @@ -1348,7 +1350,7 @@ async def fake_api_request( result = asyncio.run(tool( action="view_test_structure", args={"test_id": TEST_ID}, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1363,7 +1365,7 @@ def test_routes_list_filter_values(self, perfecto_token, monkeypatch): result = asyncio.run(tool( action="list_filter_values", args={"filter_names": ["test_name", "owner_list"]}, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1383,7 +1385,7 @@ def test_routes_save_test_as(self, perfecto_token, monkeypatch): "visibility": "PUBLIC", "comment": "v2", }, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1404,7 +1406,7 @@ def test_routes_add_test_variable(self, perfecto_token, monkeypatch): "value": 3, "set_at_runtime": True, }, - ctx=None, + ctx=make_ctx(), )) assert result.error is None @@ -1417,7 +1419,7 @@ def test_dispatcher_routes_registered_action(self, perfecto_token, monkeypatch, _setup_dispatcher_mocks(monkeypatch) tool = _register_tool(perfecto_token) - result = asyncio.run(tool(action=action, args=args, ctx=None)) + result = asyncio.run(tool(action=action, args=args, ctx=make_ctx())) assert "not found in AI Scriptless manager tool" not in (result.error or "") @@ -1434,5 +1436,5 @@ def decorator(fn): return decorator mcp = _McpStub() - ai_scriptless_manager.register(mcp, token) + ai_scriptless_manager.register(mcp, build_runtime("stdio", startup_token=token)) return mcp.tools["perfecto_ai_scriptless"] diff --git a/tests/test_http_auth.py b/tests/test_http_auth.py new file mode 100644 index 0000000..af059e8 --- /dev/null +++ b/tests/test_http_auth.py @@ -0,0 +1,337 @@ +""" +Copyright 2025 Perforce Software, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.routing import Route +from starlette.testclient import TestClient + +from config.auth import ( + AuthError, + PERFECTO_CLOUD_NAME_HEADER, + PERFECTO_TOKEN_STATE_ATTR, + PERFECTO_USER_CONFIG_STATE_ATTR, + BearerAuthMiddleware, + HttpAuthProvider, + StdioAuthProvider, + parse_authorization_header, + resolve_cloud_name, +) +from config.perfecto import PERFECTO_CLOUD_NAME_ENV_NAME +from config.runtime import build_runtime +from config.token import PerfectoToken, PerfectoTokenError +from models.manager import Manager + + +class TestBearerCredentialParsing: + def test_security_token_with_cloud_name(self): + token = PerfectoToken.from_bearer_credentials("security-token", "demo") + assert token.token == "security-token" + assert token.cloud_name == "demo" + + def test_security_token_without_cloud_name(self): + token = PerfectoToken.from_bearer_credentials("security-token") + assert token.token == "security-token" + assert token.cloud_name is None + + def test_blank_cloud_name_is_normalized_to_none(self): + token = PerfectoToken.from_bearer_credentials("security-token", " ") + assert token.cloud_name is None + + def test_empty_raises(self): + with pytest.raises(PerfectoTokenError): + PerfectoToken.from_bearer_credentials(" ") + + +class TestResolveCloudName: + def test_header_wins_over_env(self, monkeypatch): + monkeypatch.setenv(PERFECTO_CLOUD_NAME_ENV_NAME, "env-cloud") + assert resolve_cloud_name("header-cloud") == "header-cloud" + + def test_falls_back_to_env(self, monkeypatch): + monkeypatch.setenv(PERFECTO_CLOUD_NAME_ENV_NAME, "env-cloud") + assert resolve_cloud_name(None) == "env-cloud" + assert resolve_cloud_name(" ") == "env-cloud" + + def test_returns_none_when_unset(self, monkeypatch): + monkeypatch.delenv(PERFECTO_CLOUD_NAME_ENV_NAME, raising=False) + assert resolve_cloud_name(None) is None + + +class TestAuthorizationHeaderParsing: + def test_bearer_security_token(self): + token = parse_authorization_header("Bearer security-token", "demo") + assert token.token == "security-token" + assert token.cloud_name == "demo" + + def test_missing_header(self): + with pytest.raises(AuthError): + parse_authorization_header(None) + + def test_wrong_scheme(self): + with pytest.raises(AuthError): + parse_authorization_header("Basic abc") + + def test_bearer_without_credentials(self): + with pytest.raises(AuthError): + parse_authorization_header("Bearer ") + + +class TestAuthProviders: + def test_stdio_returns_startup_token(self, perfecto_token): + provider = StdioAuthProvider(perfecto_token) + assert provider.get_token(ctx=None) is perfecto_token + + def test_stdio_allows_none(self): + assert StdioAuthProvider(None).get_token(ctx=None) is None + + def test_http_reads_request_state(self): + token_a = PerfectoToken("token-a", "cloud-a") + token_b = PerfectoToken("token-b", "cloud-b") + provider = HttpAuthProvider() + + def ctx_with(token: PerfectoToken): + request = SimpleNamespace(state=SimpleNamespace(**{PERFECTO_TOKEN_STATE_ATTR: token})) + request_context = SimpleNamespace(request=request) + return SimpleNamespace(request_context=request_context) + + assert provider.get_token(ctx_with(token_a)).cloud_name == "cloud-a" + assert provider.get_token(ctx_with(token_b)).cloud_name == "cloud-b" + + def test_http_concurrent_tokens_are_isolated(self): + """Two contexts with different Bearer-derived tokens resolve independently.""" + provider = HttpAuthProvider() + token_a = PerfectoToken("token-a", "cloud-a") + token_b = PerfectoToken("token-b", "cloud-b") + + def make_request_ctx(token: PerfectoToken): + request = MagicMock() + setattr(request.state, PERFECTO_TOKEN_STATE_ATTR, token) + ctx = MagicMock() + ctx.request_context.request = request + return ctx + + assert provider.get_token(make_request_ctx(token_a)).token == "token-a" + assert provider.get_token(make_request_ctx(token_b)).token == "token-b" + + +class TestManagerTokenResolution: + def test_manager_falls_back_to_request_state_token(self, perfecto_token): + request = SimpleNamespace(state=SimpleNamespace(**{PERFECTO_TOKEN_STATE_ATTR: perfecto_token})) + ctx = SimpleNamespace(request_context=SimpleNamespace(request=request)) + + manager = Manager(ctx) + + assert manager.token is perfecto_token + + +class _StrictCtx: + """Mimics FastMCP Context where arbitrary attrs are disallowed.""" + + def __init__(self, request_context): + object.__setattr__(self, "request_context", request_context) + + def __setattr__(self, name, value): + if name == "user_config": + raise ValueError('"Context" object has no field "user_config"') + object.__setattr__(self, name, value) + + +class TestBearerAuthMiddleware: + def _app(self): + async def ok(request: Request): + token = getattr(request.state, PERFECTO_TOKEN_STATE_ATTR, None) + user_config = getattr(request.state, PERFECTO_USER_CONFIG_STATE_ATTR, {}) + return JSONResponse( + { + "token": token.token if token else None, + "cloud_name": user_config.get("cloud_name"), + } + ) + + return BearerAuthMiddleware(Starlette(routes=[Route("/mcp", endpoint=ok, methods=["POST"])])) + + def test_missing_authorization_returns_401(self): + client = TestClient(self._app()) + response = client.post("/mcp") + assert response.status_code == 401 + assert response.json()["error"] == "Unauthorized" + assert response.headers["WWW-Authenticate"] == "Bearer" + + def test_wrong_scheme_returns_401(self): + client = TestClient(self._app()) + response = client.post("/mcp", headers={"Authorization": "Basic abc"}) + assert response.status_code == 401 + + def test_valid_bearer_attaches_token(self, monkeypatch): + monkeypatch.delenv(PERFECTO_CLOUD_NAME_ENV_NAME, raising=False) + client = TestClient(self._app()) + response = client.post( + "/mcp", + headers={ + "Authorization": "Bearer security-token", + PERFECTO_CLOUD_NAME_HEADER: "demo", + }, + ) + assert response.status_code == 200 + assert response.json()["token"] == "security-token" + assert response.json()["cloud_name"] == "demo" + + def test_cloud_name_falls_back_to_env(self, monkeypatch): + monkeypatch.setenv(PERFECTO_CLOUD_NAME_ENV_NAME, "env-cloud") + client = TestClient(self._app()) + response = client.post("/mcp", headers={"Authorization": "Bearer security-token"}) + assert response.status_code == 200 + assert response.json()["cloud_name"] == "env-cloud" + + def test_missing_cloud_name_is_not_rejected_at_the_gate(self, monkeypatch): + """Tools surface the missing cloud as a config error, the same as stdio.""" + monkeypatch.delenv(PERFECTO_CLOUD_NAME_ENV_NAME, raising=False) + client = TestClient(self._app()) + response = client.post("/mcp", headers={"Authorization": "Bearer security-token"}) + assert response.status_code == 200 + assert response.json()["cloud_name"] is None + + def test_cloud_name_not_persisted_between_requests(self, monkeypatch): + monkeypatch.delenv(PERFECTO_CLOUD_NAME_ENV_NAME, raising=False) + client = TestClient(self._app()) + first = client.post( + "/mcp", + headers={ + "Authorization": "Bearer security-token", + PERFECTO_CLOUD_NAME_HEADER: "demo", + }, + ) + assert first.json()["cloud_name"] == "demo" + + second = client.post("/mcp", headers={"Authorization": "Bearer security-token"}) + assert second.json()["cloud_name"] is None + + def test_options_bypasses_auth(self): + async def ok(_request: Request): + return JSONResponse({"ok": True}) + + app = BearerAuthMiddleware(Starlette(routes=[Route("/mcp", endpoint=ok, methods=["OPTIONS"])])) + client = TestClient(app) + assert client.options("/mcp").status_code == 200 + + def test_health_bypasses_auth(self): + async def health(_request: Request): + return JSONResponse({"status": "ok"}) + + app = BearerAuthMiddleware( + Starlette( + routes=[ + Route("/health", endpoint=health, methods=["GET"]), + Route("/healthz", endpoint=health, methods=["GET"]), + ] + ) + ) + client = TestClient(app) + assert client.get("/health").status_code == 200 + assert client.get("/health").json()["status"] == "ok" + assert client.get("/healthz").status_code == 200 + + +class TestBuildRuntime: + def test_build_runtime_stdio_and_http(self, perfecto_token): + stdio = build_runtime("stdio", startup_token=perfecto_token) + assert stdio.transport == "stdio" + assert isinstance(stdio.auth, StdioAuthProvider) + assert stdio.user_config["token"] is perfecto_token + assert stdio.user_config["cloud_name"] == "demo" + + http = build_runtime("streamable-http") + assert http.transport == "streamable-http" + assert isinstance(http.auth, HttpAuthProvider) + assert http.user_config == {} + + def test_unknown_transport_raises(self): + with pytest.raises(ValueError, match="Unknown transport"): + build_runtime("carrier-pigeon") + + def test_configure_context_injects_request_context_for_stdio(self, perfecto_token): + runtime = build_runtime("stdio", startup_token=perfecto_token) + ctx = SimpleNamespace(request_context=SimpleNamespace(request=None)) + + user_config = runtime.configure_context(ctx) + + assert user_config["token"] is perfecto_token + assert getattr(ctx.request_context, PERFECTO_USER_CONFIG_STATE_ATTR, None) == user_config + assert Manager(ctx).token is perfecto_token + + def test_configure_context_merges_http_request_state(self): + runtime = build_runtime("streamable-http") + token = PerfectoToken("security-token", "demo") + request = SimpleNamespace( + state=SimpleNamespace( + **{ + PERFECTO_TOKEN_STATE_ATTR: token, + PERFECTO_USER_CONFIG_STATE_ATTR: {"token": token, "cloud_name": "demo"}, + } + ) + ) + ctx = SimpleNamespace(request_context=SimpleNamespace(request=request)) + + user_config = runtime.configure_context(ctx) + + assert user_config["token"] is token + assert user_config["cloud_name"] == "demo" + assert getattr(ctx.request_context, PERFECTO_USER_CONFIG_STATE_ATTR, None) == user_config + assert getattr(request.state, PERFECTO_USER_CONFIG_STATE_ATTR, None) == user_config + + def test_configure_context_hydrates_request_context_when_ctx_is_strict(self): + runtime = build_runtime("streamable-http") + token = PerfectoToken("security-token", "demo") + request = SimpleNamespace( + state=SimpleNamespace( + **{ + PERFECTO_TOKEN_STATE_ATTR: token, + PERFECTO_USER_CONFIG_STATE_ATTR: {"token": token, "cloud_name": "demo"}, + } + ) + ) + ctx = _StrictCtx(request_context=SimpleNamespace(request=request)) + + user_config = runtime.configure_context(ctx) + manager = Manager(ctx) + + assert user_config["token"] is token + assert getattr(ctx.request_context, PERFECTO_USER_CONFIG_STATE_ATTR, None) == user_config + assert manager.token is token + + def test_http_runtime_isolates_concurrent_sessions(self): + """Each request context resolves only its own Bearer-derived token.""" + runtime = build_runtime("streamable-http") + + def ctx_for(token: PerfectoToken): + request = SimpleNamespace(state=SimpleNamespace(**{PERFECTO_TOKEN_STATE_ATTR: token})) + return SimpleNamespace(request_context=SimpleNamespace(request=request)) + + token_a = PerfectoToken("token-a", "cloud-a") + token_b = PerfectoToken("token-b", "cloud-b") + ctx_a = ctx_for(token_a) + ctx_b = ctx_for(token_b) + + runtime.configure_context(ctx_a) + runtime.configure_context(ctx_b) + + assert Manager(ctx_a).token is token_a + assert Manager(ctx_b).token is token_b diff --git a/tests/test_main_transport.py b/tests/test_main_transport.py new file mode 100644 index 0000000..8674114 --- /dev/null +++ b/tests/test_main_transport.py @@ -0,0 +1,169 @@ +import pytest + +import main +from config.auth import HttpAuthProvider, StdioAuthProvider +from config.runtime import AppRuntime + + +class _DummyFastMCP: + def __init__(self, *args, **kwargs): + self.args = args + self.kwargs = kwargs + self.run_calls = [] + + def run(self, transport="stdio", mount_path=None): + self.run_calls.append({"transport": transport, "mount_path": mount_path}) + + +def _patch_mcp_server_dependencies(monkeypatch): + monkeypatch.setattr(main, "init_telemetry", lambda *a, **k: None) + monkeypatch.setattr(main, "get_token", lambda: None) + monkeypatch.setattr(main, "register_tools", lambda *a, **k: None) + monkeypatch.setattr(main, "FastMCP", _DummyFastMCP) + + +class TestResolveMcpTransport: + def test_precedence_cli_then_env_then_default(self, monkeypatch): + monkeypatch.delenv("PERFECTO_MCP_TRANSPORT", raising=False) + assert main.resolve_mcp_transport("") == "stdio" + + monkeypatch.setenv("PERFECTO_MCP_TRANSPORT", "http") + assert main.resolve_mcp_transport("") == "http" + + # CLI value must override environment fallback. + assert main.resolve_mcp_transport("docker") == "docker" + + def test_invalid_transport_raises_clear_error(self, monkeypatch): + monkeypatch.delenv("PERFECTO_MCP_TRANSPORT", raising=False) + + with pytest.raises(ValueError, match="Invalid MCP transport"): + main.resolve_mcp_transport("banana") + + +class TestToWireTransport: + def test_maps_logical_transports(self): + assert main.to_wire_transport("http") == "streamable-http" + assert main.to_wire_transport("stdio") == "stdio" + assert main.to_wire_transport("docker") == "stdio" + + +class TestBuildMcpServerHttp: + def test_http_uses_env_settings_and_stateful_http(self, monkeypatch): + _patch_mcp_server_dependencies(monkeypatch) + monkeypatch.setenv("FASTMCP_HOST", "0.0.0.0") + monkeypatch.setenv("FASTMCP_PORT", "8012") + monkeypatch.setenv("FASTMCP_STREAMABLE_HTTP_PATH", "/custom-mcp") + + mcp, runtime_transport = main.build_mcp_server(transport="http") + + assert runtime_transport == "streamable-http" + assert isinstance(mcp, _DummyFastMCP) + assert mcp.kwargs["host"] == "0.0.0.0" + assert mcp.kwargs["port"] == 8012 + assert mcp.kwargs["streamable_http_path"] == "/custom-mcp" + assert mcp.kwargs["stateless_http"] is False + + def test_http_falls_back_to_cloud_run_port(self, monkeypatch): + _patch_mcp_server_dependencies(monkeypatch) + monkeypatch.delenv("FASTMCP_PORT", raising=False) + monkeypatch.setenv("PORT", "8080") + monkeypatch.setenv("FASTMCP_HOST", "0.0.0.0") + + mcp, _ = main.build_mcp_server(transport="http") + + assert mcp.kwargs["port"] == 8080 + + +class TestBuildMcpServerTransportMapping: + def test_transport_mapping_keeps_docker_stdio_and_http_streamable(self, monkeypatch): + _patch_mcp_server_dependencies(monkeypatch) + + _mcp_http, runtime_transport_http = main.build_mcp_server(transport="http") + _mcp_docker, runtime_transport_docker = main.build_mcp_server(transport="docker") + _mcp_stdio, runtime_transport_stdio = main.build_mcp_server(transport="stdio") + + assert runtime_transport_http == "streamable-http" + assert runtime_transport_docker == "stdio" + assert runtime_transport_stdio == "stdio" + + +class TestBuildMcpServerAuthWiring: + def test_http_registers_http_auth_provider(self, monkeypatch): + captured = {} + + def capture_register(mcp, runtime): + captured["runtime"] = runtime + + _patch_mcp_server_dependencies(monkeypatch) + monkeypatch.setattr(main, "register_tools", capture_register) + + main.build_mcp_server(transport="http") + + runtime = captured["runtime"] + assert isinstance(runtime, AppRuntime) + assert runtime.transport == "streamable-http" + assert isinstance(runtime.auth, HttpAuthProvider) + assert runtime.user_config == {} + + def test_stdio_and_docker_register_stdio_auth_provider(self, monkeypatch, perfecto_token): + captured = {} + + def capture_register(mcp, runtime): + captured.setdefault("runtimes", []).append(runtime) + + _patch_mcp_server_dependencies(monkeypatch) + monkeypatch.setattr(main, "get_token", lambda: perfecto_token) + monkeypatch.setattr(main, "register_tools", capture_register) + + main.build_mcp_server(transport="stdio") + main.build_mcp_server(transport="docker") + + assert len(captured["runtimes"]) == 2 + for runtime in captured["runtimes"]: + assert isinstance(runtime.auth, StdioAuthProvider) + assert runtime.auth.get_token(ctx=None) is perfecto_token + assert runtime.user_config["token"] is perfecto_token + assert runtime.user_config["cloud_name"] == "demo" + + +class TestRunTransportDispatch: + def test_http_uses_bearer_middleware_server(self, monkeypatch): + calls = {"stdio": 0, "http": 0} + + class _Mcp: + def run(self, transport="stdio", mount_path=None): + calls["stdio"] += 1 + + _patch_mcp_server_dependencies(monkeypatch) + monkeypatch.setattr(main, "build_mcp_server", lambda **k: (_Mcp(), "streamable-http")) + monkeypatch.setattr( + main, + "run_streamable_http", + lambda mcp: calls.__setitem__("http", calls["http"] + 1), + ) + + main.run(transport="http") + + assert calls["http"] == 1 + assert calls["stdio"] == 0 + + def test_stdio_uses_mcp_run(self, monkeypatch): + calls = {"stdio": 0, "http": 0} + + class _Mcp: + def run(self, transport="stdio", mount_path=None): + calls["stdio"] += 1 + assert transport == "stdio" + + _patch_mcp_server_dependencies(monkeypatch) + monkeypatch.setattr(main, "build_mcp_server", lambda **k: (_Mcp(), "stdio")) + monkeypatch.setattr( + main, + "run_streamable_http", + lambda mcp: calls.__setitem__("http", calls["http"] + 1), + ) + + main.run(transport="stdio") + + assert calls["stdio"] == 1 + assert calls["http"] == 0 diff --git a/tests/test_tools_manager.py b/tests/test_tools_manager.py index 99cf12a..2792592 100644 --- a/tests/test_tools_manager.py +++ b/tests/test_tools_manager.py @@ -18,6 +18,7 @@ import httpx +from tests.conftest import make_ctx from tools.tools_manager import ( ToolsManager, _match_recommended_asset, @@ -26,12 +27,8 @@ ) -def _make_ctx(): - return MagicMock() - - def test_version_returns_current_build_metadata(perfecto_token): - manager = ToolsManager(perfecto_token, _make_ctx()) + manager = ToolsManager(make_ctx(perfecto_token)) result = asyncio.run(manager.version()) assert result.error is None @@ -101,7 +98,7 @@ def test_check_updates_when_latest_is_newer(perfecto_token): patch("tools.tools_manager.__version__", "1.0.0"), \ patch("tools.tools_manager.platform.system", return_value="Darwin"), \ patch("tools.tools_manager.platform.machine", return_value="arm64"): - manager = ToolsManager(perfecto_token, _make_ctx()) + manager = ToolsManager(make_ctx(perfecto_token)) result = asyncio.run(manager.check_updates()) assert result.error is None @@ -135,7 +132,7 @@ def test_check_updates_when_up_to_date(perfecto_token): with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client), \ patch("tools.tools_manager.__version__", "1.1.1"): - manager = ToolsManager(perfecto_token, _make_ctx()) + manager = ToolsManager(make_ctx(perfecto_token)) result = asyncio.run(manager.check_updates()) assert result.error is None @@ -158,7 +155,7 @@ def test_check_updates_http_error(perfecto_token): mock_client.__aexit__ = AsyncMock(return_value=False) with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client): - manager = ToolsManager(perfecto_token, _make_ctx()) + manager = ToolsManager(make_ctx(perfecto_token)) result = asyncio.run(manager.check_updates()) assert result.error is None @@ -182,7 +179,7 @@ def test_check_updates_connect_error(perfecto_token): with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client), \ patch("tools.tools_manager.__version__", "1.1.1"): - manager = ToolsManager(perfecto_token, _make_ctx()) + manager = ToolsManager(make_ctx(perfecto_token)) result = asyncio.run(manager.check_updates()) assert result.error is None @@ -201,7 +198,7 @@ def test_check_updates_timeout(perfecto_token): mock_client.__aexit__ = AsyncMock(return_value=False) with patch("tools.tools_manager.httpx.AsyncClient", return_value=mock_client): - manager = ToolsManager(perfecto_token, _make_ctx()) + manager = ToolsManager(make_ctx(perfecto_token)) result = asyncio.run(manager.check_updates()) assert result.error is None diff --git a/tools/ai_scriptless_manager.py b/tools/ai_scriptless_manager.py index 15629e5..942daf0 100644 --- a/tools/ai_scriptless_manager.py +++ b/tools/ai_scriptless_manager.py @@ -8,7 +8,8 @@ from config import perfecto from config.perfecto import TOOLS_PREFIX, SUPPORT_MESSAGE -from config.token import PerfectoToken, token_verify +from config.runtime import AppRuntime +from config.token import token_verify from formatters.ai_scriptless import format_ai_scriptless_tests, \ format_ai_scriptless_tests_filter_values, command_selection_policy_info, \ format_command_catalog, format_command_definitions, format_snapshots_list, \ @@ -81,8 +82,8 @@ def _append_ui_access_info( class AiScriptlessManager(Manager): - def __init__(self, token: Optional[PerfectoToken], ctx: Context): - super().__init__(token, ctx) + def __init__(self, ctx: Context): + super().__init__(ctx) @token_verify async def list_tests(self, args: dict[str, Any]) -> BaseResult: @@ -643,7 +644,7 @@ def mutator(script: dict[str, Any]) -> None: return result -def register(mcp, token: Optional[PerfectoToken]): +def register(mcp, runtime: AppRuntime): @mcp.tool( name=f"{TOOLS_PREFIX}_ai_scriptless", description=""" @@ -817,7 +818,8 @@ async def ai_scriptless( ) -> BaseResult: if args is None: args = {} - ai_scriptless_manager = AiScriptlessManager(token, ctx) + runtime.configure_context(ctx) + ai_scriptless_manager = AiScriptlessManager(ctx) async def _dispatch(): match action: diff --git a/tools/device_manager.py b/tools/device_manager.py index 23c84b9..0de1fa7 100644 --- a/tools/device_manager.py +++ b/tools/device_manager.py @@ -1,4 +1,4 @@ -from typing import Optional, Any, Dict +from typing import Any, Dict import httpx from mcp.server.fastmcp import Context @@ -6,7 +6,8 @@ from config import perfecto from config.perfecto import TOOLS_PREFIX, SUPPORT_MESSAGE -from config.token import PerfectoToken, token_verify +from config.runtime import AppRuntime +from config.token import token_verify from formatters.device import format_real_device, format_virtual_device from formatters.grid import format_grid_info from models.manager import Manager @@ -16,8 +17,8 @@ class DeviceManager(Manager): - def __init__(self, token: Optional[PerfectoToken], ctx: Context): - super().__init__(token, ctx) + def __init__(self, ctx: Context): + super().__init__(ctx) @token_verify async def read_selenium_grid_info(self) -> BaseResult: @@ -60,7 +61,7 @@ async def list_desktop_devices(self) -> BaseResult: return await api_request(self.token, "GET", endpoint=virtual_web_url) -def register(mcp, token: Optional[PerfectoToken]): +def register(mcp, runtime: AppRuntime): @mcp.tool( name=f"{TOOLS_PREFIX}_devices", description=""" @@ -82,7 +83,8 @@ async def devices( ) -> BaseResult: if args is None: args = {} - device_manager = DeviceManager(token, ctx) + runtime.configure_context(ctx) + device_manager = DeviceManager(ctx) async def _dispatch(): match action: diff --git a/tools/execution_manager.py b/tools/execution_manager.py index 7206966..8d0e442 100644 --- a/tools/execution_manager.py +++ b/tools/execution_manager.py @@ -1,5 +1,5 @@ from datetime import datetime, timedelta -from typing import Optional, Any, Dict +from typing import Any, Dict import httpx from mcp.server.fastmcp import Context @@ -7,7 +7,8 @@ from config import perfecto from config.perfecto import TOOLS_PREFIX, SUPPORT_MESSAGE -from config.token import PerfectoToken, token_verify +from config.runtime import AppRuntime +from config.token import token_verify from formatters.execution import format_executions from models.manager import Manager from models.result import BaseResult, PaginationResult @@ -16,8 +17,8 @@ class ExecutionManager(Manager): - def __init__(self, token: Optional[PerfectoToken], ctx: Context): - super().__init__(token, ctx) + def __init__(self, ctx: Context): + super().__init__(ctx) self.metadata_map = { "tag_list": "tags_v2", @@ -206,7 +207,7 @@ async def red_report_execution(self, execution_id: str) -> BaseResult: result_formatter_params={"cloud_name": self.token.cloud_name}) -def register(mcp, token: Optional[PerfectoToken]): +def register(mcp, runtime: AppRuntime): @mcp.tool( name=f"{TOOLS_PREFIX}_execution", description=""" @@ -261,7 +262,8 @@ async def execution( ) -> BaseResult: if args is None: args = {} - execution_manager = ExecutionManager(token, ctx) + runtime.configure_context(ctx) + execution_manager = ExecutionManager(ctx) async def _dispatch(): match action: diff --git a/tools/help_manager.py b/tools/help_manager.py index 7322311..5440d27 100644 --- a/tools/help_manager.py +++ b/tools/help_manager.py @@ -1,7 +1,7 @@ import asyncio from copy import deepcopy from itertools import chain -from typing import Optional, Any, Dict, List +from typing import Any, Dict, List import httpx from mcp.server.fastmcp import Context @@ -9,7 +9,7 @@ from config.perfecto import TOOLS_PREFIX, SUPPORT_MESSAGE, get_real_devices_extended_commands_help_url, \ get_real_devices_extended_command_base_help_url, HELP_INDEX_URL, HELP_TOC_URL, HELP_BASE_CONTENT_URL -from config.token import PerfectoToken +from config.runtime import AppRuntime from formatters.help import format_list_real_devices_extended_commands_info, \ format_read_real_devices_extended_command_info, format_help_info from models.manager import Manager @@ -28,8 +28,8 @@ class HelpManager(Manager): "Help content is sourced from curated Perfecto documentation domains and is trusted by design." ) - def __init__(self, token: Optional[PerfectoToken], ctx: Context): - super().__init__(token, ctx) + def __init__(self, ctx: Context): + super().__init__(ctx) async def _load_help_tree(self): help_index_url = HELP_INDEX_URL @@ -241,7 +241,7 @@ async def read_real_devices_extended_command_info(command_id: str) -> BaseResult result_formatter_params={"base_url": real_devices_extended_command_help_url}) -def register(mcp, token: Optional[PerfectoToken]): +def register(mcp, runtime: AppRuntime): @mcp.tool( name=f"{TOOLS_PREFIX}_help", description=""" @@ -272,7 +272,8 @@ async def help_main( ) -> BaseResult: if args is None: args = {} - help_manager = HelpManager(token, ctx) + runtime.configure_context(ctx) + help_manager = HelpManager(ctx) async def _dispatch(): match action: diff --git a/tools/tools_manager.py b/tools/tools_manager.py index 6bb7ced..cb547eb 100644 --- a/tools/tools_manager.py +++ b/tools/tools_manager.py @@ -16,7 +16,7 @@ TOOLS_PREFIX, WEBSITE, ) -from config.token import PerfectoToken +from config.runtime import AppRuntime from config.version import __bundle__, __executable__, __uvx__, __version__ from models.manager import Manager from models.result import BaseResult @@ -199,8 +199,8 @@ def _github_access_failure_result(exc: Exception) -> BaseResult: class ToolsManager(Manager): - def __init__(self, token: Optional[PerfectoToken], ctx: Context): - super().__init__(token, ctx) + def __init__(self, ctx: Context): + super().__init__(ctx) async def version(self) -> BaseResult: platform_data = _platform_info() @@ -314,7 +314,7 @@ async def check_updates(self) -> BaseResult: ) -def register(mcp, token: Optional[PerfectoToken]): +def register(mcp, runtime: AppRuntime): @mcp.tool( name=f"{TOOLS_PREFIX}_tools", description=""" @@ -341,7 +341,8 @@ async def tools( ) -> BaseResult: if args is None: args = {} - tools_manager = ToolsManager(token, ctx) + runtime.configure_context(ctx) + tools_manager = ToolsManager(ctx) async def _dispatch(): match action: diff --git a/tools/user_manager.py b/tools/user_manager.py index f7b40d5..c37ead4 100644 --- a/tools/user_manager.py +++ b/tools/user_manager.py @@ -1,4 +1,4 @@ -from typing import Optional, Any, Dict +from typing import Any, Dict import httpx from mcp.server.fastmcp import Context @@ -6,7 +6,8 @@ from config import perfecto from config.perfecto import TOOLS_PREFIX, SUPPORT_MESSAGE, get_cloud_app_url -from config.token import PerfectoToken, token_verify +from config.runtime import AppRuntime +from config.token import token_verify from formatters.user import format_users from models.manager import Manager from models.result import BaseResult @@ -15,8 +16,8 @@ class UserManager(Manager): - def __init__(self, token: Optional[PerfectoToken], ctx: Context): - super().__init__(token, ctx) + def __init__(self, ctx: Context): + super().__init__(ctx) @token_verify async def read_user(self) -> BaseResult: @@ -35,7 +36,7 @@ async def read_user(self) -> BaseResult: return result -def register(mcp, token: Optional[PerfectoToken]): +def register(mcp, runtime: AppRuntime): @mcp.tool( name=f"{TOOLS_PREFIX}_user", description=""" @@ -53,7 +54,8 @@ async def user( ) -> BaseResult: if args is None: args = {} - user_manager = UserManager(token, ctx) + runtime.configure_context(ctx) + user_manager = UserManager(ctx) async def _dispatch(): match action: diff --git a/uv.lock b/uv.lock index c233035..1bea189 100644 --- a/uv.lock +++ b/uv.lock @@ -805,7 +805,7 @@ wheels = [ [[package]] name = "perfecto-mcp" -version = "1.1.1" +version = "1.1.2" source = { virtual = "." } dependencies = [ { name = "httpx", extra = ["http2"] },