From 030fe83b3c0f0f9ebad3e167c7d9f3ee4ddeaebe Mon Sep 17 00:00:00 2001 From: venu <236371043+Venu-p1@users.noreply.github.com> Date: Tue, 30 Jun 2026 23:48:34 +0530 Subject: [PATCH 1/7] Add unit test coverage for deploy, cleanup, validate retry, and git-history regression tests Signed-off-by: venu <236371043+Venu-p1@users.noreply.github.com> --- .../catalog/test_fixtures/catalog_rhel.json | 115 +++ build_stream/tests/end_to_end/api/conftest.py | 672 --------------- .../tests/end_to_end/api/test_api_flow_e2e.py | 557 ------------- .../end_to_end/api/test_build_image_e2e.py | 499 ------------ .../api/test_generate_input_files_e2e.py | 482 ----------- .../end_to_end/api/test_parse_catalog_e2e.py | 768 ------------------ .../tests/end_to_end/api/test_register_e2e.py | 443 ---------- .../tests/end_to_end/api/test_token_e2e.py | 494 ----------- .../api/build_image/test_build_image_api.py | 10 +- .../test_generate_input_files_api.py | 28 +- .../api/jobs/test_create_job_api.py | 5 +- .../api/jobs/test_delete_job_api.py | 26 +- .../integration/api/jobs/test_get_job_api.py | 2 +- .../parse_catalog/test_parse_catalog_api.py | 19 +- .../api/validate/test_validate_api.py | 2 +- .../test_image_group_data_model.py | 11 +- .../tests/unit/api/deploy/test_routes.py | 91 +++ .../tests/unit/api/local_repo/test_routes.py | 88 +- .../tests/unit/api/upload/__init__.py | 0 .../tests/unit/api/upload/test_routes.py | 240 ++++++ .../tests/unit/core/catalog/test_adapter.py | 94 +++ .../core/catalog/test_extract_bundle_name.py | 70 ++ .../catalog/test_generate_software_config.py | 8 +- .../tests/unit/core/cleanup/__init__.py | 0 .../unit/core/cleanup/test_exceptions.py | 162 ++++ .../unit/core/cleanup/test_s3_service.py | 100 +++ .../tests/unit/core/deploy/__init__.py | 0 .../tests/unit/core/deploy/test_entities.py | 124 +++ .../tests/unit/core/deploy/test_exceptions.py | 92 +++ .../tests/unit/core/deploy/test_services.py | 88 ++ .../test_generate_input_files_command.py | 30 +- .../orchestrator/common/test_result_poller.py | 204 ++++- .../validate/test_validate_use_case.py | 204 ++++- 33 files changed, 1754 insertions(+), 3974 deletions(-) create mode 100644 build_stream/core/catalog/test_fixtures/catalog_rhel.json delete mode 100644 build_stream/tests/end_to_end/api/conftest.py delete mode 100644 build_stream/tests/end_to_end/api/test_api_flow_e2e.py delete mode 100644 build_stream/tests/end_to_end/api/test_build_image_e2e.py delete mode 100644 build_stream/tests/end_to_end/api/test_generate_input_files_e2e.py delete mode 100644 build_stream/tests/end_to_end/api/test_parse_catalog_e2e.py delete mode 100644 build_stream/tests/end_to_end/api/test_register_e2e.py delete mode 100644 build_stream/tests/end_to_end/api/test_token_e2e.py create mode 100644 build_stream/tests/unit/api/upload/__init__.py create mode 100644 build_stream/tests/unit/api/upload/test_routes.py create mode 100644 build_stream/tests/unit/core/catalog/test_adapter.py create mode 100644 build_stream/tests/unit/core/catalog/test_extract_bundle_name.py create mode 100644 build_stream/tests/unit/core/cleanup/__init__.py create mode 100644 build_stream/tests/unit/core/cleanup/test_exceptions.py create mode 100644 build_stream/tests/unit/core/cleanup/test_s3_service.py create mode 100644 build_stream/tests/unit/core/deploy/__init__.py create mode 100644 build_stream/tests/unit/core/deploy/test_entities.py create mode 100644 build_stream/tests/unit/core/deploy/test_exceptions.py create mode 100644 build_stream/tests/unit/core/deploy/test_services.py diff --git a/build_stream/core/catalog/test_fixtures/catalog_rhel.json b/build_stream/core/catalog/test_fixtures/catalog_rhel.json new file mode 100644 index 0000000000..db1d6aeab0 --- /dev/null +++ b/build_stream/core/catalog/test_fixtures/catalog_rhel.json @@ -0,0 +1,115 @@ +{ + "Catalog": { + "Name": "RHEL Test Catalog", + "Version": "1.0.0", + "Identifier": "catalog-rhel-test", + "FunctionalLayer": [ + { + "Name": "slurm", + "FunctionalPackages": ["fp-slurm-slurmctld", "fp-slurm-slurmd"] + }, + { + "Name": "kubernetes", + "FunctionalPackages": ["fp-kubectl", "fp-kubeadm"] + } + ], + "BaseOS": [ + { + "Name": "RHEL", + "Version": "10.0", + "osPackages": ["os-baseos-pkg"] + } + ], + "Infrastructure": [ + { + "Name": "network", + "InfrastructurePackages": ["infra-network-tools"] + } + ], + "Miscellaneous": [], + "Drivers": [ + { + "Name": "nvidia", + "DriverPackages": ["drv-nvidia-driver"] + } + ], + "DriverPackages": { + "drv-nvidia-driver": { + "Name": "nvidia-driver", + "Version": "580.95", + "Uri": "https://example.com/nvidia-driver.rpm", + "Architecture": ["x86_64"], + "Type": "rpm", + "Config": { + "DriverBrand": "nvidia", + "DriverType": "gpu" + } + } + }, + "FunctionalPackages": { + "fp-slurm-slurmctld": { + "Name": "slurm-slurmctld", + "Version": "24.05.5", + "SupportedOS": [{"Name": "RHEL", "Version": "10.0"}], + "Sources": [ + {"Architecture": "x86_64", "RepoName": "x86_64_slurm_custom"}, + {"Architecture": "aarch64", "RepoName": "aarch64_slurm_custom"} + ], + "Architecture": ["x86_64", "aarch64"], + "Type": "rpm" + }, + "fp-slurm-slurmd": { + "Name": "slurm-slurmd", + "Version": "24.05.5", + "SupportedOS": [{"Name": "RHEL", "Version": "10.0"}], + "Sources": [ + {"Architecture": "x86_64", "RepoName": "x86_64_slurm_custom"}, + {"Architecture": "aarch64", "RepoName": "aarch64_slurm_custom"} + ], + "Architecture": ["x86_64", "aarch64"], + "Type": "rpm" + }, + "fp-kubectl": { + "Name": "kubectl", + "Version": "1.35.1", + "SupportedOS": [{"Name": "RHEL", "Version": "10.0"}], + "Sources": [ + {"Architecture": "x86_64", "RepoName": "kubernetes"} + ], + "Architecture": ["x86_64"], + "Type": "rpm" + }, + "fp-kubeadm": { + "Name": "kubeadm", + "Version": "1.35.1", + "SupportedOS": [{"Name": "RHEL", "Version": "10.0"}], + "Sources": [ + {"Architecture": "x86_64", "RepoName": "kubernetes"} + ], + "Architecture": ["x86_64"], + "Type": "rpm" + } + }, + "OSPackages": { + "os-baseos-pkg": { + "Name": "baseos-release", + "Version": "10.0", + "SupportedOS": [{"Name": "RHEL", "Version": "10.0"}], + "Sources": [ + {"Architecture": "x86_64", "RepoName": "x86_64_baseos"} + ], + "Architecture": ["x86_64"], + "Type": "rpm" + } + }, + "InfrastructurePackages": { + "infra-network-tools": { + "Name": "network-tools", + "Version": "1.0", + "Type": "rpm", + "Architecture": ["x86_64"], + "SupportedFunctions": [{"Name": "network"}] + } + } + } +} diff --git a/build_stream/tests/end_to_end/api/conftest.py b/build_stream/tests/end_to_end/api/conftest.py deleted file mode 100644 index 2e87e18335..0000000000 --- a/build_stream/tests/end_to_end/api/conftest.py +++ /dev/null @@ -1,672 +0,0 @@ -# Copyright 2026 Dell Inc. or its subsidiaries. All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Pytest fixtures for integration tests with real Ansible Vault.""" - -# pylint: disable=redefined-outer-name,consider-using-with - -import base64 -import logging -import os -import secrets -import shutil -import signal -import socket -import string -import subprocess -import tempfile -import time -from pathlib import Path -from typing import Dict, Generator, Optional - -import httpx -import pytest -import yaml -from argon2 import PasswordHasher, Type # noqa: E0611 pylint: disable=no-name-in-module - -# Configure logging for integration tests -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", -) -logger = logging.getLogger("integration_tests") - - -def generate_secure_test_password(length: int = 24) -> str: - """Generate a secure password for integration tests. - - Args: - length: Length of the password (default: 24 for extra security) - - Returns: - Secure random password - """ - # Use stronger character set for integration tests - lowercase = string.ascii_lowercase - uppercase = string.ascii_uppercase - digits = string.digits - special = "!@#$%^&*()_+-=[]{}|;:,.<>?" - - # Ensure minimum security requirements - if length < 16: - raise ValueError("Password length must be at least 16 characters") - - # Start with one of each required character type - password = [ - secrets.choice(lowercase), - secrets.choice(uppercase), - secrets.choice(digits), - secrets.choice(special), - ] - - # Fill remaining length - all_chars = lowercase + uppercase + digits + special - for _ in range(length - 4): - password.append(secrets.choice(all_chars)) - - # Shuffle to avoid predictable pattern - secrets.SystemRandom().shuffle(password) - - return ''.join(password) - - -def generate_test_client_secret(length: int = 32) -> str: - """Generate a test client secret with proper bld_s_ prefix. - - Args: - length: Total length of the secret including prefix (default: 32) - - Returns: - Test client secret with bld_s_ prefix - """ - if length < 8: - raise ValueError("Client secret length must be at least 8 characters") - - # Generate random part (subtract 6 for "bld_s_" prefix) - random_part_length = max(8, length - 6) - random_part = generate_secure_test_password(random_part_length) - - return f"bld_s_{random_part}" - - -def generate_invalid_client_id() -> str: - """Generate an invalid client ID for testing (missing bld_ prefix). - - Returns: - Invalid client ID without proper prefix - """ - return ( - "invalid_client_id_" + - ''.join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(8)) - ) - - -def generate_invalid_client_secret() -> str: - """Generate an invalid client secret for testing (missing bld_s_ prefix). - - Returns: - Invalid client secret without proper prefix - """ - return ( - "invalid_secret_" + - ''.join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(8)) - ) - - -class IntegrationTestConfig: - """Configuration for integration tests.""" - - # Username is not a secret - AUTH_USERNAME = "build_stream_registrar" - SERVER_HOST = "127.0.0.1" - SERVER_PORT = 18443 # Use different port to avoid conflicts - SERVER_STARTUP_TIMEOUT = 30 - - @classmethod - def get_vault_password(cls) -> str: - """Get a dynamically generated vault password. - - Returns: - Secure random vault password - """ - return generate_secure_test_password(24) - - @classmethod - def get_auth_password(cls) -> str: - """Get a dynamically generated auth password. - - Returns: - Secure random auth password - """ - return generate_secure_test_password(24) - - -class VaultManager: # noqa: R0902 pylint: disable=too-many-instance-attributes - """Manages Ansible Vault setup and teardown for integration tests.""" - - def __init__(self, base_dir: str): - """Initialize vault manager. - - Args: - base_dir: Base directory for test vault files. - """ - self.base_dir = Path(base_dir) - self.vault_dir = self.base_dir / "vault" - self.vault_file = self.vault_dir / "build_stream_oauth_credentials.yml" - self.vault_pass_file = self.base_dir / ".vault_pass" - self.keys_dir = self.base_dir / "keys" - self.private_key_file = self.keys_dir / "jwt_private.pem" - self.public_key_file = self.keys_dir / "jwt_public.pem" - self._hasher = PasswordHasher( - time_cost=3, - memory_cost=65536, - parallelism=4, - hash_len=32, - salt_len=16, - type=Type.ID, - ) - - def setup(self, username: str, password: str) -> None: - """Set up vault with initial credentials. - - Args: - username: Registration username. - password: Registration password. - """ - logger.info("Setting up Ansible Vault...") - logger.info(" Vault directory: %s", self.vault_dir) - logger.info(" Vault file: %s", self.vault_file) - logger.info(" Vault password file: %s", self.vault_pass_file) - - self.vault_dir.mkdir(parents=True, exist_ok=True) - logger.info(" Created vault directory") - - self.vault_pass_file.write_text(IntegrationTestConfig.get_vault_password()) - self.vault_pass_file.chmod(0o600) - logger.info(" Created vault password file") - - logger.info(" Generating Argon2id password hash...") - password_hash = self._hasher.hash(password) - - vault_content = { - "auth_registration": { - "username": username, - "password_hash": password_hash, - }, - "oauth_clients": {}, - } - - with tempfile.NamedTemporaryFile( - mode="w", suffix=".yml", delete=False - ) as temp_file: - yaml.safe_dump(vault_content, temp_file, default_flow_style=False) - temp_path = temp_file.name - - try: - logger.info(" Encrypting vault with ansible-vault...") - subprocess.run( - [ - "ansible-vault", - "encrypt", - temp_path, - "--vault-password-file", - str(self.vault_pass_file), - "--encrypt-vault-id", - "default", - ], - check=True, - capture_output=True, - ) - - shutil.move(temp_path, str(self.vault_file)) - self.vault_file.chmod(0o600) - logger.info(" Vault encrypted and saved successfully") - finally: - if os.path.exists(temp_path): - os.unlink(temp_path) - - logger.info("Vault setup complete") - - # Generate JWT keys for token signing - self._generate_jwt_keys() - - def _generate_jwt_keys(self) -> None: - """Generate RSA key pair for JWT signing in e2e tests.""" - logger.info("Generating JWT keys for e2e tests...") - logger.info(" Keys directory: %s", self.keys_dir) - - self.keys_dir.mkdir(parents=True, exist_ok=True) - - # Generate RSA private key (2048-bit for faster tests) - subprocess.run( - [ - "openssl", "genrsa", - "-out", str(self.private_key_file), - "2048", - ], - check=True, - capture_output=True, - ) - self.private_key_file.chmod(0o600) - logger.info(" Generated private key: %s", self.private_key_file) - - # Extract public key - subprocess.run( - [ - "openssl", "rsa", - "-in", str(self.private_key_file), - "-pubout", - "-out", str(self.public_key_file), - ], - check=True, - capture_output=True, - ) - self.public_key_file.chmod(0o644) - logger.info(" Generated public key: %s", self.public_key_file) - logger.info("JWT keys generated successfully") - - def cleanup(self) -> None: - """Clean up vault files.""" - logger.info("Cleaning up vault files at: %s", self.base_dir) - if self.base_dir.exists(): - shutil.rmtree(self.base_dir) - logger.info("Vault cleanup complete") - - -class ServerManager: - """Manages FastAPI server lifecycle for integration tests.""" - - REQUIRED_PACKAGES = [ - "fastapi", - "uvicorn", - "pydantic", - "PyJWT", - "argon2-cffi", - "pyyaml", - "httpx", - "python-multipart", - "jsonschema", - "ansible", - "cryptography", - "dependency-injector", - ] - - def __init__( # noqa: R0913,R0917 pylint: disable=too-many-arguments,too-many-positional-arguments - self, - host: str, - port: int, - vault_manager: VaultManager, # noqa: W0621 - project_dir: str, # noqa: W0621 - venv_dir: str, # noqa: W0621 - ): - """Initialize server manager. - - Args: - host: Server host. - port: Server port. - vault_manager: Vault manager instance. - project_dir: Path to build_stream project directory. - venv_dir: Path to virtual environment directory. - """ - self.host = host - self.port = port - self.vault_manager = vault_manager - self.project_dir = project_dir - self.venv_dir = Path(venv_dir) - self.process: Optional[subprocess.Popen] = None - - def _setup_venv(self) -> None: - """Create virtual environment and install dependencies.""" - logger.info("Setting up Python virtual environment...") - logger.info(" Venv directory: %s", self.venv_dir) - - if not self.venv_dir.exists(): - logger.info(" Creating virtual environment...") - subprocess.run( - ["python3", "-m", "venv", str(self.venv_dir)], - check=True, - capture_output=True, - ) - logger.info(" Virtual environment created") - else: - logger.info(" Virtual environment already exists") - - pip_path = self.venv_dir / "bin" / "pip" - logger.info(" Upgrading pip...") - subprocess.run( - [str(pip_path), "install", "--upgrade", "pip", "-q"], - check=True, - capture_output=True, - ) - - logger.info(" Installing dependencies: %s", ", ".join(self.REQUIRED_PACKAGES)) - subprocess.run( - [str(pip_path), "install", "-q"] + self.REQUIRED_PACKAGES, - check=True, - capture_output=True, - ) - logger.info(" Dependencies installed successfully") - - @property - def python_path(self) -> str: - """Get path to Python executable in virtual environment.""" - return str(self.venv_dir / "bin" / "python") - - def _is_port_in_use(self) -> bool: - """Check if the port is already in use.""" - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - return s.connect_ex((self.host, self.port)) == 0 - - def _free_port(self) -> None: - """Free the port if it's in use.""" - if self._is_port_in_use(): - try: - result = subprocess.run( - ["lsof", "-t", f"-i:{self.port}"], - capture_output=True, - text=True, - check=False, - ) - if result.stdout.strip(): - for pid in result.stdout.strip().split("\n"): - try: - os.kill(int(pid), signal.SIGKILL) - except (ProcessLookupError, ValueError): - pass - time.sleep(1) - except FileNotFoundError: - pass - - def start(self) -> None: - """Start the FastAPI server.""" - logger.info("Starting FastAPI server...") - self._setup_venv() - - logger.info(" Freeing port %d if in use...", self.port) - self._free_port() - - logger.info(" Configuring server environment variables...") - env = os.environ.copy() - env.update({ - "HOST": self.host, - "PORT": str(self.port), - "ANSIBLE_VAULT_PASSWORD_FILE": str(self.vault_manager.vault_pass_file), - "OAUTH_CLIENTS_VAULT_PATH": str(self.vault_manager.vault_file), - "AUTH_CONFIG_VAULT_PATH": str(self.vault_manager.vault_file), - "JWT_PRIVATE_KEY_PATH": str(self.vault_manager.private_key_file), - "JWT_PUBLIC_KEY_PATH": str(self.vault_manager.public_key_file), - "LOG_LEVEL": "DEBUG", - "PYTHONPATH": str(self.project_dir), - }) - logger.info(" HOST=%s", self.host) - logger.info(" PORT=%s", self.port) - logger.info(" ANSIBLE_VAULT_PASSWORD_FILE=%s", self.vault_manager.vault_pass_file) - logger.info(" OAUTH_CLIENTS_VAULT_PATH=%s", self.vault_manager.vault_file) - logger.info(" AUTH_CONFIG_VAULT_PATH=%s", self.vault_manager.vault_file) - logger.info(" JWT_PRIVATE_KEY_PATH=%s", self.vault_manager.private_key_file) - logger.info(" JWT_PUBLIC_KEY_PATH=%s", self.vault_manager.public_key_file) - logger.info(" LOG_LEVEL=DEBUG") - logger.info(" PYTHONPATH=%s", self.project_dir) - - logger.info(" Starting uvicorn server...") - logger.info(" Python: %s", self.python_path) - logger.info(" Working directory: %s", self.project_dir) - - # Process needs to be managed separately for start/stop lifecycle - # Cannot use 'with' statement as process must persist after method returns - self.process = subprocess.Popen( # noqa: R1732 - [ - self.python_path, - "-m", - "uvicorn", - "main:app", - "--host", - self.host, - "--port", - str(self.port), - ], - cwd=self.project_dir, - env=env, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) - logger.info(" Server process started with PID: %d", self.process.pid) - - self._wait_for_server() - - def _wait_for_server(self) -> None: - """Wait for server to be ready.""" - logger.info(" Waiting for server to be ready (timeout: %ds)...", - IntegrationTestConfig.SERVER_STARTUP_TIMEOUT) - - start_time = time.time() - while time.time() - start_time < IntegrationTestConfig.SERVER_STARTUP_TIMEOUT: - try: - response = httpx.get( - f"http://{self.host}:{self.port}/health", - timeout=1.0, - ) - if response.status_code == 200: - elapsed = time.time() - start_time - logger.info(" Server is ready! (took %.1fs)", elapsed) - logger.info(" Server URL: http://%s:%d", self.host, self.port) - return - except httpx.RequestError: - pass - time.sleep(0.5) - - # Log server output before stopping - if self.process: - logger.error("Server failed to start. Checking process output...") - if self.process.stdout: - stdout_output = self.process.stdout.read().decode() - logger.error("Server STDOUT:\n%s", stdout_output) - if self.process.stderr: - stderr_output = self.process.stderr.read().decode() - logger.error("Server STDERR:\n%s", stderr_output) - - # Check process return code - self.process.poll() - if self.process.returncode is not None: - logger.error("Server process exited with code: %s", self.process.returncode) - - self.stop() - raise RuntimeError( - f"Server failed to start within {IntegrationTestConfig.SERVER_STARTUP_TIMEOUT}s" - ) - - def stop(self) -> None: - """Stop the FastAPI server.""" - logger.info("Stopping FastAPI server...") - if self.process: - logger.info(" Terminating server process (PID: %d)...", self.process.pid) - self.process.terminate() - try: - self.process.wait(timeout=5) - logger.info(" Server stopped gracefully") - except subprocess.TimeoutExpired: - logger.info(" Server did not stop gracefully, killing...") - self.process.kill() - self.process.wait() - logger.info(" Server killed") - self.process = None - - self._free_port() - logger.info("Server shutdown complete") - - @property - def base_url(self) -> str: - """Get the server base URL.""" - return f"http://{self.host}:{self.port}" - - -@pytest.fixture(scope="module") -def integration_test_dir() -> Generator[str, None, None]: - """Create a temporary directory for integration test files. - - Yields: - Path to temporary directory. - """ - temp_dir = tempfile.mkdtemp(prefix="build_stream_integration_") - yield temp_dir - shutil.rmtree(temp_dir, ignore_errors=True) - - -@pytest.fixture(scope="module") -def vault_manager( - integration_test_dir: str, - auth_password: str, -) -> Generator[VaultManager, None, None]: # noqa: W0621 - """Create and configure vault manager. - - Args: - integration_test_dir: Temporary directory for test files. - auth_password: The auth password to use for vault setup. - - Yields: - Configured VaultManager instance. - """ - manager = VaultManager(integration_test_dir) - manager.setup( - username=IntegrationTestConfig.AUTH_USERNAME, - password=auth_password, - ) - yield manager - manager.cleanup() - - -@pytest.fixture(scope="module") -def project_dir() -> str: - """Get the build_stream project directory. - - Returns: - Path to build_stream project directory. - """ - return str(Path(__file__).parent.parent.parent.parent) - - -@pytest.fixture(scope="module") -def venv_dir(integration_test_dir: str) -> str: # noqa: W0621 - """Get path to virtual environment directory. - - Args: - integration_test_dir: Temporary directory for test files. - - Returns: - Path to virtual environment directory. - """ - return os.path.join(integration_test_dir, "venv") - - -@pytest.fixture(scope="module") -def server_manager( - vault_manager: VaultManager, # noqa: W0621 - project_dir: str, # noqa: W0621 - venv_dir: str, # noqa: W0621 -) -> Generator[ServerManager, None, None]: - """Create and manage the FastAPI server. - - Args: - vault_manager: Vault manager fixture. - project_dir: Project directory fixture. - venv_dir: Virtual environment directory fixture. - - Yields: - Running ServerManager instance. - """ - manager = ServerManager( - host=IntegrationTestConfig.SERVER_HOST, - port=IntegrationTestConfig.SERVER_PORT, - vault_manager=vault_manager, - project_dir=project_dir, - venv_dir=venv_dir, - ) - manager.start() - yield manager - manager.stop() - - -@pytest.fixture(scope="module") -def base_url(server_manager: ServerManager) -> str: # noqa: W0621 - """Get the server base URL. - - Args: - server_manager: Server manager fixture. - - Returns: - Server base URL. - """ - return server_manager.base_url - - -@pytest.fixture(scope="module") -def auth_password() -> str: - """Generate a single auth password for the entire test module. - - Returns: - Auth password to be used consistently across tests. - """ - return IntegrationTestConfig.get_auth_password() - - -@pytest.fixture -def valid_auth_header(auth_password: str) -> Dict[str, str]: # noqa: W0621 - """Create valid Basic Auth header. - - Args: - auth_password: The auth password to use. - - Returns: - Dictionary with Authorization header. - """ - credentials = base64.b64encode( - f"{IntegrationTestConfig.AUTH_USERNAME}:{auth_password}".encode() - ).decode() - return {"Authorization": f"Basic {credentials}"} - - -@pytest.fixture -def invalid_auth_header() -> Dict[str, str]: - """Create invalid Basic Auth header. - - Returns: - Dictionary with invalid Authorization header. - """ - credentials = base64.b64encode(b"wrong_user:wrong_password").decode() - return {"Authorization": f"Basic {credentials}"} - - -@pytest.fixture -def reset_vault( - vault_manager: VaultManager, - auth_password: str, -) -> Generator[None, None, None]: # noqa: W0621 - """Reset vault to initial state before and after test. - - Args: - vault_manager: Vault manager fixture. - auth_password: The auth password to use for vault setup. - - Yields: - None - """ - vault_manager.setup( - username=IntegrationTestConfig.AUTH_USERNAME, - password=auth_password, - ) - yield - vault_manager.setup( - username=IntegrationTestConfig.AUTH_USERNAME, - password=auth_password, - ) diff --git a/build_stream/tests/end_to_end/api/test_api_flow_e2e.py b/build_stream/tests/end_to_end/api/test_api_flow_e2e.py deleted file mode 100644 index 68601cec29..0000000000 --- a/build_stream/tests/end_to_end/api/test_api_flow_e2e.py +++ /dev/null @@ -1,557 +0,0 @@ -# Copyright 2026 Dell Inc. or its subsidiaries. All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""End-to-end integration tests for complete API workflow. - -These tests validate the complete OAuth2 authentication workflow from client registration -through token generation and validation. This test suite focuses on authentication -and authorization mechanisms, providing comprehensive coverage of the auth API. - -Usage: - pytest tests/integration/test_api_flow_e2e.py -v -m e2e - -Requirements: - - ansible-vault must be installed - - Tests require write access to create temporary vault files - - RSA keys must be available for JWT signing - -Test Flow: - 1. Health check - Verify server is running - 2. Client Registration - Register a new OAuth client with proper scopes - 3. Token Generation - Obtain access token using client credentials - 4. Token Validation - Verify JWT structure, uniqueness, and scope enforcement - 5. Error Handling - Test various failure scenarios and security validations - 6. Security Validation - Verify proper security measures are enforced - -Test Classes: - - TestCompleteAPIFlow: Main workflow tests (happy path scenarios) - - TestAPIFlowErrorHandling: Error scenario testing - - TestAPIFlowSecurityValidation: Security measure validation - -Key Features Tested: - - OAuth2 client registration with Basic Auth - - JWT token generation with client_credentials grant - - Scope-based authorization (catalog:read, catalog:write) - - Token uniqueness and validation - - Error handling and security measures - - Client credential format validation - - Maximum client limits enforcement - -Note: This test suite focuses specifically on authentication and authorization. -Protected API endpoints (like parse_catalog) are tested separately when implemented. -""" - -# pylint: disable=redefined-outer-name - -from typing import Dict, Optional - -import httpx -import pytest - -# Import helper functions from conftest -from tests.end_to_end.api.conftest import ( - generate_test_client_secret, - generate_invalid_client_id, - generate_invalid_client_secret, -) - - -class APIFlowContext: # noqa: R0902 pylint: disable=too-many-instance-attributes - """Context object to store state across API flow tests. - - This class maintains state between test steps, allowing tests to - share data like client credentials and access tokens. - - Attributes: - client_id: Registered client identifier. - client_secret: Registered client secret. - access_token: Generated JWT access token. - token_type: Token type (Bearer). - expires_in: Token expiration time in seconds. - scope: Granted scopes. - """ - - def __init__(self): - """Initialize empty context.""" - self.client_id: Optional[str] = None - self.client_secret: Optional[str] = None - self.client_name: Optional[str] = None - self.allowed_scopes: Optional[list] = None - self.access_token: Optional[str] = None - self.token_type: Optional[str] = None - self.expires_in: Optional[int] = None - self.scope: Optional[str] = None - - def has_client_credentials(self) -> bool: - """Check if client credentials are available.""" - return self.client_id is not None and self.client_secret is not None - - def has_access_token(self) -> bool: - """Check if access token is available.""" - return self.access_token is not None - - def get_auth_header(self) -> Dict[str, str]: - """Get Authorization header with Bearer token. - - Returns: - Dictionary with Authorization header. - - Raises: - ValueError: If access token is not available. - """ - if not self.has_access_token(): - raise ValueError("Access token not available") - return {"Authorization": f"Bearer {self.access_token}"} - - -@pytest.fixture(scope="class") -def api_flow_context(): - """Create a shared context for API flow tests. - - Returns: - APIFlowContext instance shared across test class. - """ - return APIFlowContext() - - -@pytest.mark.e2e -@pytest.mark.integration -class TestCompleteAPIFlow: - """End-to-end test suite for complete OAuth2 authentication workflow. - - Tests are ordered to follow the natural authentication flow: - 1. Health check - Verify server is running - 2. Client registration - Register OAuth client with scopes - 3. Token generation - Obtain JWT access token - 4. Token validation - Verify token structure and scopes - 5. Scope enforcement - Test subset and unauthorized scope requests - 6. Security validation - Test invalid credentials and token uniqueness - - Each test builds on the previous, storing state in the shared context. - This covers the complete authentication and authorization workflow. - - Note: Protected API endpoints are not tested here - they are implemented - separately when the actual endpoints are available. - """ - - def test_01_health_check( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Step 1: Verify server health endpoint is accessible. - - This confirms the server is running and ready to accept requests. - """ - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.get("/health") - - assert response.status_code == 200, f"Health check failed: {response.text}" - - data = response.json() - assert data["status"] == "healthy" - - def test_02_register_client( - self, - base_url: str, - valid_auth_header: Dict[str, str], - api_flow_context: APIFlowContext, # noqa: W0621 - ): - """Step 2: Register a new OAuth client. - - This creates a client that will be used for subsequent token requests. - Client credentials are stored in the shared context. - """ - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/register", - headers=valid_auth_header, - json={ - "client_name": "api-flow-test-client", - "description": "Client for complete API flow testing", - "allowed_scopes": ["catalog:read", "catalog:write"], - }, - ) - - assert response.status_code == 201, f"Registration failed: {response.text}" - - data = response.json() - - # Verify response structure - assert "client_id" in data - assert "client_secret" in data - assert data["client_id"].startswith("bld_") - assert data["client_secret"].startswith("bld_s_") - - # Store credentials in context for subsequent tests - api_flow_context.client_id = data["client_id"] - api_flow_context.client_secret = data["client_secret"] - api_flow_context.client_name = data["client_name"] - api_flow_context.allowed_scopes = data["allowed_scopes"] - - def test_03_request_token( - self, - base_url: str, - api_flow_context: APIFlowContext, # noqa: W0621 - ): - """Step 3: Request access token using client credentials. - - Uses the client credentials from registration to obtain a JWT token. - Token is stored in the shared context for subsequent API calls. - """ - assert api_flow_context.has_client_credentials(), ( - "Client credentials not available. Run test_02_register_client first." - ) - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": api_flow_context.client_id, - "client_secret": api_flow_context.client_secret, - }, - ) - - assert response.status_code == 200, f"Token request failed: {response.text}" - - data = response.json() - - # Verify response structure - assert "access_token" in data - assert data["token_type"] == "Bearer" - assert data["expires_in"] > 0 - assert "scope" in data - - # Verify JWT structure - parts = data["access_token"].split(".") - assert len(parts) == 3, "Token should be valid JWT format" - - # Store token in context for subsequent tests - api_flow_context.access_token = data["access_token"] - api_flow_context.token_type = data["token_type"] - api_flow_context.expires_in = data["expires_in"] - api_flow_context.scope = data["scope"] - - def test_04_token_contains_granted_scopes( - self, - api_flow_context: APIFlowContext, # noqa: W0621 - ): - """Step 4: Verify token contains the expected scopes. - - Confirms that the granted scopes match the client's allowed scopes. - """ - assert api_flow_context.has_access_token(), ( - "Access token not available. Run test_03_request_token first." - ) - - # Verify scopes match what was registered - granted_scopes = api_flow_context.scope.split() - for scope in api_flow_context.allowed_scopes: - assert scope in granted_scopes, f"Expected scope '{scope}' not in token" - - def test_05_request_token_with_subset_scope( - self, - base_url: str, - api_flow_context: APIFlowContext, # noqa: W0621 - ): - """Step 5: Request token with a subset of allowed scopes. - - Verifies that clients can request fewer scopes than allowed. - """ - assert api_flow_context.has_client_credentials(), ( - "Client credentials not available. Run test_02_register_client first." - ) - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": api_flow_context.client_id, - "client_secret": api_flow_context.client_secret, - "scope": "catalog:read", - }, - ) - - assert response.status_code == 200, f"Token request failed: {response.text}" - - data = response.json() - assert data["scope"] == "catalog:read" - - def test_06_reject_unauthorized_scope( - self, - base_url: str, - api_flow_context: APIFlowContext, # noqa: W0621 - ): - """Step 6: Verify unauthorized scope is rejected. - - Confirms that clients cannot request scopes beyond their allowed set. - """ - assert api_flow_context.has_client_credentials(), ( - "Client credentials not available. Run test_02_register_client first." - ) - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": api_flow_context.client_id, - "client_secret": api_flow_context.client_secret, - "scope": "admin:full", - }, - ) - - assert response.status_code == 400, f"Expected 400, got: {response.text}" - - data = response.json() - assert data["detail"]["error"] == "invalid_scope" - - def test_07_reject_invalid_credentials( - self, - base_url: str, - api_flow_context: APIFlowContext, # noqa: W0621 - ): - """Step 7: Verify invalid credentials are rejected. - - Confirms that token requests with wrong credentials fail properly. - """ - - assert api_flow_context.has_client_credentials(), ( - "Client credentials not available. Run test_02_register_client first." - ) - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": api_flow_context.client_id, - "client_secret": generate_test_client_secret(), - }, - ) - - assert response.status_code == 401, f"Expected 401, got: {response.text}" - - data = response.json() - assert data["detail"]["error"] == "invalid_client" - - def test_08_multiple_tokens_are_unique( - self, - base_url: str, - api_flow_context: APIFlowContext, # noqa: W0621 - ): - """Step 8: Verify each token request generates a unique token. - - Confirms that tokens have unique identifiers (jti claim). - """ - assert api_flow_context.has_client_credentials(), ( - "Client credentials not available. Run test_02_register_client first." - ) - - tokens = [] - with httpx.Client(base_url=base_url, timeout=30.0) as client: - for _ in range(3): - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": api_flow_context.client_id, - "client_secret": api_flow_context.client_secret, - }, - ) - assert response.status_code == 200 - tokens.append(response.json()["access_token"]) - - # All tokens should be unique - assert len(set(tokens)) == 3, "All tokens should be unique" - - -@pytest.mark.e2e -@pytest.mark.integration -class TestAPIFlowErrorHandling: - """Test error handling across the OAuth2 authentication flow. - - These tests verify proper error responses for various failure scenarios: - - Registration without/with invalid authentication - - Token requests for unregistered clients - - Invalid grant types and credentials - - Format validation for client credentials - - Each test ensures that error responses are appropriate and secure, - without exposing sensitive information. - """ - - def test_register_without_auth_fails( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify registration without authentication fails.""" - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/register", - json={"client_name": "unauthorized-client"}, - ) - - assert response.status_code == 401, f"Expected 401, got: {response.text}" - - def test_register_with_invalid_auth_fails( - self, - base_url: str, - invalid_auth_header: Dict[str, str], - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify registration with invalid credentials fails.""" - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/register", - headers=invalid_auth_header, - json={"client_name": "invalid-auth-client"}, - ) - - assert response.status_code == 401, f"Expected 401, got: {response.text}" - - def test_token_without_registration_fails( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify token request for unregistered client fails.""" - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": "bld_nonexistent_client_12345678", - "client_secret": generate_test_client_secret(), - }, - ) - - assert response.status_code == 401, f"Expected 401, got: {response.text}" - - data = response.json() - assert data["detail"]["error"] == "invalid_client" - - def test_token_with_invalid_grant_type_fails( - self, - base_url: str, - valid_auth_header: Dict[str, str], - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify token request with unsupported grant type fails.""" - # First register a client - with httpx.Client(base_url=base_url, timeout=30.0) as client: - reg_response = client.post( - "/api/v1/auth/register", - headers=valid_auth_header, - json={"client_name": "grant-type-test-client"}, - ) - assert reg_response.status_code == 201 - - creds = reg_response.json() - - # Try token with invalid grant type - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "authorization_code", - "client_id": creds["client_id"], - "client_secret": creds["client_secret"], - }, - ) - - assert response.status_code == 422, f"Expected 422, got: {response.text}" - - -@pytest.mark.e2e -@pytest.mark.integration -class TestAPIFlowSecurityValidation: - """Security validation tests for the OAuth2 authentication flow. - - These tests verify that security measures are properly enforced: - - Client credential format validation - - Maximum client limits enforcement - - Proper error handling without information disclosure - - Token security and uniqueness validation - - These tests ensure the authentication system follows security best practices - and does not expose sensitive information in error responses. - """ - - def test_client_credentials_format_validation( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify client credential format validation.""" - with httpx.Client(base_url=base_url, timeout=30.0) as client: - # Invalid client_id format - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": generate_invalid_client_id(), - "client_secret": generate_test_client_secret(), - }, - ) - - assert response.status_code == 422, f"Expected 422, got: {response.text}" - - def test_client_secret_format_validation( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify client secret format validation.""" - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": "bld_valid_format_client_id", - "client_secret": generate_invalid_client_secret(), - }, - ) - - assert response.status_code == 422, f"Expected 422, got: {response.text}" - - def test_max_clients_limit_enforced( - self, - base_url: str, - valid_auth_header: Dict[str, str], - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify maximum client limit is enforced.""" - with httpx.Client(base_url=base_url, timeout=30.0) as client: - # Register first client - response1 = client.post( - "/api/v1/auth/register", - headers=valid_auth_header, - json={"client_name": "first-client"}, - ) - assert response1.status_code == 201 - - # Try to register second client - response2 = client.post( - "/api/v1/auth/register", - headers=valid_auth_header, - json={"client_name": "second-client"}, - ) - - assert response2.status_code == 409, f"Expected 409, got: {response2.text}" - - data = response2.json() - assert data["detail"]["error"] == "max_clients_reached" diff --git a/build_stream/tests/end_to_end/api/test_build_image_e2e.py b/build_stream/tests/end_to_end/api/test_build_image_e2e.py deleted file mode 100644 index 33a9148047..0000000000 --- a/build_stream/tests/end_to_end/api/test_build_image_e2e.py +++ /dev/null @@ -1,499 +0,0 @@ -# Copyright 2026 Dell Inc. or its subsidiaries. All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""End-to-end tests for Build Image API.""" - -import json -import subprocess -import time -from pathlib import Path -from typing import Dict, Any - -import pytest -import requests - - -class TestBuildImageE2E: - """End-to-end tests for build image workflow.""" - - BASE_URL = "http://localhost:8000" - API_PREFIX = "/api/v1" - AUTH_TOKEN = "test-e2e-token" - REQUEST_TIMEOUT = 30 - - @classmethod - def setup_class(cls): - """Setup class with server startup.""" - # Start the API server in background - cls.server_process = subprocess.Popen( - ["python", "main.py"], - cwd="/opt/omnia/omnia/omnia_code/build_stream", - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) - # Wait for server to start - time.sleep(5) - - # Verify server is running - try: - response = requests.get( - f"{cls.BASE_URL}/health", - timeout=cls.REQUEST_TIMEOUT, - ) - assert response.status_code == 200 - except requests.exceptions.ConnectionError: - pytest.skip("API server not available") - - @classmethod - def teardown_class(cls): - """Cleanup by stopping server.""" - if hasattr(cls, 'server_process'): - cls.server_process.terminate() - cls.server_process.wait() - - def get_headers(self, correlation_id: str = None) -> Dict[str, str]: - """Get request headers.""" - headers = { - "Authorization": f"Bearer {self.AUTH_TOKEN}", - "Content-Type": "application/json", - } - if correlation_id: - headers["X-Correlation-Id"] = correlation_id - return headers - - def test_full_build_image_workflow_x86_64(self): - """Test complete build image workflow for x86_64.""" - correlation_id = "e2e-test-x86_64" - headers = self.get_headers(correlation_id) - - # Step 1: Create a job - create_job_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs", - json={ - "stage": "build-image", - "input_parameters": { - "architecture": "x86_64", - "image_key": "e2e-test-image", - "functional_groups": [ - "slurm_control_node_x86_64", - "slurm_node_x86_64", - "login_node_x86_64" - ] - } - }, - headers=headers, - timeout=self.REQUEST_TIMEOUT, - ) - assert create_job_response.status_code == 201 - job_data = create_job_response.json() - job_id = job_data["job_id"] - assert job_id - - # Step 2: Verify job was created with build-image stage - get_job_response = requests.get( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}", - headers=headers, - timeout=self.REQUEST_TIMEOUT, - ) - assert get_job_response.status_code == 200 - job_detail = get_job_response.json() - stages = {stage["stage_name"]: stage for stage in job_detail["stages"]} - assert "build-image" in stages - assert stages["build-image"]["status"] == "PENDING" - - # Step 3: Trigger build image stage - build_image_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}/stages/build-image", - json={ - "architecture": "x86_64", - "image_key": "e2e-test-image", - "functional_groups": [ - "slurm_control_node_x86_64", - "slurm_node_x86_64", - "login_node_x86_64" - ] - }, - headers=headers - ) - assert build_image_response.status_code == 202 - build_data = build_image_response.json() - assert build_data["job_id"] == job_id - assert build_data["stage"] == "build-image" - assert build_data["status"] == "accepted" - assert build_data["architecture"] == "x86_64" - assert build_data["image_key"] == "e2e-test-image" - assert len(build_data["functional_groups"]) == 3 - - # Step 4: Verify stage is now STARTED - get_job_response2 = requests.get( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}", - headers=headers, - timeout=self.REQUEST_TIMEOUT, - ) - assert get_job_response2.status_code == 200 - job_detail2 = get_job_response2.json() - stages2 = {stage["stage_name"]: stage for stage in job_detail2["stages"]} - assert stages2["build-image"]["status"] == "STARTED" - - # Step 5: Verify request file in queue - queue_dir = Path("/opt/omnia/build_stream/queue/requests") - request_files = list(queue_dir.glob(f"{job_id}_build-image_*.json")) - assert len(request_files) == 1 - - # Verify request file content - request_data = json.loads(request_files[0].read_text()) - assert request_data["job_id"] == job_id - assert request_data["architecture"] == "x86_64" - assert request_data["image_key"] == "e2e-test-image" - assert request_data["functional_groups"] == [ - "slurm_control_node_x86_64", - "slurm_node_x86_64", - "login_node_x86_64" - ] - assert request_data["playbook_path"] == "/omnia/build_image_x86_64/build_image_x86_64.yml" - assert request_data["correlation_id"] == correlation_id - - # Step 6: Verify playbook command generation - with open(request_files[0], "r", encoding="utf-8") as f: - request_content = json.load(f) - - # The request should contain all necessary fields for playbook execution - assert "request_id" in request_content - assert "timeout_minutes" in request_content - assert "submitted_at" in request_content - assert "inventory_file_path" not in request_content # Not needed for x86_64 - - # Step 7: Verify stage naming (should be build-image-x86_64) - assert request_content["stage_name"] == "build-image-x86_64" - - def test_full_build_image_workflow_aarch64(self): - """Test complete build image workflow for aarch64.""" - correlation_id = "e2e-test-aarch64" - headers = self.get_headers(correlation_id) - - # Step 1: Create a job - create_job_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs", - json={ - "stage": "build-image", - "input_parameters": { - "architecture": "aarch64", - "image_key": "e2e-test-image-arm", - "functional_groups": [ - "slurm_control_node_aarch64", - "slurm_node_aarch64" - ] - } - }, - headers=headers - ) - assert create_job_response.status_code == 201 - job_data = create_job_response.json() - job_id = job_data["job_id"] - - # Step 2: Create build_stream_config.yml with inventory host - # Use the consolidated repository path structure - input_dir = Path("/opt/omnia/input/project_default") - input_dir.mkdir(parents=True, exist_ok=True) - - # Create default.yml for project name resolution - default_file = Path("/opt/omnia/input/default.yml") - default_file.write_text("project_name: project_default\n", encoding="utf-8") - - config_file = input_dir / "build_stream_config.yml" - config_file.write_text("aarch64_inventory_host: 10.3.0.170\n", encoding="utf-8") - - # Step 3: Trigger build image stage - build_image_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}/stages/build-image", - json={ - "architecture": "aarch64", - "image_key": "e2e-test-image-arm", - "functional_groups": [ - "slurm_control_node_aarch64", - "slurm_node_aarch64" - ] - }, - headers=headers - ) - assert build_image_response.status_code == 202 - build_data = build_image_response.json() - assert build_data["architecture"] == "aarch64" - - # Step 4: Verify request file and inventory file creation - queue_dir = Path("/opt/omnia/build_stream/queue/requests") - request_files = list(queue_dir.glob(f"{job_id}_build-image_*.json")) - assert len(request_files) == 1 - - request_data = json.loads(request_files[0].read_text(encoding="utf-8")) - assert request_data["playbook_path"] == "build_image_aarch64.yml" # Only filename, not full path - - # Step 5: Verify inventory file was created by consolidated repository - inventory_dir = Path("/opt/omnia/build_stream_inv") - inventory_file = inventory_dir / job_id / "inv" - assert inventory_file.exists(), "Inventory file should be created" - - # Verify inventory file content - with open(inventory_file, 'r') as f: - inventory_content = f.read() - assert "10.3.0.170" in inventory_content, f"Inventory file should contain host IP: {inventory_content}" - assert "[build_hosts]" in inventory_content, f"Inventory file should have proper format: {inventory_content}" - - # Step 6: Verify stage naming (should be build-image-aarch64) - with open(request_files[0], "r", encoding="utf-8") as f: - request_content = json.load(f) - assert request_content["stage_name"] == "build-image-aarch64" - - # Step 7: Verify inventory_file_path is included in request - assert "inventory_file_path" in request_content - assert request_content["inventory_file_path"] == str(inventory_file) - - def test_consolidated_repository_functionality(self): - """Test consolidated NfsInputRepository functionality.""" - correlation_id = "e2e-test-consolidated-repo" - headers = self.get_headers(correlation_id) - - # Step 1: Create a job - create_job_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs", - json={ - "stage": "build-image", - "input_parameters": { - "architecture": "aarch64", - "image_key": "e2e-consolidated-test", - "functional_groups": ["slurm_control_node_aarch64"] - } - }, - headers=headers - ) - assert create_job_response.status_code == 201 - job_data = create_job_response.json() - job_id = job_data["job_id"] - - # Step 2: Setup consolidated repository paths - input_dir = Path("/opt/omnia/input") - input_dir.mkdir(parents=True, exist_ok=True) - - # Create default.yml for project name resolution - default_file = input_dir / "default.yml" - default_file.write_text("project_name: project_default\n", encoding="utf-8") - - # Create config with correct key name - config_file = input_dir / "project_default" / "build_stream_config.yml" - config_file.parent.mkdir(parents=True, exist_ok=True) - config_file.write_text("aarch64_inventory_host: 192.168.1.200\n", encoding="utf-8") - - # Step 3: Trigger build image stage - build_image_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}/stages/build-image", - json={ - "architecture": "aarch64", - "image_key": "e2e-consolidated-test", - "functional_groups": ["slurm_control_node_aarch64"] - }, - headers=headers - ) - assert build_image_response.status_code == 202 - - # Step 4: Verify consolidated repository functionality - # 4a: Verify config reading works - queue_dir = Path("/opt/omnia/build_stream/queue/requests") - request_files = list(queue_dir.glob(f"{job_id}_build-image_*.json")) - assert len(request_files) == 1 - - # 4b: Verify inventory file creation - inventory_dir = Path("/opt/omnia/build_stream_inv") - inventory_file = inventory_dir / job_id / "inv" - assert inventory_file.exists(), "Consolidated repository should create inventory file" - - # 4c: Verify inventory file content - with open(inventory_file, 'r') as f: - content = f.read() - assert "192.168.1.200" in content - assert "[build_hosts]" in content - - # 4d: Verify input directory paths work - build_stream_dir = Path("/opt/omnia/build_stream") - source_path = build_stream_dir / job_id / "input" - dest_path = input_dir / "project_default" - - # These paths should be accessible through the consolidated repository - assert dest_path.exists(), "Destination input directory should exist" - - # 4e: Verify request contains correct playbook filename (not full path) - with open(request_files[0], "r", encoding="utf-8") as f: - request_content = json.load(f) - assert request_content["playbook_path"] == "build_image_aarch64.yml" - assert request_content["stage_name"] == "build-image-aarch64" - assert "inventory_file_path" in request_content - - def test_build_image_error_cases(self): - """Test various error scenarios.""" - correlation_id = "e2e-test-errors" - headers = self.get_headers(correlation_id) - - # Test 1: Invalid architecture - create_job_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs", - json={ - "stage": "build-image", - "input_parameters": { - "architecture": "x86_64", - "image_key": "test-image", - "functional_groups": ["group1"] - } - }, - headers=headers - ) - job_id = create_job_response.json()["job_id"] - - error_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}/stages/build-image", - json={ - "architecture": "invalid_arch", - "image_key": "test-image", - "functional_groups": ["group1"] - }, - headers=headers - ) - assert error_response.status_code == 400 - assert error_response.json()["error"] == "INVALID_ARCHITECTURE" - - # Test 2: Missing inventory host for aarch64 - create_job_response2 = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs", - json={ - "stage": "build-image", - "input_parameters": { - "architecture": "aarch64", - "image_key": "test-image", - "functional_groups": ["group1"] - } - }, - headers=headers - ) - job_id2 = create_job_response2.json()["job_id"] - - # Don't create config file (no inventory host) - error_response2 = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id2}/stages/build-image", - json={ - "architecture": "aarch64", - "image_key": "test-image", - "functional_groups": ["group1"] - }, - headers=headers - ) - assert error_response2.status_code == 400 - assert error_response2.json()["error"] == "INVENTORY_HOST_MISSING" - - def test_build_image_concurrent_requests(self): - """Test handling concurrent build image requests.""" - correlation_id = "e2e-test-concurrent" - headers = self.get_headers(correlation_id) - - # Create multiple jobs - job_ids = [] - for i in range(3): - response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs", - json={ - "stage": "build-image", - "input_parameters": { - "architecture": "x86_64", - "image_key": f"concurrent-image-{i}", - "functional_groups": [f"group{i}"] - } - }, - headers=headers, - timeout=self.REQUEST_TIMEOUT, - ) - job_ids.append(response.json()["job_id"]) - - # Submit build image requests concurrently - import concurrent.futures - - def submit_build_image(job_id): - return requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}/stages/build-image", - json={ - "architecture": "x86_64", - "image_key": f"concurrent-image-{job_id}", - "functional_groups": [f"group{job_id}"] - }, - headers=headers - ) - - with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: - futures = [executor.submit(submit_build_image, job_id) for job_id in job_ids] - responses = [future.result() for future in futures] - - # All requests should succeed - for response in responses: - assert response.status_code == 202 - - # Verify all requests are in queue - queue_dir = Path("/opt/omnia/build_stream/queue/requests") - request_files = list(queue_dir.glob("*_build-image_*.json")) - assert len(request_files) >= 3 # At least our 3 requests - - def test_build_image_audit_trail(self): - """Test that build image operations create audit events.""" - correlation_id = "e2e-test-audit" - headers = self.get_headers(correlation_id) - - # Create job and trigger build image - create_job_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs", - json={ - "stage": "build-image", - "input_parameters": { - "architecture": "x86_64", - "image_key": "audit-test-image", - "functional_groups": ["group1"] - } - }, - headers=headers - ) - job_id = create_job_response.json()["job_id"] - - build_image_response = requests.post( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}/stages/build-image", - json={ - "architecture": "x86_64", - "image_key": "audit-test-image", - "functional_groups": ["group1"] - }, - headers=headers - ) - assert build_image_response.status_code == 202 - - # Check audit events - audit_response = requests.get( - f"{self.BASE_URL}{self.API_PREFIX}/jobs/{job_id}/audit", - headers=headers, - timeout=self.REQUEST_TIMEOUT, - ) - assert audit_response.status_code == 200 - audit_events = audit_response.json() - - # Should have STAGE_STARTED event for build-image - build_image_events = [ - event for event in audit_events - if event["event_type"] == "STAGE_STARTED" and - event["details"]["stage_name"] == "build-image" - ] - assert len(build_image_events) == 1 - assert build_image_events[0]["details"]["architecture"] == "x86_64" - assert build_image_events[0]["details"]["image_key"] == "audit-test-image" diff --git a/build_stream/tests/end_to_end/api/test_generate_input_files_e2e.py b/build_stream/tests/end_to_end/api/test_generate_input_files_e2e.py deleted file mode 100644 index 2fbed30d9d..0000000000 --- a/build_stream/tests/end_to_end/api/test_generate_input_files_e2e.py +++ /dev/null @@ -1,482 +0,0 @@ -# Copyright 2026 Dell Inc. or its subsidiaries. All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""End-to-end tests for Generate Input Files complete workflow. - -These tests validate the complete generate input files workflow using real OAuth2 -authentication instead of mocks. The tests follow the chronological order: -1. Health check -2. Client registration -3. Token generation -4. Job creation -5. Parse catalog execution (prerequisite) -6. Generate input files execution -7. Error handling and edge cases - -Requirements: - - ansible-vault must be installed - - Tests require write access to create temporary vault files - - RSA keys must be available for JWT signing -""" - -import json -import os -import uuid -from typing import Dict, Any, Optional - -import pytest -import httpx - -from core.jobs.value_objects import CorrelationId - - -class GenerateInputFilesContext: - """Context object to store state across generate input files tests. - - This class maintains state between test steps, allowing tests to - share data like client credentials, access tokens, and job IDs. - - Attributes: - client_id: Registered client identifier. - client_secret: Registered client secret. - access_token: Generated JWT access token. - job_id: Created job ID for generate input files testing. - catalog_content: Valid catalog content for testing. - """ - - def __init__(self): - """Initialize empty context.""" - self.client_id: Optional[str] = None - self.client_secret: Optional[str] = None - self.client_name: Optional[str] = None - self.allowed_scopes: Optional[list] = None - self.access_token: Optional[str] = None - self.token_type: Optional[str] = None - self.expires_in: Optional[int] = None - self.scope: Optional[str] = None - self.job_id: Optional[str] = None - self.catalog_content: Optional[bytes] = None - - def has_client_credentials(self) -> bool: - """Check if client credentials are available.""" - return self.client_id is not None and self.client_secret is not None - - def has_access_token(self) -> bool: - """Check if access token is available.""" - return self.access_token is not None - - def has_job_id(self) -> bool: - """Check if job ID is available.""" - return self.job_id is not None - - def get_auth_header(self) -> Dict[str, str]: - """Get Authorization header with Bearer token. - - Returns: - Dictionary with Authorization header. - - Raises: - ValueError: If access token is not available. - """ - if not self.has_access_token(): - raise ValueError("Access token not available") - return {"Authorization": f"Bearer {self.access_token}"} - - def set_job_id(self, job_id: str) -> None: - """Set the job ID for testing.""" - self.job_id = job_id - - def load_catalog_content(self) -> str: - """Load catalog content for testing. - - Returns: - JSON string of catalog content. - """ - # Use the proper catalog_rhel fixture instead of a minimal catalog - catalog_path = os.path.join( - os.path.dirname(__file__), - "..", "..", "fixtures", "catalogs", "catalog_rhel.json" - ) - - with open(catalog_path, "r", encoding="utf-8") as f: - content = f.read() - # Store the content as bytes for upload - self.catalog_content = content.encode('utf-8') - return content - - def get_catalog_bytes(self) -> bytes: - """Get catalog content as bytes.""" - return self.catalog_content - - -@pytest.fixture(scope="class") -def generate_input_files_context(): - """Create a shared context for generate input files tests. - - Returns: - GenerateInputFilesContext instance for sharing state across tests. - """ - return GenerateInputFilesContext() - - -class TestGenerateInputFilesE2E: - - """End-to-end tests for Generate Input Files complete workflow. - - Tests are ordered to follow the natural workflow: - 1. Health check - Verify server is running - 2. Client registration - Register OAuth client with catalog scopes - 3. Token generation - Obtain JWT access token - 4. Job creation - Create a job for generate input files - 5. Parse catalog execution - Execute parse catalog stage (prerequisite) - 6. Generate input files execution - Execute generate input files stage - 7. Error handling - Test various failure scenarios - - Tests use pytest.mark.e2e and depend on fixtures from conftest.py. - """ - - @pytest.mark.e2e - def test_01_health_check(self, base_url: str): - """Step 1: Verify server health. - - Confirms the API server is running and accessible before proceeding - with authentication and workflow tests. - """ - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.get("/health") - - assert response.status_code == 200, f"Health check failed: {response.text}" - - data = response.json() - assert data["status"] == "healthy" - - @pytest.mark.e2e - def test_02_register_client_for_generate_input_files( - self, - base_url: str, - valid_auth_header: Dict[str, str], - generate_input_files_context: GenerateInputFilesContext, # noqa: W0621 - ): - """Step 2: Register a new OAuth client for generate input files access. - - This creates a client that will be used for subsequent generate input files requests. - Client credentials are stored in the shared context. - """ - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/register", - headers=valid_auth_header, - json={ - "client_name": "generate-input-files-test-client", - "description": "Client for generate input files testing", - "allowed_scopes": ["catalog:read", "catalog:write"], - }, - ) - - assert response.status_code == 201, f"Registration failed: {response.text}" - - data = response.json() - - # Verify response structure - assert "client_id" in data - assert "client_secret" in data - assert data["client_id"].startswith("bld_") - assert data["client_secret"].startswith("bld_s_") - - # Store credentials in context for subsequent tests - generate_input_files_context.client_id = data["client_id"] - generate_input_files_context.client_secret = data["client_secret"] - generate_input_files_context.client_name = data["client_name"] - generate_input_files_context.allowed_scopes = data["allowed_scopes"] - - @pytest.mark.e2e - def test_03_request_token_for_generate_input_files( - self, - base_url: str, - generate_input_files_context: GenerateInputFilesContext, # noqa: W0621 - ): - """Step 3: Request access token for generate input files API. - - Uses the client credentials from registration to obtain a JWT token. - Token is stored in the shared context for subsequent API calls. - """ - assert generate_input_files_context.has_client_credentials(), ( - "Client credentials not available. Run test_02_register_client_for_generate_input_files first." - ) - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": generate_input_files_context.client_id, - "client_secret": generate_input_files_context.client_secret, - }, - ) - - assert response.status_code == 200, f"Token request failed: {response.text}" - - data = response.json() - - # Verify response structure - assert "access_token" in data - assert data["token_type"] == "Bearer" - assert data["expires_in"] > 0 - assert "scope" in data - - # Verify JWT structure - parts = data["access_token"].split(".") - assert len(parts) == 3, "Token should be valid JWT format" - - # Store token in context for subsequent tests - generate_input_files_context.access_token = data["access_token"] - generate_input_files_context.token_type = data["token_type"] - generate_input_files_context.expires_in = data["expires_in"] - generate_input_files_context.scope = data["scope"] - - @pytest.mark.e2e - def test_04_create_job_for_generate_input_files( - self, - base_url: str, - generate_input_files_context: GenerateInputFilesContext, # noqa: W0621 - ): - """Step 4: Create a new job for generate input files testing. - - Tests job creation with proper validation and idempotency. - """ - assert generate_input_files_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_generate_input_files first." - ) - - # Prepare job creation request - job_data = { - "client_id": generate_input_files_context.client_id, - "client_name": "Generate Input Files Test Client" - } - - idempotency_key = str(uuid.uuid4()) - headers = generate_input_files_context.get_auth_header() - headers["Idempotency-Key"] = idempotency_key - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/jobs", - json=job_data, - headers=headers, - ) - - assert response.status_code == 201, f"Job creation failed: {response.text}" - - data = response.json() - - # Verify response structure - assert "job_id" in data - assert "job_state" in data - assert "created_at" in data - assert "correlation_id" in data - - # Verify job ID format (UUID) - uuid.UUID(data["job_id"]) # This will raise ValueError if not valid UUID - - # Store job ID in context - generate_input_files_context.set_job_id(data["job_id"]) - - # Verify job state - assert data["job_state"] == "CREATED" - - @pytest.mark.e2e - def test_05_parse_catalog_prerequisite( - self, - base_url: str, - generate_input_files_context: GenerateInputFilesContext, # noqa: W0621 - ): - """Step 5: Execute parse catalog as prerequisite for generate input files. - - Parse catalog must be executed successfully before generate input files - can be run, as it depends on the catalog artifacts. - """ - assert generate_input_files_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_generate_input_files first." - ) - assert generate_input_files_context.has_job_id(), ( - "Job ID not available. Run test_04_create_job_for_generate_input_files first." - ) - - # Load catalog content - generate_input_files_context.load_catalog_content() - assert generate_input_files_context.catalog_content is not None - - headers = generate_input_files_context.get_auth_header() - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{generate_input_files_context.job_id}/stages/parse-catalog", - files={ - "file": ( - "catalog.json", - generate_input_files_context.catalog_content, - "application/json" - ) - }, - headers=headers, - ) - - # The response should indicate the stage was processed successfully - assert response.status_code == 200, ( - f"Parse catalog failed: {response.text}" - ) - - # Get response data for verification - response_data = response.json() - - # Verify the response structure - assert "status" in response_data - assert response_data["status"] == "success" - assert "message" in response_data - - @pytest.mark.e2e - def test_06_generate_input_files_success( - self, - base_url: str, - generate_input_files_context: GenerateInputFilesContext, # noqa: W0621 - ): - """Step 6: Execute generate input files successfully. - - Tests the complete generate input files workflow with default policy. - This depends on parse catalog having been executed first. - """ - assert generate_input_files_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_generate_input_files first." - ) - assert generate_input_files_context.has_job_id(), ( - "Job ID not available. Run test_04_create_job_for_generate_input_files first." - ) - - headers = generate_input_files_context.get_auth_header() - - # Execute generate input files with default policy - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{generate_input_files_context.job_id}/stages/generate-input-files", - headers=headers, - ) - - # Should process the request successfully - # Tests should fail on any error (including 500) - assert response.status_code == 200, ( - f"Generate input files failed with status {response.status_code}: {response.text}" - ) - - # Verify minimal response structure - response_data = response.json() - assert "stage_state" in response_data - assert response_data["stage_state"] in ["COMPLETED", "FAILED"] - - if response_data["stage_state"] == "COMPLETED": - # Should have only these three fields - assert "job_id" in response_data - assert "message" in response_data - assert "stage_state" in response_data - print(f"✅ Generate input files completed successfully!") - print(f"Response: {response_data}") - else: - print(f"⚠️ Generate input files completed with stage state: {response_data['stage_state']}") - - - @pytest.mark.e2e - def test_07_generate_input_files_with_custom_policy( - self, - base_url: str, - generate_input_files_context: GenerateInputFilesContext, # noqa: W0621 - ): - - """Step 7: Test generate input files with custom adapter policy. - - Tests error handling and various policy path scenarios. - """ - assert generate_input_files_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_generate_input_files first." - ) - assert generate_input_files_context.has_job_id(), ( - "Job ID not available. Run test_04_create_job_for_generate_input_files first." - ) - - headers = generate_input_files_context.get_auth_header() - - # Test with invalid policy path - invalid_request = { - "adapter_policy_path": "../../../etc/passwd" - } - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - error_response = client.post( - f"/api/v1/jobs/{generate_input_files_context.job_id}/stages/generate-input-files", - json=invalid_request, - headers=headers, - ) - - # Should reject invalid path - assert error_response.status_code in [400, 422], ( - f"Expected rejection of invalid policy path: {error_response.text}" - ) - # Create a fresh job to avoid STAGE_ALREADY_COMPLETED - job_data = { - "client_id": generate_input_files_context.client_id, - "client_name": "Generate Input Files Test Client (recovery)" - } - - new_idempotency_key = str(uuid.uuid4()) - new_headers = headers.copy() - new_headers["Idempotency-Key"] = new_idempotency_key - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - job_response = client.post( - "/api/v1/jobs", - json=job_data, - headers=new_headers, - ) - - assert job_response.status_code == 201, f"Job creation failed: {job_response.text}" - new_job_id = job_response.json()["job_id"] - - # Parse catalog for the new job (prerequisite) - generate_input_files_context.load_catalog_content() - with httpx.Client(base_url=base_url, timeout=30.0) as client: - parse_response = client.post( - f"/api/v1/jobs/{new_job_id}/stages/parse-catalog", - files={ - "file": ( - "catalog.json", - generate_input_files_context.catalog_content, - "application/json", - ) - }, - headers=headers, - ) - - assert parse_response.status_code == 200, ( - f"Parse catalog failed for recovery job: {parse_response.text}" - ) - - # Test with valid request (default policy) on the fresh job - with httpx.Client(base_url=base_url, timeout=3000.0) as client: - recovery_response = client.post( - f"/api/v1/jobs/{new_job_id}/stages/generate-input-files", - headers=headers, - ) - - # Should process the valid request - assert recovery_response.status_code in [200, 400, 422, 500], ( - f"Valid request failed: {recovery_response.text}" - ) diff --git a/build_stream/tests/end_to_end/api/test_parse_catalog_e2e.py b/build_stream/tests/end_to_end/api/test_parse_catalog_e2e.py deleted file mode 100644 index 2197bdb3c8..0000000000 --- a/build_stream/tests/end_to_end/api/test_parse_catalog_e2e.py +++ /dev/null @@ -1,768 +0,0 @@ -# Copyright 2026 Dell Inc. or its subsidiaries. All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""End-to-end tests for Parse Catalog workflow with real authentication. - -These tests validate the complete parse catalog workflow using real OAuth2 -authentication instead of mocks. The tests follow the chronological order: -1. Health check -2. Client registration -3. Token generation -4. Job creation -5. Parse catalog execution -6. Error handling and edge cases - -Usage: - pytest tests/end_to_end/api/test_parse_catalog_e2e.py -v -m e2e - -Requirements: - - ansible-vault must be installed - - Tests require write access to create temporary vault files - - RSA keys must be available for JWT signing -""" - -import json -import os -import uuid -from typing import Dict, Optional - -import httpx -import pytest - - -class ParseCatalogContext: # pylint: disable=too-many-instance-attributes - """Context object to store state across parse catalog tests. - - This class maintains state between test steps, allowing tests to - share data like client credentials, access tokens, and job IDs. - - Attributes: - client_id: Registered client identifier. - client_secret: Registered client secret. - access_token: Generated JWT access token. - job_id: Created job ID for parse catalog testing. - catalog_content: Valid catalog content for testing. - """ - - def __init__(self): - """Initialize empty context.""" - self.client_id: Optional[str] = None - self.client_secret: Optional[str] = None - self.client_name: Optional[str] = None - self.allowed_scopes: Optional[list] = None - self.access_token: Optional[str] = None - self.token_type: Optional[str] = None - self.expires_in: Optional[int] = None - self.scope: Optional[str] = None - self.job_id: Optional[str] = None - self.catalog_content: Optional[bytes] = None - - def has_client_credentials(self) -> bool: - """Check if client credentials are available.""" - return self.client_id is not None and self.client_secret is not None - - def has_access_token(self) -> bool: - """Check if access token is available.""" - return self.access_token is not None - - def has_job_id(self) -> bool: - """Check if job ID is available.""" - return self.job_id is not None - - def get_auth_header(self) -> Dict[str, str]: - """Get Authorization header with Bearer token. - - Returns: - Dictionary with Authorization header. - - Raises: - ValueError: If access token is not available. - """ - if not self.has_access_token(): - raise ValueError("Access token not available") - return {"Authorization": f"Bearer {self.access_token}"} - - def set_job_id(self, job_id: str) -> None: - """Set the job ID for testing.""" - self.job_id = job_id - - def load_catalog_content(self) -> None: - """Load valid catalog content from fixtures.""" - here = os.path.dirname(__file__) - # Go up from end_to_end/api/ to tests/ then to fixtures/ - fixtures_dir = os.path.dirname(os.path.dirname(here)) - catalog_path = os.path.join(fixtures_dir, "fixtures", "catalogs", "catalog_rhel.json") - - with open(catalog_path, 'r', encoding='utf-8') as f: - catalog_data = json.load(f) - - self.catalog_content = json.dumps(catalog_data, indent=2).encode('utf-8') - - -@pytest.fixture(scope="class") -def parse_catalog_context(): - """Create a shared context for parse catalog tests. - - Returns: - ParseCatalogContext instance shared across test class. - """ - return ParseCatalogContext() - - -@pytest.mark.e2e -@pytest.mark.integration -class TestParseCatalogWorkflow: - """End-to-end test suite for parse catalog workflow. - - Tests are ordered to follow the natural workflow: - 1. Health check - Verify server is running - 2. Client registration - Register OAuth client with catalog scopes - 3. Token generation - Obtain JWT access token - 4. Job creation - Create a job for parse catalog - 5. Parse catalog execution - Execute parse catalog stage - 6. Error handling - Test various failure scenarios - - Each test builds on the previous, storing state in the shared context. - """ - - def test_01_health_check( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Step 1: Verify server health endpoint is accessible. - - This confirms the server is running and ready to accept requests. - """ - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.get("/health") - - assert response.status_code == 200, f"Health check failed: {response.text}" - - data = response.json() - assert data["status"] == "healthy" - - def test_02_register_client_for_parse_catalog( - self, - base_url: str, - valid_auth_header: Dict[str, str], - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 2: Register a new OAuth client for parse catalog access. - - This creates a client that will be used for subsequent parse catalog requests. - Client credentials are stored in the shared context. - """ - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/register", - headers=valid_auth_header, - json={ - "client_name": "parse-catalog-test-client", - "description": "Client for parse catalog testing", - "allowed_scopes": ["catalog:read", "catalog:write"], - }, - ) - - assert response.status_code == 201, f"Registration failed: {response.text}" - - data = response.json() - - # Verify response structure - assert "client_id" in data - assert "client_secret" in data - assert data["client_id"].startswith("bld_") - assert data["client_secret"].startswith("bld_s_") - - # Store credentials in context for subsequent tests - parse_catalog_context.client_id = data["client_id"] - parse_catalog_context.client_secret = data["client_secret"] - parse_catalog_context.client_name = data["client_name"] - parse_catalog_context.allowed_scopes = data["allowed_scopes"] - - def test_03_request_token_for_parse_catalog( - self, - base_url: str, - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 3: Request access token for parse catalog API. - - Uses the client credentials from registration to obtain a JWT token. - Token is stored in the shared context for subsequent API calls. - """ - assert parse_catalog_context.has_client_credentials(), ( - "Client credentials not available. Run test_02_register_client_for_parse_catalog first." - ) - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": parse_catalog_context.client_id, - "client_secret": parse_catalog_context.client_secret, - }, - ) - - assert response.status_code == 200, f"Token request failed: {response.text}" - - data = response.json() - - # Verify response structure - assert "access_token" in data - assert data["token_type"] == "Bearer" - assert data["expires_in"] > 0 - assert "scope" in data - - # Verify JWT structure - parts = data["access_token"].split(".") - assert len(parts) == 3, "Token should be valid JWT format" - - # Store token in context for subsequent tests - parse_catalog_context.access_token = data["access_token"] - parse_catalog_context.token_type = data["token_type"] - parse_catalog_context.expires_in = data["expires_in"] - parse_catalog_context.scope = data["scope"] - - def test_04_create_job_for_parse_catalog( - self, - base_url: str, - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 4: Create a new job for parse catalog testing. - - Tests job creation with proper validation and idempotency. - """ - assert parse_catalog_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_parse_catalog first." - ) - - # Prepare job creation request - job_data = { - "client_id": parse_catalog_context.client_id, - "client_name": "Parse Catalog Test Client" - } - - idempotency_key = str(uuid.uuid4()) - headers = parse_catalog_context.get_auth_header() - headers["Idempotency-Key"] = idempotency_key - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - "/api/v1/jobs", - json=job_data, - headers=headers, - ) - - assert response.status_code == 201, f"Job creation failed: {response.text}" - - data = response.json() - - # Verify response structure - assert "job_id" in data - assert "job_state" in data - assert "created_at" in data - assert "correlation_id" in data - - # Verify job ID format (UUID) - uuid.UUID(data["job_id"]) # This will raise ValueError if not valid UUID - - # Store job ID in context - parse_catalog_context.set_job_id(data["job_id"]) - - # Verify job state - assert data["job_state"] == "CREATED" - - def test_05_parse_catalog_success( - self, - base_url: str, - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 5: Execute parse catalog successfully. - - Tests the complete parse catalog workflow with a valid catalog file. - """ - assert parse_catalog_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_parse_catalog first." - ) - assert parse_catalog_context.has_job_id(), ( - "Job ID not available. Run test_04_create_job_for_parse_catalog first." - ) - - # Load catalog content - parse_catalog_context.load_catalog_content() - assert parse_catalog_context.catalog_content is not None - - headers = parse_catalog_context.get_auth_header() - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{parse_catalog_context.job_id}/stages/parse-catalog", - files={ - "file": ( - "catalog.json", - parse_catalog_context.catalog_content, - "application/json" - ) - }, - headers=headers, - ) - - # The response should indicate the stage was processed - # It might fail due to missing dependencies, but the workflow should be complete - assert response.status_code in [200, 400, 422, 500], ( - f"Parse catalog failed: {response.text}" - ) - - # Get response data for verification - response_data = response.json() if response.status_code == 200 else None - - # If successful, verify the response structure - if response.status_code == 200 and response_data: - assert "status" in response_data - assert response_data["status"] == "success" - assert "message" in response_data - - def test_06_parse_catalog_with_invalid_data( - self, - base_url: str, - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 6: Test parse catalog with invalid catalog data. - - Tests error handling when invalid catalog data is provided. - """ - assert parse_catalog_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_parse_catalog first." - ) - - # Create a new job for this test since the previous job might be in a processed state - job_data = { - "client_id": parse_catalog_context.client_id, - "client_name": "Parse Catalog Test Client" - } - - idempotency_key = str(uuid.uuid4()) - headers = parse_catalog_context.get_auth_header() - headers["Idempotency-Key"] = idempotency_key - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - job_response = client.post( - "/api/v1/jobs", - json=job_data, - headers=headers, - ) - - assert job_response.status_code == 201 - new_job_id = job_response.json()["job_id"] - - # Create invalid catalog data - invalid_catalog = b'{"invalid": "catalog"}' - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{new_job_id}/stages/parse-catalog", - files={"file": ("invalid.json", invalid_catalog, "application/json")}, - headers=headers, - ) - - # Should handle the error gracefully - assert response.status_code in [400, 422, 500, 409], ( - f"Expected error response, got: {response.status_code}" - ) - - def test_07_parse_catalog_with_oversized_file( - self, - base_url: str, - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 7: Test parse catalog with oversized file. - - Tests file upload limits are enforced. - """ - assert parse_catalog_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_parse_catalog first." - ) - assert parse_catalog_context.has_job_id(), ( - "Job ID not available. Run test_04_create_job_for_parse_catalog first." - ) - - # Create a new job for this test since the previous job might be in a failed state - job_data = { - "client_id": parse_catalog_context.client_id, - "client_name": "Parse Catalog Test Client" - } - - idempotency_key = str(uuid.uuid4()) - headers = parse_catalog_context.get_auth_header() - headers["Idempotency-Key"] = idempotency_key - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - job_response = client.post( - "/api/v1/jobs", - json=job_data, - headers=headers, - ) - - assert job_response.status_code == 201 - new_job_id = job_response.json()["job_id"] - - # Test with an oversized file - oversized_content = b'x' * (10 * 1024 * 1024) # 10MB - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{new_job_id}/stages/parse-catalog", - files={"file": ("oversized.json", oversized_content, "application/json")}, - headers=headers, - ) - - # Should reject oversized files - assert response.status_code in [400, 413, 422], ( - f"Expected file size error, got: {response.status_code}" - ) - - def test_08_parse_catalog_job_status_integration( - self, - base_url: str, - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 8: Test parse catalog integration with job status. - - Tests that parse catalog properly updates job status and state. - """ - assert parse_catalog_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_parse_catalog first." - ) - assert parse_catalog_context.has_job_id(), ( - "Job ID not available. Run test_04_create_job_for_parse_catalog first." - ) - - headers = parse_catalog_context.get_auth_header() - - # Check job status - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.get( - f"/api/v1/jobs/{parse_catalog_context.job_id}", - headers=headers, - ) - - # Job status should be accessible - assert response.status_code in [200, 404], ( - f"Job status check failed: {response.status_code}" - ) - - if response.status_code == 200: - job_data = response.json() - assert "job_state" in job_data - assert "created_at" in job_data - - def test_09_parse_catalog_with_nonexistent_job_fails( - self, - base_url: str, - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 9: Test parse catalog with nonexistent job fails. - - Tests error handling when trying to parse catalog for a job that doesn't exist. - """ - assert parse_catalog_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_parse_catalog first." - ) - - headers = parse_catalog_context.get_auth_header() - nonexistent_job_id = str(uuid.uuid4()) - catalog_content = b'{"test": "catalog"}' - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{nonexistent_job_id}/stages/parse-catalog", - files={"file": ("catalog.json", catalog_content, "application/json")}, - headers=headers, - ) - - assert response.status_code == 404, f"Expected 404, got: {response.status_code}" - - def test_10_parse_catalog_with_oversized_file_security_check( - self, - base_url: str, - parse_catalog_context: ParseCatalogContext, # noqa: W0621 - ): - """Step 10: Test parse catalog security with oversized file. - - Tests file upload limits are enforced for security. - """ - assert parse_catalog_context.has_access_token(), ( - "Access token not available. Run test_03_request_token_for_parse_catalog first." - ) - - # Create a new job for this test - job_data = { - "client_id": parse_catalog_context.client_id, - "client_name": "Parse Catalog Security Test Client" - } - - idempotency_key = str(uuid.uuid4()) - headers = parse_catalog_context.get_auth_header() - headers["Idempotency-Key"] = idempotency_key - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - job_response = client.post( - "/api/v1/jobs", - json=job_data, - headers=headers, - ) - - assert job_response.status_code == 201 - new_job_id = job_response.json()["job_id"] - - # Test with an oversized file (security check) - oversized_content = b'x' * (10 * 1024 * 1024) # 10MB - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{new_job_id}/stages/parse-catalog", - files={"file": ("oversized.json", oversized_content, "application/json")}, - headers=headers, - ) - - # Should reject oversized files for security - assert response.status_code in [400, 413, 422], ( - f"Expected file size error, got: {response.status_code}" - ) - - -@pytest.mark.e2e -@pytest.mark.integration -class TestParseCatalogErrorHandling: - """Error handling tests for parse catalog API. - - These tests ensure the parse catalog API handles errors gracefully - and does not expose sensitive information in error responses. - """ - - def test_parse_catalog_without_authentication_fails( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify parse catalog without authentication fails.""" - job_id = str(uuid.uuid4()) - catalog_content = b'{"test": "catalog"}' - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{job_id}/stages/parse-catalog", - files={ - "file": ("catalog.json", catalog_content, "application/json") - }, - ) - - # Should fail with either 401 (auth) or 422 (validation before auth) - assert response.status_code in [401, 422], ( - f"Expected 401 or 422, got: {response.status_code}" - ) - - def test_parse_catalog_with_invalid_token_fails( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify parse catalog with invalid token fails.""" - headers = {"Authorization": "Bearer invalid_token"} - job_id = str(uuid.uuid4()) - catalog_content = b'{"test": "catalog"}' - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{job_id}/stages/parse-catalog", - files={"file": ("catalog.json", catalog_content, "application/json")}, - headers=headers, - ) - - assert response.status_code == 401, ( - f"Expected 401, got: {response.status_code}" - ) - - - -@pytest.mark.e2e -@pytest.mark.integration -@pytest.mark.skip( - reason=( - "Security validation tests have vault setup conflicts - " - "skipping to focus on core functionality" - ) -) -class TestParseCatalogSecurityValidation: - """Security validation tests for parse catalog API. - - These tests verify that security measures are properly enforced: - - Input validation and sanitization - - File type validation - - Path traversal prevention - - NOTE: This class is skipped due to vault setup conflicts in independent test execution. - Core security validation is covered in the main workflow tests. - """ - - def test_parse_catalog_with_malicious_content( - self, - base_url: str, - reset_vault, # noqa: W0613 pylint: disable=unused-argument - ): - """Verify parse catalog handles malicious content safely.""" - - pytest.skip() - # Use unique client name to avoid conflicts - unique_client_id = str(uuid.uuid4())[:8] - client_name = f"malicious-content-test-{unique_client_id}" - - # Register client and get token first - with httpx.Client(base_url=base_url, timeout=30.0) as client: - # Register client - reg_response = client.post( - "/api/v1/auth/register", - headers={"Authorization": "Basic dGVzdDp0ZXN0"}, # test:test - json={ - "client_name": client_name, - "allowed_scopes": ["catalog:write"], - }, - ) - assert reg_response.status_code == 201 - creds = reg_response.json() - - # Get token - token_response = client.post( - "/api/v1/auth/token", - data={ - "grant_type": "client_credentials", - "client_id": creds["client_id"], - "client_secret": creds["client_secret"], - }, - ) - assert token_response.status_code == 200 - token_data = token_response.json() - - # Create a job - job_response = client.post( - "/api/v1/jobs", - json={ - "client_id": creds["client_id"], - "client_name": client_name - }, - headers={ - "Authorization": f"Bearer {token_data['access_token']}", - "Idempotency-Key": str(uuid.uuid4()) - }, - ) - assert job_response.status_code == 201 - job_id = job_response.json()["job_id"] - - headers = {"Authorization": f"Bearer {token_data['access_token']}"} - - # Test with malicious content - malicious_content = b'{"Catalog": {"Name": ""}}' - - with httpx.Client(base_url=base_url, timeout=30.0) as client: - response = client.post( - f"/api/v1/jobs/{job_id}/stages/parse-catalog", - files={"file": ("malicious.json", malicious_content, "application/json")}, - headers=headers, - ) - - # Should handle malicious content safely - assert response.status_code in [400, 422, 500], ( - f"Expected error for malicious content, got: {response.status_code}" - ) - - # Response should not contain the malicious content - if response.status_code in [400, 422]: - response_text = response.text.lower() - assert "