diff --git a/.gitignore b/.gitignore index 66a9f61..8ba604a 100644 --- a/.gitignore +++ b/.gitignore @@ -67,3 +67,7 @@ Thumbs.db # Agent local data .letsbe-agent/ pending_results.json + +# Claude Code & MCP +.claude/ +.serena/ diff --git a/Dockerfile b/Dockerfile index 8383a6c..5a3291f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,10 +6,29 @@ WORKDIR /app # Install system dependencies # - Docker CLI for docker executor # - curl for health checks +# - Playwright browser dependencies RUN apt-get update && \ apt-get install -y --no-install-recommends \ docker-cli \ curl \ + # Playwright Chromium dependencies + libnss3 \ + libnspr4 \ + libatk1.0-0 \ + libatk-bridge2.0-0 \ + libcups2 \ + libdrm2 \ + libdbus-1-3 \ + libxkbcommon0 \ + libatspi2.0-0 \ + libxcomposite1 \ + libxdamage1 \ + libxfixes3 \ + libxrandr2 \ + libgbm1 \ + libasound2 \ + libpango-1.0-0 \ + libcairo2 \ && rm -rf /var/lib/apt/lists/* # Install Docker Compose plugin (not in Debian repos, download from Docker) @@ -24,13 +43,20 @@ COPY requirements.txt . # Install Python dependencies RUN pip install --no-cache-dir -r requirements.txt +# Install Playwright browsers (Chromium only for smaller image) +# Skip install-deps as we manually install required libs above +# and the automatic deps installer uses outdated Ubuntu package names +RUN playwright install chromium + # Copy application code COPY app/ ./app/ # Create non-root user for security RUN useradd -m -s /bin/bash agent && \ mkdir -p /home/agent/.letsbe-agent && \ - chown -R agent:agent /home/agent/.letsbe-agent + mkdir -p /opt/letsbe/playwright-artifacts && \ + chown -R agent:agent /home/agent/.letsbe-agent && \ + chown -R agent:agent /opt/letsbe/playwright-artifacts # Environment ENV PYTHONUNBUFFERED=1 diff --git a/ROADMAP.md b/ROADMAP.md index 7c6ef7d..1a35a8a 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -30,7 +30,7 @@ This document tracks Agent-specific work for the AI SysAdmin system. | DOCKER_RELOAD | Pull + up -d compose stacks | 26 | Done | | COMPOSITE | Chain multiple executors | ✅ | Done | | NEXTCLOUD | Nextcloud-specific tasks | ✅ | Done | -| PLAYWRIGHT | Browser automation (stub) | - | Stub | +| PLAYWRIGHT | Browser automation | ✅ | Done | ### Security - [x] Path sandboxing to `/opt/letsbe/` @@ -93,14 +93,47 @@ No new executors needed - existing executors support all Phase 1 tool playbooks --- -### Phase 5: Playwright Automation +### Phase 5: Playwright Browser Automation ✅ -Currently stubbed. Full implementation needs: +**Completed:** -- [ ] Playwright installation in container -- [ ] Browser automation for initial tool setup -- [ ] Screenshot capture for verification -- [ ] Form filling for admin account creation +- [x] Playwright installation in container +- [x] Scenario-based executor architecture +- [x] Domain allowlist security (mandatory) +- [x] Screenshot capture for success/failure +- [x] Artifact storage with per-task isolation +- [x] Route interception for domain blocking +- [x] Unit tests for validation logic + +**Available Scenarios:** + +| Scenario | Purpose | Status | +|----------|---------|--------| +| `echo` | Test connectivity and page load | ✅ Done | +| `nextcloud_initial_setup` | Automate Nextcloud admin setup wizard | ✅ Done | + +**Usage Example:** +```json +{ + "type": "PLAYWRIGHT", + "payload": { + "scenario": "nextcloud_initial_setup", + "inputs": { + "base_url": "https://cloud.example.com", + "admin_username": "admin", + "admin_password": "secret123" + }, + "options": { + "allowed_domains": ["cloud.example.com"], + "screenshot_on_success": true + } + } +} +``` + +**Remaining Work:** +- [ ] MCP sidecar service for exploratory browser control +- [ ] Additional tool setup scenarios (Keycloak, Poste, etc.) --- diff --git a/app/config.py b/app/config.py index 425b91f..efb788b 100644 --- a/app/config.py +++ b/app/config.py @@ -103,6 +103,24 @@ class Settings(BaseSettings): description="Path for persisting agent credentials after registration" ) + # Playwright browser automation + playwright_artifacts_dir: str = Field( + default="/opt/letsbe/playwright-artifacts", + description="Directory for screenshots, traces, and other browser artifacts" + ) + playwright_default_timeout_ms: int = Field( + default=60000, ge=5000, le=300000, + description="Default timeout for Playwright actions in milliseconds" + ) + playwright_navigation_timeout_ms: int = Field( + default=120000, ge=10000, le=300000, + description="Timeout for page navigation in milliseconds" + ) + mcp_service_url: Optional[str] = Field( + default=None, + description="URL for Playwright MCP sidecar service (for exploratory mode)" + ) + @lru_cache def get_settings() -> Settings: diff --git a/app/executors/playwright_executor.py b/app/executors/playwright_executor.py index 74b0180..7ccd8db 100644 --- a/app/executors/playwright_executor.py +++ b/app/executors/playwright_executor.py @@ -1,31 +1,50 @@ -"""Playwright browser automation executor (stub for MVP).""" +"""Playwright browser automation executor. +Executes deterministic, scenario-based browser automation tasks. +Each scenario is a reusable workflow registered in the scenario registry. +""" + +import time +import uuid +from pathlib import Path from typing import Any +from playwright.async_api import async_playwright, Route, Request + +from app.config import get_settings from app.executors.base import BaseExecutor, ExecutionResult +from app.playwright_scenarios import get_scenario, get_scenario_names, ScenarioOptions +from app.utils.validation import is_domain_allowed, validate_allowed_domains, ValidationError class PlaywrightExecutor(BaseExecutor): - """Browser automation executor using Playwright. + """Browser automation executor using Playwright scenarios. - This is a stub for MVP. Future implementation will support: - - Flow definitions with steps - - Screenshot capture - - Form filling - - Navigation - - Element interaction - - Waiting conditions + Executes pre-defined browser automation scenarios with strict security controls. + Each execution creates an isolated browser context with domain restrictions. - Payload (future): + Payload: { - "flow": [ - {"action": "goto", "url": "https://example.com"}, - {"action": "fill", "selector": "#email", "value": "test@example.com"}, - {"action": "click", "selector": "#submit"}, - {"action": "screenshot", "path": "/tmp/result.png"} - ], - "timeout": 30 + "scenario": "nextcloud_initial_setup", # Required: registered scenario name + "inputs": { # Required: scenario-specific inputs + "base_url": "https://cloud.example.com", + "admin_username": "admin", + "admin_password": "secret123" + }, + "options": { # Optional configuration + "timeout_ms": 60000, # Action timeout (default: 60000) + "screenshot_on_failure": true, # Screenshot on fail (default: true) + "screenshot_on_success": false, # Screenshot on success (default: false) + "save_trace": false, # Save trace file (default: false) + "allowed_domains": ["cloud.example.com"] # REQUIRED: domain allowlist + } } + + Security: + - allowed_domains is REQUIRED - blocks all requests to non-listed domains + - Browser runs in headless mode only (not configurable) + - Each execution gets an isolated browser context + - Artifacts are stored in per-task directories """ @property @@ -33,21 +52,278 @@ class PlaywrightExecutor(BaseExecutor): return "PLAYWRIGHT" async def execute(self, payload: dict[str, Any]) -> ExecutionResult: - """Stub: Playwright automation is not yet implemented. + """Execute a Playwright scenario. Args: - payload: Flow definition (ignored in stub) + payload: Task payload with scenario, inputs, and options Returns: - ExecutionResult indicating not implemented + ExecutionResult with scenario output and artifact paths """ - self.logger.info("playwright_stub_called", payload_keys=list(payload.keys())) + start_time = time.time() + settings = get_settings() - return ExecutionResult( - success=False, - data={ - "status": "NOT_IMPLEMENTED", - "message": "Playwright executor is planned for a future release", - }, - error="Playwright executor not yet implemented", - ) + try: + # Validate required fields + self.validate_payload(payload, ["scenario", "inputs"]) + + scenario_name = payload["scenario"] + inputs = payload["inputs"] + options_dict = payload.get("options", {}) + + # Validate allowed_domains is present + allowed_domains = options_dict.get("allowed_domains") + if not allowed_domains: + return ExecutionResult( + success=False, + data={"scenario": scenario_name}, + error="Security error: 'allowed_domains' is required in options", + duration_ms=(time.time() - start_time) * 1000, + ) + + # Validate domain patterns + try: + allowed_domains = validate_allowed_domains(allowed_domains) + except ValidationError as e: + return ExecutionResult( + success=False, + data={"scenario": scenario_name}, + error=f"Invalid allowed_domains: {e}", + duration_ms=(time.time() - start_time) * 1000, + ) + + # Get scenario from registry + scenario = get_scenario(scenario_name) + if scenario is None: + available = get_scenario_names() + return ExecutionResult( + success=False, + data={ + "scenario": scenario_name, + "available_scenarios": available, + }, + error=f"Unknown scenario: '{scenario_name}'. Available: {available}", + duration_ms=(time.time() - start_time) * 1000, + ) + + # Validate scenario inputs + missing_inputs = scenario.validate_inputs(inputs) + if missing_inputs: + return ExecutionResult( + success=False, + data={ + "scenario": scenario_name, + "missing_inputs": missing_inputs, + "required_inputs": scenario.required_inputs, + }, + error=f"Missing required inputs: {missing_inputs}", + duration_ms=(time.time() - start_time) * 1000, + ) + + # Create artifacts directory for this execution + task_id = str(uuid.uuid4())[:8] + artifacts_dir = Path(settings.playwright_artifacts_dir) / f"task-{task_id}" + artifacts_dir.mkdir(parents=True, exist_ok=True) + + # Build scenario options + scenario_options = ScenarioOptions( + timeout_ms=options_dict.get("timeout_ms", settings.playwright_default_timeout_ms), + screenshot_on_failure=options_dict.get("screenshot_on_failure", True), + screenshot_on_success=options_dict.get("screenshot_on_success", False), + save_trace=options_dict.get("save_trace", False), + allowed_domains=allowed_domains, + artifacts_dir=artifacts_dir, + ) + + self.logger.info( + "playwright_scenario_starting", + scenario=scenario_name, + task_id=task_id, + allowed_domains=allowed_domains, + ) + + # Execute scenario with browser + result = await self._run_scenario( + scenario=scenario, + inputs=inputs, + options=scenario_options, + task_id=task_id, + ) + + duration_ms = (time.time() - start_time) * 1000 + + self.logger.info( + "playwright_scenario_completed", + scenario=scenario_name, + success=result.success, + duration_ms=duration_ms, + ) + + return ExecutionResult( + success=result.success, + data={ + "scenario": scenario_name, + "result": result.data, + "screenshots": result.screenshots, + "artifacts_dir": str(artifacts_dir), + "trace_path": result.trace_path, + }, + error=result.error, + duration_ms=duration_ms, + ) + + except ValueError as e: + # Validation errors + return ExecutionResult( + success=False, + data={}, + error=str(e), + duration_ms=(time.time() - start_time) * 1000, + ) + except Exception as e: + self.logger.error( + "playwright_executor_error", + error=str(e), + error_type=type(e).__name__, + ) + return ExecutionResult( + success=False, + data={}, + error=f"Playwright executor error: {e}", + duration_ms=(time.time() - start_time) * 1000, + ) + + async def _run_scenario( + self, + scenario, + inputs: dict[str, Any], + options: ScenarioOptions, + task_id: str, + ): + """Run a scenario with browser and domain restrictions. + + Args: + scenario: The scenario instance to execute + inputs: Scenario inputs + options: Scenario options + task_id: Task identifier for logging + + Returns: + ScenarioResult from the scenario execution + """ + from app.playwright_scenarios import ScenarioResult + + settings = get_settings() + blocked_requests: list[str] = [] + + async def route_handler(route: Route, request: Request) -> None: + """Block requests to non-allowed domains.""" + url = request.url + + if is_domain_allowed(url, options.allowed_domains): + await route.continue_() + else: + blocked_requests.append(url) + self.logger.warning( + "playwright_blocked_request", + url=url, + task_id=task_id, + ) + await route.abort("blockedbyclient") + + async with async_playwright() as p: + # Launch browser in headless mode (always) + browser = await p.chromium.launch( + headless=True, + args=[ + "--no-sandbox", + "--disable-setuid-sandbox", + "--disable-dev-shm-usage", + "--disable-gpu", + ], + ) + + try: + # Create isolated context + context = await browser.new_context( + viewport={"width": 1280, "height": 720}, + user_agent="LetsBe-SysAdmin-Agent/1.0 Playwright", + ) + + # Set default timeouts + context.set_default_timeout(options.timeout_ms) + context.set_default_navigation_timeout( + settings.playwright_navigation_timeout_ms + ) + + # Start tracing if enabled + if options.save_trace and options.artifacts_dir: + await context.tracing.start( + screenshots=True, + snapshots=True, + ) + + # Apply domain restrictions via route interception + await context.route("**/*", route_handler) + + # Create page + page = await context.new_page() + + try: + # Run scenario setup hook + await scenario.setup(page, options) + + # Execute the scenario + result = await scenario.execute(page, inputs, options) + + # Take success screenshot if enabled + if options.screenshot_on_success and options.artifacts_dir: + screenshot_path = options.artifacts_dir / "success.png" + await page.screenshot(path=str(screenshot_path)) + result.screenshots.append(str(screenshot_path)) + + except Exception as e: + # Capture failure screenshot + screenshots = [] + if options.screenshot_on_failure and options.artifacts_dir: + try: + screenshot_path = options.artifacts_dir / "failure.png" + await page.screenshot(path=str(screenshot_path)) + screenshots.append(str(screenshot_path)) + except Exception as screenshot_error: + self.logger.warning( + "playwright_screenshot_failed", + error=str(screenshot_error), + ) + + result = ScenarioResult( + success=False, + data={"blocked_requests": blocked_requests}, + screenshots=screenshots, + error=str(e), + ) + + finally: + # Run scenario teardown hook + try: + await scenario.teardown(page, options) + except Exception as teardown_error: + self.logger.warning( + "playwright_teardown_error", + error=str(teardown_error), + ) + + # Stop tracing and save + if options.save_trace and options.artifacts_dir: + trace_path = options.artifacts_dir / "trace.zip" + await context.tracing.stop(path=str(trace_path)) + result.trace_path = str(trace_path) + + # Add blocked requests info + if blocked_requests: + result.data["blocked_requests"] = blocked_requests + + return result + + finally: + await browser.close() diff --git a/app/playwright_scenarios/__init__.py b/app/playwright_scenarios/__init__.py new file mode 100644 index 0000000..1bb833e --- /dev/null +++ b/app/playwright_scenarios/__init__.py @@ -0,0 +1,109 @@ +"""Playwright scenario registry. + +This module provides the central registry for all available Playwright scenarios. +Scenarios are registered at import time and looked up by name during execution. + +Usage: + from app.playwright_scenarios import get_scenario, list_scenarios + + # Get a specific scenario + scenario = get_scenario("nextcloud_initial_setup") + + # List all available scenarios + available = list_scenarios() +""" + +from typing import Optional + +from app.playwright_scenarios.base import BaseScenario, ScenarioOptions, ScenarioResult + +# Registry mapping scenario names to scenario classes +_SCENARIO_REGISTRY: dict[str, type[BaseScenario]] = {} + + +def register_scenario(scenario_class: type[BaseScenario]) -> type[BaseScenario]: + """Decorator to register a scenario class. + + Usage: + @register_scenario + class MyScenario(BaseScenario): + ... + + Args: + scenario_class: The scenario class to register + + Returns: + The scenario class (unchanged) + + Raises: + ValueError: If a scenario with the same name is already registered + """ + # Create instance to get the name + instance = scenario_class() + name = instance.name + + if name in _SCENARIO_REGISTRY: + raise ValueError( + f"Scenario '{name}' is already registered by {_SCENARIO_REGISTRY[name].__name__}" + ) + + _SCENARIO_REGISTRY[name] = scenario_class + return scenario_class + + +def get_scenario(name: str) -> Optional[BaseScenario]: + """Get a scenario instance by name. + + Args: + name: The scenario name (e.g., 'nextcloud_initial_setup') + + Returns: + Scenario instance if found, None otherwise + """ + scenario_class = _SCENARIO_REGISTRY.get(name) + if scenario_class is None: + return None + return scenario_class() + + +def list_scenarios() -> list[dict[str, str]]: + """List all registered scenarios with their metadata. + + Returns: + List of dictionaries with scenario name, description, and required inputs + """ + result = [] + for name, scenario_class in sorted(_SCENARIO_REGISTRY.items()): + instance = scenario_class() + result.append({ + "name": name, + "description": instance.description, + "required_inputs": instance.required_inputs, + "optional_inputs": instance.optional_inputs, + }) + return result + + +def get_scenario_names() -> list[str]: + """Get list of all registered scenario names. + + Returns: + Sorted list of scenario names + """ + return sorted(_SCENARIO_REGISTRY.keys()) + + +# Import scenario modules to trigger registration +# Add imports here as new scenarios are created: +from app.playwright_scenarios import echo # noqa: F401 +from app.playwright_scenarios.nextcloud import initial_setup # noqa: F401 + +__all__ = [ + "BaseScenario", + "ScenarioOptions", + "ScenarioResult", + "register_scenario", + "get_scenario", + "list_scenarios", + "get_scenario_names", +] diff --git a/app/playwright_scenarios/base.py b/app/playwright_scenarios/base.py new file mode 100644 index 0000000..011e942 --- /dev/null +++ b/app/playwright_scenarios/base.py @@ -0,0 +1,162 @@ +"""Base classes for Playwright scenario execution. + +Scenarios are deterministic, reusable browser automation sequences +that execute specific UI workflows against tenant applications. +""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +from playwright.async_api import Page + + +@dataclass +class ScenarioOptions: + """Configuration options for scenario execution. + + Attributes: + timeout_ms: Default timeout for actions in milliseconds + screenshot_on_failure: Capture screenshot when scenario fails + screenshot_on_success: Capture screenshot when scenario succeeds + save_trace: Save Playwright trace for debugging + allowed_domains: List of domains the scenario can access (REQUIRED for security) + artifacts_dir: Directory to save screenshots and traces + """ + timeout_ms: int = 60000 + screenshot_on_failure: bool = True + screenshot_on_success: bool = False + save_trace: bool = False + allowed_domains: list[str] = field(default_factory=list) + artifacts_dir: Optional[Path] = None + + def __post_init__(self) -> None: + if self.artifacts_dir and isinstance(self.artifacts_dir, str): + self.artifacts_dir = Path(self.artifacts_dir) + + +@dataclass +class ScenarioResult: + """Result of a scenario execution. + + Attributes: + success: Whether the scenario completed successfully + data: Scenario-specific result data + screenshots: List of paths to captured screenshots + error: Error message if scenario failed + trace_path: Path to trace file if tracing was enabled + """ + success: bool + data: dict[str, Any] + screenshots: list[str] = field(default_factory=list) + error: Optional[str] = None + trace_path: Optional[str] = None + + +class BaseScenario(ABC): + """Abstract base class for Playwright scenarios. + + Each scenario implements a specific UI automation workflow. + Scenarios are registered by name and dispatched by the PlaywrightExecutor. + + Example implementation: + class NextcloudInitialSetup(BaseScenario): + @property + def name(self) -> str: + return "nextcloud_initial_setup" + + @property + def required_inputs(self) -> list[str]: + return ["base_url", "admin_username", "admin_password"] + + async def execute(self, page, inputs, options) -> ScenarioResult: + # Perform setup steps... + return ScenarioResult(success=True, data={"setup": "complete"}) + """ + + @property + @abstractmethod + def name(self) -> str: + """Unique name identifying this scenario. + + This name is used in task payloads to select the scenario. + Convention: lowercase_with_underscores (e.g., 'nextcloud_initial_setup') + """ + ... + + @property + @abstractmethod + def required_inputs(self) -> list[str]: + """List of required input keys for this scenario. + + The executor validates that all required inputs are present + before executing the scenario. + """ + ... + + @property + def optional_inputs(self) -> list[str]: + """List of optional input keys for this scenario. + + Override this property to declare optional inputs with defaults. + """ + return [] + + @property + def description(self) -> str: + """Human-readable description of what this scenario does. + + Override this property to provide documentation. + """ + return f"Scenario: {self.name}" + + @abstractmethod + async def execute( + self, + page: Page, + inputs: dict[str, Any], + options: ScenarioOptions, + ) -> ScenarioResult: + """Execute the scenario against the provided page. + + Args: + page: Playwright Page object with domain restrictions applied + inputs: Dictionary of input values (validated by executor) + options: Scenario options including timeout and artifact settings + + Returns: + ScenarioResult with success status and any result data + + Note: + - Domain restrictions are already enforced by the executor + - Screenshots on failure are handled by the executor + - Focus on the business logic of the UI workflow + """ + ... + + async def setup(self, page: Page, options: ScenarioOptions) -> None: + """Optional setup hook called before execute(). + + Override to perform setup actions like setting viewport size, + configuring page settings, etc. + """ + pass + + async def teardown(self, page: Page, options: ScenarioOptions) -> None: + """Optional teardown hook called after execute(). + + Override to perform cleanup actions. Called even if execute() fails. + """ + pass + + def validate_inputs(self, inputs: dict[str, Any]) -> list[str]: + """Validate inputs and return list of missing required keys. + + Args: + inputs: Dictionary of inputs to validate + + Returns: + List of missing required input keys (empty if all present) + """ + return [key for key in self.required_inputs if key not in inputs] diff --git a/app/playwright_scenarios/echo.py b/app/playwright_scenarios/echo.py new file mode 100644 index 0000000..cd8a98b --- /dev/null +++ b/app/playwright_scenarios/echo.py @@ -0,0 +1,120 @@ +"""Echo scenario for testing Playwright executor. + +This simple scenario navigates to a URL and verifies the page loads. +Useful for testing the Playwright infrastructure without complex workflows. +""" + +from typing import Any + +from playwright.async_api import Page + +from app.playwright_scenarios import register_scenario +from app.playwright_scenarios.base import BaseScenario, ScenarioOptions, ScenarioResult + + +@register_scenario +class EchoScenario(BaseScenario): + """Simple echo scenario for testing Playwright executor. + + This scenario navigates to a URL and returns basic page information. + Useful for verifying: + - Playwright is installed and working + - Domain restrictions are enforced + - Screenshots are captured correctly + + Required inputs: + url: The URL to navigate to + + Optional inputs: + wait_for_selector: CSS selector to wait for (default: body) + expected_title: Expected page title (optional validation) + + Result data: + title: Page title after load + url: Final URL after any redirects + content_length: Approximate content length + """ + + @property + def name(self) -> str: + return "echo" + + @property + def required_inputs(self) -> list[str]: + return ["url"] + + @property + def optional_inputs(self) -> list[str]: + return ["wait_for_selector", "expected_title"] + + @property + def description(self) -> str: + return "Navigate to URL and return page info (test scenario)" + + async def execute( + self, + page: Page, + inputs: dict[str, Any], + options: ScenarioOptions, + ) -> ScenarioResult: + """Navigate to URL and capture page information. + + Args: + page: Playwright Page object + inputs: Scenario inputs (url, optional wait_for_selector) + options: Scenario options + + Returns: + ScenarioResult with page information + """ + url = inputs["url"] + wait_for_selector = inputs.get("wait_for_selector", "body") + expected_title = inputs.get("expected_title") + + screenshots = [] + result_data = {} + + try: + # Navigate to the URL + response = await page.goto(url, wait_until="networkidle") + + # Wait for specified selector + if wait_for_selector: + await page.wait_for_selector(wait_for_selector, timeout=options.timeout_ms) + + # Collect page information + result_data = { + "title": await page.title(), + "url": page.url, + "status_code": response.status if response else None, + "content_length": len(await page.content()), + } + + # Validate title if expected + if expected_title and result_data["title"] != expected_title: + return ScenarioResult( + success=False, + data=result_data, + screenshots=screenshots, + error=f"Title mismatch: expected '{expected_title}', got '{result_data['title']}'", + ) + + # Take screenshot if requested + if options.screenshot_on_success and options.artifacts_dir: + screenshot_path = options.artifacts_dir / "echo_result.png" + await page.screenshot(path=str(screenshot_path)) + screenshots.append(str(screenshot_path)) + + return ScenarioResult( + success=True, + data=result_data, + screenshots=screenshots, + ) + + except Exception as e: + return ScenarioResult( + success=False, + data=result_data, + screenshots=screenshots, + error=f"Echo scenario failed: {str(e)}", + ) diff --git a/app/playwright_scenarios/nextcloud/__init__.py b/app/playwright_scenarios/nextcloud/__init__.py new file mode 100644 index 0000000..c92ad1b --- /dev/null +++ b/app/playwright_scenarios/nextcloud/__init__.py @@ -0,0 +1,5 @@ +"""Nextcloud browser automation scenarios.""" + +from app.playwright_scenarios.nextcloud.initial_setup import NextcloudInitialSetup + +__all__ = ["NextcloudInitialSetup"] diff --git a/app/playwright_scenarios/nextcloud/initial_setup.py b/app/playwright_scenarios/nextcloud/initial_setup.py new file mode 100644 index 0000000..9e2b4b4 --- /dev/null +++ b/app/playwright_scenarios/nextcloud/initial_setup.py @@ -0,0 +1,231 @@ +"""Nextcloud initial setup scenario. + +Automates the first-time setup wizard for a fresh Nextcloud installation. +This scenario: +1. Navigates to the Nextcloud instance +2. Creates the admin account +3. Optionally skips recommended apps installation +4. Verifies successful login to the dashboard +""" + +from typing import Any + +from playwright.async_api import Page, expect + +from app.playwright_scenarios import register_scenario +from app.playwright_scenarios.base import BaseScenario, ScenarioOptions, ScenarioResult + + +@register_scenario +class NextcloudInitialSetup(BaseScenario): + """Automate Nextcloud first-time setup wizard. + + This scenario handles the initial admin account creation when + Nextcloud is freshly installed. It's idempotent - if setup is + already complete, it will detect this and succeed. + + Required inputs: + base_url: The Nextcloud instance URL (e.g., https://cloud.example.com) + admin_username: Username for the admin account + admin_password: Password for the admin account + + Optional inputs: + skip_recommended_apps: Skip the recommended apps step (default: True) + + Result data: + admin_created: Whether a new admin was created (False if already setup) + login_successful: Whether login to dashboard succeeded + setup_skipped: True if Nextcloud was already configured + """ + + @property + def name(self) -> str: + return "nextcloud_initial_setup" + + @property + def required_inputs(self) -> list[str]: + return ["base_url", "admin_username", "admin_password"] + + @property + def optional_inputs(self) -> list[str]: + return ["skip_recommended_apps"] + + @property + def description(self) -> str: + return "Automate Nextcloud first-time admin setup wizard" + + async def execute( + self, + page: Page, + inputs: dict[str, Any], + options: ScenarioOptions, + ) -> ScenarioResult: + """Execute the Nextcloud initial setup. + + Args: + page: Playwright Page object + inputs: Scenario inputs (base_url, admin_username, admin_password) + options: Scenario options + + Returns: + ScenarioResult with setup status + """ + base_url = inputs["base_url"].rstrip("/") + admin_username = inputs["admin_username"] + admin_password = inputs["admin_password"] + skip_recommended_apps = inputs.get("skip_recommended_apps", True) + + screenshots = [] + result_data = { + "admin_created": False, + "login_successful": False, + "setup_skipped": False, + } + + try: + # Navigate to Nextcloud + await page.goto(base_url, wait_until="networkidle") + + # Check if we're on the setup page or login page + current_url = page.url + + # Detect if setup is already complete (redirects to login) + if "/login" in current_url or await page.locator('input[name="user"]').count() > 0: + # Already configured, try to login + result_data["setup_skipped"] = True + login_success = await self._try_login( + page, admin_username, admin_password + ) + result_data["login_successful"] = login_success + + return ScenarioResult( + success=login_success, + data=result_data, + screenshots=screenshots, + error=None if login_success else "Login failed - check credentials", + ) + + # We're on the setup page - create admin account + # Wait for the setup form to be visible + admin_user_input = page.locator('input[id="adminlogin"], input[name="adminlogin"]') + await admin_user_input.wait_for(state="visible", timeout=10000) + + # Fill in admin credentials + await admin_user_input.fill(admin_username) + + admin_pass_input = page.locator('input[id="adminpass"], input[name="adminpass"]') + await admin_pass_input.fill(admin_password) + + # Check for data directory input (may or may not be present) + data_dir_input = page.locator('input[id="directory"]') + if await data_dir_input.count() > 0 and await data_dir_input.is_visible(): + # Keep default data directory + pass + + # Click install/finish setup button + # Nextcloud uses various button texts depending on version + install_button = page.locator( + 'input[type="submit"][value*="Install"], ' + 'input[type="submit"][value*="Finish"], ' + 'button:has-text("Install"), ' + 'button:has-text("Finish setup")' + ) + await install_button.click() + + # Wait for installation to complete (this can take a while) + # Look for either dashboard or recommended apps screen + try: + await page.wait_for_url( + lambda url: "/apps" in url or "/index.php" in url or "dashboard" in url.lower(), + timeout=120000, # 2 minutes for installation + ) + except Exception: + # May be on recommended apps screen + pass + + result_data["admin_created"] = True + + # Handle recommended apps screen if present + if skip_recommended_apps: + skip_button = page.locator( + 'button:has-text("Skip"), ' + 'a:has-text("Skip"), ' + '.skip-button' + ) + if await skip_button.count() > 0: + await skip_button.first.click() + await page.wait_for_load_state("networkidle") + + # Verify we're logged in by checking for user menu or dashboard elements + dashboard_indicators = page.locator( + '#user-menu, ' + '.user-menu, ' + '[data-id="dashboard"], ' + '#nextcloud, ' + '.app-dashboard' + ) + + try: + await dashboard_indicators.first.wait_for(state="visible", timeout=30000) + result_data["login_successful"] = True + except Exception: + # Try one more check - look for any indication we're logged in + if await page.locator('.header-menu').count() > 0: + result_data["login_successful"] = True + + # Take a screenshot of the final state if requested + if options.screenshot_on_success and options.artifacts_dir: + screenshot_path = options.artifacts_dir / "setup_complete.png" + await page.screenshot(path=str(screenshot_path)) + screenshots.append(str(screenshot_path)) + + success = result_data["admin_created"] and result_data["login_successful"] + return ScenarioResult( + success=success, + data=result_data, + screenshots=screenshots, + error=None if success else "Setup completed but verification failed", + ) + + except Exception as e: + return ScenarioResult( + success=False, + data=result_data, + screenshots=screenshots, + error=f"Nextcloud setup failed: {str(e)}", + ) + + async def _try_login(self, page: Page, username: str, password: str) -> bool: + """Attempt to login to an already-configured Nextcloud. + + Args: + page: Playwright Page object (should be on login page) + username: Username to login with + password: Password to login with + + Returns: + True if login succeeded, False otherwise + """ + try: + # Fill login form + await page.locator('input[name="user"]').fill(username) + await page.locator('input[name="password"]').fill(password) + + # Submit login + await page.locator('input[type="submit"], button[type="submit"]').click() + + # Wait for redirect to dashboard + await page.wait_for_url( + lambda url: "/login" not in url, + timeout=30000, + ) + + # Check for login error message + error_msg = page.locator('.warning, .error, [class*="error"]') + if await error_msg.count() > 0 and await error_msg.first.is_visible(): + return False + + return True + + except Exception: + return False diff --git a/app/utils/validation.py b/app/utils/validation.py index a442aad..4d02275 100644 --- a/app/utils/validation.py +++ b/app/utils/validation.py @@ -268,3 +268,135 @@ def validate_env_key(key: str) -> bool: ) return True + + +def is_domain_allowed(url: str, allowed_domains: list[str]) -> bool: + """Check if a URL's domain is in the allowed list. + + Supports: + - Exact domain match: "cloud.example.com" + - Wildcard subdomain: "*.example.com" (matches any subdomain) + - Port specification: "cloud.example.com:8443" + + Args: + url: The URL to check + allowed_domains: List of allowed domain patterns + + Returns: + True if the domain is allowed, False otherwise + + Examples: + >>> is_domain_allowed("https://cloud.example.com/path", ["cloud.example.com"]) + True + >>> is_domain_allowed("https://sub.example.com", ["*.example.com"]) + True + >>> is_domain_allowed("https://evil.com", ["example.com"]) + False + """ + from urllib.parse import urlparse + + if not url or not allowed_domains: + return False + + try: + parsed = urlparse(url) + url_host = parsed.netloc.lower() + + # Handle URLs without scheme (shouldn't happen, but be safe) + if not url_host and parsed.path: + # URL might be like "example.com/path" without scheme + url_host = parsed.path.split("/")[0].lower() + + if not url_host: + return False + + # Extract port if present in URL + if ":" in url_host: + url_domain, url_port = url_host.rsplit(":", 1) + else: + url_domain = url_host + url_port = None + + for pattern in allowed_domains: + pattern = pattern.lower().strip() + + # Extract port from pattern if present + if ":" in pattern and not pattern.startswith("*."): + pattern_domain, pattern_port = pattern.rsplit(":", 1) + elif ":" in pattern: + # Handle "*.example.com:8443" + parts = pattern.split(":") + pattern_domain = parts[0] + pattern_port = parts[1] if len(parts) > 1 else None + else: + pattern_domain = pattern + pattern_port = None + + # If pattern specifies a port, URL must match that port + if pattern_port and url_port != pattern_port: + continue + + # Wildcard subdomain match + if pattern_domain.startswith("*."): + suffix = pattern_domain[2:] # Remove "*." + # Match the suffix or the exact domain without subdomain + if url_domain == suffix or url_domain.endswith("." + suffix): + return True + else: + # Exact match + if url_domain == pattern_domain: + return True + + return False + + except Exception: + return False + + +def validate_allowed_domains(domains: list[str]) -> list[str]: + """Validate and normalize a list of allowed domains. + + Args: + domains: List of domain patterns to validate + + Returns: + List of normalized domain patterns + + Raises: + ValidationError: If any domain pattern is invalid + """ + if not domains: + raise ValidationError("allowed_domains cannot be empty") + + normalized = [] + for domain in domains: + domain = domain.strip().lower() + + if not domain: + raise ValidationError("Empty domain in allowed_domains list") + + # Basic format validation + if domain.startswith("http://") or domain.startswith("https://"): + raise ValidationError( + f"Domain should not include protocol: {domain}. " + "Use 'example.com' not 'https://example.com'" + ) + + # Wildcard validation + if "*" in domain: + if not domain.startswith("*."): + raise ValidationError( + f"Invalid wildcard pattern: {domain}. " + "Wildcards must be at the start: '*.example.com'" + ) + # Ensure there's something after the wildcard + suffix = domain[2:] + if "." not in suffix or suffix.startswith("."): + raise ValidationError( + f"Invalid wildcard pattern: {domain}. " + "Must have a valid domain after '*.' like '*.example.com'" + ) + + normalized.append(domain) + + return normalized diff --git a/docker-compose.yml b/docker-compose.yml index 224f513..cd94895 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -37,12 +37,19 @@ services: - MAX_FILE_SIZE=${MAX_FILE_SIZE:-10485760} - SHELL_TIMEOUT=${SHELL_TIMEOUT:-60} + # Playwright browser automation + - PLAYWRIGHT_ARTIFACTS_DIR=${PLAYWRIGHT_ARTIFACTS_DIR:-/opt/letsbe/playwright-artifacts} + - PLAYWRIGHT_DEFAULT_TIMEOUT_MS=${PLAYWRIGHT_DEFAULT_TIMEOUT_MS:-60000} + - PLAYWRIGHT_NAVIGATION_TIMEOUT_MS=${PLAYWRIGHT_NAVIGATION_TIMEOUT_MS:-120000} + volumes: # Docker socket for docker executor - /var/run/docker.sock:/var/run/docker.sock # Hot reload in development - ./app:/app/app:ro + - ./tests:/app/tests:ro + - ./pytest.ini:/app/pytest.ini:ro # Host directory mounts for real infrastructure access - /opt/letsbe/env:/opt/letsbe/env @@ -52,22 +59,31 @@ services: # Pending results persistence - agent_home:/home/agent/.letsbe-agent + # Playwright artifacts storage + - playwright_artifacts:/opt/letsbe/playwright-artifacts + + # Security options for Chromium sandboxing + security_opt: + - seccomp=unconfined + # Run as root for Docker socket access in dev # In production, use Docker group membership instead user: root restart: unless-stopped - # Resource limits + # Resource limits (increased for Playwright browser automation) deploy: resources: limits: - cpus: '0.5' - memory: 256M + cpus: '1.5' + memory: 1G reservations: - cpus: '0.1' - memory: 64M + cpus: '0.25' + memory: 256M volumes: agent_home: name: letsbe-agent-home + playwright_artifacts: + name: letsbe-playwright-artifacts diff --git a/requirements.txt b/requirements.txt index 7c061bd..34d6274 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,6 +4,9 @@ python-dotenv>=1.0.0 pydantic>=2.0.0 pydantic-settings>=2.0.0 +# Browser automation +playwright==1.49.1 + # Testing pytest>=8.0.0 pytest-asyncio>=0.23.0 diff --git a/tests/executors/test_playwright_executor.py b/tests/executors/test_playwright_executor.py new file mode 100644 index 0000000..bda0ae9 --- /dev/null +++ b/tests/executors/test_playwright_executor.py @@ -0,0 +1,450 @@ +"""Unit tests for PlaywrightExecutor. + +These tests focus on validation logic without launching browsers. +Browser-based integration tests are skipped by default (SKIP_BROWSER_TESTS=true). +""" + +import os +import sys +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# Mock playwright module before any imports that might use it +sys.modules["playwright"] = MagicMock() +sys.modules["playwright.async_api"] = MagicMock() + +# Patch the logger before importing the executor +with patch("app.utils.logger.get_logger", return_value=MagicMock()): + from app.utils.validation import is_domain_allowed, validate_allowed_domains, ValidationError + + +class TestDomainValidation: + """Test domain allowlist validation functions.""" + + # ==================== is_domain_allowed Tests ==================== + + def test_exact_domain_match(self): + """Test exact domain matching.""" + assert is_domain_allowed("https://cloud.example.com/path", ["cloud.example.com"]) is True + assert is_domain_allowed("https://cloud.example.com", ["cloud.example.com"]) is True + assert is_domain_allowed("http://cloud.example.com", ["cloud.example.com"]) is True + + def test_exact_domain_no_match(self): + """Test exact domain non-matching.""" + assert is_domain_allowed("https://evil.com/path", ["cloud.example.com"]) is False + assert is_domain_allowed("https://sub.cloud.example.com", ["cloud.example.com"]) is False + + def test_wildcard_subdomain_match(self): + """Test wildcard subdomain matching.""" + assert is_domain_allowed("https://sub.example.com", ["*.example.com"]) is True + assert is_domain_allowed("https://deep.sub.example.com", ["*.example.com"]) is True + assert is_domain_allowed("https://example.com", ["*.example.com"]) is True + + def test_wildcard_subdomain_no_match(self): + """Test wildcard subdomain non-matching.""" + assert is_domain_allowed("https://evil.com", ["*.example.com"]) is False + assert is_domain_allowed("https://example.org", ["*.example.com"]) is False + + def test_domain_with_port(self): + """Test domain matching with port specification.""" + assert is_domain_allowed("https://cloud.example.com:8443/path", ["cloud.example.com:8443"]) is True + assert is_domain_allowed("https://cloud.example.com:8443", ["cloud.example.com:8443"]) is True + # Wrong port should not match + assert is_domain_allowed("https://cloud.example.com:9000", ["cloud.example.com:8443"]) is False + # No port in URL should not match port-specific pattern + assert is_domain_allowed("https://cloud.example.com", ["cloud.example.com:8443"]) is False + + def test_multiple_allowed_domains(self): + """Test with multiple allowed domains.""" + allowed = ["cloud.example.com", "mail.example.com", "*.internal.com"] + assert is_domain_allowed("https://cloud.example.com", allowed) is True + assert is_domain_allowed("https://mail.example.com", allowed) is True + assert is_domain_allowed("https://app.internal.com", allowed) is True + assert is_domain_allowed("https://evil.com", allowed) is False + + def test_empty_inputs(self): + """Test with empty inputs.""" + assert is_domain_allowed("", ["example.com"]) is False + assert is_domain_allowed("https://example.com", []) is False + assert is_domain_allowed("", []) is False + + def test_case_insensitive(self): + """Test case-insensitive matching.""" + assert is_domain_allowed("https://Cloud.Example.COM", ["cloud.example.com"]) is True + assert is_domain_allowed("https://cloud.example.com", ["Cloud.Example.COM"]) is True + + # ==================== validate_allowed_domains Tests ==================== + + def test_validate_valid_domains(self): + """Test validation of valid domain patterns.""" + result = validate_allowed_domains(["example.com", "cloud.example.com"]) + assert result == ["example.com", "cloud.example.com"] + + def test_validate_wildcard_domains(self): + """Test validation of wildcard domain patterns.""" + result = validate_allowed_domains(["*.example.com", "*.internal.org"]) + assert result == ["*.example.com", "*.internal.org"] + + def test_validate_with_ports(self): + """Test validation of domains with ports.""" + result = validate_allowed_domains(["example.com:8080", "cloud.example.com:8443"]) + assert result == ["example.com:8080", "cloud.example.com:8443"] + + def test_validate_empty_list_raises(self): + """Test that empty list raises ValidationError.""" + with pytest.raises(ValidationError, match="cannot be empty"): + validate_allowed_domains([]) + + def test_validate_protocol_raises(self): + """Test that domains with protocol raise ValidationError.""" + with pytest.raises(ValidationError, match="should not include protocol"): + validate_allowed_domains(["https://example.com"]) + + def test_validate_invalid_wildcard_raises(self): + """Test that invalid wildcards raise ValidationError.""" + with pytest.raises(ValidationError, match="Wildcards must be at the start"): + validate_allowed_domains(["example.*.com"]) + + with pytest.raises(ValidationError, match="Wildcards must be at the start"): + validate_allowed_domains(["*"]) + + def test_validate_normalizes_case(self): + """Test that validation normalizes to lowercase.""" + result = validate_allowed_domains(["Example.COM", "CLOUD.Example.com"]) + assert result == ["example.com", "cloud.example.com"] + + +class TestPlaywrightExecutor: + """Test suite for PlaywrightExecutor.""" + + @pytest.fixture + def executor(self): + """Create executor instance with mocked logger.""" + with patch("app.executors.base.get_logger", return_value=MagicMock()): + from app.executors.playwright_executor import PlaywrightExecutor + return PlaywrightExecutor() + + @pytest.fixture + def mock_settings(self, tmp_path): + """Mock settings with temporary paths.""" + settings = MagicMock() + settings.playwright_artifacts_dir = str(tmp_path / "playwright-artifacts") + settings.playwright_default_timeout_ms = 60000 + settings.playwright_navigation_timeout_ms = 120000 + return settings + + # ==================== Validation Error Tests ==================== + + @pytest.mark.asyncio + async def test_missing_scenario_field(self, executor, mock_settings): + """Test that missing scenario field returns error.""" + with patch("app.executors.playwright_executor.get_settings", return_value=mock_settings): + result = await executor.execute({ + "inputs": {"base_url": "https://example.com"}, + "options": {"allowed_domains": ["example.com"]} + }) + + assert result.success is False + assert "Missing required fields: scenario" in result.error + + @pytest.mark.asyncio + async def test_missing_inputs_field(self, executor, mock_settings): + """Test that missing inputs field returns error.""" + with patch("app.executors.playwright_executor.get_settings", return_value=mock_settings): + result = await executor.execute({ + "scenario": "test_scenario", + "options": {"allowed_domains": ["example.com"]} + }) + + assert result.success is False + assert "Missing required fields: inputs" in result.error + + @pytest.mark.asyncio + async def test_missing_allowed_domains(self, executor, mock_settings): + """Test that missing allowed_domains returns security error.""" + with patch("app.executors.playwright_executor.get_settings", return_value=mock_settings): + result = await executor.execute({ + "scenario": "test_scenario", + "inputs": {"base_url": "https://example.com"}, + "options": {} + }) + + assert result.success is False + assert "allowed_domains" in result.error + assert "required" in result.error.lower() + + @pytest.mark.asyncio + async def test_missing_options_means_no_domains(self, executor, mock_settings): + """Test that missing options dict means no allowed_domains.""" + with patch("app.executors.playwright_executor.get_settings", return_value=mock_settings): + result = await executor.execute({ + "scenario": "test_scenario", + "inputs": {"base_url": "https://example.com"}, + }) + + assert result.success is False + assert "allowed_domains" in result.error + + @pytest.mark.asyncio + async def test_invalid_allowed_domains_format(self, executor, mock_settings): + """Test that invalid domain patterns return error.""" + with patch("app.executors.playwright_executor.get_settings", return_value=mock_settings): + result = await executor.execute({ + "scenario": "test_scenario", + "inputs": {"base_url": "https://example.com"}, + "options": {"allowed_domains": ["https://example.com"]} # Protocol not allowed + }) + + assert result.success is False + assert "Invalid allowed_domains" in result.error + + @pytest.mark.asyncio + async def test_unknown_scenario(self, executor, mock_settings): + """Test that unknown scenario returns error with available list.""" + with patch("app.executors.playwright_executor.get_settings", return_value=mock_settings): + result = await executor.execute({ + "scenario": "nonexistent_scenario", + "inputs": {"base_url": "https://example.com"}, + "options": {"allowed_domains": ["example.com"]} + }) + + assert result.success is False + assert "Unknown scenario" in result.error + assert "nonexistent_scenario" in result.error + assert "available_scenarios" in result.data + + # ==================== Task Type Tests ==================== + + def test_task_type_is_playwright(self, executor): + """Test that executor reports correct task type.""" + assert executor.task_type == "PLAYWRIGHT" + + +class TestScenarioRegistry: + """Test scenario registration and lookup.""" + + def test_register_and_get_scenario(self): + """Test registering and retrieving a scenario.""" + from app.playwright_scenarios import get_scenario, get_scenario_names, _SCENARIO_REGISTRY + from app.playwright_scenarios import register_scenario, BaseScenario, ScenarioResult + + # Clear registry for clean test + original_registry = _SCENARIO_REGISTRY.copy() + _SCENARIO_REGISTRY.clear() + + try: + @register_scenario + class TestScenario(BaseScenario): + @property + def name(self) -> str: + return "test_scenario" + + @property + def required_inputs(self) -> list[str]: + return ["base_url"] + + async def execute(self, page, inputs, options) -> ScenarioResult: + return ScenarioResult(success=True, data={}) + + # Should find the registered scenario + scenario = get_scenario("test_scenario") + assert scenario is not None + assert scenario.name == "test_scenario" + + # Should be in the list + names = get_scenario_names() + assert "test_scenario" in names + + finally: + # Restore original registry + _SCENARIO_REGISTRY.clear() + _SCENARIO_REGISTRY.update(original_registry) + + def test_get_unknown_scenario_returns_none(self): + """Test that unknown scenario lookup returns None.""" + from app.playwright_scenarios import get_scenario + + scenario = get_scenario("definitely_does_not_exist_xyz123") + assert scenario is None + + +class TestScenarioOptions: + """Test ScenarioOptions dataclass.""" + + def test_default_values(self): + """Test default option values.""" + from app.playwright_scenarios import ScenarioOptions + + options = ScenarioOptions() + assert options.timeout_ms == 60000 + assert options.screenshot_on_failure is True + assert options.screenshot_on_success is False + assert options.save_trace is False + assert options.allowed_domains == [] + assert options.artifacts_dir is None + + def test_custom_values(self): + """Test custom option values.""" + from app.playwright_scenarios import ScenarioOptions + + options = ScenarioOptions( + timeout_ms=30000, + screenshot_on_failure=False, + screenshot_on_success=True, + save_trace=True, + allowed_domains=["example.com"], + artifacts_dir=Path("/tmp/artifacts"), + ) + assert options.timeout_ms == 30000 + assert options.screenshot_on_failure is False + assert options.screenshot_on_success is True + assert options.save_trace is True + assert options.allowed_domains == ["example.com"] + assert options.artifacts_dir == Path("/tmp/artifacts") + + def test_string_artifacts_dir_converted(self): + """Test that string artifacts_dir is converted to Path.""" + from app.playwright_scenarios import ScenarioOptions + + options = ScenarioOptions(artifacts_dir="/tmp/artifacts") + assert isinstance(options.artifacts_dir, Path) + # Path separators differ by OS, just check it's a valid Path + assert options.artifacts_dir == Path("/tmp/artifacts") + + +class TestScenarioResult: + """Test ScenarioResult dataclass.""" + + def test_success_result(self): + """Test successful result creation.""" + from app.playwright_scenarios import ScenarioResult + + result = ScenarioResult( + success=True, + data={"setup": "complete"}, + screenshots=["/tmp/success.png"], + ) + assert result.success is True + assert result.data == {"setup": "complete"} + assert result.screenshots == ["/tmp/success.png"] + assert result.error is None + + def test_failure_result(self): + """Test failure result creation.""" + from app.playwright_scenarios import ScenarioResult + + result = ScenarioResult( + success=False, + data={}, + error="Element not found", + ) + assert result.success is False + assert result.error == "Element not found" + + +class TestBaseScenario: + """Test BaseScenario ABC.""" + + def test_validate_inputs_missing(self): + """Test input validation returns missing keys.""" + from app.playwright_scenarios import BaseScenario, ScenarioResult + + class TestScenario(BaseScenario): + @property + def name(self) -> str: + return "test" + + @property + def required_inputs(self) -> list[str]: + return ["base_url", "username", "password"] + + async def execute(self, page, inputs, options) -> ScenarioResult: + return ScenarioResult(success=True, data={}) + + scenario = TestScenario() + + # Missing all inputs + missing = scenario.validate_inputs({}) + assert "base_url" in missing + assert "username" in missing + assert "password" in missing + + # Missing some inputs + missing = scenario.validate_inputs({"base_url": "https://example.com"}) + assert "base_url" not in missing + assert "username" in missing + assert "password" in missing + + # All inputs present + missing = scenario.validate_inputs({ + "base_url": "https://example.com", + "username": "admin", + "password": "secret", + }) + assert missing == [] + + def test_default_optional_inputs(self): + """Test default optional inputs is empty.""" + from app.playwright_scenarios import BaseScenario, ScenarioResult + + class TestScenario(BaseScenario): + @property + def name(self) -> str: + return "test" + + @property + def required_inputs(self) -> list[str]: + return ["base_url"] + + async def execute(self, page, inputs, options) -> ScenarioResult: + return ScenarioResult(success=True, data={}) + + scenario = TestScenario() + assert scenario.optional_inputs == [] + + def test_default_description(self): + """Test default description uses name.""" + from app.playwright_scenarios import BaseScenario, ScenarioResult + + class TestScenario(BaseScenario): + @property + def name(self) -> str: + return "my_test_scenario" + + @property + def required_inputs(self) -> list[str]: + return [] + + async def execute(self, page, inputs, options) -> ScenarioResult: + return ScenarioResult(success=True, data={}) + + scenario = TestScenario() + assert "my_test_scenario" in scenario.description + + +# Skip browser tests by default +SKIP_BROWSER_TESTS = os.environ.get("SKIP_BROWSER_TESTS", "true").lower() == "true" + + +@pytest.mark.skipif(SKIP_BROWSER_TESTS, reason="Browser tests skipped (set SKIP_BROWSER_TESTS=false to run)") +class TestPlaywrightExecutorIntegration: + """Integration tests that require a real browser. + + These tests are skipped by default. Set SKIP_BROWSER_TESTS=false to run. + """ + + @pytest.fixture + def executor(self): + """Create executor instance.""" + with patch("app.executors.base.get_logger", return_value=MagicMock()): + from app.executors.playwright_executor import PlaywrightExecutor + return PlaywrightExecutor() + + @pytest.mark.asyncio + async def test_domain_blocking_in_browser(self, executor, tmp_path): + """Test that blocked domains are actually blocked in browser.""" + # This would require a mock HTTP server and real browser + # Implementation deferred to manual testing + pass