From 72b838380090be01c1287438b8a8d410a8e50732 Mon Sep 17 00:00:00 2001 From: Eduard Dumitru Date: Thu, 2 Jul 2026 12:04:42 +0200 Subject: [PATCH] feat(cli): add opt-in uipath-ipc transport to uipath server [ROBO-5779] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `uipath server` now supports two transports behind one shared job core; HTTP stays the DEFAULT and IPC is opt-in via an env var, so the switch is reversible and backward compatible. - Transport = the UIPATH_SERVER_IPC env var: its presence selects uipath-ipc, gated by _ipc_capability_available() (an isolated, non-failing seam — today "is uipath_ipc importable") so an env var set against a runtime without IPC support falls back to HTTP instead of failing. Unset ⇒ HTTP. - The endpoint arg is shared: --server-socket is the pipe name under IPC and the UDS path under HTTP. The executor never changes how it spawns, and an older runtime that predates IPC just ignores the env var and treats it as a UDS path. - HTTP edge (default, unchanged): create_app / /health / /jobs/{job_key}/start, Unix socket or TCP, ready-ACK, host-header validation. - IPC edge (opt-in): IPythonRuntimeServer (Ping / StartJob) over NamedPipeServerTransport, Windows Proactor loop. Both edges run jobs through _run_command_isolated (per-job env/cwd isolation, serialized by _state.lock). Cancellation is out of scope for Phase 1 (StopJobAsync stays a no-op); a future phase can add it. `uipath-ipc` stays a hard dependency (internal Azure Artifacts feed for now). Publishing it to public PyPI is a prerequisite for a public `pip install uipath`. Tests: test_server.py (HTTP), test_server_ipc.py (IPC), test_server_transport.py (env-var selection, capability fallback, endpoint guard). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/uipath/pyproject.toml | 14 + packages/uipath/src/uipath/_cli/cli_server.py | 388 ++++++++++++++---- packages/uipath/tests/cli/test_server_ipc.py | 185 +++++++++ .../uipath/tests/cli/test_server_transport.py | 68 +++ packages/uipath/uv.lock | 12 + 5 files changed, 577 insertions(+), 90 deletions(-) create mode 100644 packages/uipath/tests/cli/test_server_ipc.py create mode 100644 packages/uipath/tests/cli/test_server_transport.py diff --git a/packages/uipath/pyproject.toml b/packages/uipath/pyproject.toml index 4628d4707..b04263751 100644 --- a/packages/uipath/pyproject.toml +++ b/packages/uipath/pyproject.toml @@ -8,6 +8,7 @@ dependencies = [ "uipath-core>=0.5.30, <0.6.0", "uipath-runtime>=0.12.2, <0.13.0", "uipath-platform>=0.2.4, <0.3.0", + "uipath-ipc>=2.5.1", "click>=8.3.1", "httpx>=0.28.1", "pyjwt>=2.10.1", @@ -171,10 +172,23 @@ exclude-newer = "2 days" uipath-core = false uipath-runtime = false uipath-platform = false +uipath-ipc = false [tool.uv.sources] uipath-core = { path = "../uipath-core", editable = true } uipath-platform = { path = "../uipath-platform", editable = true } +# uipath-ipc is published to the (anonymously accessible) UiPath-Internal Azure +# Artifacts feed, not PyPI. Pin it to that index so uv fetches ONLY uipath-ipc +# there — this avoids a dependency-confusion exposure from a plain extra-index. +# NOTE: this is uv/dev-and-CI resolution config only (not baked into the wheel); +# `pip install uipath` from PyPI still needs uipath-ipc to be resolvable +# (public-PyPI release or an optional extra) — a separate decision. +uipath-ipc = { index = "uipath-internal" } + +[[tool.uv.index]] +name = "uipath-internal" +url = "https://pkgs.dev.azure.com/uipath/Public.Feeds/_packaging/UiPath-Internal/pypi/simple/" +explicit = true [[tool.uv.index]] name = "testpypi" diff --git a/packages/uipath/src/uipath/_cli/cli_server.py b/packages/uipath/src/uipath/_cli/cli_server.py index 06450c401..3a1f32abb 100644 --- a/packages/uipath/src/uipath/_cli/cli_server.py +++ b/packages/uipath/src/uipath/_cli/cli_server.py @@ -1,3 +1,23 @@ +"""`uipath server` — serves the pre-warmed Python runtime to the .NET job executor. + +Two transports, one job core: + +* **HTTP over a local socket** (default) — an aiohttp server on a Unix socket + (or TCP on Windows / ``--tcp``) exposing ``POST /jobs/{job_key}/start`` and + ``GET /health``, with a ready-ACK pushed back to ``--client-socket``. +* **uipath-ipc named pipe** (opt-in) — hosts ``IPythonRuntimeServer`` (Ping / + StartJob) on a named pipe (``$TMPDIR/CoreFxPipe_`` UDS on Linux, a Win32 + pipe on Windows), using ``--server-socket`` as the pipe name. + +Both forward StartJob to run/debug/eval through the same isolated job core +(``_run_command_isolated``): per-job ``os.environ``/cwd save-set-restore, +serialized by ``_state.lock``. The transport is chosen by the +``UIPATH_SERVER_IPC`` env var (its presence selects IPC), gated by a capability +check that falls back to HTTP — so HTTP stays the default and an older runtime +that predates IPC simply ignores the env var and keeps working. The endpoint arg +(``--server-socket``) is shared, so the executor never changes how it spawns. +""" + import asyncio import importlib import json @@ -6,12 +26,14 @@ import sys import tempfile import time -from importlib.metadata import entry_points +from abc import ABC, abstractmethod +from importlib.metadata import PackageNotFoundError, entry_points, version from importlib.util import find_spec from typing import Any import click from aiohttp import ClientSession, UnixConnector, web +from uipath_ipc import IpcServer, NamedPipeServerTransport from ._telemetry import track_command from ._utils._console import ConsoleLogger @@ -21,11 +43,18 @@ console = ConsoleLogger() +IS_WINDOWS = sys.platform == "win32" + SOCKET_ENV_VAR = "UIPATH_SERVER_SOCKET" DEFAULT_SOCKET_PATH = "/tmp/uipath-server.sock" DEFAULT_PORT = 8765 -IS_WINDOWS = sys.platform == "win32" +# Opt-in switch for the uipath-ipc transport: its mere PRESENCE selects IPC; +# unset ⇒ the traditional HTTP-over-socket transport. Read by both the .NET +# executor (which sets it in the child env) and this server. Selection is gated +# by _should_use_ipc / _ipc_capability_available so an env var set against a +# runtime without IPC support falls back to HTTP rather than failing. +IPC_ENV_VAR = "UIPATH_SERVER_IPC" COMMANDS = { "run": run, @@ -97,7 +126,7 @@ def preload_modules() -> None: def generate_socket_path() -> str: - """Generate a unique socket path for the server to listen on.""" + """Generate a unique socket path for the HTTP server to listen on.""" return os.path.join(tempfile.gettempdir(), f"uipath-server-{os.getpid()}.sock") @@ -120,6 +149,87 @@ def parse_args(args: str | list[str] | None) -> list[str]: return [] +def _runtime_version() -> str: + try: + return version("uipath") + except PackageNotFoundError: + return "unknown" + + +async def _run_command_isolated( + cmd: Any, + args: list[str], + env_vars: dict[str, str], + working_dir: str | None, +) -> dict[str, Any]: + """Run one command with per-job env/cwd isolation (the shared job core). + + Serialized by ``_state.lock``, with ``os.environ`` reset to the server + baseline + request vars and cwd swapped/restored around the run. Returns a + dict both transports project from: + ``{ExitCode, Error, Result, Unexpected}`` — ``Result`` is the command's + return value on the clean-success path and ``Unexpected`` marks an + unexpected exception (the HTTP edge maps that to a 500). The rich result is + still written off-channel (``output.json``). + """ + if _state.lock is None or _state.baseline_env is None: + raise RuntimeError("Server state not initialized") + + async with _state.lock: + original_cwd = os.getcwd() + try: + # Start from server baseline + request env vars only, so nothing from + # a previous job leaks through. + os.environ.clear() + os.environ.update(_state.baseline_env) + if isinstance(env_vars, dict): + os.environ.update(env_vars) + + if working_dir and isinstance(working_dir, str): + try: + os.chdir(working_dir) + except (FileNotFoundError, NotADirectoryError, PermissionError) as e: + return { + "ExitCode": 1, + "Error": f"Cannot change to working directory: {e}", + "Result": None, + "Unexpected": False, + } + + result_value = await asyncio.to_thread( + cmd.main, args, standalone_mode=False + ) + return { + "ExitCode": 0, + "Error": None, + "Result": result_value, + "Unexpected": False, + } + except SystemExit as e: + exit_code = e.code if isinstance(e.code, int) else 1 + return { + "ExitCode": exit_code, + "Error": None if exit_code == 0 else f"Exit code: {exit_code}", + "Result": None, + "Unexpected": False, + } + except Exception as e: # report any job failure as a result, not a fault + return {"ExitCode": 1, "Error": str(e), "Result": None, "Unexpected": True} + finally: + # Restore to server baseline. + try: + os.chdir(original_cwd) + except OSError: + pass + os.environ.clear() + os.environ.update(_state.baseline_env) + + +# --------------------------------------------------------------------------- # +# HTTP transport (default) — aiohttp over a Unix socket / TCP, with ready-ACK # +# --------------------------------------------------------------------------- # + + async def send_ack(ack_socket_path: str, server_socket_path: str) -> None: """Send acknowledgment via HTTP POST to the ack socket.""" ack_message: dict[str, str] = { @@ -150,7 +260,7 @@ async def handle_health(request: web.Request) -> web.Response: async def handle_start(request: web.Request) -> web.Response: - """Handle POST /jobs/{job_key}/start endpoint.""" + """Handle POST /jobs/{job_key}/start — runs a job via the shared core.""" job_key = request.match_info.get("job_key") if not job_key: return web.json_response( @@ -173,13 +283,18 @@ async def handle_start(request: web.Request) -> web.Response: status=400, ) - args_raw = get_field(message, "args", "Args") - args = parse_args(args_raw) - + args = parse_args(get_field(message, "args", "Args")) env_vars = get_field(message, "environmentVariables", "EnvironmentVariables") or {} working_dir = get_field(message, "workingDirectory", "WorkingDirectory") - console.info(f"Starting job {job_key}: {command_name} {args}") + if env_vars and not isinstance(env_vars, dict): + return web.json_response( + { + "success": False, + "error": "Invalid field: 'environmentVariables' must be a dict", + }, + status=400, + ) cmd = COMMANDS.get(command_name) if cmd is None: @@ -188,78 +303,22 @@ async def handle_start(request: web.Request) -> web.Response: status=400, ) - console.info(f"Original cwd: {os.getcwd()}") - console.info(f"Requested working_dir: {working_dir}") + console.info(f"Starting job {job_key}: {command_name} {args}") - if _state.lock is None or _state.baseline_env is None: - raise RuntimeError("Server state not initialized") + result = await _run_command_isolated(cmd, args, env_vars, working_dir) - # Validate environmentVariables type early - if env_vars and not isinstance(env_vars, dict): + if result["Unexpected"]: return web.json_response( - { - "success": False, - "error": "Invalid field: 'environmentVariables' must be a dict", - }, - status=400, + {"success": False, "job_key": job_key, "error": result["Error"]}, + status=500, ) - - # Serialize command execution to prevent concurrent os.environ mutation - async with _state.lock: - original_cwd = os.getcwd() - - try: - # Start from server baseline + request env vars only. - # This ensures no env vars from previous requests leak through. - os.environ.clear() - os.environ.update(_state.baseline_env) - if isinstance(env_vars, dict): - os.environ.update(env_vars) - - if working_dir and isinstance(working_dir, str): - try: - os.chdir(working_dir) - except (FileNotFoundError, NotADirectoryError, PermissionError) as e: - return web.json_response( - { - "success": False, - "job_key": job_key, - "error": f"Cannot change to working directory: {e}", - }, - status=400, - ) - - result = await asyncio.to_thread(cmd.main, args, standalone_mode=False) - - return web.json_response( - { - "success": True, - "job_key": job_key, - "result": result, - } - ) - except SystemExit as e: - exit_code = e.code if isinstance(e.code, int) else 1 - return web.json_response( - { - "success": exit_code == 0, - "job_key": job_key, - "error": None if exit_code == 0 else f"Exit code: {exit_code}", - } - ) - except Exception as e: - return web.json_response( - {"success": False, "job_key": job_key, "error": str(e)}, - status=500, - ) - finally: - # Restore to server baseline - try: - os.chdir(original_cwd) - except OSError: - pass - os.environ.clear() - os.environ.update(_state.baseline_env) + if result["ExitCode"] == 0: + return web.json_response( + {"success": True, "job_key": job_key, "result": result["Result"]} + ) + return web.json_response( + {"success": False, "job_key": job_key, "error": result["Error"]} + ) ALLOWED_HOSTS = {"127.0.0.1", "localhost", "[::1]"} @@ -350,29 +409,132 @@ async def start_tcp_server(host: str, port: int) -> None: await runner.cleanup() +# --------------------------------------------------------------------------- # +# uipath-ipc transport (opt-in via --pipe-name) # +# --------------------------------------------------------------------------- # + + +class IPythonRuntimeServer(ABC): + """Contract the .NET job executor calls over uipath-ipc. + + ``__name__`` is the wire endpoint (matching the .NET ``IPythonRuntimeServer``); + DTOs cross as PascalCase dicts. + """ + + @abstractmethod + async def Ping(self) -> dict[str, Any]: + """Readiness probe → ``{Status, RuntimeVersion}`` (replaces the push-ACK).""" + + @abstractmethod + async def StartJob(self, request: dict[str, Any]) -> dict[str, Any]: + """Run a job (``request`` = PythonRunRequest) → ``{ExitCode, Error}``.""" + + +class PythonRuntimeService: + """``IPythonRuntimeServer`` implementation backed by run/debug/eval.""" + + async def Ping(self) -> dict[str, Any]: + return {"Status": "ready", "RuntimeVersion": _runtime_version()} + + async def StartJob(self, request: dict[str, Any]) -> dict[str, Any]: + job_key = str(get_field(request, "JobKey", "jobKey") or "") + + command_name = get_field(request, "Command", "command") + if not isinstance(command_name, str): + return {"ExitCode": 1, "Error": "Missing or invalid field: 'Command'"} + + cmd = COMMANDS.get(command_name) + if cmd is None: + return {"ExitCode": 1, "Error": f"Unknown command: {command_name}"} + + args = parse_args(get_field(request, "Args", "args")) + env_vars = ( + get_field(request, "EnvironmentVariables", "environmentVariables") or {} + ) + working_dir = get_field(request, "WorkingDirectory", "workingDirectory") + + console.info(f"Starting job {job_key}: {command_name} {args}") + + result = await _run_command_isolated(cmd, args, env_vars, working_dir) + # IPC contract (PythonRunResult) carries only ExitCode + Error. + return {"ExitCode": result["ExitCode"], "Error": result["Error"]} + + +async def start_ipc_server(pipe_name: str) -> None: + """Serve the Python runtime over a uipath-ipc named pipe until it is closed.""" + _state.init() + server = IpcServer( + transport=NamedPipeServerTransport(pipe_name), + services={IPythonRuntimeServer: PythonRuntimeService()}, + request_timeout=None, # jobs are long-running; no server-side timeout + ) + console.success(f"IPC server listening on pipe '{pipe_name}'") + async with server: + await server.serve_forever() + + +# --------------------------------------------------------------------------- # +# Transport selection # +# --------------------------------------------------------------------------- # + + +def _ipc_capability_available() -> bool: + """Whether this runtime can actually serve the uipath-ipc transport. + + Isolated, deliberately non-failing seam for capability detection: even when + ``UIPATH_SERVER_IPC`` is set we only switch to IPC if the capability is + genuinely present, otherwise we fall back to HTTP. Today ``uipath-ipc`` is a + hard dependency, so this just checks it is importable (effectively always + ``True``); the concern is quarantined here so it can later grow into a real + version/marker probe without touching the selection logic below. + """ + return find_spec("uipath_ipc") is not None + + +def _should_use_ipc() -> bool: + """Select the transport: IPC iff ``UIPATH_SERVER_IPC`` is set AND supported. + + Presence of the env var is the executor's opt-in signal; the capability + check guards against running the IPC path where it is unavailable — in which + case we fall back to HTTP instead of failing. + """ + if IPC_ENV_VAR not in os.environ: + return False + return _ipc_capability_available() + + +# --------------------------------------------------------------------------- # +# CLI # +# --------------------------------------------------------------------------- # + + @click.command() @click.option( "--client-socket", type=str, default=None, - help=f"Unix socket path to send ready ack to (default: ${SOCKET_ENV_VAR} or {DEFAULT_SOCKET_PATH})", + help=f"HTTP: Unix socket to send the ready ack to (default: ${SOCKET_ENV_VAR} " + f"or {DEFAULT_SOCKET_PATH}).", ) @click.option( "--server-socket", type=str, default=None, - help="Unix socket path the server listens on (default: auto-generated in tmp dir)", + help=f"Server endpoint. With ${IPC_ENV_VAR} set this is the uipath-ipc pipe " + "name (a $TMPDIR/CoreFxPipe_ UDS on Linux, a Win32 pipe on Windows); " + "otherwise the Unix socket path the HTTP server listens on (default: " + "auto-generated in tmp).", ) @click.option( "--port", type=int, default=None, - help=f"TCP port, used on Windows or when --tcp flag is set (default: {DEFAULT_PORT})", + help=f"HTTP: TCP port, used on Windows or with --tcp (default: {DEFAULT_PORT}).", ) @click.option( "--tcp", is_flag=True, - help="Force TCP mode even on Unix systems", + help="HTTP: force TCP mode even on Unix systems.", ) @track_command("server") def server( @@ -381,20 +543,41 @@ def server( port: int | None, tcp: bool, ) -> None: - """Start an HTTP server that forwards commands to run/debug/eval. + """Serve the pre-warmed runtime to the .NET job executor. + + The transport is chosen by the ``UIPATH_SERVER_IPC`` env var (opt-in): when + set — and the uipath-ipc capability is available — it serves over a + uipath-ipc named pipe using ``--server-socket`` as the pipe name; otherwise + it serves the HTTP path (Unix socket, or TCP on Windows / --tcp) exposing + POST /jobs/{job_key}/start and GET /health. Both forward run/debug/eval and + isolate env/cwd per job. + + The single endpoint arg keeps the command backward compatible: an older + runtime that predates IPC simply ignores the env var and treats + ``--server-socket`` as a UDS path, so the executor never has to change how it + spawns the server. + """ + preload_modules() - Creates its own socket to listen on and sends an ack to --client-socket with: - {"status": "ready", "socket": "/path/to/server.sock"} + if _should_use_ipc(): + if not server_socket: + raise click.UsageError( + "--server-socket (used as the pipe name) is required when " + f"{IPC_ENV_VAR} is set." + ) + _run_ipc_server(server_socket) + else: + _run_http_server(client_socket, server_socket, port, tcp) - Endpoint: POST /jobs/{job_key}/start - Body: {"command": "run", "args": "agent.json '{}'", "environmentVariables": {}, "workingDirectory": "/path"} - Endpoint: GET /health - """ +def _run_http_server( + client_socket: str | None, + server_socket: str | None, + port: int | None, + tcp: bool, +) -> None: + """Run the HTTP server: Unix socket by default, TCP on Windows or with --tcp.""" use_tcp = IS_WINDOWS or tcp - - preload_modules() - try: if use_tcp: asyncio.run(start_tcp_server("127.0.0.1", port or DEFAULT_PORT)) @@ -405,3 +588,28 @@ def server( asyncio.run(start_unix_server(ack_socket_path, server_socket)) except KeyboardInterrupt: console.info("Shutting down") + + +def _run_ipc_server(pipe_name: str) -> None: + """Run the IPC server, on a Proactor event loop when on Windows. + + Windows named pipes require the Proactor loop (the Selector loop can't do pipe + I/O). Proactor is the default since Python 3.8, but we build it explicitly + rather than trust the ambient policy, which another library in the process + (e.g. socketio) could have flipped to Selector. + """ + try: + # Gate on the sys.platform literal (not IS_WINDOWS) so mypy narrows this + # per-platform: ProactorEventLoop is Windows-only, so on Linux this branch + # is unreachable-typed and needs no attr-defined ignore. + if sys.platform == "win32": + loop = asyncio.ProactorEventLoop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(start_ipc_server(pipe_name)) + finally: + loop.close() + else: + asyncio.run(start_ipc_server(pipe_name)) + except KeyboardInterrupt: + console.info("Shutting down") diff --git a/packages/uipath/tests/cli/test_server_ipc.py b/packages/uipath/tests/cli/test_server_ipc.py new file mode 100644 index 000000000..b50efe95e --- /dev/null +++ b/packages/uipath/tests/cli/test_server_ipc.py @@ -0,0 +1,185 @@ +"""Tests for the uipath-ipc runtime server (opt-in via ``UIPATH_SERVER_IPC``). + +Mirrors ``test_server.py`` (which covers the HTTP path) but drives the new +``IpcServer`` over a named pipe with a Python ``uipath-ipc`` client. Adds the +readiness (``Ping``) coverage the HTTP/ack path never had. + +Requires ``uipath-ipc`` to be installed. ``StartJob`` success runs the real +runtime (like ``test_server.test_start_job_success``); the rest exercise the IPC +wiring and env isolation without it. +""" + +import asyncio +import json +import os +import threading +import time +from typing import Any, Awaitable, Callable + +import click +import pytest +from uipath_ipc import IpcClient, NamedPipeClientTransport + +from uipath._cli import cli_server +from uipath._cli.cli_server import ( + IPythonRuntimeServer, + start_ipc_server, +) + +_pipe_counter = 0 + + +def _unique_pipe() -> str: + global _pipe_counter + _pipe_counter += 1 + return f"uipath-ipc-test-{os.getpid()}-{_pipe_counter}" + + +def _serve_in_background(pipe_name: str) -> None: + """Run the IPC server on its own event loop in a daemon thread. + + ``asyncio.new_event_loop()`` yields the per-OS default loop — Proactor on + Windows (required for named pipes), Selector on Linux (CoreFxPipe UDS). + """ + + def run_server() -> None: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + loop.run_until_complete(start_ipc_server(pipe_name)) + except asyncio.CancelledError: + pass + finally: + loop.close() + + thread = threading.Thread(target=run_server, daemon=True) + thread.start() + time.sleep(0.5) + + +async def _with_proxy(pipe_name: str, fn: Callable[[Any], Awaitable[Any]]) -> Any: + """Connect a uipath-ipc client to the pipe, run ``fn(proxy)``, then close.""" + client = IpcClient(transport=NamedPipeClientTransport(pipe_name)) + try: + proxy = client.get_proxy(IPythonRuntimeServer) + return await fn(proxy) + finally: + await client.aclose() + + +def create_uipath_json(script_path: str, entrypoint_name: str = "main") -> dict: + return {"functions": {entrypoint_name: f"{script_path}:main"}} + + +SIMPLE_SCRIPT = """ +from dataclasses import dataclass + +@dataclass +class Input: + message: str + repeat: int = 1 + +def main(input: Input) -> str: + return (input.message + " ") * input.repeat +""" + + +class TestIpcServer: + @pytest.fixture + def pipe(self): + pipe_name = _unique_pipe() + _serve_in_background(pipe_name) + # Daemon thread; the server blocks in serve_forever and is torn down when + # the process exits (mirrors test_server.py's background HTTP server). + yield pipe_name + + def test_ping_reports_ready(self, pipe): + result = asyncio.run(_with_proxy(pipe, lambda p: p.Ping())) + assert result["Status"] == "ready" + assert isinstance(result["RuntimeVersion"], str) + assert result["RuntimeVersion"] + + def test_start_job_success(self, pipe, temp_dir): + """A real 'run' job executes and writes output.json (needs the runtime).""" + script_file = "entrypoint.py" + with open(os.path.join(temp_dir, script_file), "w") as f: + f.write(SIMPLE_SCRIPT) + with open(os.path.join(temp_dir, "uipath.json"), "w") as f: + json.dump(create_uipath_json(script_file), f) + + input_file = os.path.join(temp_dir, "input.json") + with open(input_file, "w") as f: + json.dump({"message": "Hello", "repeat": 3}, f) + output_file = os.path.join(temp_dir, "output.json") + + request = { + "JobKey": "job-123", + "Command": "run", + "Args": ["main", "--input-file", input_file, "--output-file", output_file], + "WorkingDirectory": temp_dir, + "EnvironmentVariables": {}, + } + result = asyncio.run(_with_proxy(pipe, lambda p: p.StartJob(request))) + + assert result["ExitCode"] == 0 + assert result["Error"] is None + assert os.path.exists(output_file) + with open(output_file, "r") as f: + assert "Hello" in f.read() + + def test_start_job_unknown_command(self, pipe): + request = {"JobKey": "job-1", "Command": "does_not_exist"} + result = asyncio.run(_with_proxy(pipe, lambda p: p.StartJob(request))) + assert result["ExitCode"] != 0 + assert "Unknown command" in (result["Error"] or "") + + +class TestIpcServerEnvIsolation: + """Env vars must not leak between sequential jobs (as on the HTTP path).""" + + @pytest.fixture + def pipe_with_spy(self): + env_snapshots: list[dict[str, str]] = [] + + @click.command() + def spy_cmd() -> None: + env_snapshots.append(dict(os.environ)) + + original = cli_server.COMMANDS.copy() + cli_server.COMMANDS["spy"] = spy_cmd + + pipe_name = _unique_pipe() + _serve_in_background(pipe_name) + try: + yield pipe_name, env_snapshots + finally: + cli_server.COMMANDS.clear() + cli_server.COMMANDS.update(original) + + def test_env_vars_do_not_leak_between_jobs(self, pipe_with_spy): + pipe_name, env_snapshots = pipe_with_spy + + async def run_two(proxy: Any) -> None: + await proxy.StartJob( + { + "JobKey": "job-1", + "Command": "spy", + "EnvironmentVariables": {"TEST_VAR_A": "a"}, + } + ) + await proxy.StartJob( + { + "JobKey": "job-2", + "Command": "spy", + "EnvironmentVariables": {"TEST_VAR_B": "b"}, + } + ) + + asyncio.run(_with_proxy(pipe_name, run_two)) + + assert len(env_snapshots) == 2 + run1, run2 = env_snapshots + assert run1["TEST_VAR_A"] == "a" + assert "TEST_VAR_B" not in run1 + assert run2["TEST_VAR_B"] == "b" + assert "TEST_VAR_A" not in run2 diff --git a/packages/uipath/tests/cli/test_server_transport.py b/packages/uipath/tests/cli/test_server_transport.py new file mode 100644 index 000000000..b4a73e59e --- /dev/null +++ b/packages/uipath/tests/cli/test_server_transport.py @@ -0,0 +1,68 @@ +"""`uipath server` picks its transport from the UIPATH_SERVER_IPC env var. + +Contract: the env var's *presence* opts into uipath-ipc (using --server-socket as +the pipe name), gated by a capability check that falls back to HTTP; unset ⇒ the +default HTTP transport. HTTP stays the default so the switch is opt-in and +reversible, and the single endpoint arg keeps the command backward compatible. +These tests stub the two runners (and telemetry) and assert only which one the +CLI dispatches to. +""" + +from click.testing import CliRunner + +import uipath._cli._telemetry as _telemetry +from uipath._cli import cli_server + + +def _stub_runners(monkeypatch) -> dict: + """Stub preload + both transport runners; disable telemetry; clear the env var. + + Returns a dict recording which runner the ``server`` command dispatched to. + """ + calls: dict = {} + monkeypatch.setattr(_telemetry, "is_telemetry_enabled", lambda: False) + monkeypatch.setattr(cli_server, "preload_modules", lambda: None) + monkeypatch.setattr( + cli_server, "_run_ipc_server", lambda pipe_name: calls.update(ipc=pipe_name) + ) + monkeypatch.setattr(cli_server, "_run_http_server", lambda *a: calls.update(http=a)) + monkeypatch.delenv(cli_server.IPC_ENV_VAR, raising=False) + return calls + + +def test_env_var_selects_ipc(monkeypatch): + calls = _stub_runners(monkeypatch) + monkeypatch.setenv(cli_server.IPC_ENV_VAR, "1") + result = CliRunner().invoke(cli_server.server, ["--server-socket", "demo-pipe"]) + assert result.exit_code == 0, result.output + assert calls.get("ipc") == "demo-pipe" # --server-socket used as the pipe name + assert "http" not in calls + + +def test_no_env_var_selects_http(monkeypatch): + calls = _stub_runners(monkeypatch) + result = CliRunner().invoke(cli_server.server, ["--server-socket", "/tmp/x.sock"]) + assert result.exit_code == 0, result.output + assert "http" in calls + assert "ipc" not in calls + + +def test_ipc_env_without_capability_falls_back_to_http(monkeypatch): + """Env set but capability absent ⇒ HTTP (fallback, not failure).""" + calls = _stub_runners(monkeypatch) + monkeypatch.setenv(cli_server.IPC_ENV_VAR, "1") + monkeypatch.setattr(cli_server, "_ipc_capability_available", lambda: False) + result = CliRunner().invoke(cli_server.server, ["--server-socket", "demo-pipe"]) + assert result.exit_code == 0, result.output + assert "http" in calls + assert "ipc" not in calls + + +def test_ipc_env_requires_endpoint(monkeypatch): + """IPC selected but no --server-socket ⇒ a clear usage error, no dispatch.""" + calls = _stub_runners(monkeypatch) + monkeypatch.setenv(cli_server.IPC_ENV_VAR, "1") + result = CliRunner().invoke(cli_server.server, []) + assert result.exit_code != 0 + assert "server-socket" in result.output + assert calls == {} diff --git a/packages/uipath/uv.lock b/packages/uipath/uv.lock index 730276088..a26611fbc 100644 --- a/packages/uipath/uv.lock +++ b/packages/uipath/uv.lock @@ -7,6 +7,7 @@ exclude-newer = "2026-07-08T17:52:11.22969Z" exclude-newer-span = "P2D" [options.exclude-newer-package] +uipath-ipc = false uipath-runtime = false uipath-platform = false uipath-core = false @@ -2619,6 +2620,7 @@ dependencies = [ { name = "tenacity" }, { name = "truststore" }, { name = "uipath-core" }, + { name = "uipath-ipc" }, { name = "uipath-platform" }, { name = "uipath-runtime" }, ] @@ -2673,6 +2675,7 @@ requires-dist = [ { name = "tenacity", specifier = ">=9.0.0" }, { name = "truststore", specifier = ">=0.10.1" }, { name = "uipath-core", editable = "../uipath-core" }, + { name = "uipath-ipc", specifier = ">=2.5.1", index = "https://pkgs.dev.azure.com/uipath/Public.Feeds/_packaging/UiPath-Internal/pypi/simple/" }, { name = "uipath-platform", editable = "../uipath-platform" }, { name = "uipath-runtime", specifier = ">=0.12.2,<0.13.0" }, ] @@ -2739,6 +2742,15 @@ dev = [ { name = "rust-just", specifier = ">=1.39.0" }, ] +[[package]] +name = "uipath-ipc" +version = "2.5.1+20260625.1" +source = { registry = "https://pkgs.dev.azure.com/uipath/Public.Feeds/_packaging/UiPath-Internal/pypi/simple/" } +sdist = { url = "https://pkgs.dev.azure.com/uipath/5b98d55c-1b14-4a03-893f-7a59746f1246/_packaging/788028a9-5a01-48ee-b925-3af51ae46294/pypi/download/uipath-ipc/2.5.1+20260625.1/uipath_ipc-2.5.1+20260625.1.tar.gz", hash = "sha256:bc202fef26d8ecae6b8bcfaccd6bcfb4675e0160faabf8d9281f1028643adee9" } +wheels = [ + { url = "https://pkgs.dev.azure.com/uipath/5b98d55c-1b14-4a03-893f-7a59746f1246/_packaging/788028a9-5a01-48ee-b925-3af51ae46294/pypi/download/uipath-ipc/2.5.1+20260625.1/uipath_ipc-2.5.1+20260625.1-py3-none-any.whl", hash = "sha256:0e90b0198b9afa8a1fa832e010aa1160914391d9b5610684a3bf582704f41773" }, +] + [[package]] name = "uipath-platform" version = "0.2.7"