feat: add E2E tests for Textual TUI and Playwright web frontend

.github/workflows/test.yml

@@ -30,6 +30,9 @@ jobs:
       - name: Install dependencies
         run: uv sync --all-extras
 
+      - name: Install Playwright browsers
+        run: uv run playwright install chromium
+
       - name: Run tests
         run: uv run pytest
 

CLAUDE.md

@@ -3,6 +3,7 @@
 ## Workflow Rules
 - NEVER commit, push, or create PRs unless explicitly asked to do so.
 - Always wait for explicit user confirmation before any git operations that affect the repository.
+- Always run `uv run ruff check src/ tests/`, `uv run ruff format --check .`, and `uv run mypy src/` before committing. Fix any errors before creating the commit.
 
 ## Project Structure
 - Monorepo: Python backend (`src/uipath/dev/`) + React frontend (`src/uipath/dev/server/frontend/`)

pyproject.toml

@@ -52,6 +52,7 @@ dev = [
   "pytest-trio>=0.8.0",
   "pytest-cov>=4.1.0",
   "pytest-mock>=3.11.1",
+  "pytest-playwright>=0.6.2",
   "pre-commit>=4.5.1",
   "filelock>=3.20.3",
   "virtualenv>=20.36.1",

tests/conftest.py

@@ -1 +1,167 @@
 """Shared pytest fixtures for all tests."""
+
+import asyncio
+from typing import Any, AsyncGenerator
+
+import pytest
+from uipath.core.tracing import UiPathTraceManager
+from uipath.runtime import (
+    UiPathExecuteOptions,
+    UiPathRuntimeEvent,
+    UiPathRuntimeFactorySettings,
+    UiPathRuntimeResult,
+    UiPathRuntimeStatus,
+    UiPathRuntimeStorageProtocol,
+    UiPathStreamOptions,
+)
+from uipath.runtime.schema import UiPathRuntimeSchema
+
+ENTRYPOINT_GREETING = "agent/greeting.py:main"
+ENTRYPOINT_NUMBERS = "agent/numbers.py:analyze"
+
+
+class _MockGreetingRuntime:
+    """Lightweight greeting runtime for tests (no OTel tracing)."""
+
+    def __init__(self, entrypoint: str = ENTRYPOINT_GREETING) -> None:
+        self.entrypoint = entrypoint
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        return UiPathRuntimeSchema(
+            filePath=self.entrypoint,
+            uniqueId="test-greeting",
+            type="agent",
+            input={
+                "type": "object",
+                "properties": {"name": {"type": "string"}},
+                "required": ["name"],
+            },
+            output={
+                "type": "object",
+                "properties": {"greeting": {"type": "string"}},
+            },
+        )
+
+    async def execute(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
+    ) -> UiPathRuntimeResult:
+        payload = input or {}
+        name = str(payload.get("name", "world"))
+        await asyncio.sleep(0.05)
+        return UiPathRuntimeResult(
+            output={"greeting": f"Hello, {name}!"},
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+        )
+
+    async def stream(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
+    ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
+        yield await self.execute(input=input, options=options)
+
+    async def dispose(self) -> None:
+        pass
+
+
+class _MockNumbersRuntime:
+    """Lightweight numbers runtime for tests (no OTel tracing)."""
+
+    def __init__(self, entrypoint: str = ENTRYPOINT_NUMBERS) -> None:
+        self.entrypoint = entrypoint
+
+    async def get_schema(self) -> UiPathRuntimeSchema:
+        return UiPathRuntimeSchema(
+            filePath=self.entrypoint,
+            uniqueId="test-numbers",
+            type="script",
+            input={
+                "type": "object",
+                "properties": {
+                    "numbers": {
+                        "type": "array",
+                        "items": {"type": "number"},
+                    },
+                    "operation": {
+                        "type": "string",
+                        "enum": ["sum", "avg", "max"],
+                        "default": "sum",
+                    },
+                },
+                "required": ["numbers"],
+            },
+            output={
+                "type": "object",
+                "properties": {
+                    "operation": {"type": "string"},
+                    "result": {"type": "number"},
+                    "count": {"type": "integer"},
+                },
+            },
+        )
+
+    async def execute(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathExecuteOptions | None = None,
+    ) -> UiPathRuntimeResult:
+        payload = input or {}
+        numbers = [float(x) for x in (payload.get("numbers") or [])]
+        operation = str(payload.get("operation", "sum")).lower()
+        await asyncio.sleep(0.05)
+
+        if operation == "avg" and numbers:
+            result = sum(numbers) / len(numbers)
+        elif operation == "max" and numbers:
+            result = max(numbers)
+        else:
+            operation = "sum"
+            result = sum(numbers)
+
+        return UiPathRuntimeResult(
+            output={"operation": operation, "result": result, "count": len(numbers)},
+            status=UiPathRuntimeStatus.SUCCESSFUL,
+        )
+
+    async def stream(
+        self,
+        input: dict[str, Any] | None = None,
+        options: UiPathStreamOptions | None = None,
+    ) -> AsyncGenerator[UiPathRuntimeEvent, None]:
+        yield await self.execute(input=input, options=options)
+
+    async def dispose(self) -> None:
+        pass
+
+
+class MockRuntimeFactory:
+    """Test runtime factory compatible with UiPathRuntimeFactoryProtocol."""
+
+    async def new_runtime(self, entrypoint: str, runtime_id: str, **kwargs):
+        if entrypoint == ENTRYPOINT_NUMBERS:
+            return _MockNumbersRuntime(entrypoint=entrypoint)
+        return _MockGreetingRuntime(entrypoint=entrypoint)
+
+    async def get_settings(self) -> UiPathRuntimeFactorySettings | None:
+        return UiPathRuntimeFactorySettings()
+
+    async def get_storage(self) -> UiPathRuntimeStorageProtocol | None:
+        return None
+
+    def discover_entrypoints(self) -> list[str]:
+        return [ENTRYPOINT_GREETING, ENTRYPOINT_NUMBERS]
+
+    async def dispose(self) -> None:
+        pass
+
+
+@pytest.fixture()
+def mock_factory():
+    return MockRuntimeFactory()
+
+
+@pytest.fixture()
+def trace_manager():
+    return UiPathTraceManager()

tests/e2e/__init__.py

@@ -0,0 +1 @@
+"""End-to-end tests for UiPath Developer Console."""

tests/e2e/conftest.py

@@ -0,0 +1,83 @@
+"""E2E-specific fixtures for Textual TUI and web server tests."""
+
+import socket
+import threading
+import time
+
+import pytest
+from uipath.core.tracing import UiPathTraceManager
+
+from tests.conftest import MockRuntimeFactory
+from uipath.dev import UiPathDeveloperConsole
+
+
+@pytest.fixture()
+def app(mock_factory, trace_manager):
+    """Create a UiPathDeveloperConsole instance for Textual pilot tests."""
+    return UiPathDeveloperConsole(
+        runtime_factory=mock_factory,
+        trace_manager=trace_manager,
+    )
+
+
+def _find_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+@pytest.fixture(scope="session")
+def live_server_url():
+    """Start a real FastAPI server in a background thread and yield its URL.
+
+    Session-scoped so the server is started only once for all web tests.
+    Uses 'live_server_url' (not 'base_url') to avoid conflicting with the
+    autouse session fixture from pytest-base-url, which would force the
+    server to start even for non-web tests.
+    """
+    try:
+        import uvicorn
+
+        from uipath.dev.server import UiPathDeveloperServer
+    except ImportError:
+        pytest.skip("server extras not installed (pip install uipath-dev[server])")
+
+    factory = MockRuntimeFactory()
+    trace_mgr = UiPathTraceManager()
+    port = _find_free_port()
+
+    server_obj = UiPathDeveloperServer(
+        runtime_factory=factory,
+        trace_manager=trace_mgr,
+        host="127.0.0.1",
+        port=port,
+        open_browser=False,
+    )
+    fastapi_app = server_obj.create_app()
+
+    config = uvicorn.Config(
+        fastapi_app,
+        host="127.0.0.1",
+        port=port,
+        log_level="warning",
+    )
+    uv_server = uvicorn.Server(config)
+
+    thread = threading.Thread(target=uv_server.run, daemon=True)
+    thread.start()
+
+    # Wait for server to be ready
+    url = f"http://127.0.0.1:{port}"
+    for _ in range(50):
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.2):
+                break
+        except OSError:
+            time.sleep(0.1)
+    else:
+        raise RuntimeError("Server did not start in time")
+
+    yield url
+
+    uv_server.should_exit = True
+    thread.join(timeout=5)