diff --git a/build_stream/api/dependencies.py b/build_stream/api/dependencies.py index 2a5bbd7e26..c8aaee2117 100644 --- a/build_stream/api/dependencies.py +++ b/build_stream/api/dependencies.py @@ -18,6 +18,7 @@ authorization, database sessions, repositories, and domain-specific use cases. """ +# pylint: disable=import-error,too-many-locals,broad-exception-caught,wrong-import-position,line-too-long,trailing-whitespace import os from typing import Annotated, Generator @@ -156,7 +157,7 @@ def scope_dependency( HTTPException: If required scope is not present. """ if required_scope not in token_data["scopes"]: - log_secure_info('warning', f"Access denied - missing required scope: {required_scope} (client: {token_data["client_id"][:8] + "..."})") + log_secure_info('warning', f'Access denied - missing required scope: {required_scope} (client: {token_data["client_id"][:8] + "..."})') raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail={ @@ -165,7 +166,8 @@ def scope_dependency( }, ) - log_secure_info('info', f"Scope validation passed for client: {token_data["client_id"][:8] + "..."}, scope: {required_scope}") + client_id_short = token_data["client_id"][:8] + "..." + log_secure_info('info', f'Scope validation passed for client: {client_id_short}, scope: {required_scope}') return token_data return scope_dependency diff --git a/build_stream/core/catalog/tests/sample.py b/build_stream/core/catalog/tests/sample.py deleted file mode 100644 index 89f2472d6c..0000000000 --- a/build_stream/core/catalog/tests/sample.py +++ /dev/null @@ -1,81 +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. - -"""Example script showing programmatic usage of the generator and adapter APIs. - -This script runs the catalog feature-list generator and adapter config generator -directly from Python, configuring logging and handling common errors. -""" - -import logging -import os - -from catalog_parser.generator import generate_root_json_from_catalog, get_functional_layer_roles_from_file, get_package_list -from catalog_parser.adapter import generate_omnia_json_from_catalog -from catalog_parser.adapter_policy import generate_configs_from_policy - -BASE_DIR = os.path.dirname(os.path.dirname(__file__)) -CATALOG_PARSER_DIR = os.path.join(BASE_DIR, "") -CATALOG_PATH = os.path.join(CATALOG_PARSER_DIR, "test_fixtures", "catalog_rhel.json") -SCHEMA_PATH = os.path.join(CATALOG_PARSER_DIR, "resources", "CatalogSchema.json") -FUNCTIONAL_LAYER_PATH = os.path.join(CATALOG_PARSER_DIR, "test_fixtures", "functional_layer.json") -ADAPTER_POLICY_PATH = os.path.join(CATALOG_PARSER_DIR, "resources", "adapter_policy_default.json") -ADAPTER_POLICY_SCHEMA_PATH = os.path.join(CATALOG_PARSER_DIR, "resources", "AdapterPolicySchema.json") - -try: - generate_root_json_from_catalog( - catalog_path=CATALOG_PATH, - schema_path=SCHEMA_PATH, - output_root="out/generator2", - configure_logging=True, - log_file="logs/generator.log", - log_level=logging.INFO, - ) - - generate_omnia_json_from_catalog( - catalog_path=CATALOG_PATH, - schema_path=SCHEMA_PATH, - output_root="out/adapter/config2", - configure_logging=True, - log_file="logs/adapter.log", - log_level=logging.INFO, - ) - - generate_configs_from_policy( - input_dir="out/generator2", - output_dir="out/adapter_policy/config2", - policy_path=ADAPTER_POLICY_PATH, - schema_path=ADAPTER_POLICY_SCHEMA_PATH, - configure_logging=True, - log_file="logs/adapter_policy.log", - log_level=logging.INFO, - ) - - roles = get_functional_layer_roles_from_file(FUNCTIONAL_LAYER_PATH) - print(f"Functional layer roles: {roles}") - - # Get packages for a specific role - result = get_package_list(FUNCTIONAL_LAYER_PATH, role="K8S Controller") - print(f"Packages for role 'K8S Controller': {result}") - - # Get packages for all roles - result = get_package_list(FUNCTIONAL_LAYER_PATH) - print(f"Packages for all roles: {result}") - -except FileNotFoundError as e: - # handle missing catalog/schema - print(f"Missing file: {e}") -except Exception as e: - # handle generic processing errors - print(f"Processing failed: {e}") \ No newline at end of file diff --git a/build_stream/core/catalog/tests/test_adapter_cli_defaults.py b/build_stream/core/catalog/tests/test_adapter_cli_defaults.py deleted file mode 100644 index 63000b69af..0000000000 --- a/build_stream/core/catalog/tests/test_adapter_cli_defaults.py +++ /dev/null @@ -1,56 +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. - -import os -import sys -import tempfile -import unittest - -HERE = os.path.dirname(__file__) -CATALOG_PARSER_DIR = os.path.dirname(HERE) -PROJECT_ROOT = os.path.dirname(CATALOG_PARSER_DIR) -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - -from catalog_parser.adapter import generate_omnia_json_from_catalog, _DEFAULT_SCHEMA_PATH - - -class TestAdapterDefaults(unittest.TestCase): - def test_default_schema_path_points_to_resources(self): - catalog_parser_dir = os.path.dirname(os.path.dirname(__file__)) - expected_schema = os.path.join(catalog_parser_dir, "resources", "CatalogSchema.json") - self.assertEqual(os.path.abspath(_DEFAULT_SCHEMA_PATH), os.path.abspath(expected_schema)) - - def test_generate_omnia_json_with_defaults_writes_output(self): - catalog_parser_dir = os.path.dirname(os.path.dirname(__file__)) - catalog_path = os.path.join(catalog_parser_dir, "test_fixtures", "catalog_rhel.json") - - with tempfile.TemporaryDirectory() as tmpdir: - generate_omnia_json_from_catalog( - catalog_path=catalog_path, - output_root=tmpdir, - ) - - # We expect some JSON files under arch/os/version - found_any_json = False - for root, dirs, files in os.walk(tmpdir): - if any(f.endswith('.json') for f in files): - found_any_json = True - break - - self.assertTrue(found_any_json, "No JSON configs generated under any arch/os/version") - - -if __name__ == "__main__": - unittest.main() diff --git a/build_stream/core/catalog/tests/test_adapter_policy.py b/build_stream/core/catalog/tests/test_adapter_policy.py deleted file mode 100644 index 26746f169b..0000000000 --- a/build_stream/core/catalog/tests/test_adapter_policy.py +++ /dev/null @@ -1,953 +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. - -"""Unit tests for adapter_policy module.""" - -import json -import os -import sys -import tempfile -import unittest - -HERE = os.path.dirname(__file__) -CATALOG_PARSER_DIR = os.path.dirname(HERE) -PROJECT_ROOT = os.path.dirname(CATALOG_PARSER_DIR) -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - -from catalog_parser.adapter_policy import ( - validate_policy_config, - discover_architectures, - discover_os_versions, - transform_package, - apply_substring_filter, - compute_common_packages, - apply_extract_common_filter, - apply_extract_unique_filter, - apply_filter, - merge_transform, - compute_common_keys_from_roles, - derive_common_role, - check_conditions, - process_target_spec, - write_config_file, - generate_configs_from_policy, - _DEFAULT_POLICY_PATH, - _DEFAULT_SCHEMA_PATH, -) -from catalog_parser import adapter_policy_schema_consts as schema - - -class TestValidatePolicyConfig(unittest.TestCase): - """Tests for validate_policy_config function.""" - - def setUp(self): - self.valid_policy = { - "version": "2.0.0", - "targets": { - "test.json": { - "sources": [ - { - "source_file": "source.json", - "pulls": [{"source_key": "role1"}] - } - ] - } - } - } - self.schema_path = _DEFAULT_SCHEMA_PATH - with open(self.schema_path, "r", encoding="utf-8") as f: - self.schema_config = json.load(f) - - def test_valid_policy_passes_validation(self): - """Valid policy should not raise any exception.""" - validate_policy_config( - self.valid_policy, - self.schema_config, - policy_path="test_policy.json", - schema_path=self.schema_path - ) - - def test_missing_version_raises_error(self): - """Policy missing required 'version' field should raise ValueError.""" - invalid_policy = {"targets": {}} - with self.assertRaises(ValueError) as ctx: - validate_policy_config( - invalid_policy, - self.schema_config, - policy_path="test_policy.json", - schema_path=self.schema_path - ) - self.assertIn("Adapter policy validation failed", str(ctx.exception)) - self.assertIn("version", str(ctx.exception)) - - def test_missing_targets_raises_error(self): - """Policy missing required 'targets' field should raise ValueError.""" - invalid_policy = {"version": "2.0.0"} - with self.assertRaises(ValueError) as ctx: - validate_policy_config( - invalid_policy, - self.schema_config, - policy_path="test_policy.json", - schema_path=self.schema_path - ) - self.assertIn("Adapter policy validation failed", str(ctx.exception)) - self.assertIn("targets", str(ctx.exception)) - - def test_invalid_target_spec_raises_error(self): - """Target spec missing 'sources' should raise ValueError.""" - invalid_policy = { - "version": "2.0.0", - "targets": { - "test.json": {} - } - } - with self.assertRaises(ValueError) as ctx: - validate_policy_config( - invalid_policy, - self.schema_config, - policy_path="test_policy.json", - schema_path=self.schema_path - ) - self.assertIn("Adapter policy validation failed", str(ctx.exception)) - - def test_allowlist_filter_policy_validates(self): - """Policy using allowlist filter type should validate against schema.""" - policy = { - "version": "2.0.0", - "targets": { - "openldap.json": { - "sources": [ - { - "source_file": "base_os.json", - "pulls": [ - { - "source_key": "Base OS", - "filter": { - "type": "allowlist", - "field": "package", - "values": ["openldap-clients"], - "case_sensitive": False, - }, - } - ], - } - ] - } - }, - } - - validate_policy_config( - policy, - self.schema_config, - policy_path="test_policy.json", - schema_path=self.schema_path, - ) - - def test_field_in_filter_policy_validates(self): - """Policy using field_in filter type should validate against schema.""" - policy = { - "version": "2.0.0", - "targets": { - "openldap.json": { - "sources": [ - { - "source_file": "base_os.json", - "pulls": [ - { - "source_key": "Base OS", - "filter": { - "type": "field_in", - "field": "feature", - "values": ["openldap"], - "case_sensitive": False, - }, - } - ], - } - ] - } - }, - } - - validate_policy_config( - policy, - self.schema_config, - policy_path="test_policy.json", - schema_path=self.schema_path, - ) - - def test_any_of_filter_requires_filters(self): - """any_of filter must define nested filters.""" - policy = { - "version": "2.0.0", - "targets": { - "openldap.json": { - "sources": [ - { - "source_file": "base_os.json", - "pulls": [ - {"source_key": "Base OS", "filter": {"type": "any_of"}} - ], - } - ] - } - }, - } - - with self.assertRaises(ValueError) as ctx: - validate_policy_config( - policy, - self.schema_config, - policy_path="test_policy.json", - schema_path=self.schema_path, - ) - self.assertIn("Adapter policy validation failed", str(ctx.exception)) - - def test_any_of_filter_policy_validates(self): - """Policy using any_of filter type should validate against schema.""" - policy = { - "version": "2.0.0", - "targets": { - "openldap.json": { - "sources": [ - { - "source_file": "base_os.json", - "pulls": [ - { - "source_key": "Base OS", - "filter": { - "type": "any_of", - "filters": [ - {"type": "substring", "values": ["ldap"]}, - {"type": "field_in", "field": "feature", "values": ["openldap"]}, - ], - }, - } - ], - } - ] - } - }, - } - - validate_policy_config( - policy, - self.schema_config, - policy_path="test_policy.json", - schema_path=self.schema_path, - ) - - -class TestDiscoverArchitectures(unittest.TestCase): - """Tests for discover_architectures function.""" - - def test_discovers_architecture_directories(self): - """Should return list of subdirectory names.""" - with tempfile.TemporaryDirectory() as tmpdir: - os.makedirs(os.path.join(tmpdir, "x86_64")) - os.makedirs(os.path.join(tmpdir, "aarch64")) - # Create a file (should be ignored) - with open(os.path.join(tmpdir, "readme.txt"), "w") as f: - f.write("test") - - archs = discover_architectures(tmpdir) - self.assertEqual(sorted(archs), ["aarch64", "x86_64"]) - - def test_returns_empty_for_nonexistent_dir(self): - """Should return empty list for non-existent directory.""" - archs = discover_architectures("/nonexistent/path") - self.assertEqual(archs, []) - - def test_returns_empty_for_empty_dir(self): - """Should return empty list for empty directory.""" - with tempfile.TemporaryDirectory() as tmpdir: - archs = discover_architectures(tmpdir) - self.assertEqual(archs, []) - - -class TestDiscoverOsVersions(unittest.TestCase): - """Tests for discover_os_versions function.""" - - def test_discovers_os_and_versions(self): - """Should return list of (os_family, version) tuples.""" - with tempfile.TemporaryDirectory() as tmpdir: - os.makedirs(os.path.join(tmpdir, "x86_64", "rhel", "9.0")) - os.makedirs(os.path.join(tmpdir, "x86_64", "rhel", "8.0")) - os.makedirs(os.path.join(tmpdir, "x86_64", "ubuntu", "22.04")) - - results = discover_os_versions(tmpdir, "x86_64") - self.assertEqual(len(results), 3) - self.assertIn(("rhel", "9.0"), results) - self.assertIn(("rhel", "8.0"), results) - self.assertIn(("ubuntu", "22.04"), results) - - def test_returns_empty_for_nonexistent_arch(self): - """Should return empty list for non-existent architecture.""" - with tempfile.TemporaryDirectory() as tmpdir: - results = discover_os_versions(tmpdir, "nonexistent") - self.assertEqual(results, []) - - -class TestTransformPackage(unittest.TestCase): - """Tests for transform_package function.""" - - def test_no_transform_returns_copy(self): - """No transform config should return a copy of the package.""" - pkg = {"name": "test", "version": "1.0"} - result = transform_package(pkg, None) - self.assertEqual(result, pkg) - self.assertIsNot(result, pkg) - - def test_exclude_fields(self): - """Should exclude specified fields.""" - pkg = {"name": "test", "version": "1.0", "architecture": "x86_64"} - transform = {schema.EXCLUDE_FIELDS: ["architecture"]} - result = transform_package(pkg, transform) - self.assertEqual(result, {"name": "test", "version": "1.0"}) - - def test_rename_fields(self): - """Should rename specified fields.""" - pkg = {"name": "test", "ver": "1.0"} - transform = {schema.RENAME_FIELDS: {"ver": "version"}} - result = transform_package(pkg, transform) - self.assertEqual(result, {"name": "test", "version": "1.0"}) - - def test_exclude_and_rename_combined(self): - """Should apply both exclude and rename.""" - pkg = {"name": "test", "ver": "1.0", "arch": "x86_64"} - transform = { - schema.EXCLUDE_FIELDS: ["arch"], - schema.RENAME_FIELDS: {"ver": "version"} - } - result = transform_package(pkg, transform) - self.assertEqual(result, {"name": "test", "version": "1.0"}) - - -class TestApplySubstringFilter(unittest.TestCase): - """Tests for apply_substring_filter function.""" - - def test_filters_by_substring(self): - """Should filter packages by substring match.""" - packages = [ - {"package": "kubernetes-client"}, - {"package": "kubernetes-server"}, - {"package": "docker-ce"}, - ] - filter_config = { - schema.FIELD: "package", - schema.VALUES: ["kubernetes"] - } - result = apply_substring_filter(packages, filter_config) - self.assertEqual(len(result), 2) - self.assertTrue(all("kubernetes" in p["package"] for p in result)) - - def test_case_insensitive_by_default(self): - """Should be case-insensitive by default.""" - packages = [ - {"package": "Kubernetes-Client"}, - {"package": "docker-ce"}, - ] - filter_config = { - schema.FIELD: "package", - schema.VALUES: ["kubernetes"] - } - result = apply_substring_filter(packages, filter_config) - self.assertEqual(len(result), 1) - - def test_case_sensitive_when_specified(self): - """Should be case-sensitive when specified.""" - packages = [ - {"package": "Kubernetes-Client"}, - {"package": "kubernetes-server"}, - ] - filter_config = { - schema.FIELD: "package", - schema.VALUES: ["kubernetes"], - schema.CASE_SENSITIVE: True - } - result = apply_substring_filter(packages, filter_config) - self.assertEqual(len(result), 1) - self.assertEqual(result[0]["package"], "kubernetes-server") - - def test_empty_values_returns_all(self): - """Empty values list should return all packages.""" - packages = [{"package": "test1"}, {"package": "test2"}] - filter_config = {schema.FIELD: "package", schema.VALUES: []} - result = apply_substring_filter(packages, filter_config) - self.assertEqual(result, packages) - - -class TestAllowlistAndFieldFilters(unittest.TestCase): - def test_allowlist_matches_exact_package_names(self): - packages = [ - {"package": "openldap-clients"}, - {"package": "openldap-servers"}, - {"package": "openmpi"}, - ] - filter_config = { - schema.TYPE: schema.ALLOWLIST_FILTER, - schema.FIELD: "package", - schema.VALUES: ["openldap-clients"], - schema.CASE_SENSITIVE: False, - } - - result = apply_filter(packages, {}, "Base OS", filter_config) - self.assertEqual([p["package"] for p in result], ["openldap-clients"]) - - def test_field_in_matches_classification_field(self): - packages = [ - {"package": "vendor-ldap", "feature": "openldap"}, - {"package": "vendor-ldap2", "feature": "other"}, - {"package": "no-feature"}, - ] - filter_config = { - schema.TYPE: schema.FIELD_IN_FILTER, - schema.FIELD: "feature", - schema.VALUES: ["openldap"], - schema.CASE_SENSITIVE: False, - } - - result = apply_filter(packages, {}, "Base OS", filter_config) - self.assertEqual([p["package"] for p in result], ["vendor-ldap"]) - - def test_any_of_combines_multiple_strategies(self): - packages = [ - {"package": "openldap-clients"}, - {"package": "vendor-ldap", "feature": "openldap"}, - {"package": "slapd-utils"}, - {"package": "unrelated"}, - ] - - filter_config = { - schema.TYPE: schema.ANY_OF_FILTER, - schema.FILTERS: [ - { - schema.TYPE: schema.ALLOWLIST_FILTER, - schema.FIELD: "package", - schema.VALUES: ["openldap-clients"], - schema.CASE_SENSITIVE: False, - }, - { - schema.TYPE: schema.FIELD_IN_FILTER, - schema.FIELD: "feature", - schema.VALUES: ["openldap"], - schema.CASE_SENSITIVE: False, - }, - { - schema.TYPE: schema.SUBSTRING_FILTER, - schema.FIELD: "package", - schema.VALUES: ["slapd"], - schema.CASE_SENSITIVE: False, - }, - ], - } - - result = apply_filter(packages, {}, "Base OS", filter_config) - self.assertEqual( - [p["package"] for p in result], - ["openldap-clients", "vendor-ldap", "slapd-utils"], - ) - - -class TestComputeCommonPackages(unittest.TestCase): - """Tests for compute_common_packages function.""" - - def test_finds_common_packages(self): - """Should find packages common across multiple keys.""" - source_data = { - "role1": {schema.PACKAGES: [ - {"name": "common-pkg", "version": "1.0"}, - {"name": "unique1", "version": "1.0"}, - ]}, - "role2": {schema.PACKAGES: [ - {"name": "common-pkg", "version": "1.0"}, - {"name": "unique2", "version": "1.0"}, - ]}, - } - common_keys, key_to_pkg = compute_common_packages( - source_data, ["role1", "role2"], min_occurrences=2 - ) - self.assertEqual(len(common_keys), 1) - - def test_respects_min_occurrences(self): - """Should respect min_occurrences threshold.""" - source_data = { - "role1": {schema.PACKAGES: [{"name": "pkg1"}]}, - "role2": {schema.PACKAGES: [{"name": "pkg1"}]}, - "role3": {schema.PACKAGES: [{"name": "pkg2"}]}, - } - common_keys, _ = compute_common_packages( - source_data, ["role1", "role2", "role3"], min_occurrences=3 - ) - self.assertEqual(len(common_keys), 0) - - -class TestMergeTransform(unittest.TestCase): - """Tests for merge_transform function.""" - - def test_none_inputs_return_none(self): - """Both None should return None.""" - self.assertIsNone(merge_transform(None, None)) - - def test_base_only(self): - """Only base should return base.""" - base = {schema.EXCLUDE_FIELDS: ["arch"]} - self.assertEqual(merge_transform(base, None), base) - - def test_override_only(self): - """Only override should return override.""" - override = {schema.EXCLUDE_FIELDS: ["arch"]} - self.assertEqual(merge_transform(None, override), override) - - def test_override_wins(self): - """Override values should win.""" - base = {schema.EXCLUDE_FIELDS: ["arch"]} - override = {schema.EXCLUDE_FIELDS: ["version"]} - result = merge_transform(base, override) - self.assertEqual(result[schema.EXCLUDE_FIELDS], ["version"]) - - -class TestCheckConditions(unittest.TestCase): - """Tests for check_conditions function.""" - - def test_no_conditions_returns_true(self): - """No conditions should always return True.""" - self.assertTrue(check_conditions(None, "x86_64", "rhel", "9.0")) - - def test_architecture_condition(self): - """Should check architecture condition.""" - conditions = {schema.ARCHITECTURES: ["x86_64"]} - self.assertTrue(check_conditions(conditions, "x86_64", "rhel", "9.0")) - self.assertFalse(check_conditions(conditions, "aarch64", "rhel", "9.0")) - - def test_os_family_condition(self): - """Should check OS family condition.""" - conditions = {schema.OS_FAMILIES: ["rhel"]} - self.assertTrue(check_conditions(conditions, "x86_64", "rhel", "9.0")) - self.assertFalse(check_conditions(conditions, "x86_64", "ubuntu", "22.04")) - - def test_os_version_condition(self): - """Should check OS version condition.""" - conditions = {schema.OS_VERSIONS: ["9.0"]} - self.assertTrue(check_conditions(conditions, "x86_64", "rhel", "9.0")) - self.assertFalse(check_conditions(conditions, "x86_64", "rhel", "8.0")) - - def test_multiple_conditions_all_must_pass(self): - """All conditions must pass.""" - conditions = { - schema.ARCHITECTURES: ["x86_64"], - schema.OS_FAMILIES: ["rhel"], - schema.OS_VERSIONS: ["9.0"] - } - self.assertTrue(check_conditions(conditions, "x86_64", "rhel", "9.0")) - self.assertFalse(check_conditions(conditions, "aarch64", "rhel", "9.0")) - - -class TestDeriveCommonRole(unittest.TestCase): - """Tests for derive_common_role function.""" - - def test_derives_common_packages(self): - """Should derive common packages into new role.""" - target_roles = { - "role1": [{"name": "common"}, {"name": "unique1"}], - "role2": [{"name": "common"}, {"name": "unique2"}], - } - derive_common_role( - target_roles, - derived_key="common_role", - from_keys=["role1", "role2"], - min_occurrences=2, - remove_from_sources=True - ) - self.assertIn("common_role", target_roles) - self.assertEqual(len(target_roles["common_role"]), 1) - self.assertEqual(target_roles["common_role"][0]["name"], "common") - - def test_removes_from_sources_when_specified(self): - """Should remove common packages from source roles.""" - target_roles = { - "role1": [{"name": "common"}, {"name": "unique1"}], - "role2": [{"name": "common"}, {"name": "unique2"}], - } - derive_common_role( - target_roles, - derived_key="common_role", - from_keys=["role1", "role2"], - min_occurrences=2, - remove_from_sources=True - ) - self.assertEqual(len(target_roles["role1"]), 1) - self.assertEqual(target_roles["role1"][0]["name"], "unique1") - - def test_keeps_sources_when_not_removing(self): - """Should keep source packages when remove_from_sources=False.""" - target_roles = { - "role1": [{"name": "common"}, {"name": "unique1"}], - "role2": [{"name": "common"}, {"name": "unique2"}], - } - derive_common_role( - target_roles, - derived_key="common_role", - from_keys=["role1", "role2"], - min_occurrences=2, - remove_from_sources=False - ) - self.assertEqual(len(target_roles["role1"]), 2) - - -class TestWriteConfigFile(unittest.TestCase): - """Tests for write_config_file function.""" - - def test_writes_valid_json(self): - """Should write valid JSON file.""" - with tempfile.TemporaryDirectory() as tmpdir: - file_path = os.path.join(tmpdir, "subdir", "test.json") - config = { - "role1": {schema.CLUSTER: [{"name": "pkg1"}]}, - "role2": {schema.CLUSTER: [{"name": "pkg2"}]}, - } - write_config_file(file_path, config) - - self.assertTrue(os.path.exists(file_path)) - with open(file_path, "r", encoding="utf-8") as f: - loaded = json.load(f) - self.assertEqual(loaded["role1"][schema.CLUSTER][0]["name"], "pkg1") - - def test_creates_parent_directories(self): - """Should create parent directories if they don't exist.""" - with tempfile.TemporaryDirectory() as tmpdir: - file_path = os.path.join(tmpdir, "a", "b", "c", "test.json") - config = {"role1": {schema.CLUSTER: []}} - write_config_file(file_path, config) - self.assertTrue(os.path.exists(file_path)) - - -class TestGenerateConfigsFromPolicy(unittest.TestCase): - """Tests for generate_configs_from_policy function.""" - - def setUp(self): - self.test_fixtures_dir = os.path.join(CATALOG_PARSER_DIR, "test_fixtures") - self.test_policy_path = os.path.join(self.test_fixtures_dir, "adapter_policy_test.json") - - def test_generates_output_files(self): - """Should generate output JSON files from valid policy.""" - with tempfile.TemporaryDirectory() as tmpdir: - # Create input directory structure - input_dir = os.path.join(tmpdir, "input") - output_dir = os.path.join(tmpdir, "output") - os.makedirs(os.path.join(input_dir, "x86_64", "rhel", "9.0")) - - # Create source file - source_data = { - "Base OS": { - schema.PACKAGES: [ - {"package": "test-pkg", "version": "1.0"} - ] - } - } - with open(os.path.join(input_dir, "x86_64", "rhel", "9.0", "base_os.json"), "w") as f: - json.dump(source_data, f) - - # Create minimal policy - policy = { - "version": "2.0.0", - "targets": { - "output.json": { - "sources": [{ - "source_file": "base_os.json", - "pulls": [{"source_key": "Base OS", "target_key": "base_role"}] - }] - } - } - } - policy_path = os.path.join(tmpdir, "policy.json") - with open(policy_path, "w") as f: - json.dump(policy, f) - - generate_configs_from_policy( - input_dir=input_dir, - output_dir=output_dir, - policy_path=policy_path, - schema_path=_DEFAULT_SCHEMA_PATH - ) - - output_file = os.path.join(output_dir, "x86_64", "rhel", "9.0", "output.json") - self.assertTrue(os.path.exists(output_file)) - - def test_generates_openldap_with_any_of_filter(self): - with tempfile.TemporaryDirectory() as tmpdir: - input_dir = os.path.join(tmpdir, "input") - output_dir = os.path.join(tmpdir, "output") - os.makedirs(os.path.join(input_dir, "x86_64", "rhel", "9.0")) - - source_data = { - "Base OS": { - schema.PACKAGES: [ - {"package": "openldap-clients", "type": "rpm", "architecture": ["x86_64"]}, - {"package": "vendor-directory-client", "type": "rpm", "architecture": ["x86_64"], "feature": "openldap"}, - {"package": "slapd-utils", "type": "rpm", "architecture": ["x86_64"]}, - {"package": "bash", "type": "rpm", "architecture": ["x86_64"]}, - ] - } - } - with open(os.path.join(input_dir, "x86_64", "rhel", "9.0", "base_os.json"), "w") as f: - json.dump(source_data, f) - - policy = { - "version": "2.0.0", - "targets": { - "openldap.json": { - "transform": {"exclude_fields": ["architecture"]}, - "sources": [ - { - "source_file": "base_os.json", - "pulls": [ - { - "source_key": "Base OS", - "target_key": "openldap", - "filter": { - "type": "any_of", - "filters": [ - {"type": "allowlist", "field": "package", "values": ["openldap-clients"], "case_sensitive": False}, - {"type": "field_in", "field": "feature", "values": ["openldap"], "case_sensitive": False}, - {"type": "substring", "field": "package", "values": ["slapd"], "case_sensitive": False}, - ], - }, - } - ], - } - ], - } - }, - } - policy_path = os.path.join(tmpdir, "policy.json") - with open(policy_path, "w") as f: - json.dump(policy, f) - - generate_configs_from_policy( - input_dir=input_dir, - output_dir=output_dir, - policy_path=policy_path, - schema_path=_DEFAULT_SCHEMA_PATH, - ) - - output_file = os.path.join(output_dir, "x86_64", "rhel", "9.0", "openldap.json") - self.assertTrue(os.path.exists(output_file)) - - with open(output_file, "r", encoding="utf-8") as f: - out_json = json.load(f) - - self.assertIn("openldap", out_json) - pkgs = out_json["openldap"][schema.CLUSTER] - - self.assertEqual( - [p.get("package") for p in pkgs], - ["openldap-clients", "vendor-directory-client", "slapd-utils"], - ) - self.assertTrue(all("architecture" not in p for p in pkgs)) - - def test_invalid_policy_raises_error(self): - """Should raise ValueError for invalid policy.""" - with tempfile.TemporaryDirectory() as tmpdir: - input_dir = os.path.join(tmpdir, "input") - output_dir = os.path.join(tmpdir, "output") - os.makedirs(input_dir) - - # Create invalid policy (missing version) - invalid_policy = {"targets": {}} - policy_path = os.path.join(tmpdir, "invalid_policy.json") - with open(policy_path, "w") as f: - json.dump(invalid_policy, f) - - with self.assertRaises(ValueError) as ctx: - generate_configs_from_policy( - input_dir=input_dir, - output_dir=output_dir, - policy_path=policy_path, - schema_path=_DEFAULT_SCHEMA_PATH - ) - self.assertIn("Adapter policy validation failed", str(ctx.exception)) - - def test_missing_input_dir_raises_file_not_found(self): - """Should raise FileNotFoundError if input_dir does not exist.""" - with tempfile.TemporaryDirectory() as tmpdir: - output_dir = os.path.join(tmpdir, "output") - missing_input_dir = os.path.join(tmpdir, "does_not_exist") - - with self.assertRaises(FileNotFoundError): - generate_configs_from_policy( - input_dir=missing_input_dir, - output_dir=output_dir, - policy_path=_DEFAULT_POLICY_PATH, - schema_path=_DEFAULT_SCHEMA_PATH, - ) - - def test_missing_policy_file_raises_file_not_found(self): - """Should raise FileNotFoundError if policy_path does not exist.""" - with tempfile.TemporaryDirectory() as tmpdir: - input_dir = os.path.join(tmpdir, "input") - output_dir = os.path.join(tmpdir, "output") - os.makedirs(input_dir) - - missing_policy_path = os.path.join(tmpdir, "missing_policy.json") - - with self.assertRaises(FileNotFoundError): - generate_configs_from_policy( - input_dir=input_dir, - output_dir=output_dir, - policy_path=missing_policy_path, - schema_path=_DEFAULT_SCHEMA_PATH, - ) - - def test_missing_schema_file_raises_file_not_found(self): - """Should raise FileNotFoundError if schema_path does not exist.""" - with tempfile.TemporaryDirectory() as tmpdir: - input_dir = os.path.join(tmpdir, "input") - output_dir = os.path.join(tmpdir, "output") - os.makedirs(input_dir) - - missing_schema_path = os.path.join(tmpdir, "missing_schema.json") - - with self.assertRaises(FileNotFoundError): - generate_configs_from_policy( - input_dir=input_dir, - output_dir=output_dir, - policy_path=_DEFAULT_POLICY_PATH, - schema_path=missing_schema_path, - ) - - -class TestDefaultPaths(unittest.TestCase): - """Tests for default path constants.""" - - def test_default_policy_path_exists(self): - """Default policy path should point to existing file.""" - self.assertTrue( - os.path.exists(_DEFAULT_POLICY_PATH), - f"Default policy file not found: {_DEFAULT_POLICY_PATH}" - ) - - def test_default_schema_path_exists(self): - """Default schema path should point to existing file.""" - self.assertTrue( - os.path.exists(_DEFAULT_SCHEMA_PATH), - f"Default schema file not found: {_DEFAULT_SCHEMA_PATH}" - ) - - def test_default_policy_validates_against_schema(self): - """Default policy should validate against default schema.""" - with open(_DEFAULT_POLICY_PATH, "r", encoding="utf-8") as f: - policy = json.load(f) - with open(_DEFAULT_SCHEMA_PATH, "r", encoding="utf-8") as f: - schema_config = json.load(f) - - # Should not raise - validate_policy_config( - policy, - schema_config, - policy_path=_DEFAULT_POLICY_PATH, - schema_path=_DEFAULT_SCHEMA_PATH - ) - - -class TestProcessTargetSpec(unittest.TestCase): - """Tests for process_target_spec function.""" - - def test_processes_simple_target(self): - """Should process a simple target specification.""" - source_files = { - "source.json": { - "role1": {schema.PACKAGES: [{"name": "pkg1"}]} - } - } - target_spec = { - "sources": [{ - "source_file": "source.json", - "pulls": [{"source_key": "role1", "target_key": "output_role"}] - }] - } - target_configs = {} - - process_target_spec( - target_file="output.json", - target_spec=target_spec, - source_files=source_files, - target_configs=target_configs, - arch="x86_64", - os_family="rhel", - os_version="9.0" - ) - - self.assertIn("output.json", target_configs) - self.assertIn("output_role", target_configs["output.json"]) - - def test_skips_when_conditions_not_met(self): - """Should skip target when conditions are not met.""" - source_files = {"source.json": {"role1": {schema.PACKAGES: []}}} - target_spec = { - "conditions": {schema.ARCHITECTURES: ["aarch64"]}, - "sources": [{ - "source_file": "source.json", - "pulls": [{"source_key": "role1"}] - }] - } - target_configs = {} - - process_target_spec( - target_file="output.json", - target_spec=target_spec, - source_files=source_files, - target_configs=target_configs, - arch="x86_64", - os_family="rhel", - os_version="9.0" - ) - - self.assertNotIn("output.json", target_configs) - - def test_applies_transform(self): - """Should apply transform to packages.""" - source_files = { - "source.json": { - "role1": {schema.PACKAGES: [ - {"name": "pkg1", "architecture": "x86_64"} - ]} - } - } - target_spec = { - "transform": {schema.EXCLUDE_FIELDS: ["architecture"]}, - "sources": [{ - "source_file": "source.json", - "pulls": [{"source_key": "role1", "target_key": "output_role"}] - }] - } - target_configs = {} - - process_target_spec( - target_file="output.json", - target_spec=target_spec, - source_files=source_files, - target_configs=target_configs, - arch="x86_64", - os_family="rhel", - os_version="9.0" - ) - - pkgs = target_configs["output.json"]["output_role"][schema.CLUSTER] - self.assertNotIn("architecture", pkgs[0]) - - -if __name__ == "__main__": - unittest.main() diff --git a/build_stream/core/catalog/tests/test_generator_cli_defaults.py b/build_stream/core/catalog/tests/test_generator_cli_defaults.py deleted file mode 100644 index 9062b8694e..0000000000 --- a/build_stream/core/catalog/tests/test_generator_cli_defaults.py +++ /dev/null @@ -1,56 +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. - -import os -import sys -import tempfile -import unittest - -HERE = os.path.dirname(__file__) -CATALOG_PARSER_DIR = os.path.dirname(HERE) -PROJECT_ROOT = os.path.dirname(CATALOG_PARSER_DIR) -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - -from catalog_parser.generator import generate_root_json_from_catalog, _DEFAULT_SCHEMA_PATH - - -class TestGeneratorDefaults(unittest.TestCase): - def test_default_schema_path_points_to_resources(self): - catalog_parser_dir = os.path.dirname(os.path.dirname(__file__)) - expected_schema = os.path.join(catalog_parser_dir, "resources", "CatalogSchema.json") - self.assertEqual(os.path.abspath(_DEFAULT_SCHEMA_PATH), os.path.abspath(expected_schema)) - - def test_generate_root_json_with_defaults_writes_output(self): - catalog_parser_dir = os.path.dirname(os.path.dirname(__file__)) - catalog_path = os.path.join(catalog_parser_dir, "test_fixtures", "catalog_rhel.json") - - with tempfile.TemporaryDirectory() as tmpdir: - generate_root_json_from_catalog( - catalog_path=catalog_path, - output_root=tmpdir, - ) - - # We expect at least one arch/os/version directory with functional_layer.json - found = False - for root, dirs, files in os.walk(tmpdir): - if "functional_layer.json" in files: - found = True - break - - self.assertTrue(found, "functional_layer.json not generated under any arch/os/version") - - -if __name__ == "__main__": - unittest.main() diff --git a/build_stream/core/catalog/tests/test_generator_package_list.py b/build_stream/core/catalog/tests/test_generator_package_list.py deleted file mode 100644 index 1fe00a4ef3..0000000000 --- a/build_stream/core/catalog/tests/test_generator_package_list.py +++ /dev/null @@ -1,224 +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. - -"""Unit tests for get_package_list function in generator module.""" - -import json -import os -import sys -import tempfile -import unittest - -from jsonschema import ValidationError - -HERE = os.path.dirname(__file__) -CATALOG_PARSER_DIR = os.path.dirname(HERE) -PROJECT_ROOT = os.path.dirname(CATALOG_PARSER_DIR) -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - -from catalog_parser.generator import ( - FeatureList, - serialize_json, - get_package_list, -) - - -class TestGetPackageList(unittest.TestCase): - """Tests for get_package_list function.""" - - def setUp(self): - """Set up test fixtures.""" - self.base_dir = os.path.dirname(__file__) - self.fixture_path = os.path.abspath( - os.path.join(self.base_dir, "..", "test_fixtures", "functional_layer.json") - ) - - def test_get_packages_for_valid_single_role(self): - """TC01: Given a valid role, returns list with one role object containing packages.""" - result = get_package_list(self.fixture_path, role="Compiler") - - self.assertIsInstance(result, list) - self.assertEqual(len(result), 1) - self.assertEqual(result[0]["roleName"], "Compiler") - self.assertIn("packages", result[0]) - self.assertIsInstance(result[0]["packages"], list) - self.assertGreater(len(result[0]["packages"]), 0) - - def test_get_packages_for_all_roles_when_role_is_none(self): - """TC02: When role is None, returns list with all role objects.""" - result = get_package_list(self.fixture_path, role=None) - - self.assertIsInstance(result, list) - # Fixture has 6 roles - expected_roles = [ - "Compiler", - "K8S Controller", - "K8S Worker", - "Login Node", - "Slurm Controller", - "Slurm Worker", - ] - actual_roles = [r["roleName"] for r in result] - self.assertCountEqual(actual_roles, expected_roles) - - def test_invalid_role_raises_value_error(self): - """TC03: Invalid/unknown role raises ValueError with clear message.""" - with self.assertRaises(ValueError) as context: - get_package_list(self.fixture_path, role="NonExistentRole") - - self.assertIn("NonExistentRole", str(context.exception)) - - def test_empty_role_raises_value_error(self): - """Empty role string is treated as invalid input.""" - with self.assertRaises(ValueError) as context: - get_package_list(self.fixture_path, role="") - - self.assertIn("non-empty", str(context.exception)) - - def test_file_not_found_raises_error(self): - """TC04: Non-existent file raises FileNotFoundError.""" - with self.assertRaises(FileNotFoundError): - get_package_list("/nonexistent/path/functional_layer.json") - - def test_malformed_json_raises_error(self): - """TC05: Malformed JSON raises json.JSONDecodeError.""" - with tempfile.TemporaryDirectory() as tmp_dir: - malformed_path = os.path.join(tmp_dir, "malformed.json") - with open(malformed_path, "w", encoding="utf-8") as f: - f.write("{ invalid json }") - - with self.assertRaises(json.JSONDecodeError): - get_package_list(malformed_path) - - def test_schema_validation_failure_raises_error(self): - """TC06: JSON that fails schema validation raises ValidationError.""" - with tempfile.TemporaryDirectory() as tmp_dir: - # Missing required 'architecture' field for a package item - invalid_json = { - "SomeRole": { - "packages": [ - { - "package": "firewalld", - "type": "rpm", - "repo_name": "x86_64_baseos", - # Missing 'architecture' field - } - ] - } - } - json_path = os.path.join(tmp_dir, "invalid_schema.json") - with open(json_path, "w", encoding="utf-8") as f: - json.dump(invalid_json, f) - - with self.assertRaises(ValidationError): - get_package_list(json_path) - - def test_empty_feature_list_returns_empty_list(self): - """TC07: Empty feature list returns empty list.""" - with tempfile.TemporaryDirectory() as tmp_dir: - empty_feature_list = FeatureList(features={}) - json_path = os.path.join(tmp_dir, "empty_functional_layer.json") - serialize_json(empty_feature_list, json_path) - - result = get_package_list(json_path) - - self.assertEqual(result, []) - - def test_package_attributes_are_complete(self): - """TC08: All package fields are present in the response.""" - result = get_package_list(self.fixture_path, role="Compiler") - - self.assertEqual(len(result), 1) - packages = result[0]["packages"] - self.assertGreater(len(packages), 0) - - # Check first package has all required fields - first_pkg = packages[0] - required_fields = ["name", "type", "repo_name", "architecture", "uri", "tag"] - for field in required_fields: - self.assertIn(field, first_pkg, f"Missing field: {field}") - - def test_package_with_uri_and_tag(self): - """Verify packages with uri and tag fields are correctly returned.""" - result = get_package_list(self.fixture_path, role="K8S Controller") - - packages = result[0]["packages"] - # Find a package with tag (image type) - image_pkgs = [p for p in packages if p["type"] == "image"] - self.assertGreater(len(image_pkgs), 0) - # Image packages should have tag - self.assertIsNotNone(image_pkgs[0].get("tag")) - - # Find a package with uri (tarball type) - tarball_pkgs = [p for p in packages if p["type"] == "tarball"] - self.assertGreater(len(tarball_pkgs), 0) - # Tarball packages should have uri - self.assertIsNotNone(tarball_pkgs[0].get("uri")) - - def test_role_with_spaces_in_name(self): - """Verify roles with spaces in name work correctly.""" - result = get_package_list(self.fixture_path, role="K8S Controller") - - self.assertEqual(len(result), 1) - self.assertEqual(result[0]["roleName"], "K8S Controller") - - def test_all_roles_returns_correct_package_counts(self): - """Verify each role returns the correct number of packages.""" - result = get_package_list(self.fixture_path, role=None) - - # Verify we have packages for each role - for role_obj in result: - self.assertIn("roleName", role_obj) - self.assertIn("packages", role_obj) - # Each role should have at least one package - self.assertGreater( - len(role_obj["packages"]), - 0, - f"Role {role_obj['roleName']} has no packages", - ) - - def test_case_insensitive_role_matching_lowercase(self): - """Verify role matching is case-insensitive with lowercase input.""" - result = get_package_list(self.fixture_path, role="compiler") - - self.assertEqual(len(result), 1) - # Should return the original role name from JSON - self.assertEqual(result[0]["roleName"], "Compiler") - - def test_case_insensitive_role_matching_uppercase(self): - """Verify role matching is case-insensitive with uppercase input.""" - result = get_package_list(self.fixture_path, role="COMPILER") - - self.assertEqual(len(result), 1) - self.assertEqual(result[0]["roleName"], "Compiler") - - def test_case_insensitive_role_matching_mixed_case(self): - """Verify role matching is case-insensitive with mixed case input.""" - result = get_package_list(self.fixture_path, role="k8s controller") - - self.assertEqual(len(result), 1) - self.assertEqual(result[0]["roleName"], "K8S Controller") - - def test_case_insensitive_role_matching_preserves_original_name(self): - """Verify the returned roleName preserves the original case from JSON.""" - result = get_package_list(self.fixture_path, role="SLURM CONTROLLER") - - self.assertEqual(len(result), 1) - # Should preserve original case from JSON - self.assertEqual(result[0]["roleName"], "Slurm Controller") - - -if __name__ == "__main__": - unittest.main() diff --git a/build_stream/core/catalog/tests/test_generator_roles.py b/build_stream/core/catalog/tests/test_generator_roles.py deleted file mode 100644 index c829bed2c0..0000000000 --- a/build_stream/core/catalog/tests/test_generator_roles.py +++ /dev/null @@ -1,89 +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. - -import os -import sys -import tempfile -import unittest -from jsonschema import ValidationError - -HERE = os.path.dirname(__file__) -CATALOG_PARSER_DIR = os.path.dirname(HERE) -PROJECT_ROOT = os.path.dirname(CATALOG_PARSER_DIR) -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - -from catalog_parser.generator import ( - FeatureList, - serialize_json, - get_functional_layer_roles_from_file, -) - - -class TestGetFunctionalLayerRolesFromFile(unittest.TestCase): - def test_returns_all_role_names_from_fixture(self): - base_dir = os.path.dirname(__file__) - fixture_path = os.path.abspath( - os.path.join(base_dir, "..", "test_fixtures", "functional_layer.json") - ) - - roles = get_functional_layer_roles_from_file(fixture_path) - - expected_roles = [ - "Compiler", - "K8S Controller", - "K8S Worker", - "Login Node", - "Slurm Controller", - "Slurm Worker", - ] - - self.assertCountEqual(roles, expected_roles) - - def test_empty_feature_list_returns_empty_roles(self): - with tempfile.TemporaryDirectory() as tmp_dir: - empty_feature_list = FeatureList(features={}) - json_path = os.path.join(tmp_dir, "functional_layer.json") - serialize_json(empty_feature_list, json_path) - - roles = get_functional_layer_roles_from_file(json_path) - - self.assertEqual(roles, []) - - def test_invalid_functional_layer_json_fails_schema_validation(self): - with tempfile.TemporaryDirectory() as tmp_dir: - # Missing required 'architecture' field for a package item - invalid_json = { - "SomeRole": { - "packages": [ - { - "package": "firewalld", - "type": "rpm", - "repo_name": "x86_64_baseos", - } - ] - } - } - json_path = os.path.join(tmp_dir, "functional_layer_invalid.json") - with open(json_path, "w") as f: - import json - - json.dump(invalid_json, f) - - with self.assertRaises(ValidationError): - get_functional_layer_roles_from_file(json_path) - - -if __name__ == "__main__": - unittest.main() diff --git a/build_stream/core/catalog/tests/test_parser_defaults.py b/build_stream/core/catalog/tests/test_parser_defaults.py deleted file mode 100644 index 923aac465b..0000000000 --- a/build_stream/core/catalog/tests/test_parser_defaults.py +++ /dev/null @@ -1,44 +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. - -import os -import sys -import unittest - -HERE = os.path.dirname(__file__) -CATALOG_PARSER_DIR = os.path.dirname(HERE) -PROJECT_ROOT = os.path.dirname(CATALOG_PARSER_DIR) -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - -from catalog_parser.parser import ParseCatalog, _DEFAULT_SCHEMA_PATH - - -class TestParseCatalogDefaults(unittest.TestCase): - def test_default_schema_path_points_to_resources(self): - catalog_parser_dir = os.path.dirname(os.path.dirname(__file__)) - expected_schema = os.path.join(catalog_parser_dir, "resources", "CatalogSchema.json") - self.assertEqual(os.path.abspath(_DEFAULT_SCHEMA_PATH), os.path.abspath(expected_schema)) - - def test_parse_catalog_with_explicit_paths_uses_fixture(self): - catalog_parser_dir = os.path.dirname(os.path.dirname(__file__)) - catalog_path = os.path.join(catalog_parser_dir, "test_fixtures", "catalog_rhel.json") - schema_path = os.path.join(catalog_parser_dir, "resources", "CatalogSchema.json") - - catalog = ParseCatalog(catalog_path, schema_path) - self.assertGreater(len(catalog.functional_packages), 0) - - -if __name__ == "__main__": - unittest.main() diff --git a/build_stream/tests/README.md b/build_stream/tests/README.md index fcd6ae3aff..fe00a4740c 100644 --- a/build_stream/tests/README.md +++ b/build_stream/tests/README.md @@ -1,80 +1,51 @@ # Build Stream Test Suite -This directory contains comprehensive unit and integration tests for all Build Stream workflows including Jobs API, Catalog Processing, Local Repository, Image Building, and Validation. +Comprehensive tests for all Build Stream workflows including Jobs API, Catalog +Processing, Local Repository, Image Building, Deploy, Restart, Cleanup, Upload, +and Validation. + +All tests (including former "integration" tests that exercise full API request/ +response cycles with a real FastAPI `TestClient` and SQLite-backed DB) now live +under `tests/unit/`. Each test subpackage carries its own `conftest.py` with the +fixtures it needs (mocked auth, temp SQLite DB, etc.) so tests remain isolated +and fast without any external services. ## Test Structure ``` tests/ -├── integration/ # Integration tests for end-to-end workflows -│ ├── api/ # API endpoint integration tests -│ │ ├── jobs/ # Jobs API tests -│ │ │ ├── conftest.py # Shared fixtures -│ │ │ ├── test_create_job_api.py # POST /jobs tests -│ │ │ ├── test_get_job_api.py # GET /jobs/{id} tests -│ │ │ └── test_delete_job_api.py # DELETE /jobs/{id} tests -│ │ ├── catalog_roles/ # Catalog processing tests -│ │ │ ├── conftest.py # Shared fixtures -│ │ │ ├── test_get_roles_api.py # GET /catalog_roles tests -│ │ │ └── test_catalog_workflow.py # End-to-end catalog tests -│ │ ├── parse_catalog/ # Catalog parsing tests -│ │ │ ├── conftest.py # Shared fixtures -│ │ │ └── test_parse_catalog_api.py # POST /parse_catalog tests -│ │ ├── local_repo/ # Local repository tests -│ │ │ ├── conftest.py # Shared fixtures -│ │ │ ├── test_create_local_repo_api.py # POST /local_repo tests -│ │ │ └── test_repo_workflow.py # End-to-end repo tests -│ │ ├── build_image/ # Image building tests -│ │ │ ├── conftest.py # Shared fixtures -│ │ │ ├── test_build_image_api.py # POST /build_image tests -│ │ │ └── test_multi_arch_build.py # Multi-architecture tests -│ │ └── validate/ # Validation tests -│ │ ├── conftest.py # Shared fixtures -│ │ └── test_validate_api.py # POST /validate tests -│ ├── core/ # Core domain integration tests -│ │ ├── jobs/ # Job entity integration tests -│ │ ├── catalog/ # Catalog entity integration tests -│ │ └── localrepo/ # Repository entity integration tests -│ └── infra/ # Infrastructure integration tests -│ ├── repositories/ # Repository integration tests -│ └── external/ # External service integration tests -├── unit/ # Unit tests for individual components -│ ├── api/ # API layer unit tests -│ │ ├── jobs/ # Jobs API unit tests -│ │ │ ├── test_schemas.py # Pydantic schema tests -│ │ │ ├── test_dependencies.py # Dependency injection tests -│ │ │ └── test_routes.py # Route handler tests -│ │ ├── catalog_roles/ # Catalog API unit tests -│ │ ├── local_repo/ # Local repo API unit tests -│ │ └── validate/ # Validation API unit tests -│ ├── core/ # Core domain unit tests -│ │ ├── jobs/ # Job entity and value object tests -│ │ ├── catalog/ # Catalog entity tests -│ │ ├── localrepo/ # Repository entity tests -│ │ └── validate/ # Validation entity tests -│ ├── orchestrator/ # Use case unit tests -│ │ ├── jobs/ # Job use case tests -│ │ ├── catalog/ # Catalog use case tests -│ │ ├── local_repo/ # Repository use case tests -│ │ └── validate/ # Validation use case tests -│ └── infra/ # Infrastructure unit tests -│ ├── repositories/ # Repository implementation tests -│ ├── artifact_store/ # Artifact store tests -│ └── db/ # Database layer tests -├── end_to_end/ # Complete workflow tests -│ ├── test_full_job_workflow.py # Complete job lifecycle -│ └── test_catalog_to_image.py # Catalog to image workflow -├── performance/ # Performance and load tests -│ └── test_load.py # Load testing scenarios -├── fixtures/ # Shared test fixtures -│ ├── job_fixtures.py # Job test data -│ └── repo_fixtures.py # Repository test data -├── mocks/ # Mock objects and data -│ ├── mock_vault.py # Vault mock -│ └── mock_registry.py # Registry mock -└── utils/ # Test utilities and helpers - ├── assertions.py # Custom assertions - └── helpers.py # Test helper functions +├── unit/ # All tests (isolated, no external services required) +│ ├── api/ +│ │ ├── auth/ # Registration & token tests +│ │ ├── build_image/ # POST /build_image route + API tests +│ │ ├── catalog_roles/ # GET /catalog_roles route + service tests +│ │ ├── deploy/ # Deploy route error handler tests +│ │ ├── generate_input_files/ # Generate input files route + API tests +│ │ ├── images/ # Images route tests +│ │ ├── jobs/ # Jobs CRUD, schema & dependency tests +│ │ ├── local_repo/ # Local repo route + API tests +│ │ ├── parse_catalog/ # Parse catalog route + API tests +│ │ ├── restart/ # Restart stage route + API tests +│ │ ├── upload/ # Upload route tests +│ │ └── validate/ # Validate route + API tests +│ ├── core/ +│ │ ├── catalog/ # Parser, adapter, generator, policy, diff regression +│ │ ├── cleanup/ # Cleanup exceptions, S3 service interface +│ │ ├── deploy/ # Deploy entities, services, exceptions +│ │ ├── image_group/ # ImageGroup entities & value objects +│ │ ├── jobs/ # Job entities, value objects, state machine +│ │ └── localrepo/ # Local repo entities +│ ├── infra/ +│ │ ├── artifact_store/ # File artifact store tests +│ │ └── db/ # SQL repository tests (skips if no live PostgreSQL) +│ └── orchestrator/ +│ ├── catalog/ # Catalog use case & command tests +│ ├── common/ # ResultPoller tests (build-image, deploy/restart failure) +│ ├── local_repo/ # Local repo use case tests +│ └── validate/ # Validate use case (retry lifecycle, guard edge cases) +├── others/ # Design rule enforcement tests +├── performance/ # Performance/load tests +└── conftest.py # Shared fixtures (auth, client, DB session) ``` ## Prerequisites @@ -96,43 +67,40 @@ Required packages: ### Run All Tests ```bash -# Run all tests -pytest tests/ -v +# Run all tests (ENV=test prevents debugpy from attaching) +ENV=test python -m pytest tests/ -v # Run with coverage -pytest tests/ --cov=api --cov=orchestrator --cov-report=html +ENV=test python -m pytest tests/ --cov=api --cov=orchestrator --cov=core --cov-report=html ``` ### Run Specific Test Suites ```bash -# Integration tests only -pytest tests/integration/ -v - -# Unit tests only -pytest tests/unit/ -v +# All tests (fast — no external services required) +python -m pytest tests/unit/ -v # API tests only -pytest tests/integration/api/ tests/unit/api/ -v +python -m pytest tests/unit/api/ -v ``` ### Run Specific Test Files ```bash # Jobs API tests -pytest tests/integration/api/jobs/test_create_job_api.py -v +pytest tests/unit/api/jobs/test_create_job_api.py -v # Catalog processing tests -pytest tests/integration/api/catalog_roles/ -v +pytest tests/unit/api/catalog_roles/ -v # Local repository tests -pytest tests/integration/api/local_repo/ -v +pytest tests/unit/api/local_repo/ -v # Image building tests -pytest tests/integration/api/build_image/ -v +pytest tests/unit/api/build_image/ -v # Validation tests -pytest tests/integration/api/validate/ -v +pytest tests/unit/api/validate/ -v # Schema validation tests pytest tests/unit/api/jobs/test_schemas.py -v @@ -145,13 +113,13 @@ pytest tests/unit/orchestrator/ -v ```bash # Run specific test class -pytest tests/integration/api/jobs/test_create_job_api.py::TestCreateJobSuccess -v +pytest tests/unit/api/jobs/test_create_job_api.py::TestCreateJobSuccess -v # Run specific test function -pytest tests/integration/api/jobs/test_create_job_api.py::TestCreateJobSuccess::test_create_job_returns_201_with_valid_request -v +pytest tests/unit/api/jobs/test_create_job_api.py::TestCreateJobSuccess::test_create_job_returns_201_with_valid_request -v # Run tests matching pattern -pytest tests/integration/ -k idempotency -v +pytest tests/unit/ -k idempotency -v ``` ## Test Types @@ -163,19 +131,14 @@ Test individual components in isolation: - **Orchestrator Layer**: Use cases and business logic - **Infrastructure Layer**: Repositories, external integrations -### Integration Tests -Test component interactions: -- **API Integration**: Full HTTP request/response cycles -- **Database Integration**: Repository operations with real DB -- **External Services**: Vault, Pulp, container registries -- **Cross-Layer**: API → Use Case → Repository flows - -### End-to-End Tests -Test complete workflows from start to finish: +### API/Workflow Tests +Full HTTP request/response cycles against a real FastAPI `TestClient`, backed +by a temporary SQLite database and mocked authentication (no live external +services required): - Full job creation and execution - Catalog parsing through role generation - Repository creation and package sync -- Image building and registry push +- Image building request handling ### Performance Tests Test system performance and scalability: @@ -188,22 +151,22 @@ Test system performance and scalability: ### Jobs Workflow Tests ```bash # All jobs tests -pytest tests/integration/api/jobs/ tests/unit/orchestrator/jobs/ -v +pytest tests/unit/api/jobs/ tests/unit/orchestrator/jobs/ -v # Job creation and idempotency -pytest tests/integration/api/jobs/test_create_job_api.py -v +pytest tests/unit/api/jobs/test_create_job_api.py -v # Job lifecycle management -pytest tests/integration/api/jobs/test_get_job_api.py -v +pytest tests/unit/api/jobs/test_get_job_api.py -v ``` ### Catalog Workflow Tests ```bash # All catalog tests -pytest tests/integration/api/catalog_roles/ tests/unit/core/catalog/ -v +pytest tests/unit/api/catalog_roles/ tests/unit/core/catalog/ -v # Catalog parsing -pytest tests/integration/api/parse_catalog/ -v +pytest tests/unit/api/parse_catalog/ -v # Role generation pytest tests/unit/orchestrator/catalog/ -v @@ -212,25 +175,25 @@ pytest tests/unit/orchestrator/catalog/ -v ### Local Repository Workflow Tests ```bash # All local repo tests -pytest tests/integration/api/local_repo/ tests/unit/core/localrepo/ -v +pytest tests/unit/api/local_repo/ tests/unit/core/localrepo/ -v # Repository creation -pytest tests/integration/api/local_repo/test_create_local_repo.py -v +pytest tests/unit/api/local_repo/test_create_local_repo_api.py -v ``` ### Image Building Workflow Tests ```bash # All build image tests -pytest tests/integration/api/build_image/ tests/unit/core/build_image/ -v +pytest tests/unit/api/build_image/ -v # Multi-architecture builds -pytest tests/integration/api/build_image/ -k multi_arch -v +pytest tests/unit/api/build_image/ -k multi_arch -v ``` ### Validation Workflow Tests ```bash # All validation tests -pytest tests/integration/api/validate/ tests/unit/core/validate/ -v +pytest tests/unit/api/validate/ tests/unit/core/validate/ -v # Schema validation pytest tests/unit/core/validate/ -k schema -v @@ -238,12 +201,26 @@ pytest tests/unit/core/validate/ -k schema -v ## Test Fixtures +### Catalog Fixtures + +Tests use the **real examples catalog** shipped with the repo: + +``` +../examples/catalog/catalog_rhel.json # Primary catalog fixture (RHEL 10.0) +core/catalog/test_fixtures/ # Remaining domain-specific fixtures: + ├── adapter_policy_test.json # Policy config for adapter_policy tests + └── functional_layer.json # Functional layer for generator tests +``` + +The stale `core/catalog/test_fixtures/catalog_rhel.json` was removed; all tests +now reference `examples/catalog/catalog_rhel.json` and skip gracefully if the +file is not present (e.g. in a shallow CI clone). + ### Shared Fixtures (conftest.py) **Authentication & Authorization:** - `client`: FastAPI TestClient with dev container - `auth_headers`: Standard authentication headers -- `admin_auth_headers`: Admin-level authentication **Idempotency & Correlation:** - `unique_idempotency_key`: Unique key per test @@ -251,7 +228,6 @@ pytest tests/unit/core/validate/ -k schema -v **Database & Storage:** - `db_session`: Database session for tests -- `clean_db`: Fresh database for each test - `artifact_store`: Test artifact storage **Mock Services:** @@ -273,6 +249,19 @@ def test_create_job(client, auth_headers, unique_idempotency_key): assert "job_id" in response.json() ``` +## Known Skips & Pre-existing Issues + +- `tests/unit/infra/db/test_sql_repositories.py` — skips/errors when no + PostgreSQL dialect is available (no live DB in local dev) +- `test_adapter_cli_defaults::test_generate_omnia_json_with_defaults_writes_output` + — skipped: legacy `generate_all_configs` adapter path is unused in production; + the current production path is `adapter_policy.generate_configs_from_policy` +- `tests/others/test_dependency_rules.py` — 2 pre-existing architectural + violation failures in `api/jobs/routes.py` (not test bugs) +- `tests/performance/test_local_repo_performance.py` — 3 pre-existing failures; + tests expect `202` from `POST /local_repo` but get `412` because the job + has no completed upstream stages + ## Coverage Report Generate HTML coverage report: @@ -516,17 +505,19 @@ def test_job_creation_with_valid_data(): assert job.status == "pending" ``` -### Adding a New Integration Test +### Adding a New API/Workflow Test -1. Create test file in appropriate `tests/integration/` subdirectory -2. Use shared fixtures from conftest.py +1. Create test file in the appropriate `tests/unit/api//` subdirectory +2. Add a `conftest.py` in that subdirectory if it needs its own `client` + fixture (mocked auth + temp SQLite DB) — see existing subdirectories + (e.g. `tests/unit/api/jobs/conftest.py`) for the pattern 3. Test full request/response cycles 4. Verify database state changes -5. Clean up test data +5. Clean up test data (handled automatically via `tmp_path` fixture) **Example:** ```python -# tests/integration/api/jobs/test_create_job_integration.py +# tests/unit/api/jobs/test_create_job_integration.py def test_create_job_integration(client, auth_headers, unique_idempotency_key): """Test complete job creation flow.""" payload = { diff --git a/build_stream/tests/conftest.py b/build_stream/tests/conftest.py index 31bddd100b..986bb417ed 100644 --- a/build_stream/tests/conftest.py +++ b/build_stream/tests/conftest.py @@ -12,19 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Shared pytest fixtures for Build Stream API tests. +"""Shared pytest fixtures for Build Stream API tests.""" -Note: This conftest is for mock-based unit/integration tests. -E2E integration tests use tests/integration/conftest.py which does not -import the app directly (it runs the server as a subprocess). -""" - -# pylint: disable=redefined-outer-name,global-statement,import-outside-toplevel,protected-access +# pylint: disable=redefined-outer-name,global-statement,import-outside-toplevel,protected-access,wrong-import-position,import-error import base64 import os import sys -from pathlib import Path from typing import Dict, Generator import pytest @@ -33,7 +27,7 @@ os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:") # Patch JSONB to JSON for SQLite compatibility (must be before any model imports) -from sqlalchemy import JSON as _sa_JSON +from sqlalchemy import JSON as _sa_JSON # noqa: E402 pylint: disable=wrong-import-position if 'sqlalchemy.dialects.postgresql' not in sys.modules: _postgresql_module = type(sys)('postgresql') @@ -43,8 +37,8 @@ # Patch infra.db.session engine creation for SQLite compatibility # SQLite does not support pool_size/max_overflow parameters -import infra.db.session as _db_session_mod -from sqlalchemy import create_engine as _sa_create_engine, event as _sa_event +import infra.db.session as _db_session_mod # noqa: E402 pylint: disable=wrong-import-position,ungrouped-imports +from sqlalchemy import create_engine as _sa_create_engine, event as _sa_event # noqa: E402 pylint: disable=wrong-import-position _sqlite_engine = _sa_create_engine("sqlite:///:memory:", echo=False) @@ -59,25 +53,30 @@ def _set_sqlite_pragma(dbapi_connection, connection_record): # pylint: disable= # Patch JWT exceptions for compatibility with newer PyJWT versions # This must be done before any imports of jwt.exceptions -import jwt.exceptions -if not hasattr(jwt.exceptions, 'DecodeError'): - jwt.exceptions.DecodeError = jwt.exceptions.JWTDecodeError -if not hasattr(jwt.exceptions, 'ExpiredSignatureError'): - class ExpiredSignatureError(jwt.exceptions.JWTDecodeError): - """Alias for expired signature errors.""" - jwt.exceptions.ExpiredSignatureError = ExpiredSignatureError -if not hasattr(jwt.exceptions, 'InvalidAudienceError'): - class InvalidAudienceError(jwt.exceptions.JWTDecodeError): - """Alias for invalid audience errors.""" - jwt.exceptions.InvalidAudienceError = InvalidAudienceError -if not hasattr(jwt.exceptions, 'InvalidIssuerError'): - class InvalidIssuerError(jwt.exceptions.JWTDecodeError): - """Alias for invalid issuer errors.""" - jwt.exceptions.InvalidIssuerError = InvalidIssuerError -if not hasattr(jwt.exceptions, 'InvalidSignatureError'): - class InvalidSignatureError(jwt.exceptions.JWTDecodeError): - """Alias for invalid signature errors.""" - jwt.exceptions.InvalidSignatureError = InvalidSignatureError +try: + import jwt.exceptions # noqa: E402 pylint: disable=wrong-import-position +except ImportError: + # jwt module not available, skip patching + pass +else: + if not hasattr(jwt.exceptions, 'DecodeError'): + jwt.exceptions.DecodeError = jwt.exceptions.JWTDecodeError + if not hasattr(jwt.exceptions, 'ExpiredSignatureError'): + class ExpiredSignatureError(jwt.exceptions.JWTDecodeError): # pylint: disable=too-few-public-methods + """Alias for expired signature errors.""" + jwt.exceptions.ExpiredSignatureError = ExpiredSignatureError + if not hasattr(jwt.exceptions, 'InvalidAudienceError'): + class InvalidAudienceError(jwt.exceptions.JWTDecodeError): # pylint: disable=too-few-public-methods + """Alias for invalid audience errors.""" + jwt.exceptions.InvalidAudienceError = InvalidAudienceError + if not hasattr(jwt.exceptions, 'InvalidIssuerError'): + class InvalidIssuerError(jwt.exceptions.JWTDecodeError): # pylint: disable=too-few-public-methods + """Alias for invalid issuer errors.""" + jwt.exceptions.InvalidIssuerError = InvalidIssuerError + if not hasattr(jwt.exceptions, 'InvalidSignatureError'): + class InvalidSignatureError(jwt.exceptions.JWTDecodeError): # pylint: disable=too-few-public-methods + """Alias for invalid signature errors.""" + jwt.exceptions.InvalidSignatureError = InvalidSignatureError # Note: pythonpath is set in pytest.ini at project root @@ -342,7 +341,7 @@ def generate_invalid_client_id() -> str: def generate_invalid_client_secret() -> str: """Generate an invalid client secret for testing. - + Returns: Invalid client secret string (too short). """ 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 "