Add Playwright E2E tests for SSH Command integration

tests/playwright/README.md

@@ -0,0 +1,99 @@
+# SSH Command Playwright E2E Tests
+
+End-to-end tests for the **SSH Command** Home Assistant custom component using
+[Playwright](https://playwright.dev/python/).
+
+## Prerequisites
+
+- Python 3.11+
+- A running Home Assistant instance (default: `http://homeassistant:8123`)
+- Two SSH test servers accessible at:
+  - `ssh_docker_test:2222` (user: `foo`, password: `pass`)
+  - `ssh_docker_test:2223` (user: `foo`, password: `pass`)
+
+The SSH test servers and Home Assistant are provided by the `docker-compose.yaml`
+in the repository root.
+
+## Quick Start
+
+```bash
+# 1. Start the test environment
+docker-compose up -d
+
+# 2. Wait for Home Assistant to complete its first-run setup, then create an
+#    admin account (username: admin, password: admin) or set HA_USERNAME/HA_PASSWORD.
+
+# 3. Install the SSH Command custom component into Home Assistant:
+docker cp . homeassistant_test:/config/custom_components/ssh_command
+
+# 4. Restart Home Assistant so it loads the component:
+docker-compose restart homeassistant
+
+# 5. Install Python dependencies:
+pip install -r tests/playwright/requirements.txt
+
+# 6. Install the Playwright browser:
+playwright install chromium
+
+# 7. Run all tests:
+pytest tests/playwright/
+```
+
+## Environment Variables
+
+| Variable | Default | Description |
+|---|---|---|
+| `HOMEASSISTANT_URL` | `http://homeassistant:8123` | Home Assistant base URL |
+| `SSH_HOST` | `ssh_docker_test` | Hostname of the SSH test servers |
+| `SSH_PORT_1` | `2222` | Port for SSH Test Server 1 |
+| `SSH_PORT_2` | `2223` | Port for SSH Test Server 2 |
+| `SSH_USER` | `foo` | SSH username |
+| `SSH_PASSWORD` | `pass` | SSH password |
+| `HA_USERNAME` | `admin` | Home Assistant admin username |
+| `HA_PASSWORD` | `admin` | Home Assistant admin password |
+
+## Running on a Local Machine (outside Docker)
+
+```bash
+export HOMEASSISTANT_URL=http://localhost:8123
+export SSH_HOST=localhost
+export SSH_PORT_1=2222
+export SSH_PORT_2=2223
+export HA_USERNAME=admin
+export HA_PASSWORD=admin
+
+pytest tests/playwright/ -v
+```
+
+## Test Modules
+
+| File | What it tests |
+|---|---|
+| `test_integration_setup.py` | Adding/removing the integration via the config flow |
+| `test_command_execution.py` | Executing SSH commands against real test servers |
+| `test_services.py` | The `ssh_command.execute` HA service interface |
+| `test_frontend.py` | Home Assistant frontend pages and UI interactions |
+| `test_configuration.py` | Configuration options (timeout, auth, known hosts, …) |
+| `test_security.py` | Security properties (auth validation, unauthenticated access, …) |
+
+## Fixtures (`conftest.py`)
+
+| Fixture | Scope | Description |
+|---|---|---|
+| `playwright_instance` | session | Playwright instance |
+| `browser` | session | Headless Chromium browser |
+| `ha_base_url` | session | Configured HA URL |
+| `ha_token` | session | Long-lived HA access token |
+| `context` | function | Authenticated browser context |
+| `page` | function | Fresh page within the authenticated context |
+| `ssh_server_1` | session | Connection params for SSH server 1 |
+| `ssh_server_2` | session | Connection params for SSH server 2 |
+| `ha_api` | function | `requests.Session` for the HA REST API |
+| `ensure_integration` | function | Ensures SSH Command is set up; tears down after test |
+
+## Notes
+
+- Tests are designed to be **idempotent** – each test cleans up after itself.
+- Tests do **not** depend on each other.
+- Browser-based tests use a headless Chromium instance.
+- API-based tests call Home Assistant's REST API directly for speed and reliability.

tests/playwright/conftest.py

@@ -0,0 +1,245 @@
+"""Pytest configuration and fixtures for SSH Command Playwright E2E tests."""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from typing import Any, Generator
+
+import pytest
+import requests
+from playwright.sync_api import Browser, BrowserContext, Page, Playwright, sync_playwright
+
+# ---------------------------------------------------------------------------
+# Environment-variable driven configuration
+# ---------------------------------------------------------------------------
+
+HA_URL: str = os.environ.get("HOMEASSISTANT_URL", "http://homeassistant:8123")
+SSH_HOST: str = os.environ.get("SSH_HOST", "ssh_docker_test")
+SSH_PORT_1: int = int(os.environ.get("SSH_PORT_1", "2222"))
+SSH_PORT_2: int = int(os.environ.get("SSH_PORT_2", "2223"))
+SSH_USER: str = os.environ.get("SSH_USER", "foo")
+SSH_PASSWORD: str = os.environ.get("SSH_PASSWORD", "pass")
+
+HA_USERNAME: str = os.environ.get("HA_USERNAME", "admin")
+HA_PASSWORD: str = os.environ.get("HA_PASSWORD", "admin")
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_HA_TOKEN: str | None = None
+
+
+def get_ha_token() -> str:
+    """Obtain a long-lived Home Assistant access token via the REST API.
+
+    On the first call the token is fetched and cached for the remainder of
+    the test session.
+    """
+    global _HA_TOKEN  # noqa: PLW0603
+    if _HA_TOKEN:
+        return _HA_TOKEN
+
+    # 1. Fetch the CSRF token from the login page
+    session = requests.Session()
+    login_page = session.get(f"{HA_URL}/auth/login_flow", timeout=30)
+    login_page.raise_for_status()
+
+    # 2. Initiate the login flow
+    flow_resp = session.post(
+        f"{HA_URL}/auth/login_flow",
+        json={"client_id": HA_URL, "handler": ["homeassistant", None], "redirect_uri": f"{HA_URL}/"},
+        timeout=30,
+    )
+    flow_resp.raise_for_status()
+    flow_id = flow_resp.json()["flow_id"]
+
+    # 3. Submit credentials
+    cred_resp = session.post(
+        f"{HA_URL}/auth/login_flow/{flow_id}",
+        json={"username": HA_USERNAME, "password": HA_PASSWORD, "client_id": HA_URL},
+        timeout=30,
+    )
+    cred_resp.raise_for_status()
+    auth_code = cred_resp.json().get("result")
+
+    # 4. Exchange code for token
+    token_resp = session.post(
+        f"{HA_URL}/auth/token",
+        data={
+            "grant_type": "authorization_code",
+            "code": auth_code,
+            "client_id": HA_URL,
+        },
+        timeout=30,
+    )
+    token_resp.raise_for_status()
+    _HA_TOKEN = token_resp.json()["access_token"]
+    return _HA_TOKEN
+
+
+def wait_for_ha(timeout: int = 120) -> None:
+    """Block until Home Assistant is ready to accept connections."""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        try:
+            resp = requests.get(f"{HA_URL}/api/", timeout=5)
+            if resp.status_code in (200, 401):
+                return
+        except requests.RequestException:
+            pass
+        time.sleep(2)
+    raise RuntimeError(f"Home Assistant did not become ready within {timeout}s")
+
+
+# ---------------------------------------------------------------------------
+# Session-scoped Playwright fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session")
+def playwright_instance() -> Generator[Playwright, None, None]:
+    """Provide a session-scoped Playwright instance."""
+    with sync_playwright() as pw:
+        yield pw
+
+
+@pytest.fixture(scope="session")
+def browser(playwright_instance: Playwright) -> Generator[Browser, None, None]:
+    """Provide a session-scoped Chromium browser."""
+    browser = playwright_instance.chromium.launch(headless=True)
+    yield browser
+    browser.close()
+
+
+@pytest.fixture(scope="session")
+def ha_base_url() -> str:
+    """Return the configured Home Assistant base URL."""
+    return HA_URL
+
+
+@pytest.fixture(scope="session")
+def ha_token() -> str:
+    """Provide a valid Home Assistant long-lived access token."""
+    wait_for_ha()
+    return get_ha_token()
+
+
+# ---------------------------------------------------------------------------
+# Per-test browser context with an authenticated HA session
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def context(browser: Browser, ha_token: str) -> Generator[BrowserContext, None, None]:
+    """Provide an authenticated browser context for Home Assistant."""
+    ctx = browser.new_context(
+        base_url=HA_URL,
+        extra_http_headers={"Authorization": f"Bearer {ha_token}"},
+    )
+    # Inject the token into localStorage so the HA frontend recognises the session.
+    # Use json.dumps to safely escape all values before embedding in JS.
+    token_json = json.dumps(ha_token)
+    ha_url_json = json.dumps(HA_URL)
+    ctx.add_init_script(
+        f"""
+        window.localStorage.setItem(
+            'hassTokens',
+            JSON.stringify({{
+                access_token: {token_json},
+                token_type: 'Bearer',
+                expires_in: 1800,
+                hassUrl: {ha_url_json},
+                clientId: {ha_url_json},
+                expires: Date.now() + 1800000,
+                refresh_token: ''
+            }})
+        );
+        """
+    )
+    yield ctx
+    ctx.close()
+
+
+@pytest.fixture()
+def page(context: BrowserContext) -> Generator[Page, None, None]:
+    """Provide a fresh page within the authenticated browser context."""
+    pg = context.new_page()
+    yield pg
+    pg.close()
+
+
+# ---------------------------------------------------------------------------
+# SSH server fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session")
+def ssh_server_1() -> dict:
+    """Return connection parameters for SSH Test Server 1."""
+    return {
+        "host": SSH_HOST,
+        "port": SSH_PORT_1,
+        "username": SSH_USER,
+        "password": SSH_PASSWORD,
+    }
+
+
+@pytest.fixture(scope="session")
+def ssh_server_2() -> dict:
+    """Return connection parameters for SSH Test Server 2."""
+    return {
+        "host": SSH_HOST,
+        "port": SSH_PORT_2,
+        "username": SSH_USER,
+        "password": SSH_PASSWORD,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Integration setup / teardown helper
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def ha_api(ha_token: str) -> requests.Session:
+    """Return a requests Session pre-configured to call the HA REST API."""
+    session = requests.Session()
+    session.headers["Authorization"] = f"Bearer {ha_token}"
+    session.headers["Content-Type"] = "application/json"
+    return session
+
+
+@pytest.fixture()
+def ensure_integration(ha_api: requests.Session) -> Generator[None, None, None]:
+    """Ensure the SSH Command integration is set up before a test runs.
+
+    Tears down the integration (removes the config entry) after the test.
+    """
+    # Check whether the integration is already configured
+    resp = ha_api.get(f"{HA_URL}/api/config/config_entries/entry")
+    resp.raise_for_status()
+    entries_before = {
+        e["entry_id"]
+        for e in resp.json()
+        if e.get("domain") == "ssh_command"
+    }
+
+    # If not present, initiate the config flow
+    if not entries_before:
+        flow_resp = ha_api.post(
+            f"{HA_URL}/api/config/config_entries/flow",
+            json={"handler": "ssh_command"},
+        )
+        flow_resp.raise_for_status()
+
+    yield
+
+    # Teardown: remove any entries that were added during the test
+    resp = ha_api.get(f"{HA_URL}/api/config/config_entries/entry")
+    resp.raise_for_status()
+    for entry in resp.json():
+        if entry.get("domain") == "ssh_command" and entry["entry_id"] not in entries_before:
+            ha_api.delete(f"{HA_URL}/api/config/config_entries/entry/{entry['entry_id']}")

tests/playwright/requirements.txt

@@ -0,0 +1,4 @@
+pytest>=8.0.0,<9.0.0
+pytest-playwright>=0.5.0,<1.0.0
+playwright>=1.44.0,<2.0.0
+requests>=2.32.0,<3.0.0