diff --git a/sunbeam-python/tests/functional/chaos/README.md b/sunbeam-python/tests/functional/chaos/README.md new file mode 100644 index 000000000..df84d69e1 --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/README.md @@ -0,0 +1,57 @@ +# Chaos Mesh functional tests + +Chaos Mesh-based resilience tests for Canonical OpenStack features. + +This directory is separate from the feature functional tests under +`tests/functional/feature` so that chaos experiments can be run independently +and expanded over time. + +## Prerequisites + +- A working Canonical OpenStack deployment (same as feature functional tests). +- `sunbeam`, `openstack` and `juju` CLIs configured for that deployment. + +Session-scoped fixtures automatically: + +- Enable the **validation** feature once per run. +- Install or verify Chaos Mesh (Helm and `kubectl` are run via `juju exec` on + `k8s/0` in the `openstack-machines` model). + +## Run outcome and reports + +Each chaos test run is **SUCCESS** or **FAIL**: + +- **FAIL** if any unit does not return to `active` within the recovery timeout, + or if the post-chaos **quick** validation test fails. +- **SUCCESS** only when all targeted units recover to `active` and the quick + test passes. + +A JSON report is written to `tests/functional/chaos/reports/` for each run. +Filenames include the outcome and a timestamp: + +- `SUCCESS__.json` +- `FAIL__.json` + +Reports include test duration, smoke test output/status, per-unit recovery +times and state sequences, and quick test output/status. + +## Layout + +- `validation/`: Chaos tests for the **validation** feature (Keystone, API pods, + DB routers, infra). +- `reports/`: JSON reports from each run. + +## Running the tests + +From the `sunbeam-python` tree: + +```bash +tox -e functional-chaos +``` + +Or run a single test: + +```bash +python -m pytest -s -vv tests/functional/chaos/validation/test_validation_keystone_chaos.py --config tests/functional/feature/test_config.yaml +``` + diff --git a/sunbeam-python/tests/functional/chaos/__init__.py b/sunbeam-python/tests/functional/chaos/__init__.py new file mode 100644 index 000000000..b9a36a80a --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/__init__.py @@ -0,0 +1,4 @@ +# SPDX-FileCopyrightText: 2026 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Chaos Mesh-based functional tests.""" diff --git a/sunbeam-python/tests/functional/chaos/conftest.py b/sunbeam-python/tests/functional/chaos/conftest.py new file mode 100644 index 000000000..f5aa33bc6 --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/conftest.py @@ -0,0 +1,150 @@ +# SPDX-FileCopyrightText: 2026 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 +# ruff: noqa: I001 + +"""Functional fixtures and config hooks for chaos tests.""" + +import logging +import subprocess + +import pytest + +from tests.functional.chaos.utils import _helm_command, _kubectl_command +from tests.functional.feature.conftest import ( # noqa: F401 + juju_client, + sunbeam_client as _feature_sunbeam_client, + test_config, +) + +logger = logging.getLogger(__name__) + + +@pytest.fixture(scope="session", autouse=True) +def ensure_chaos_mesh_installed() -> None: + """Ensure Chaos Mesh is installed and ready for chaos tests. + + This follows the documented Helm installation path, using: + + - sudo snap install helm --classic + - helm repo add chaos-mesh https://charts.chaos-mesh.org + - helm upgrade --install chaos-mesh chaos-mesh/chaos-mesh + """ + + def _has_chaos_mesh() -> bool: + try: + subprocess.run( + _kubectl_command(["get", "pods", "-n", "chaos-mesh"]), + check=True, + capture_output=True, + text=True, + ) + except subprocess.CalledProcessError: + return False + + return True + + if _has_chaos_mesh(): + logger.info("Chaos Mesh already present in namespace 'chaos-mesh'.") + return + + logger.info("Chaos Mesh not detected; attempting to install via Helm...") + + try: + subprocess.run( + _helm_command(["version"]), + check=True, + capture_output=True, + text=True, + ) + except subprocess.CalledProcessError: + logger.info( + "Helm not found or not working in k8s/0@openstack-machines; " + "attempting to install helm snap in that unit...", + ) + subprocess.run( + [ + "juju", + "exec", + "--unit", + "k8s/0", + "-m", + "openstack-machines", + "--", + "sudo", + "snap", + "install", + "helm", + "--classic", + ], + check=True, + text=True, + ) + # Re-check helm availability; if this fails, surface a clear error. + try: + subprocess.run( + _helm_command(["version"]), + check=True, + capture_output=True, + text=True, + ) + except subprocess.CalledProcessError as exc: # pragma: no cover + msg = ( + "Helm is still not available in k8s/0@openstack-machines after " + "attempted snap installation. Please log into that unit and " + "ensure 'helm' is installed and configured." + ) + raise RuntimeError(msg) from exc + + subprocess.run( + _helm_command(["repo", "add", "chaos-mesh", "https://charts.chaos-mesh.org"]), + check=False, + text=True, + ) + subprocess.run( + _helm_command(["repo", "update"]), + check=False, + text=True, + ) + + subprocess.run( + _helm_command( + [ + "upgrade", + "--install", + "chaos-mesh", + "chaos-mesh/chaos-mesh", + "--namespace", + "chaos-mesh", + "--create-namespace", + ], + ), + check=True, + text=True, + ) + + if not _has_chaos_mesh(): + logger.warning( + "Chaos Mesh could not be verified as running in 'chaos-mesh' " + "namespace after attempted installation. " + "Continuing anyway; PodChaos operations may fail if Chaos Mesh " + "is not fully ready.", + ) + + +@pytest.fixture(scope="session", autouse=True) +def ensure_validation_enabled_once(sunbeam_client) -> None: + """Enable the validation feature once for all chaos tests. + + Chaos scenarios assume that the validation feature is enabled and they + merely start ``sunbeam validation run smoke`` during fault injection. + """ + logger.info("Ensuring 'validation' feature is enabled for chaos tests...") + try: + sunbeam_client.enable_feature("validation") + except ( + subprocess.CalledProcessError + ) as exc: # pragma: no cover - environment-specific + logger.warning( + "Validation feature could not be enabled; chaos tests may fail: %s", + exc, + ) diff --git a/sunbeam-python/tests/functional/chaos/utils.py b/sunbeam-python/tests/functional/chaos/utils.py new file mode 100644 index 000000000..3abf54f1a --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/utils.py @@ -0,0 +1,592 @@ +# SPDX-FileCopyrightText: 2026 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Shared helpers for Chaos Mesh functional tests. + +These utilities centralise common operations so that multiple chaos scenarios +can reuse the same logic for: + +- Enabling Sunbeam features. +- Inspecting Juju status via Jubilant. +- Waiting for units to become active again. +- Applying and deleting Chaos Mesh PodChaos resources. +""" + +from __future__ import annotations + +import base64 +import json +import logging +import subprocess +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import List, Sequence, Tuple + +import jubilant + +logger = logging.getLogger(__name__) + + +def get_leader_and_non_leaders( + juju_client, + app_name: str, +) -> Tuple[str, List[str]]: + """Return (leader_unit_name, [non_leader_unit_names]) for a Juju app.""" + logger.info("Querying Juju status for application '%s' units...", app_name) + + status: jubilant.Status = juju_client.juju.status() + try: + app = status.apps[app_name] + except KeyError as exc: + available_apps = ", ".join(sorted(status.apps.keys())) + raise RuntimeError( + f"Application '{app_name}' not found in Juju status. " + f"Available applications: {available_apps}" + ) from exc + + leader_unit: str | None = None + non_leaders: List[str] = [] + for unit_name, unit_data in app.units.items(): + if getattr(unit_data, "leader", False): + leader_unit = unit_name + else: + non_leaders.append(unit_name) + + if leader_unit is None: + raise AssertionError( + f"No leader unit found for application '{app_name}' in Juju status." + ) + + return leader_unit, non_leaders + + +def wait_for_unit_active( + juju_client, + app_name: str, + unit_name: str, + timeout: int = 600, +) -> float: + """Wait until the given Juju unit's workload status is 'active'. + + Returns the time (in seconds) taken for the unit to become active again. + Raises AssertionError if the timeout is exceeded or the app enters error. + """ + logger.info( + "Waiting for unit %s (app '%s') to become active again...", + unit_name, + app_name, + ) + start = time.time() + + try: + juju_client.juju.wait( + lambda status: is_unit_active(status, app_name, unit_name), + error=lambda status: app_has_error(status, app_name), + timeout=timeout, + delay=5.0, + ) + except jubilant.WaitError as exc: + raise AssertionError( + f"Application '{app_name}' entered error state while waiting for " + f"{unit_name} to recover." + ) from exc + + elapsed = time.time() - start + logger.info( + "Unit %s (app '%s') is active again after %.1f seconds.", + unit_name, + app_name, + elapsed, + ) + return elapsed + + +def run_validation_command( + cmd: List[str], + timeout: int = 600, +) -> Tuple[float, str, bool]: + """Run a validation command (e.g. sunbeam validation run quick). + + Returns (duration_seconds, output, success). + """ + start = time.time() + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + ) + duration = time.time() - start + output = (result.stdout or "") + (result.stderr or "") + return (round(duration, 1), output, result.returncode == 0) + + +def wait_for_unit_active_with_tracking( + juju_client, + app_name: str, + unit_name: str, + timeout: int = 600, + poll_interval: int = 10, +) -> Tuple[float | None, List[dict]]: + """Poll until unit is active or timeout. + + Returns time_to_return_active_seconds (or None) and state_sequence. + + state_sequence: list of {timestamp_iso, state, message} when not active. + """ + state_sequence: List[dict] = [] + poll_start = time.time() + left_active_at: float | None = None + + while (time.time() - poll_start) < timeout: + status = juju_client.juju.status() + current, message = get_unit_workload_status(status, app_name, unit_name) + now = time.time() + ts_iso = datetime.fromtimestamp(now, tz=timezone.utc).isoformat() + + if current != "active": + if left_active_at is None: + left_active_at = now + state_sequence.append( + { + "timestamp": ts_iso, + "state": current, + "message": message, + } + ) + else: + if left_active_at is not None: + return (round(now - left_active_at, 1), state_sequence) + return (0.0, state_sequence) + + time.sleep(poll_interval) + + return (None, state_sequence) + + +def get_unit_workload_status( + status: jubilant.Status, + app_name: str, + unit_name: str, +) -> Tuple[str, str]: + """Return (workload_status.current, workload_status.message) for the unit. + + Returns ("unknown", "") if the unit or workload_status is missing. + """ + units = status.get_units(app_name) + unit = units.get(unit_name) + if not unit: + return ("unknown", "") + workload = getattr(unit, "workload_status", None) + current = getattr(workload, "current", None) or "unknown" + message = getattr(workload, "message", None) or "" + return (str(current), str(message)) + + +def is_unit_active( + status: jubilant.Status, + app_name: str, + unit_name: str, +) -> bool: + """Return True if the given unit's workload status is 'active'.""" + current, _ = get_unit_workload_status(status, app_name, unit_name) + return current == "active" + + +def app_has_error(status: jubilant.Status, app_name: str) -> bool: + """Return True if any unit in the given app is in error.""" + return jubilant.any_error(status, app_name) + + +def get_status_json_for_apps(juju_client, app_names: List[str]) -> dict: + """Return juju status as a dict restricted to the given application names. + + Runs ``juju status --format json`` and returns a structure with only + the requested applications (for use in report snapshots). + """ + try: + raw = juju_client.juju.cli("status", "--format", "json") + full = json.loads(raw) if isinstance(raw, str) else json.loads(raw.decode()) + except (json.JSONDecodeError, TypeError) as exc: + logger.warning("Could not get juju status JSON: %s", exc) + return {"_error": str(exc)} + + apps = full.get("applications") or {} + return {"applications": {name: apps[name] for name in app_names if name in apps}} + + +def assert_apps_healthy(juju_client, app_names: List[str]) -> None: + """Assert that the given applications have no units in error. + + If none of the applications are present in the model, this function logs + a warning and returns without failing the test. This allows the same test + suite to run against deployments that may not include all optional apps. + """ + status: jubilant.Status = juju_client.juju.status() + present_apps = [name for name in app_names if name in status.apps] + + if not present_apps: + logger.warning( + "None of the apps %s found in Juju model; " + "skipping health assertion for them.", + app_names, + ) + return + + for app_name in present_apps: + if jubilant.any_error(status, app_name): + raise AssertionError( + f"Application '{app_name}' has units in error state during chaos." + ) + logger.info( + "Application '%s' is healthy during chaos (no units in error).", app_name + ) + + +def unit_name_to_pod_name(unit_name: str) -> str: + """Map a Juju unit name (e.g. 'keystone/1') to a pod name (e.g. 'keystone-1'). + + For Kubernetes charms, Juju unit names and pod names follow this convention. + """ + return unit_name.replace("/", "-") + + +def pod_chaos_name_for_pod(app_name: str, pod_name: str) -> str: + """Return a deterministic PodChaos name for a given pod.""" + return f"{app_name}-{pod_name}-pod-kill" + + +def _kubectl_command(args: List[str]) -> List[str]: + """Build a kubectl command suitable for the environment. + + ``juju exec --unit -m --stdin -- sudo k8s kubectl ...`` + """ + k8s_unit = "k8s/0" + k8s_model = "openstack-machines" + return [ + "juju", + "exec", + "--unit", + k8s_unit, + "-m", + k8s_model, + "--", + "sudo", + "k8s", + "kubectl", + *args, + ] + + +def _helm_command(args: List[str]) -> List[str]: + """Build a helm command targeting the Sunbeam K8s cluster. + + ``juju exec --unit -m -- sudo helm ...`` + """ + k8s_unit = "k8s/0" + k8s_model = "openstack-machines" + return [ + "juju", + "exec", + "--unit", + k8s_unit, + "-m", + k8s_model, + "--", + "sudo", + "helm", + *args, + ] + + +def apply_pod_chaos_for_pod( + app_namespace: str, + pod_name: str, + chaos_namespace: str = "chaos-mesh", + *, + duration: str = "30s", + action: str = "pod-kill", +) -> str: + """Create a PodChaos resource targeting a single pod. + + Returns the name of the created PodChaos resource. + """ + chaos_name = pod_chaos_name_for_pod(app_namespace, pod_name) + manifest = f""" +apiVersion: chaos-mesh.org/v1alpha1 +kind: PodChaos +metadata: + name: {chaos_name} + namespace: {chaos_namespace} +spec: + action: {action} + mode: one + duration: "{duration}" + selector: + pods: + {app_namespace}: + - {pod_name} +""".lstrip() + logger.info( + "Applying PodChaos for pod %s in namespace %s (resource: %s)", + pod_name, + app_namespace, + chaos_name, + ) + + manifest_b64 = base64.b64encode(manifest.encode("utf-8")).decode("ascii") + cmd = [ + "juju", + "exec", + "--unit", + "k8s/0", + "-m", + "openstack-machines", + "--", + "bash", + "-c", + 'echo "$1" | base64 -d | sudo k8s kubectl apply -f -', + "_", + manifest_b64, + ] + result = subprocess.run( + cmd, + check=False, + capture_output=True, + text=True, + ) + if result.returncode != 0: + logger.error( + "Failed to apply PodChaos %s (exit code %s).\nstdout:\n%s\nstderr:\n%s", + chaos_name, + result.returncode, + result.stdout, + result.stderr, + ) + raise RuntimeError( + f"kubectl apply for PodChaos '{chaos_name}' failed with exit code " + f"{result.returncode}: {result.stderr.strip()}" + ) + return chaos_name + + +def delete_pod_chaos(chaos_name: str, chaos_namespace: str = "chaos-mesh") -> None: + """Delete a PodChaos resource by name.""" + logger.info( + "Deleting PodChaos resource: %s (namespace: %s)", chaos_name, chaos_namespace + ) + subprocess.run( + _kubectl_command( + [ + "delete", + "podchaos", + chaos_name, + "-n", + chaos_namespace, + "--ignore-not-found=true", + ] + ), + check=False, + capture_output=True, + text=True, + ) + + +def _write_chaos_json_report(report_name: str, data: dict) -> Path: + """Write a single JSON report file; name includes timestamp. Returns path.""" + reports_dir = Path(__file__).parent / "reports" + reports_dir.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d_%H-%M-%S") + path = reports_dir / f"{report_name}_{timestamp}.json" + with path.open("w", encoding="utf-8") as f: + json.dump(data, f, indent=2, sort_keys=True) + return path + + +def run_validation_with_pod_chaos( # noqa: C901 + juju_client, + targets: Sequence[tuple[str, List[str]]], + *, + suite_name: str, + report_name: str | None = None, + openstack_namespace: str = "openstack", + chaos_namespace: str = "chaos-mesh", + validation_timeout: int = 3600, + initial_delay: int = 60, + recovery_timeout: int = 600, + poll_interval: int = 10, + quick_test_timeout: int = 600, +) -> None: + """Run validation with PodChaos and optional JSON reporting. + + Smoke runs in parallel with chaos; quick run and JSON report + are executed after chaos when report_name is provided. + """ + logger.info( + "Starting 'sunbeam validation run smoke' for %s chaos suite...", + suite_name, + ) + validation_proc = subprocess.Popen( + ["sunbeam", "validation", "run", "smoke"], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + + run_start_monotonic = time.time() + failed_recoveries: List[dict] = [] + apps_in_error: List[dict] = [] + recovery_per_unit: List[dict] = [] + validation_return_code: int | None = None + error_summary: str | None = None + validation_output: str | None = None + smoke_duration: float | None = None + + chaos_resources: List[str] = [] + + try: + if initial_delay > 0: + logger.info( + "Sleeping %s seconds before starting PodChaos injections to allow " + "Tempest discover-tempest-config/bootstrap to complete.", + initial_delay, + ) + time.sleep(initial_delay) + + for app_name, dependent_apps in targets: + leader_unit, non_leader_units = get_leader_and_non_leaders( + juju_client, + app_name, + ) + logger.info( + "%s leader unit: %s; non-leaders: %s", + app_name, + leader_unit, + non_leader_units, + ) + + for unit_name in non_leader_units: + pod_name = unit_name_to_pod_name(unit_name) + chaos_name = apply_pod_chaos_for_pod( + openstack_namespace, + pod_name, + chaos_namespace=chaos_namespace, + duration="30s", + ) + chaos_resources.append(chaos_name) + + time_to_return_active_seconds, state_sequence = ( + wait_for_unit_active_with_tracking( + juju_client, + app_name, + unit_name, + timeout=recovery_timeout, + poll_interval=poll_interval, + ) + ) + recovery_per_unit.append( + { + "app": app_name, + "unit": unit_name, + "time_to_return_active_seconds": time_to_return_active_seconds, + } + ) + if state_sequence: + apps_in_error.append( + { + "app": app_name, + "unit": unit_name, + "state_sequence": state_sequence, + } + ) + if time_to_return_active_seconds is None: + failed_recoveries.append( + { + "app": app_name, + "unit": unit_name, + "pod": pod_name, + "error": "timeout", + } + ) + break + if dependent_apps: + assert_apps_healthy(juju_client, dependent_apps) + + logger.info( + "Waiting for validation smoke run to complete after %s chaos suite...", + suite_name, + ) + try: + stdout_data, _ = validation_proc.communicate(timeout=validation_timeout) + validation_output = stdout_data or "" + validation_return_code = validation_proc.returncode + smoke_duration = time.time() - run_start_monotonic + except subprocess.TimeoutExpired: + validation_proc.kill() + stdout_data, _ = validation_proc.communicate() + validation_output = stdout_data or "" + smoke_duration = time.time() - run_start_monotonic + validation_return_code = None + error_summary = ( + "sunbeam validation run smoke did not complete within the timeout." + ) + + if report_name: + quick_duration, quick_output, quick_success = run_validation_command( + ["sunbeam", "validation", "run", "quick"], + timeout=quick_test_timeout, + ) + test_duration = time.time() - run_start_monotonic + final_status = "SUCCESS" + if failed_recoveries or not quick_success: + final_status = "FAIL" + report_data = { + "status": final_status, + "test_duration_seconds": round(test_duration, 1), + "smoke_test": { + "duration_seconds": round(smoke_duration or 0, 1), + "output": (validation_output or "")[:10000], + "success": validation_return_code == 0, + }, + "apps_in_error": apps_in_error, + "recovery_per_unit": recovery_per_unit, + "quick_test": { + "duration_seconds": quick_duration, + "output": (quick_output or "")[:10000], + "success": quick_success, + }, + } + report_path = _write_chaos_json_report( + f"{final_status}_{report_name}", + report_data, + ) + logger.info("Chaos report written to %s", report_path) + + if not quick_success: + # Quick validation failure makes the chaos run a FAIL. + raise AssertionError( + "Quick validation test failed after chaos. See reports/." + ) + + if failed_recoveries and error_summary is None: + failed_labels = ", ".join( + f"{fr['app']}/{fr['unit']}" for fr in failed_recoveries + ) + error_summary = ( + f"One or more chaos targets did not recover cleanly: {failed_labels}" + ) + raise AssertionError(error_summary) + except Exception as exc: # noqa: BLE001 + if error_summary is None: + error_summary = repr(exc) + raise + finally: + for chaos_name in chaos_resources: + try: + delete_pod_chaos(chaos_name, chaos_namespace=chaos_namespace) + except Exception as exc: # noqa: BLE001 + logger.warning("Failed to clean up PodChaos %s: %s", chaos_name, exc) + + if validation_proc.poll() is None: + validation_proc.terminate() diff --git a/sunbeam-python/tests/functional/chaos/validation/__init__.py b/sunbeam-python/tests/functional/chaos/validation/__init__.py new file mode 100644 index 000000000..15cff7ecb --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/validation/__init__.py @@ -0,0 +1,4 @@ +# SPDX-FileCopyrightText: 2026 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Chaos tests for the validation feature.""" diff --git a/sunbeam-python/tests/functional/chaos/validation/test_validation_api_pod_chaos.py b/sunbeam-python/tests/functional/chaos/validation/test_validation_api_pod_chaos.py new file mode 100644 index 000000000..efeb43863 --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/validation/test_validation_api_pod_chaos.py @@ -0,0 +1,47 @@ +# SPDX-FileCopyrightText: 2026 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Generic API control-plane pod loss chaos tests.""" + +from __future__ import annotations + +import logging + +import pytest + +from tests.functional.chaos.utils import run_validation_with_pod_chaos + +logger = logging.getLogger(__name__) + + +API_APPS: list[str] = [ + "nova", + "neutron", + "glance", + "cinder", + "placement", + # "aodh", + # "ceilometer", + # "gnocchi", + # "masakari", + # "watcher", + "horizon", +] + + +@pytest.mark.functional +def test_validation_resilient_to_non_leader_api_pod_kills( + sunbeam_client, + juju_client, +) -> None: + """Validation 'smoke' profile should tolerate non-leader API pod kills.""" + run_validation_with_pod_chaos( + juju_client, + targets=[(app, []) for app in API_APPS], + suite_name="API pod", + report_name="test_validation_api_pod_chaos", + initial_delay=60, + recovery_timeout=1800, + poll_interval=10, + quick_test_timeout=1800, + ) diff --git a/sunbeam-python/tests/functional/chaos/validation/test_validation_db_router_chaos.py b/sunbeam-python/tests/functional/chaos/validation/test_validation_db_router_chaos.py new file mode 100644 index 000000000..797f7f467 --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/validation/test_validation_db_router_chaos.py @@ -0,0 +1,50 @@ +# SPDX-FileCopyrightText: 2026 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Chaos tests for database access-path degradation (mysql-router pods).""" + +from __future__ import annotations + +import logging + +import pytest + +from tests.functional.chaos.utils import run_validation_with_pod_chaos + +logger = logging.getLogger(__name__) + + +ROUTER_APPS: list[str] = [ + "nova-api-mysql-router", + "nova-cell-mysql-router", + "nova-mysql-router", + "cinder-mysql-router", + "cinder-volume-mysql-router", + "neutron-mysql-router", + "keystone-mysql-router", + "glance-mysql-router", + "placement-mysql-router", + # "aodh-mysql-router", + # "gnocchi-mysql-router", + # "masakari-mysql-router", + # "watcher-mysql-router", + "horizon-mysql-router", +] + + +@pytest.mark.functional +def test_validation_resilient_to_mysql_router_pod_kills( + sunbeam_client, + juju_client, +) -> None: + """Validation 'smoke' profile should tolerate mysql-router pod kills.""" + run_validation_with_pod_chaos( + juju_client, + targets=[(app, []) for app in ROUTER_APPS], + suite_name="mysql-router", + report_name="test_validation_db_router_chaos", + initial_delay=60, + recovery_timeout=1800, + poll_interval=10, + quick_test_timeout=1800, + ) diff --git a/sunbeam-python/tests/functional/chaos/validation/test_validation_infra_chaos.py b/sunbeam-python/tests/functional/chaos/validation/test_validation_infra_chaos.py new file mode 100644 index 000000000..0d5b9f262 --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/validation/test_validation_infra_chaos.py @@ -0,0 +1,41 @@ +# SPDX-FileCopyrightText: 2026 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Chaos tests for core infrastructure services (MySQL, RabbitMQ, Traefik).""" + +from __future__ import annotations + +import logging + +import pytest + +from tests.functional.chaos.utils import run_validation_with_pod_chaos + +logger = logging.getLogger(__name__) + + +INFRA_APPS: list[str] = [ + "mysql", + "rabbitmq", + "traefik-public", + "traefik", + "traefik-rgw", +] + + +@pytest.mark.functional +def test_validation_resilient_to_infra_pod_kills( + sunbeam_client, + juju_client, +) -> None: + """Validation 'smoke' profile should tolerate infra pod/unit loss.""" + run_validation_with_pod_chaos( + juju_client, + targets=[(app, []) for app in INFRA_APPS], + suite_name="infra", + report_name="test_validation_infra_chaos", + initial_delay=60, + recovery_timeout=1800, + poll_interval=10, + quick_test_timeout=1800, + ) diff --git a/sunbeam-python/tests/functional/chaos/validation/test_validation_keystone_chaos.py b/sunbeam-python/tests/functional/chaos/validation/test_validation_keystone_chaos.py new file mode 100644 index 000000000..32fcbf67a --- /dev/null +++ b/sunbeam-python/tests/functional/chaos/validation/test_validation_keystone_chaos.py @@ -0,0 +1,36 @@ +# SPDX-FileCopyrightText: 2026 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + + +"""Keystone-specific chaos tests for the validation feature.""" + +from __future__ import annotations + +import logging + +import pytest + +from tests.functional.chaos.utils import run_validation_with_pod_chaos + +logger = logging.getLogger(__name__) + +KEYSTONE_APP = "keystone" +TRAEFIK_APPS = ["traefik-public", "traefik-internal"] + + +@pytest.mark.functional +def test_validation_resilient_to_non_leader_keystone_pod_kills( + sunbeam_client, + juju_client, +) -> None: + """Run smoke + quick validation around non-leader Keystone pod chaos.""" + run_validation_with_pod_chaos( + juju_client, + targets=[(KEYSTONE_APP, TRAEFIK_APPS)], + suite_name="Keystone API", + report_name="test_validation_keystone_chaos", + initial_delay=60, + recovery_timeout=1800, + poll_interval=10, + quick_test_timeout=1800, + ) diff --git a/sunbeam-python/tests/functional/feature/.gitignore b/sunbeam-python/tests/functional/feature/.gitignore new file mode 100644 index 000000000..db362e0b5 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/.gitignore @@ -0,0 +1,21 @@ +test_config.yaml +features/adminrc +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python + +venv/ +env/ +ENV/ + +.vscode/ +.idea/ +*.swp +*.swo + +.pytest_cache/ +.coverage +htmlcov/ +*.log diff --git a/sunbeam-python/tests/functional/feature/README.md b/sunbeam-python/tests/functional/feature/README.md new file mode 100644 index 000000000..7befa7276 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/README.md @@ -0,0 +1,106 @@ +# Sunbeam Feature Functional Tests + +Functional tests for Sunbeam feature enablement/disablement. These tests +connect to an **existing Sunbeam deployment** and run the enable/verify/disable +lifecycle for each feature, logging timing and basic behaviour checks. + +The suite is designed to be run via `tox` from the `sunbeam-python` tree. + +## Prerequisites + +- **Existing Sunbeam deployment** already bootstrapped and reachable +- `sunbeam` CLI on `PATH` and configured to talk to that deployment + - e.g. `sunbeam deployment list` shows your deployment +- `openstack` CLI configured for that cloud + - e.g. `openstack endpoint list` works +- `juju` CLI installed and able to access the controller/model that backs the + Sunbeam deployment + +## Configuration + +Create a config file from the example: + +```bash +cd sunbeam-python +cp tests/functional/feature/test_config.yaml.example tests/functional/feature/test_config.yaml +``` + +Then edit `tests/functional/feature/test_config.yaml`: + +```yaml +sunbeam: + deployment_name: "ps6" # Name shown by `sunbeam deployment list` + +juju: + model: "openstack" # Juju model backing the cloud + # controller: "my-controller" # Optional; auto-detected if omitted +``` + +### Run the full feature functional suite + +```bash +tox -e functional-feature +``` + +### Run a single feature functional test + +You can pass standard `pytest` selectors through tox via `posargs`. For example: + +- **Instance Recovery**: + + ```bash + tox -e functional-feature -- tests/functional/feature/test_features.py::test_instance_recovery + ``` + +- **TLS CA**: + + ```bash + tox -e functional-feature -- tests/functional/feature/test_features.py::test_tls_ca + ``` + +## Feature coverage and dependencies + +### Features in this suite + +- **Enabled in current flow** + - `instance-recovery` + - `caas` (Containers as a Service) + - `dns` + - `images-sync` + - `loadbalancer` + - `resource-optimization` + - `shared-filesystem` + - `telemetry` + - `observability` + - `tls` (CA mode) + - `vault` + - `validation` + - `secrets` + +- **Present but intentionally disabled for now** + - `baremetal` + - `ldap` + - `maintenance` + - `pro` + +### Feature dependencies + +Some features have explicit dependencies: + +- **CaaS (`caas`)** + - Depends on: **`secrets`**, **`loadbalancer`** + - The CaaS test ensures these dependencies are enabled before running. + +- **Secrets as a Service (`secrets`)** + - Depends on: **`vault`** + - The Secrets test ensures the Vault feature is enabled before running. + +- **TLS (Vault-backed)** + - TLS can also be deployed in a Vault-backed mode which implicitly depends on + the **`vault`** feature. This suite currently exercises only the TLS CA + mode (`test_tls_ca`). + +## Notes + +- Disable failures are **logged and ignored** so that the suite continues + to the next feature, matching the behaviour of the original tests. diff --git a/sunbeam-python/tests/functional/feature/__init__.py b/sunbeam-python/tests/functional/feature/__init__.py new file mode 100644 index 000000000..f3adc6677 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/__init__.py @@ -0,0 +1,8 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Sunbeam feature functional test suite. + +These tests exercise `sunbeam enable/disable` for individual features +against an existing Sunbeam deployment. +""" diff --git a/sunbeam-python/tests/functional/feature/conftest.py b/sunbeam-python/tests/functional/feature/conftest.py new file mode 100644 index 000000000..ef5d4cc5e --- /dev/null +++ b/sunbeam-python/tests/functional/feature/conftest.py @@ -0,0 +1,72 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Pytest configuration and fixtures for Sunbeam feature functional tests.""" + +from pathlib import Path + +import pytest +import yaml + +from .utils.juju import JujuClient +from .utils.sunbeam import SunbeamClient + + +def pytest_addoption(parser): + """Add custom command-line options.""" + parser.addoption( + "--config", + action="store", + default="test_config.yaml", + help="Path to test configuration file", + ) + + +@pytest.fixture(scope="session") +def test_config(request): + """Load test configuration from YAML file.""" + config_path = request.config.getoption("--config") + # Resolve relative to this feature functional directory + config_file = Path(__file__).parent / config_path + + if not config_file.exists(): + msg = ( + f"Configuration file not found: {config_file}. " + "Copy tests/functional/feature/test_config.yaml.example to " + "tests/functional/feature/test_config.yaml and set sunbeam.deployment_name, juju.model." + ) + pytest.skip(msg) + + with open(config_file, "r") as f: + config = yaml.safe_load(f) + + return config + + +@pytest.fixture(scope="session") +def sunbeam_client(test_config): + """Create Sunbeam client for test session.""" + deployment_name = test_config.get("sunbeam", {}).get("deployment_name") + if not deployment_name: + pytest.skip("deployment_name not configured in test_config.yaml") + + client = SunbeamClient(deployment_name) + + if not client.is_connected(): + pytest.skip(f"Cannot connect to Sunbeam deployment '{deployment_name}'.") + + return client + + +@pytest.fixture(scope="session") +def juju_client(test_config): + """Create Juju client for test session.""" + model = test_config.get("juju", {}).get("model", "openstack") + controller = test_config.get("juju", {}).get("controller") + + client = JujuClient(model=model, controller=controller) + + if not client.is_connected(): + pytest.skip(f"Cannot connect to Juju model '{model}'.") + + return client diff --git a/sunbeam-python/tests/functional/feature/features/__init__.py b/sunbeam-python/tests/functional/feature/features/__init__.py new file mode 100644 index 000000000..0f78a8aa4 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/__init__.py @@ -0,0 +1,4 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Feature test classes for Sunbeam feature functional tests.""" diff --git a/sunbeam-python/tests/functional/feature/features/baremetal.py b/sunbeam-python/tests/functional/feature/features/baremetal.py new file mode 100644 index 000000000..57d99bfa8 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/baremetal.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for baremetal feature. + +Baremetal provides Ironic-based bare metal provisioning. +Functionality is validated via the Ironic (baremetal) API. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class BaremetalTest(BaseFeatureTest): + """Test baremetal feature enablement/disablement.""" + + feature_name = "baremetal" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that the Baremetal (Ironic) API is reachable.""" + logger.info("Verifying Baremetal (Ironic) service is available...") + try: + subprocess.run( + ["openstack", "baremetal", "driver", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying Baremetal service: %s", exc) + raise AssertionError( + f"Baremetal service verification failed: {exc}" + ) from exc + + logger.info("Baremetal service verified via `openstack baremetal driver list`") diff --git a/sunbeam-python/tests/functional/feature/features/base.py b/sunbeam-python/tests/functional/feature/features/base.py new file mode 100644 index 000000000..87f647c53 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/base.py @@ -0,0 +1,219 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Base class for Sunbeam feature functional tests.""" + +import logging +import os +import time +from pathlib import Path +from typing import Dict, List, Optional + +from ..utils.juju import JujuClient +from ..utils.sunbeam import SunbeamClient + +logger = logging.getLogger(__name__) + + +class BaseFeatureTest: + """Base class for testing Sunbeam features.""" + + feature_name: str = "" + expected_applications: List[str] = [] + timeout_seconds: int = 300 + enable_args: List[str] = [] + disable_args: List[str] = [] + + def __init__( + self, + sunbeam_client: SunbeamClient, + juju_client: JujuClient, + config: Optional[Dict] = None, + ): + self.sunbeam = sunbeam_client + self.juju = juju_client + self.config = config or {} + + feature_config = self.config.get("features", {}).get(self.feature_name, {}) + self.expected_applications = feature_config.get( + "expected_applications", + self.expected_applications, + ) + self.timeout_seconds = feature_config.get( + "timeout_seconds", + self.timeout_seconds, + ) + self.enable_args = feature_config.get("enable_args", self.enable_args) + self.disable_args = feature_config.get("disable_args", self.disable_args) + + self._ensure_openstack_env() + + def enable(self) -> bool: + """Enable the feature.""" + logger.info("Enabling feature: '%s'", self.feature_name) + return self.sunbeam.enable_feature( + self.feature_name, + extra_args=self.enable_args, + ) + + def disable(self) -> bool: + """Disable the feature. + + Returns True if successful, False otherwise. + """ + logger.info("Disabling feature: '%s'", self.feature_name) + try: + return self.sunbeam.disable_feature( + self.feature_name, + extra_args=self.disable_args, + ) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Failed to disable feature '%s': %s", + self.feature_name, + exc, + ) + return False + + def run_full_lifecycle(self) -> bool: + """Run enable/disable lifecycle with timing. + + Disable failures are logged but do not fail the overall test. + """ + logger.info("Starting lifecycle test for feature: '%s'", self.feature_name) + + enable_start = time.time() + logger.info("[ENABLE] Starting enable for '%s'...", self.feature_name) + enable_success = self.enable() + enable_duration = time.time() - enable_start + if enable_success: + logger.info( + "[ENABLE] SUCCESS for '%s' - Time taken: %.2f seconds", + self.feature_name, + enable_duration, + ) + else: + logger.error( + "[ENABLE] FAILED for '%s' - Time taken: %.2f seconds", + self.feature_name, + enable_duration, + ) + return False + + try: + self.verify_validate_feature_behavior() + except Exception: # noqa: BLE001 + logger.exception( + "Validation failed for feature '%s' after enable", self.feature_name + ) + # Best-effort cleanup – if disable also fails, log and continue. + try: + self.disable() + except Exception: # noqa: BLE001 + logger.warning( + "Disable also failed while handling validation error for '%s'", + self.feature_name, + ) + return False + + disable_start = time.time() + logger.info("[DISABLE] Starting disable for '%s'...", self.feature_name) + disable_success = self.disable() + disable_duration = time.time() - disable_start + if disable_success: + logger.info( + "[DISABLE] SUCCESS for '%s' - Time taken: %.2f seconds", + self.feature_name, + disable_duration, + ) + else: + logger.warning( + "[DISABLE] FAILED for '%s' - Time taken: %.2f seconds (continuing anyway)", + self.feature_name, + disable_duration, + ) + + total_duration = time.time() - enable_start + logger.info( + "[SUMMARY] Feature '%s' - Enable: %.2fs, Disable: %.2fs (%s), Total: %.2fs", + self.feature_name, + enable_duration, + disable_duration, + "SUCCESS" if disable_success else "FAILED", + total_duration, + ) + return True + + def verify_enabled(self) -> None: + """Verify that expected applications and units are present. + + This is a boilerplate method for future use. Currently not called + by default, but can be overridden in subclasses to add verification. + """ + pass + + def validate_feature_behavior(self) -> None: + """Validate that the feature is working correctly. + + This is a boilerplate method for future use. Currently not called + by default, but can be overridden in subclasses to add functionality tests. + """ + pass + + def verify_validate_feature_behavior(self) -> None: + """Simple verification that feature is enabled and basic check passes. + + This is a simple method that can be called after enable to verify + the feature is working. Override in subclasses for feature-specific checks. + Subclasses can override validate_feature_behavior() for behavior checks; + that is invoked from here before the application presence checks. + """ + logger.info("Verifying feature '%s' is enabled...", self.feature_name) + self.validate_feature_behavior() + if self.expected_applications: + for app in self.expected_applications: + if self.juju.has_application(app): + logger.info("Application '%s' found", app) + else: + logger.warning( + "Application '%s' not found (may still be deploying)", app + ) + logger.info("Basic verification completed for feature '%s'", self.feature_name) + + def _ensure_openstack_env(self) -> None: + """Load OpenStack credentials from adminrc if needed. + + This avoids repeating sourcing logic across tests and keeps credentials + out of the code. If OS_AUTH_URL is already set, this is a no-op. + """ + if os.environ.get("OS_AUTH_URL"): + return + + adminrc_path = Path(__file__).resolve().parent / "adminrc" + if not adminrc_path.exists(): + logger.debug( + "adminrc file not found at %s; relying on existing environment", + adminrc_path, + ) + return + + try: + for line in adminrc_path.read_text().splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + if not line.startswith("export "): + continue + _, rest = line.split("export ", 1) + if "=" not in rest: + continue + key, value = rest.split("=", 1) + key = key.strip() + value = value.strip().strip('"').strip("'") + os.environ.setdefault(key, value) + logger.info("Loaded OpenStack credentials from %s", adminrc_path) + except Exception: # noqa: BLE001 + logger.exception( + "Failed to load OpenStack credentials from %s", + adminrc_path, + ) diff --git a/sunbeam-python/tests/functional/feature/features/caas.py b/sunbeam-python/tests/functional/feature/features/caas.py new file mode 100644 index 000000000..02dfa08f2 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/caas.py @@ -0,0 +1,83 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for caas feature. + +Container as a Service (Magnum) allows managing Kubernetes clusters via OpenStack. +Functionality is validated via the Magnum (COE) API. +""" + +import logging +import subprocess + +import pytest + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class CaaSTest(BaseFeatureTest): + """Test caas feature enablement/disablement.""" + + feature_name = "caas" + expected_units: list[str] = [] + expected_applications: list[str] = [] + timeout_seconds = 600 + + def _ensure_dependency_enabled(self, feature: str) -> bool: + """Best-effort enable a required dependency feature. + + If enabling the dependency fails (for example, missing Vault for + Secrets), we treat this as an unsatisfied dependency and skip. + """ + logger.info("Ensuring dependency feature '%s' is enabled for CaaS...", feature) + try: + self.sunbeam.enable_feature(feature) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Failed to enable dependency '%s' required by CaaS: %s", + feature, + exc, + ) + return False + return True + + def verify_validate_feature_behavior(self) -> None: + """Validate that the Magnum (COE) API is reachable. + + We call `openstack coe cluster list` to confirm the API is up. + """ + logger.info("Verifying CaaS (Magnum) service is available...") + try: + subprocess.run( + ["openstack", "coe", "cluster", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except subprocess.CalledProcessError as exc: + logger.warning("Failed to list COE clusters: %s", exc.stderr) + raise AssertionError( + f"CaaS (Magnum) service not accessible: {exc.stderr}" + ) from exc + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying CaaS service: %s", exc) + raise AssertionError(f"CaaS service verification failed: {exc}") from exc + + logger.info("CaaS (Magnum) service verified via `openstack coe cluster list`") + + def run_full_lifecycle(self) -> bool: + """Ensure dependencies then run the standard enable/verify/disable flow. + + CaaS depends on the Secrets and Load Balancer features. + """ + for dep in ("secrets", "loadbalancer"): + if not self._ensure_dependency_enabled(dep): + pytest.skip( + f"Skipping CaaS feature test: dependency '{dep}' " + "could not be enabled" + ) + + return super().run_full_lifecycle() diff --git a/sunbeam-python/tests/functional/feature/features/dns.py b/sunbeam-python/tests/functional/feature/features/dns.py new file mode 100644 index 000000000..6aa90802a --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/dns.py @@ -0,0 +1,40 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for dns feature. + +DNS requires nameservers as arguments, so we use dummy nameservers for testing. +DNS is a simple feature with no direct feature dependencies (besides the required +nameservers argument). Functionality is validated via the Designate (DNS) API. +""" + +import logging + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class DnsTest(BaseFeatureTest): + """Test dns feature enablement/disablement.""" + + feature_name = "dns" + # DNS requires nameservers argument - using dummy values for testing + enable_args: list[str] = ["ns1.example.com.", "ns2.example.com."] + expected_applications: list[str] = [] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that DNS as a Service is reachable. + + We call `sunbeam dns address` to confirm that the + Designate service is registered and accessible. + """ + logger.info("Verifying DNS service endpoints are available...") + try: + self.sunbeam.run(["dns", "address"]) + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying DNS service: %s", exc) + raise AssertionError(f"DNS service verification failed: {exc}") from exc + + logger.info("DNS service endpoints verified via `sunbeam dns address`") diff --git a/sunbeam-python/tests/functional/feature/features/images_sync.py b/sunbeam-python/tests/functional/feature/features/images_sync.py new file mode 100644 index 000000000..288f8fece --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/images_sync.py @@ -0,0 +1,43 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for images-sync feature. + +Images-sync is a simple feature with no dependencies. +Functionality is validated via the OpenStack Image API. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class ImagesSyncTest(BaseFeatureTest): + """Test images-sync feature enablement/disablement.""" + + feature_name = "images-sync" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that the Image service is reachable. + + We call `openstack image list` to confirm that Glance is responding. + """ + logger.info("Verifying Image service (Glance) is available...") + try: + subprocess.run( + ["openstack", "image", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying Image service: %s", exc) + raise AssertionError(f"Image service verification failed: {exc}") from exc + + logger.info("Image service verified via `openstack image list`") diff --git a/sunbeam-python/tests/functional/feature/features/instance_recovery.py b/sunbeam-python/tests/functional/feature/features/instance_recovery.py new file mode 100644 index 000000000..0f51f60ca --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/instance_recovery.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for instance-recovery feature.""" + +import subprocess + +from .base import BaseFeatureTest + + +class InstanceRecoveryTest(BaseFeatureTest): + """Test instance-recovery feature enablement/disablement.""" + + # CLI feature name + feature_name = "instance-recovery" + expected_applications = [ + "masakari", + "masakari-mysql-router", + "consul-management", + "consul-storage", + "consul-tenant", + ] + timeout_seconds = 900 + + def validate_feature_behavior(self) -> None: + """Run a small smoke test against the Masakari API. + + We call `openstack segment list` to confirm Masakari is responding + and that the CLI can talk to the Instance Recovery control plane. + """ + cmd = [ + "openstack", + "segment", + "list", + "-c", + "name", + "-c", + "service_type", + ] + result = subprocess.run(cmd, check=True, capture_output=True, text=True) + if not result.stdout.strip(): + raise AssertionError("openstack segment list returned no data") diff --git a/sunbeam-python/tests/functional/feature/features/ldap.py b/sunbeam-python/tests/functional/feature/features/ldap.py new file mode 100644 index 000000000..21aae49d5 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/ldap.py @@ -0,0 +1,40 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for ldap feature. + +LDAP integration configures Keystone to authenticate against LDAP. +Functionality is minimally validated via the Identity API. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class LdapTest(BaseFeatureTest): + """Test ldap feature enablement/disablement.""" + + feature_name = "ldap" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that the Identity API is reachable.""" + logger.info("Verifying Identity (Keystone) service is available for LDAP...") + try: + subprocess.run( + ["openstack", "domain", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying Identity service: %s", exc) + raise AssertionError(f"LDAP feature verification failed: {exc}") from exc + + logger.info("Identity service verified via `openstack domain list`") diff --git a/sunbeam-python/tests/functional/feature/features/loadbalancer.py b/sunbeam-python/tests/functional/feature/features/loadbalancer.py new file mode 100644 index 000000000..1ff3549b9 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/loadbalancer.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for loadbalancer feature. + +Loadbalancer is a simple feature with no dependencies. +Deploys Octavia, the OpenStack Load Balancer as a Service. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class LoadbalancerTest(BaseFeatureTest): + """Test loadbalancer feature enablement/disablement.""" + + feature_name = "loadbalancer" + expected_applications: list[str] = ["octavia"] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that loadbalancer service (Octavia) is working.""" + logger.info("Verifying loadbalancer service (Octavia) is available...") + + try: + result = subprocess.run( + ["openstack", "loadbalancer", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + logger.info("Loadbalancer service (Octavia) is accessible") + logger.debug("Loadbalancer list output: %s", result.stdout[:200]) + + except Exception as e: + logger.warning("Error checking loadbalancer service: %s", e) + raise AssertionError(f"Loadbalancer service verification failed: {e}") diff --git a/sunbeam-python/tests/functional/feature/features/maintenance.py b/sunbeam-python/tests/functional/feature/features/maintenance.py new file mode 100644 index 000000000..a2eeac6ee --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/maintenance.py @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for maintenance feature.""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class MaintenanceTest(BaseFeatureTest): + """Test maintenance feature enablement/disablement.""" + + feature_name = "maintenance" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that the Compute API is reachable.""" + logger.info("Verifying Compute service is available for maintenance...") + try: + subprocess.run( + ["openstack", "compute", "service", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying Compute service: %s", exc) + raise AssertionError( + f"Maintenance feature verification failed: {exc}" + ) from exc + + logger.info("Compute service verified via `openstack compute service list`") diff --git a/sunbeam-python/tests/functional/feature/features/observability.py b/sunbeam-python/tests/functional/feature/features/observability.py new file mode 100644 index 000000000..9f6366cd3 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/observability.py @@ -0,0 +1,51 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for observability feature. + +Observability integrates Canonical OpenStack with COS. + +For this functional test we exercise the simple embedded workflow from the +documentation: + +1. `sunbeam enable observability embedded` +2. `sunbeam observability dashboard-url` +3. `sunbeam disable observability embedded` +""" + +import logging + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class ObservabilityTest(BaseFeatureTest): + """Test observability feature enablement/disablement.""" + + feature_name = "observability" + enable_args: list[str] = ["embedded"] + disable_args: list[str] = ["embedded"] + expected_applications: list[str] = [] + timeout_seconds = 900 + + def verify_validate_feature_behavior(self) -> None: + """Validate that the observability dashboard URL is available. + + This uses `sunbeam observability dashboard-url` from the docs to + confirm that the embedded COS deployment is responding. + """ + logger.info("Fetching observability dashboard URL...") + try: + result = self.sunbeam.run(["observability", "dashboard-url"]) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Error while retrieving observability dashboard URL: %s", + exc, + ) + raise AssertionError( + f"Observability feature verification failed: {exc}" + ) from exc + + url = result.stdout.strip() + logger.info("Observability dashboard URL: %s", url) diff --git a/sunbeam-python/tests/functional/feature/features/orchestration.py b/sunbeam-python/tests/functional/feature/features/orchestration.py new file mode 100644 index 000000000..b09601f56 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/orchestration.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for orchestration feature. + +Orchestration is a simple feature with no dependencies. +Deploys Heat, the OpenStack Orchestration service. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class OrchestrationTest(BaseFeatureTest): + """Test orchestration feature enablement/disablement.""" + + feature_name = "orchestration" + expected_applications: list[str] = ["heat"] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that orchestration service (Heat) is working.""" + logger.info("Verifying orchestration service (Heat) is available...") + + try: + result = subprocess.run( + ["openstack", "stack", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + logger.info("Orchestration service (Heat) is accessible") + logger.debug("Stack list output: %s", result.stdout[:200]) + + except Exception as e: + logger.warning("Error checking orchestration service: %s", e) + raise AssertionError(f"Orchestration service verification failed: {e}") diff --git a/sunbeam-python/tests/functional/feature/features/pro.py b/sunbeam-python/tests/functional/feature/features/pro.py new file mode 100644 index 000000000..7948bde8f --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/pro.py @@ -0,0 +1,61 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for pro feature. + +Ubuntu Pro integrates subscription/entitlement with the deployment. +Functionality is minimally validated via a generic OpenStack service call. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class ProTest(BaseFeatureTest): + """Test pro feature enablement/disablement.""" + + feature_name = "pro" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def __init__(self, *args, **kwargs) -> None: + """Initialise Pro test with a token argument for enable. + + The token is taken from the functional test configuration, if present. + If no token is configured, a dummy placeholder is used. + """ + super().__init__(*args, **kwargs) + pro_cfg = self.config.get("pro", {}) if self.config is not None else {} + token = pro_cfg.get("token", "DUMMY-UBUNTU-PRO-TOKEN") + self.enable_args = ["--token", token] + + def verify_validate_feature_behavior(self) -> None: + """Validate that OpenStack APIs remain reachable under Pro.""" + logger.info("Verifying OpenStack service catalog for Ubuntu Pro...") + try: + result = subprocess.run( + ["openstack", "service", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except subprocess.CalledProcessError as exc: + logger.warning("Failed to list services: %s", exc.stderr) + raise AssertionError( + f"OpenStack service catalog not accessible: {exc.stderr}" + ) from exc + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying OpenStack services: %s", exc) + raise AssertionError( + f"Ubuntu Pro feature verification failed: {exc}" + ) from exc + + if not result.stdout.strip(): + raise AssertionError("Service list returned no data") + + logger.info("OpenStack service catalog verified via `openstack service list`") diff --git a/sunbeam-python/tests/functional/feature/features/resource_optimization.py b/sunbeam-python/tests/functional/feature/features/resource_optimization.py new file mode 100644 index 000000000..8db74bbb2 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/resource_optimization.py @@ -0,0 +1,50 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for resource-optimization feature. + +Resource Optimization provides Watcher as a service. +Functionality is validated via the Watcher (optimize) API. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class ResourceOptimizationTest(BaseFeatureTest): + """Test resource-optimization feature enablement/disablement.""" + + feature_name = "resource-optimization" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that the Watcher (resource optimization) API is reachable. + + We call `openstack optimize goal list` to confirm the API is up. + """ + logger.info("Verifying Resource Optimization (Watcher) service is available...") + try: + subprocess.run( + ["openstack", "optimize", "goal", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Error while verifying Resource Optimization service: %s", + exc, + ) + raise AssertionError( + f"Resource Optimization service verification failed: {exc}" + ) from exc + + logger.info( + "Resource Optimization service verified via `openstack optimize goal list`" + ) diff --git a/sunbeam-python/tests/functional/feature/features/secrets.py b/sunbeam-python/tests/functional/feature/features/secrets.py new file mode 100644 index 000000000..49fdf12da --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/secrets.py @@ -0,0 +1,57 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for secrets feature.""" + +import logging +import subprocess + +import pytest + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class SecretsTest(BaseFeatureTest): + """Test secrets feature enablement/disablement.""" + + feature_name = "secrets" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def _ensure_vault_enabled(self) -> bool: + """Ensure the Vault feature is enabled before Secrets.""" + logger.info("Ensuring 'vault' feature is enabled before 'secrets'...") + try: + self.sunbeam.enable_feature("vault") + except Exception as exc: # noqa: BLE001 + logger.warning("Failed to enable required dependency 'vault': %s", exc) + return False + return True + + def verify_validate_feature_behavior(self) -> None: + """Validate that the Secrets (Barbican) API is reachable.""" + logger.info("Verifying Secrets (Barbican) service is available...") + try: + subprocess.run( + ["openstack", "secret", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying Secrets service: %s", exc) + raise AssertionError(f"Secrets service verification failed: {exc}") from exc + + logger.info("Secrets service verified via `openstack secret list`") + + def run_full_lifecycle(self) -> bool: + """Enable Vault first, then run the Secrets lifecycle.""" + if not self._ensure_vault_enabled(): + pytest.skip( + "Skipping Secrets feature test: dependency 'vault' not available" + ) + + return super().run_full_lifecycle() diff --git a/sunbeam-python/tests/functional/feature/features/shared_filesystem.py b/sunbeam-python/tests/functional/feature/features/shared_filesystem.py new file mode 100644 index 000000000..086cdbe41 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/shared_filesystem.py @@ -0,0 +1,48 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for shared-filesystem feature. + +Shared Filesystems provides Manila-based file share services. +Functionality is validated via the Manila API. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class SharedFilesystemTest(BaseFeatureTest): + """Test shared-filesystem feature enablement/disablement.""" + + feature_name = "shared-filesystem" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that the Shared Filesystems (Manila) API is reachable. + + We call `openstack share list` to confirm the API is up. + """ + logger.info("Verifying Shared Filesystems (Manila) service is available...") + try: + subprocess.run( + ["openstack", "share", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Error while verifying Shared Filesystems service: %s", + exc, + ) + raise AssertionError( + f"Shared Filesystems service verification failed: {exc}" + ) from exc + + logger.info("Shared Filesystems service verified via `openstack share list`") diff --git a/sunbeam-python/tests/functional/feature/features/telemetry.py b/sunbeam-python/tests/functional/feature/features/telemetry.py new file mode 100644 index 000000000..191867230 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/telemetry.py @@ -0,0 +1,43 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for telemetry feature. + +Telemetry is a simple feature with no dependencies. +Deploys Ceilometer, Aodh, Gnocchi, and OpenStack Exporter. +""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class TelemetryTest(BaseFeatureTest): + """Test telemetry feature enablement/disablement.""" + + feature_name = "telemetry" + expected_applications: list[str] = ["ceilometer", "gnocchi", "aodh"] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that telemetry services are working.""" + logger.info("Verifying telemetry services are available...") + + # Check if alarm service (Aodh) is accessible + try: + result = subprocess.run( + ["openstack", "alarm", "list"], + capture_output=True, + text=True, + timeout=30, + check=True, + ) + logger.info("Telemetry alarm service (Aodh) is accessible") + logger.debug("Alarm list output: %s", result.stdout[:200]) + + except Exception as e: + logger.warning("Error checking telemetry services: %s", e) + raise AssertionError(f"Telemetry service verification failed: {e}") diff --git a/sunbeam-python/tests/functional/feature/features/tls.py b/sunbeam-python/tests/functional/feature/features/tls.py new file mode 100644 index 000000000..cfc8775d5 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/tls.py @@ -0,0 +1,125 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for tls feature (CA mode).""" + +import base64 +import logging +import subprocess +import tempfile +from pathlib import Path +from typing import Tuple + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +def generate_self_signed_ca_certificate() -> Tuple[str, str]: + """Generate a self-signed CA certificate. + + Returns a tuple of (ca_cert_base64, ca_chain_base64). For a simple self-signed CA, + the chain is the same as the cert. TLS CA currently only uses the CA certificate. + """ + with tempfile.TemporaryDirectory() as tmpdir: + tmp_path = Path(tmpdir) + + key_path = tmp_path / "ca.key" + subprocess.run( + ["openssl", "genrsa", "-out", str(key_path), "4096"], + check=True, + capture_output=True, + ) + + cert_path = tmp_path / "ca.crt" + subprocess.run( + [ + "openssl", + "req", + "-new", + "-x509", + "-days", + "365", + "-key", + str(key_path), + "-out", + str(cert_path), + "-subj", + "/C=US/ST=State/L=City/O=TestOrg/CN=TestCA", + "-extensions", + "v3_ca", + "-config", + "/dev/stdin", + ], + input=b"""[req] +distinguished_name = req_distinguished_name +[req_distinguished_name] +[v3_ca] +basicConstraints = critical,CA:TRUE +keyUsage = critical,keyCertSign,cRLSign +subjectKeyIdentifier = hash +authorityKeyIdentifier = keyid:always,issuer +""", + check=True, + capture_output=True, + ) + + ca_cert = cert_path.read_text() + ca_cert_base64 = base64.b64encode(ca_cert.encode()).decode() + + ca_chain_base64 = ca_cert_base64 + + return (ca_cert_base64, ca_chain_base64) + + +class TlsCaTest(BaseFeatureTest): + """Test TLS CA mode enablement/disablement. + + TLS CA mode uses Certificate Authority certificates for TLS. + This test verifies that: + - TLS CA can be enabled (with self-signed CA certificates) + - Endpoints are exposed over HTTPS (both public and internal) + - Basic OpenStack operations work (e.g., listing images) + """ + + feature_name = "tls" + enable_args: list[str] = [] + disable_args: list[str] = ["ca"] + expected_applications = [ + "manual-tls-certificates", + ] + timeout_seconds = 600 + + def __init__(self, *args, **kwargs): + """Initialize and generate CA certificates.""" + super().__init__(*args, **kwargs) + self.ca_cert_base64, _ = generate_self_signed_ca_certificate() + self.enable_args = [ + "ca", + "--ca", + self.ca_cert_base64, + ] + + def enable(self) -> bool: + """Enable TLS CA feature (without --accept-defaults flag).""" + logger.info("Enabling feature: '%s'", self.feature_name) + return self.sunbeam.enable_feature( + self.feature_name, + extra_args=self.enable_args, + ) + + def disable(self) -> bool: + """Disable TLS CA feature (without --accept-defaults flag).""" + logger.info("Disabling feature: '%s'", self.feature_name) + try: + return self.sunbeam.disable_feature( + self.feature_name, + extra_args=self.disable_args, + ) + except Exception as exc: # noqa: BLE001 + logger.warning( + "Failed to disable feature '%s': %s", + self.feature_name, + exc, + ) + return False diff --git a/sunbeam-python/tests/functional/feature/features/validation.py b/sunbeam-python/tests/functional/feature/features/validation.py new file mode 100644 index 000000000..c752af5f9 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/validation.py @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for validation feature.""" + +import logging +import subprocess + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class ValidationTest(BaseFeatureTest): + """Test validation feature enablement/disablement.""" + + feature_name = "validation" + expected_applications: list[str] = [] + timeout_seconds = 900 + + def verify_validate_feature_behavior(self) -> None: + """Validate that the validation CLI is usable.""" + logger.info("Verifying validation feature via `sunbeam validation profiles`...") + try: + subprocess.run( + ["sunbeam", "validation", "profiles"], + capture_output=True, + text=True, + timeout=60, + check=True, + ) + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying validation feature: %s", exc) + raise AssertionError( + f"Validation feature verification failed: {exc}" + ) from exc + + logger.info("Validation feature verified via `sunbeam validation profiles`") diff --git a/sunbeam-python/tests/functional/feature/features/vault.py b/sunbeam-python/tests/functional/feature/features/vault.py new file mode 100644 index 000000000..26b71a5d6 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/features/vault.py @@ -0,0 +1,33 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Test for vault feature. + +Vault provides the HashiCorp Vault service used by other features. +Functionality is validated via the `sunbeam vault status` command. +""" + +import logging + +from .base import BaseFeatureTest + +logger = logging.getLogger(__name__) + + +class VaultTest(BaseFeatureTest): + """Test vault feature enablement/disablement.""" + + feature_name = "vault" + expected_applications: list[str] = [] + timeout_seconds = 600 + + def verify_validate_feature_behavior(self) -> None: + """Validate that Vault is reachable via sunbeam.""" + logger.info("Verifying Vault status via `sunbeam vault status`...") + try: + self.sunbeam.run(["vault", "status"]) + except Exception as exc: # noqa: BLE001 + logger.warning("Error while verifying Vault service: %s", exc) + raise AssertionError(f"Vault service verification failed: {exc}") from exc + + logger.info("Vault service verified via `sunbeam vault status`") diff --git a/sunbeam-python/tests/functional/feature/pytest.ini b/sunbeam-python/tests/functional/feature/pytest.ini new file mode 100644 index 000000000..acba1481b --- /dev/null +++ b/sunbeam-python/tests/functional/feature/pytest.ini @@ -0,0 +1,22 @@ +[pytest] +# Pytest configuration for feature functional tests + +# Test discovery patterns +python_files = test_*.py +python_classes = Test* +python_functions = test_* + +markers = + functional: marks tests as functional (deselect with '-m "not functional"') + slow: marks tests as slow (deselect with '-m "not slow"') + +addopts = + -v + --tb=short + --strict-markers + +timeout = 1800 + +log_cli = true +log_cli_level = INFO + diff --git a/sunbeam-python/tests/functional/feature/requirements.txt b/sunbeam-python/tests/functional/feature/requirements.txt new file mode 100644 index 000000000..7ef618ac2 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/requirements.txt @@ -0,0 +1,4 @@ +pytest>=7.4.0 +pytest-timeout>=2.1.0 +pyyaml>=6.0 +jubilant>=1.0.0 diff --git a/sunbeam-python/tests/functional/feature/test_config.yaml.example b/sunbeam-python/tests/functional/feature/test_config.yaml.example new file mode 100644 index 000000000..2507a8535 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/test_config.yaml.example @@ -0,0 +1,13 @@ +# Sunbeam Feature Functional Test Configuration +# Copy this file to test_config.yaml and fill in your values + +sunbeam: + # Deployment name in Sunbeam (from `sunbeam deployment list`) + deployment_name: "ps6" + +juju: + # Juju model name (default: "openstack") + model: "openstack" + # Juju controller (auto-detected from sunbeam if not specified) + # controller: "your-controller" + diff --git a/sunbeam-python/tests/functional/feature/test_features.py b/sunbeam-python/tests/functional/feature/test_features.py new file mode 100644 index 000000000..d6522f627 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/test_features.py @@ -0,0 +1,186 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Functional tests for Sunbeam features. + +These tests connect to an existing Sunbeam cluster and test feature +enablement/disablement lifecycle. +""" + +import logging + +import pytest + +from .features.baremetal import BaremetalTest +from .features.caas import CaaSTest +from .features.dns import DnsTest +from .features.images_sync import ImagesSyncTest +from .features.instance_recovery import InstanceRecoveryTest +from .features.ldap import LdapTest +from .features.loadbalancer import LoadbalancerTest +from .features.maintenance import MaintenanceTest +from .features.observability import ObservabilityTest +from .features.orchestration import OrchestrationTest +from .features.pro import ProTest +from .features.resource_optimization import ResourceOptimizationTest +from .features.secrets import SecretsTest +from .features.shared_filesystem import SharedFilesystemTest +from .features.telemetry import TelemetryTest +from .features.tls import TlsCaTest +from .features.validation import ValidationTest +from .features.vault import VaultTest + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +@pytest.mark.functional +def test_instance_recovery(sunbeam_client, juju_client, test_config): + """Test instance-recovery feature lifecycle (enable/disable with verification).""" + feature_test = InstanceRecoveryTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Instance recovery feature test failed" + + +@pytest.mark.functional +@pytest.mark.skip( + reason=( + "Baremetal feature test is present but intentionally disabled in the " + "current feature flow (enable later when ready)." + ) +) +def test_baremetal(sunbeam_client, juju_client, test_config): + """Test baremetal feature lifecycle (enable/disable only).""" + feature_test = BaremetalTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Baremetal feature test failed" + + +@pytest.mark.functional +def test_caas(sunbeam_client, juju_client, test_config): + """Test caas feature lifecycle (enable/disable only).""" + feature_test = CaaSTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "CaaS feature test failed" + + +@pytest.mark.functional +def test_dns(sunbeam_client, juju_client, test_config): + """Test dns feature lifecycle (enable/disable only).""" + feature_test = DnsTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "DNS feature test failed" + + +@pytest.mark.functional +def test_images_sync(sunbeam_client, juju_client, test_config): + """Test images-sync feature lifecycle (enable/disable only).""" + feature_test = ImagesSyncTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Images-sync feature test failed" + + +@pytest.mark.functional +@pytest.mark.skip( + reason=( + "LDAP feature test is present but intentionally disabled in the " + "current feature flow (enable later when ready)." + ) +) +def test_ldap(sunbeam_client, juju_client, test_config): + """Test ldap feature lifecycle (enable/disable only).""" + feature_test = LdapTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "LDAP feature test failed" + + +@pytest.mark.functional +def test_loadbalancer(sunbeam_client, juju_client, test_config): + """Test loadbalancer feature lifecycle (enable/disable only).""" + feature_test = LoadbalancerTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Loadbalancer feature test failed" + + +@pytest.mark.functional +def test_orchestration(sunbeam_client, juju_client, test_config): + """Test orchestration feature lifecycle (enable/disable only).""" + feature_test = OrchestrationTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Orchestration feature test failed" + + +@pytest.mark.functional +def test_resource_optimization(sunbeam_client, juju_client, test_config): + """Test resource-optimization feature lifecycle (enable/disable only).""" + feature_test = ResourceOptimizationTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), ( + "Resource-optimization feature test failed" + ) + + +@pytest.mark.functional +def test_shared_filesystem(sunbeam_client, juju_client, test_config): + """Test shared-filesystem feature lifecycle (enable/disable only).""" + feature_test = SharedFilesystemTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Shared-filesystem feature test failed" + + +@pytest.mark.functional +def test_secrets(sunbeam_client, juju_client, test_config): + """Test secrets feature lifecycle (enable/disable only).""" + feature_test = SecretsTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Secrets feature test failed" + + +@pytest.mark.functional +def test_telemetry(sunbeam_client, juju_client, test_config): + """Test telemetry feature lifecycle (enable/disable only).""" + feature_test = TelemetryTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Telemetry feature test failed" + + +@pytest.mark.functional +def test_observability(sunbeam_client, juju_client, test_config): + """Test observability feature lifecycle (enable/disable only).""" + feature_test = ObservabilityTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Observability feature test failed" + + +@pytest.mark.functional +@pytest.mark.skip( + reason=( + "Maintenance feature test is present but intentionally disabled in the " + "current feature flow (enable later when ready)." + ) +) +def test_maintenance(sunbeam_client, juju_client, test_config): + """Test maintenance feature lifecycle (enable/disable only).""" + feature_test = MaintenanceTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Maintenance feature test failed" + + +@pytest.mark.functional +@pytest.mark.skip( + reason=( + "Pro feature test is present but intentionally disabled in the " + "current feature flow (enable later when ready)." + ) +) +def test_pro(sunbeam_client, juju_client, test_config): + """Test pro feature lifecycle (enable/disable only).""" + feature_test = ProTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Pro feature test failed" + + +@pytest.mark.functional +def test_tls_ca(sunbeam_client, juju_client, test_config): + """Test TLS CA mode lifecycle (enable/disable with verification).""" + feature_test = TlsCaTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "TLS CA feature test failed" + + +@pytest.mark.functional +def test_vault(sunbeam_client, juju_client, test_config): + """Test vault feature lifecycle (enable/disable only).""" + feature_test = VaultTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Vault feature test failed" + + +@pytest.mark.functional +def test_validation(sunbeam_client, juju_client, test_config): + """Test validation feature lifecycle (enable/disable only).""" + feature_test = ValidationTest(sunbeam_client, juju_client, test_config) + assert feature_test.run_full_lifecycle(), "Validation feature test failed" diff --git a/sunbeam-python/tests/functional/feature/utils/__init__.py b/sunbeam-python/tests/functional/feature/utils/__init__.py new file mode 100644 index 000000000..42e1068cd --- /dev/null +++ b/sunbeam-python/tests/functional/feature/utils/__init__.py @@ -0,0 +1,4 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Utility wrappers for Sunbeam feature functional tests.""" diff --git a/sunbeam-python/tests/functional/feature/utils/juju.py b/sunbeam-python/tests/functional/feature/utils/juju.py new file mode 100644 index 000000000..448e85da2 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/utils/juju.py @@ -0,0 +1,135 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Juju CLI wrapper using Jubilant library for feature functional tests.""" + +import json +import logging +from typing import Dict, List, Optional, Set + +from jubilant import Juju + +logger = logging.getLogger(__name__) + + +class JujuClient: + """Client for interacting with Juju using Jubilant.""" + + def __init__(self, model: str = "openstack", controller: Optional[str] = None): + self.model = model + self.controller = controller + self._juju: Optional[Juju] = None + + @property + def juju(self) -> Juju: + """Get or create Jubilant Juju instance.""" + if self._juju is None: + self._juju = Juju() + if self.model: + try: + self._juju.cli("switch", self.model) + except Exception as exc: # noqa: BLE001 + # Log but continue - the model might already be active + logger.debug("Could not switch to model %s: %s", self.model, exc) + return self._juju + + def is_connected(self) -> bool: + """Check if we can connect to Juju.""" + result = self.juju.cli("status", "--format", "json") + return bool(result) + + def get_applications(self) -> Set[str]: + """Get list of all applications in the model.""" + result_str = self.juju.cli("status", "--format", "json") + status = json.loads(result_str) + applications: Set[str] = set() + + if "applications" in status: + applications.update(status["applications"].keys()) + + return applications + + def get_units(self) -> Set[str]: + """Get list of all units in the model.""" + result_str = self.juju.cli("status", "--format", "json") + status = json.loads(result_str) + units: Set[str] = set() + + if "applications" in status: + for app_data in status["applications"].values(): + if "units" in app_data: + for unit_name in app_data["units"].keys(): + units.add(unit_name) + + return units + + def has_application(self, application_name: str) -> bool: + """Check if an application exists.""" + applications = self.get_applications() + return application_name in applications + + def has_unit(self, unit_name: str) -> bool: + """Check if a unit exists.""" + units = self.get_units() + return unit_name in units + + def wait_for_application(self, application_name: str, timeout: int = 300) -> bool: + """Wait for an application to appear using Jubilant's wait mechanism.""" + if self.has_application(application_name): + logger.info( + "Application '%s' already exists, skipping wait", + application_name, + ) + return True + + def app_exists(status) -> bool: + return hasattr(status, "apps") and application_name in status.apps + + self.juju.wait(app_exists, timeout=timeout, delay=1.0) + return True + + def wait_for_unit(self, unit_name: str, timeout: int = 300) -> bool: + """Wait for a unit to appear using Jubilant's wait mechanism.""" + if self.has_unit(unit_name): + logger.info("Unit '%s' already exists, skipping wait", unit_name) + return True + + def unit_exists(status) -> bool: + if not hasattr(status, "apps"): + return False + for app_data in status.apps.values(): + if hasattr(app_data, "units") and unit_name in app_data.units: + return True + return False + + self.juju.wait(unit_exists, timeout=timeout, delay=1.0) + return True + + def wait_for_application_ready( + self, + application_name: str, + timeout: int = 600, + ) -> bool: + """Wait for an application to be in 'active' state.""" + + def app_active(status) -> bool: + if not hasattr(status, "apps") or application_name not in status.apps: + return False + app = status.apps[application_name] + return hasattr(app, "app_status") and app.app_status.current == "active" + + self.juju.wait(app_active, timeout=timeout, delay=1.0) + return True + + def verify_applications_exist( + self, + expected_applications: List[str], + ) -> Dict[str, bool]: + """Verify that expected applications exist.""" + actual_applications = self.get_applications() + results: Dict[str, bool] = {} + + for app in expected_applications: + results[app] = app in actual_applications + + return results diff --git a/sunbeam-python/tests/functional/feature/utils/sunbeam.py b/sunbeam-python/tests/functional/feature/utils/sunbeam.py new file mode 100644 index 000000000..7982f3de4 --- /dev/null +++ b/sunbeam-python/tests/functional/feature/utils/sunbeam.py @@ -0,0 +1,87 @@ +# SPDX-FileCopyrightText: 2024 - Canonical Ltd +# SPDX-License-Identifier: Apache-2.0 + +"""Sunbeam CLI wrapper for feature functional tests.""" + +import logging +import subprocess +from typing import List, Optional + +logger = logging.getLogger(__name__) + + +class SunbeamClient: + """Client for interacting with Sunbeam CLI.""" + + def __init__(self, deployment_name: str): + self.deployment_name = deployment_name + self._sunbeam_cmd = "/snap/bin/sunbeam" + + def _run_command(self, command: List[str]) -> subprocess.CompletedProcess: + """Run a sunbeam command and return the result.""" + full_command = [self._sunbeam_cmd] + command + logger.debug("Running: %s", " ".join(full_command)) + + result = subprocess.run( + full_command, + capture_output=True, + text=True, + check=False, + timeout=3600, + ) + + if result.returncode != 0: + logger.error( + "Command failed with exit code %d: %s", + result.returncode, + " ".join(full_command), + ) + if result.stderr: + logger.error("stderr: %s", result.stderr) + if result.stdout: + logger.error("stdout: %s", result.stdout) + result.check_returncode() + + return result + + def run(self, command: List[str]) -> subprocess.CompletedProcess: + """Public helper to run arbitrary sunbeam subcommands.""" + return self._run_command(command) + + def is_connected(self) -> bool: + """Check if we can connect to the Sunbeam deployment.""" + result = subprocess.run( + ["sunbeam", "deployment", "list"], + capture_output=True, + text=True, + timeout=30, + ) + return result.returncode == 0 and self.deployment_name in result.stdout + + def enable_feature( + self, + feature_name: str, + extra_args: Optional[List[str]] = None, + ) -> bool: + """Enable a Sunbeam feature.""" + cmd: List[str] = ["enable", feature_name] + if extra_args: + cmd.extend(extra_args) + + self._run_command(cmd) + logger.info("Feature '%s' enabled successfully", feature_name) + return True + + def disable_feature( + self, + feature_name: str, + extra_args: Optional[List[str]] = None, + ) -> bool: + """Disable a Sunbeam feature.""" + cmd: List[str] = ["disable", feature_name] + if extra_args: + cmd.extend(extra_args) + + self._run_command(cmd) + logger.info("Feature '%s' disabled successfully", feature_name) + return True diff --git a/sunbeam-python/tox.ini b/sunbeam-python/tox.ini index 8dc81acfa..2e89e14a8 100644 --- a/sunbeam-python/tox.ini +++ b/sunbeam-python/tox.ini @@ -1,9 +1,6 @@ [tox] envlist = unit,pep8,mypy skipsdist = True -# Automatic envs (pyXX) will only use the python version appropriate to that -# env and ignore basepython inherited from [testenv] if we set -# ignore_basepython_conflict. ignore_basepython_conflict = True [vars] @@ -25,14 +22,8 @@ setenv = OS_STDOUT_CAPTURE=1 description = Sunbeam unit tests commands = uv run {[vars]uv_flags} python -m pytest -vv tests/unit {posargs} -# The functional tests may have specific hardware requirements and are currently -# skipped by default. [testenv:functional] description = Sunbeam functional tests -# The snap can't access /tmp, we'll need to place manifests and other temporary -# files in the home directory. At the same time, we need to expose USER/LOGNAME, -# otherwise the Sunbeam group won't be initialized correctly and the Sunbeam -# commands will fail due to missing privileges. passenv = USER LOGNAME USERNAME @@ -42,6 +33,29 @@ commands = uv run {[vars]uv_flags} \ --basetemp={env:HOME}/.local/share/openstack/tmp \ {posargs} +[testenv:functional-feature] +description = Sunbeam feature functional tests (existing deployment) +passenv = USER + LOGNAME + USERNAME + HOME +commands = uv run {[vars]uv_flags} \ + python -m pytest -s -vv tests/functional/feature \ + --config=test_config.yaml \ + --basetemp={env:HOME}/.local/share/openstack/tmp \ + {posargs} + +[testenv:functional-chaos] +description = Sunbeam Chaos Mesh functional tests +passenv = USER + LOGNAME + USERNAME + HOME +commands = uv run {[vars]uv_flags} \ + python -m pytest -s -vv tests/functional/chaos \ + --basetemp={env:HOME}/.local/share/openstack/tmp \ + {posargs} + [testenv:fmt] description = Apply coding style standards to code deps = @@ -60,9 +74,6 @@ commands = [testenv:mypy] commands = uv run {[vars]uv_flags} mypy {[vars]src_path}/sunbeam - # TODO: consider uncommenting the following line once - # the unit tests pass the mypy check. - # uv run {[vars]uv_flags} mypy {[vars]tst_path}/unit uv run {[vars]uv_flags} mypy {[vars]tst_path}/functional [testenv:cover] @@ -87,7 +98,6 @@ deps = commands = sphinx-build -a -E -W -d doc/build/doctrees -b html doc/source doc/build/html sphinx-build -a -E -W -d doc/build/doctrees -b man doc/source doc/build/man - # Validate redirects (must be done after the docs build whereto doc/build/html/.htaccess doc/test/redirect-tests.txt [testenv:releasenotes]