diff --git a/requirements-ui-tests.txt b/requirements-ui-tests.txt new file mode 100644 index 0000000..3cdf293 --- /dev/null +++ b/requirements-ui-tests.txt @@ -0,0 +1,4 @@ +playwright>=1.44.0 +pytest-playwright>=0.5.0 +requests>=2.31.0 +nodeenv>=1.8.0 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py new file mode 100644 index 0000000..4a6908b --- /dev/null +++ b/tests/e2e/conftest.py @@ -0,0 +1,243 @@ +""" +Pytest configuration for Playwright E2E UI tests. + +Each test session finds a free TCP port, writes a temporary conf/local.yml +(so multiple VENVs / CI jobs can run simultaneously without port conflicts), +starts Caldera on that port, waits until healthy, provides an authenticated +Playwright page, then tears everything down. + +Prerequisites (installed by the `ui` tox environment): + pip install playwright pytest-playwright requests + playwright install chromium + +Run: + pytest plugins/magma/tests/e2e -v --browser chromium + +Environment variables: + CALDERA_PORT Force a specific port (default: auto-detect a free port) + CALDERA_USER Username to log in as (default: admin) + CALDERA_PASS Password (default: admin) + CALDERA_EXTERNAL Set to '1' to skip server startup and use CALDERA_URL + CALDERA_URL Base URL when CALDERA_EXTERNAL=1 + CALDERA_STARTUP_TIMEOUT Seconds to wait for server (default: 90) +""" + +import os +import shutil +import socket +import subprocess +import sys +import tempfile +import time + +import pytest +import requests +import yaml + +# --------------------------------------------------------------------------- +# Paths +# --------------------------------------------------------------------------- + +_HERE = os.path.dirname(__file__) +CALDERA_ROOT = os.path.normpath(os.path.join(_HERE, '..', '..', '..', '..')) +CONF_DIR = os.path.join(CALDERA_ROOT, 'conf') +DEFAULT_YML = os.path.join(CONF_DIR, 'default.yml') +LOCAL_YML = os.path.join(CONF_DIR, 'local.yml') + +CALDERA_USER = os.environ.get('CALDERA_USER', 'admin') +CALDERA_PASS = os.environ.get('CALDERA_PASS', 'admin') +STARTUP_TIMEOUT = int(os.environ.get('CALDERA_STARTUP_TIMEOUT', '90')) + + +# --------------------------------------------------------------------------- +# Port helpers +# --------------------------------------------------------------------------- + +def _find_free_port() -> int: + """Bind to port 0 to let the OS assign a free ephemeral port.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(('127.0.0.1', 0)) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + return s.getsockname()[1] + + +def _port_in_use(port: int) -> bool: + """Return True if something is already listening on *port*.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(1) + return s.connect_ex(('127.0.0.1', port)) == 0 + + +# --------------------------------------------------------------------------- +# Config helpers +# --------------------------------------------------------------------------- + +def _write_test_local_yml(port: int) -> None: + """ + Write conf/local.yml based on conf/default.yml but with: + - port overridden to *port* + - host set to 127.0.0.1 (test-only, no external exposure) + + If conf/local.yml already exists it is backed up first so the original + is restored in teardown. + """ + with open(DEFAULT_YML, 'r', encoding='utf-8') as fh: + config = yaml.safe_load(fh) + + config['port'] = port + config['host'] = '127.0.0.1' + + with open(LOCAL_YML, 'w', encoding='utf-8') as fh: + yaml.safe_dump(config, fh, default_flow_style=False) + + +# --------------------------------------------------------------------------- +# Server lifecycle +# --------------------------------------------------------------------------- + +@pytest.fixture(scope='session') +def caldera_server(): + """ + Start Caldera on a free port using conf/local.yml, wait until healthy, + yield the base URL, then terminate and clean up. + + Set CALDERA_EXTERNAL=1 to skip startup and connect to an already-running + instance at CALDERA_URL instead. + """ + if os.environ.get('CALDERA_EXTERNAL') == '1': + yield os.environ.get('CALDERA_URL', 'http://localhost:8888') + return + + # Choose port — env var overrides auto-detection + port = int(os.environ.get('CALDERA_PORT', _find_free_port())) + base_url = f'http://127.0.0.1:{port}' + + # Safety: refuse to start if something is already on this port. + # Caldera loads all state into memory; writing conf/local.yml while a + # running instance holds config in memory would be silently ignored and + # could leave the config file in an inconsistent state. + if _port_in_use(port): + pytest.fail( + f'Port {port} is already in use. ' + 'Stop any running Caldera instance before running UI tests, ' + 'or set CALDERA_PORT to a free port.' + ) + + # Back up existing local.yml if present + local_yml_backup = None + if os.path.exists(LOCAL_YML): + local_yml_backup = LOCAL_YML + '.e2e_backup' + shutil.copy2(LOCAL_YML, local_yml_backup) + + _write_test_local_yml(port) + + env = os.environ.copy() + env['PYTHONPATH'] = CALDERA_ROOT + + proc = subprocess.Popen( + # -E local → uses conf/local.yml we just wrote + # -l ERROR → suppress startup noise in test output + [sys.executable, 'server.py', '-E', 'local', '-l', 'ERROR'], + cwd=CALDERA_ROOT, + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + health_url = f'{base_url}/api/v2/health' + deadline = time.time() + STARTUP_TIMEOUT + ready = False + while time.time() < deadline: + try: + r = requests.get(health_url, timeout=2) + if r.status_code in (200, 401): + ready = True + break + except requests.RequestException: + pass + time.sleep(1) + + if not ready: + proc.terminate() + # Restore backup before failing + if local_yml_backup: + shutil.move(local_yml_backup, LOCAL_YML) + else: + os.remove(LOCAL_YML) + pytest.fail( + f'Caldera did not become healthy within {STARTUP_TIMEOUT}s ' + f'(checked {health_url}). ' + f'Port {port} was selected.' + ) + + yield base_url + + # Teardown + proc.terminate() + try: + proc.wait(timeout=15) + except subprocess.TimeoutExpired: + proc.kill() + + if local_yml_backup: + shutil.move(local_yml_backup, LOCAL_YML) + elif os.path.exists(LOCAL_YML): + os.remove(LOCAL_YML) + + +# --------------------------------------------------------------------------- +# API session (requests) +# --------------------------------------------------------------------------- + +@pytest.fixture(scope='session') +def api_session(caldera_server): + """ + Authenticated requests.Session for direct API calls in tests. + Used to set up / verify data independently of the browser. + """ + session = requests.Session() + resp = session.post( + f'{caldera_server}/enter', + data={'username': CALDERA_USER, 'password': CALDERA_PASS}, + allow_redirects=False, + ) + assert resp.status_code in (200, 302), ( + f'Login failed: HTTP {resp.status_code}. ' + f'Verify CALDERA_USER/CALDERA_PASS match conf/local.yml credentials.' + ) + return session + + +@pytest.fixture(scope='session') +def base_url(caldera_server): + """Base URL of the running Caldera instance (e.g. http://127.0.0.1:54321).""" + return caldera_server + + +# --------------------------------------------------------------------------- +# Playwright fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def auth_page(page, caldera_server, api_session): + """ + Playwright ``page`` pre-loaded with authentication cookies. + + Copies the requests.Session cookies into the Playwright browser context + so tests start already logged in without going through the login form. + The page is NOT yet navigated — tests must call ``page.goto(...)`` first. + """ + pw_cookies = [ + { + 'name': c.name, + 'value': c.value, + 'domain': '127.0.0.1', + 'path': '/', + 'httpOnly': False, + 'secure': False, + } + for c in api_session.cookies + ] + if pw_cookies: + page.context.add_cookies(pw_cookies) + yield page diff --git a/tests/e2e/test_abilities.py b/tests/e2e/test_abilities.py new file mode 100644 index 0000000..36da344 --- /dev/null +++ b/tests/e2e/test_abilities.py @@ -0,0 +1,264 @@ +""" +E2E tests for the Caldera Abilities page (AbilitiesView.vue). + +These tests cover page structure, filter controls, the ability list, and the +create-ability modal. All tests use the `auth_page` fixture so auth cookies +are already present before navigation. + +Run with: + pytest plugins/magma/tests/e2e/test_abilities.py -v --browser chromium +""" + +import pytest +from playwright.sync_api import expect, Page + + +# --------------------------------------------------------------------------- +# 1. h2 heading "Abilities" is visible +# --------------------------------------------------------------------------- + +def test_abilities_page_heading(auth_page: Page, base_url: str) -> None: + """ + Navigating to /abilities must render an

whose text is "Abilities". + """ + auth_page.goto(base_url + '/abilities') + auth_page.wait_for_load_state('networkidle') + + heading = auth_page.locator('h2', has_text='Abilities') + expect(heading).to_be_visible() + + +# --------------------------------------------------------------------------- +# 2. Introductory ATT&CK description paragraph is visible +# --------------------------------------------------------------------------- + +def test_abilities_page_description(auth_page: Page, base_url: str) -> None: + """ + The page must display a descriptive paragraph that mentions ATT&CK — the + text matches the static copy rendered just below the

. + """ + auth_page.goto(base_url + '/abilities') + auth_page.wait_for_load_state('networkidle') + + # The paragraph contains the phrase used in the Vue template + description = auth_page.locator('p', has_text='ATT&CK') + expect(description).to_be_visible() + + +# --------------------------------------------------------------------------- +# 3. "Create an Ability" button is visible +# --------------------------------------------------------------------------- + +def test_create_ability_button_visible(auth_page: Page, base_url: str) -> None: + """ + The primary call-to-action button labelled "Create an Ability" must be + present and visible in the left sidebar column. + """ + auth_page.goto(base_url + '/abilities') + auth_page.wait_for_load_state('networkidle') + + create_btn = auth_page.locator('button', has_text='Create an Ability') + expect(create_btn).to_be_visible() + + +# --------------------------------------------------------------------------- +# 4. Search input with placeholder "Find an ability..." is visible +# --------------------------------------------------------------------------- + +def test_search_input_visible(auth_page: Page, base_url: str) -> None: + """ + A text input with placeholder "Find an ability..." must be rendered in the + filter sidebar so users can search the ability list by name. + """ + auth_page.goto(base_url + '/abilities') + auth_page.wait_for_load_state('networkidle') + + search_input = auth_page.locator('input[placeholder="Find an ability..."]') + expect(search_input).to_be_visible() + + +# --------------------------------------------------------------------------- +# 5. Tactic filter dropdown is visible and contains an "All" option +# --------------------------------------------------------------------------- + +def test_tactic_filter_dropdown_visible(auth_page: Page, base_url: str) -> None: + """ + The Tactic filter must be rendered as a for source selection + heading = auth_page.locator('h2', has_text='Fact Sources') + expect(heading).to_be_visible() + + # The source selector dropdown should also be visible + source_select = auth_page.locator('select') + expect(source_select.first).to_be_visible() + + +# --------------------------------------------------------------------------- +# 3. "New Source" button is visible +# --------------------------------------------------------------------------- + +def test_new_source_button_visible(auth_page: Page, base_url: str) -> None: + """ + The primary action button labelled "New Source" must be present and + visible in the left sidebar column regardless of how many sources exist. + """ + auth_page.goto(base_url + '/factsources') + auth_page.wait_for_load_state('networkidle') + + new_btn = auth_page.locator('button', has_text='New Source') + expect(new_btn).to_be_visible() + + +# --------------------------------------------------------------------------- +# 4. Source list item count matches the API count +# --------------------------------------------------------------------------- + +def test_fact_sources_api_count_matches_ui( + auth_page: Page, base_url: str, api_session +) -> None: + """ + The number of source entries rendered in the selector dropdown + (excluding the placeholder "Select a source" option with disabled/empty value). + When the API returns zero sources the test passes trivially (empty selector). + """ + resp = api_session.get(base_url + '/api/v2/sources') + assert resp.status_code == 200, ( + f'GET /api/v2/sources returned HTTP {resp.status_code}' + ) + api_sources = resp.json() + api_count = len(api_sources) + + auth_page.goto(base_url + '/factsources') + auth_page.wait_for_load_state('networkidle') + + # Sources are rendered as