From f2c4833c7d613fbba4499a54cae7ba528df9ff13 Mon Sep 17 00:00:00 2001 From: Ryan Haasken Date: Mon, 22 Apr 2024 12:26:09 -0500 Subject: [PATCH] CRAYSAT-1841: Add support for CFS v2 or v3 to `sat bootprep` Add support for using either the CFS v2 or v3 API in `sat bootprep` based on the value of the command-line option `--cfs-version` or the config-file option `cfs.api_version`. Before this change, `sat bootprep` was hard-coded to always use CFS v2. Update the `InputConfiguration`, `InputConfigurationLayer`, and `AdditionalInventory` classes to keep a reference to the instance of the version-specific subclass of `CFSClientBase`, so that they can delegate operations to the appropriate version of the CFS API. This removes a lot of code that required knowledge of the CFS API and instead uses classes and methods defined in the `csm_api_client` library, which has already been updated to support both CFS v2 and v3. This should reduce code duplication across `sat` and `csm_api_client`. Some unit tests are no longer needed here because that functionality is tested in the `csm_api_client` instead. Other unit tests are updated to include the `cfs_client` instance that needs to be passed into the input layer and configuration objects. This also makes the `playbook` property required in the schema because the CFS v3 API requires that the playbook be specified. This is technically a backwards-incompatible schema change because it can invalidate certain bootprep input files if they define layers without playbooks. The default bootprep files contain a playbook for every layer and do not depend on the CFS default playbook value, so this should be a pretty safe and non-disruptive change. Test Description: Unit tests only so far. Still needs to be tested on a real system. --- CHANGELOG.md | 12 + sat/cli/bootprep/input/configuration.py | 250 ++++------ sat/cli/bootprep/main.py | 2 +- sat/data/schema/bootprep_schema.yaml | 8 +- .../cli/bootprep/input/test_configuration.py | 461 +++++++++--------- tests/cli/bootprep/test_validate.py | 23 +- 6 files changed, 363 insertions(+), 393 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0760af28..e2093f92 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [3.33.0] - 2024-11-15 + +### Added +- Added support for using either the CFS v2 or v3 API in `sat bootprep`, + depending on the value of the `--cfs-version` command-line option or the + `cfs.api_version` config-file option. + +### Changed +- Modified the `sat bootprep` input file schema to require the `playbook` field + be specified for each layer of a CFS configuration. The CFS v3 API requires a + playbook for each layer. + ## [3.32.10] - 2024-11-04 ### Fixed diff --git a/sat/cli/bootprep/input/configuration.py b/sat/cli/bootprep/input/configuration.py index f3387217..4ae0c6e4 100644 --- a/sat/cli/bootprep/input/configuration.py +++ b/sat/cli/bootprep/input/configuration.py @@ -26,11 +26,9 @@ """ from abc import ABC, abstractmethod import logging -from urllib.parse import urlparse, urlunparse -from cray_product_catalog.query import ProductCatalogError from csm_api_client.service.gateway import APIError -from csm_api_client.service.vcs import VCSError, VCSRepo +from csm_api_client.service.cfs import CFSConfigurationError from sat.cached_property import cached_property from sat.cli.bootprep.constants import LATEST_VERSION_VALUE @@ -41,24 +39,13 @@ jinja_rendered, ) from sat.config import get_config_value -from sat.util import get_val_by_path, set_val_by_path +from sat.util import get_val_by_path LOGGER = logging.getLogger(__name__) class InputConfigurationLayerBase(ABC): """An object representing data in common between layers and additional inventory""" - REQUIRED_CFS_PROPERTIES = { - 'cloneUrl': 'clone_url' - } - # Mapping from CFS property name to class property name for optional properties - # If the property value is None, the property will be omitted from the CFS layer - # Nested CFS properties are specified by separating each property name with '.' - OPTIONAL_CFS_PROPERTIES = { - 'branch': 'branch', - 'commit': 'commit', - 'name': 'name' - } # CRAYSAT-1174: Specifying a 'branch' in a CFS configuration layer is not # supported until CSM 1.2. Toggling this variable will change the behavior @@ -68,16 +55,19 @@ class InputConfigurationLayerBase(ABC): # The jinja_rendered properties here are only rendered at item creation time template_render_err = InputItemCreateError - def __init__(self, layer_data, jinja_env): + def __init__(self, layer_data, jinja_env, cfs_client): """Create a new configuration layer. Args: layer_data (dict): the layer data from the input instance jinja_env (jinja2.Environment): the Jinja2 environment in which fields supporting Jinja2 templating should be rendered. + cfs_client (csm_api_client.service.cfs.CFSClientBase): the CFS API + client to use when creating the layer in CFS """ self.layer_data = layer_data self.jinja_env = jinja_env + self.cfs_client = cfs_client @property @jinja_rendered @@ -85,12 +75,6 @@ def name(self): """str or None: the name specified for the layer""" return self.layer_data.get('name') - @property - @abstractmethod - def clone_url(self): - """str: the git clone URL for this layer""" - pass - @property @abstractmethod def branch(self): @@ -103,6 +87,7 @@ def commit(self): """str or None: the commit for this layer""" pass + @abstractmethod def get_cfs_api_data(self): """Get the data to pass to the CFS API to create this layer. @@ -114,60 +99,18 @@ def get_cfs_api_data(self): InputItemCreateError: if there was a failure to obtain the data needed to create the layer in CFS. """ - cfs_layer_data = {cfs_property: getattr(self, self_property) - for cfs_property, self_property in self.REQUIRED_CFS_PROPERTIES.items()} - for cfs_property, self_property in self.OPTIONAL_CFS_PROPERTIES.items(): - # CRAYSAT-1174: Ignore branch property if not supported by CFS - if self.resolve_branches and cfs_property == 'branch': - continue - - property_value = getattr(self, self_property) - if property_value is not None: - set_val_by_path(cfs_layer_data, cfs_property, property_value) - - return cfs_layer_data - - def resolve_commit_hash(self, branch): - """Query VCS to determine the commit hash at the head of the branch. - - Args: - branch (str): the name of the branch to look up - - Returns: - str: the commit hash corresponding to the HEAD commit of the branch. - - Raises: - InputItemCreateError: if there is no such branch on the remote - repository. - """ - try: - commit_hash = VCSRepo(self.clone_url).get_commit_hash_for_branch(branch) - except VCSError as err: - raise InputItemCreateError(f'Could not query VCS to resolve branch name "{branch}": ' - f'{err}') - - if commit_hash is None: - raise InputItemCreateError(f'Could not retrieve HEAD commit for branch "{branch}"; ' - 'no matching branch was found on remote VCS repo.') - return commit_hash + pass class InputConfigurationLayer(InputConfigurationLayerBase, ABC): """A CFS configuration layer as defined in the bootprep input file""" - OPTIONAL_CFS_PROPERTIES = { - 'branch': 'branch', - 'commit': 'commit', - 'name': 'name', - 'playbook': 'playbook', - 'specialParameters.imsRequireDkms': 'ims_require_dkms' - } - @property @jinja_rendered def playbook(self): - """str or None: the playbook specified in the layer""" - return self.layer_data.get('playbook') + """str: the playbook specified in the layer""" + # playbook is now required by the schema + return self.layer_data['playbook'] @property def ims_require_dkms(self): @@ -175,7 +118,7 @@ def ims_require_dkms(self): return get_val_by_path(self.layer_data, 'special_parameters.ims_require_dkms') @staticmethod - def get_configuration_layer(layer_data, jinja_env, product_catalog): + def get_configuration_layer(layer_data, jinja_env, cfs_client, product_catalog): """Get and return a new InputConfigurationLayer for the given layer data. Args: @@ -183,6 +126,8 @@ def get_configuration_layer(layer_data, jinja_env, product_catalog): the bootprep input file schema. jinja_env (jinja2.Environment): the Jinja2 environment in which fields supporting Jinja2 templating should be rendered. + cfs_client (csm_api_client.service.cfs.CFSClientBase): the CFS API + client to use when creating the layer in CFS product_catalog (cray_product_catalog.query.ProductCatalog): the product catalog object @@ -192,9 +137,9 @@ def get_configuration_layer(layer_data, jinja_env, product_catalog): properly validated against the schema. """ if 'git' in layer_data: - return GitInputConfigurationLayer(layer_data, jinja_env) + return GitInputConfigurationLayer(layer_data, jinja_env, cfs_client) elif 'product' in layer_data: - return ProductInputConfigurationLayer(layer_data, jinja_env, product_catalog) + return ProductInputConfigurationLayer(layer_data, jinja_env, cfs_client, product_catalog) else: raise ValueError('Unrecognized type of configuration layer') @@ -218,28 +163,55 @@ def branch(self): @property def commit(self): # The 'commit' property is optional - if self.resolve_branches and self.branch is not None: - # If given a branch, we can look up the commit dynamically. - return self.resolve_commit_hash(self.branch) return self.layer_data['git'].get('commit') + def get_cfs_api_data(self): + """Get the data to pass to the CFS API to create this layer. + + Returns: + dict: The dictionary of data to pass to the CFS API to create the + layer. + + Raises: + InputItemCreateError: if there was a failure to obtain the data + needed to create the layer in CFS. + """ + layer_cls = self.cfs_client.configuration_cls.cfs_config_layer_cls + try: + layer = layer_cls.from_clone_url( + clone_url=self.clone_url, branch=self.branch, commit=self.commit, + name=self.name, playbook=self.playbook, ims_require_dkms=self.ims_require_dkms + ) + except ValueError as err: + raise InputItemCreateError(str(err)) + + if self.resolve_branches: + try: + layer.resolve_branch_to_commit_hash() + except CFSConfigurationError as err: + raise InputItemCreateError(str(err)) + + return layer.req_payload + class ProductInputConfigurationLayer(InputConfigurationLayer): """ A configuration layer that is defined with the name of a product and the version or branch. """ - def __init__(self, layer_data, jinja_env, product_catalog): + def __init__(self, layer_data, jinja_env, cfs_client, product_catalog): """Create a new ProductInputConfigurationLayer. Args: layer_data (dict): the layer data from the input instance jinja_env (jinja2.Environment): the Jinja2 environment in which fields supporting Jinja2 templating should be rendered. + cfs_client (csm_api_client.service.cfs.CFSClientBase): the CFS API + client to use when creating the layer in CFS product_catalog (cray_product_catalog.query.ProductCatalog or None): the product catalog object """ - super().__init__(layer_data, jinja_env) + super().__init__(layer_data, jinja_env, cfs_client) self.product_catalog = product_catalog @property @@ -252,42 +224,8 @@ def product_name(self): @jinja_rendered def product_version(self): """str: the version specified for the product""" - # The 'version' property is optional. If not specified, assume latest - return self.layer_data['product'].get('version', LATEST_VERSION_VALUE) - - @cached_property - def matching_product(self): - """sat.software_inventory.products.InstalledProductVersion: the matching installed product""" - if self.product_catalog is None: - raise InputItemCreateError(f'Product catalog data is not available.') - - try: - if self.product_version == LATEST_VERSION_VALUE: - return self.product_catalog.get_product(self.product_name) - return self.product_catalog.get_product(self.product_name, self.product_version) - except ProductCatalogError as err: - raise InputItemCreateError(f'Unable to get product data from product catalog: {err}') - - @staticmethod - def substitute_url_hostname(url): - """Substitute the hostname in a URL with the configured API gateway hostname. - - Args: - url (str): The URL to substitute. - - Returns: - str: The URL with the hostname portion replaced. - """ - return urlunparse(urlparse(url)._replace( - netloc=get_config_value('api_gateway.host') - )) - - @cached_property - def clone_url(self): - if self.matching_product.clone_url is None: - raise InputItemCreateError(f'No clone URL present for version {self.product_version} ' - f'of product {self.product_name}') - return self.substitute_url_hostname(self.matching_product.clone_url) + # The 'version' property is optional + return self.layer_data['product'].get('version') @cached_property @jinja_rendered @@ -297,33 +235,35 @@ def branch(self): @cached_property def commit(self): - # There are a few ways to determine the proper commit hash for a - # layer, if necessary. Their precedence is as follows. - # 1. If the commit for the product was specified explicitly in the - # input file, that should be returned. - # 2. If a branch for the product was specified in the input file, the - # branch needs to be resolved to a commit hash, and that commit hash - # should be returned. If branch resolving is disabled (i.e. with - # --no-resolve-branches), then presumably CFS supports branch names, - # and so this property should be None in order for a branch name to - # be passed to CFS. - # 3. If neither a commit nor a branch was specified, then consult the - # product catalog for the an associated commit hash and return - # that. - - input_commit = self.layer_data['product'].get('commit') - if input_commit: - return input_commit - - if self.branch is not None: + # The 'commit' property is optional + return self.layer_data['product'].get('commit') + + def get_cfs_api_data(self): + """Get the data to pass to the CFS API to create this layer. + + Returns: + dict: The dictionary of data to pass to the CFS API to create the + layer. + + Raises: + InputItemCreateError: if there was a failure to obtain the data + needed to create the layer in CFS. + """ + layer_cls = self.cfs_client.configuration_cls.cfs_config_layer_cls + + try: + layer = layer_cls.from_product_catalog( + product_name=self.product_name, api_gw_host=get_config_value('api_gateway.host'), + product_version=self.product_version, commit=self.commit, branch=self.branch, + name=self.name, playbook=self.playbook, ims_require_dkms=self.ims_require_dkms, + product_catalog=self.product_catalog + ) if self.resolve_branches: - return self.resolve_commit_hash(self.branch) - return None + layer.resolve_branch_to_commit_hash() + except (ValueError, CFSConfigurationError) as err: + raise InputItemCreateError(str(err)) - if self.matching_product.commit is None: - raise InputItemCreateError(f'No commit present for version {self.product_version} ' - f'of product {self.product_name}') - return self.matching_product.commit + return layer.req_payload class AdditionalInventory(InputConfigurationLayerBase): @@ -337,9 +277,6 @@ def clone_url(self): @property def commit(self): """str or None: the commit hash or None if branch was specified instead""" - if self.resolve_branches and self.branch is not None: - # If given a branch, we can look up the commit dynamically. - return self.resolve_commit_hash(self.branch) return self.layer_data.get('commit') @property @@ -352,6 +289,32 @@ def name(self): """str or None: the optional name of the additional inventory""" return self.layer_data.get('name') + def get_cfs_api_data(self): + """Get the data to pass to the CFS API to create this layer. + + Returns: + dict: The dictionary of data to pass to the CFS API to create the + layer. + + Raises: + InputItemCreateError: if there was a failure to obtain the data + needed to create the layer in CFS. + """ + layer_cls = self.cfs_client.configuration_cls.cfs_additional_inventory_cls + + try: + layer = layer_cls.from_clone_url(clone_url=self.clone_url, name=self.name, + branch=self.branch, commit=self.commit) + except ValueError as err: + raise InputItemCreateError(str(err)) + try: + if self.resolve_branches: + layer.resolve_branch_to_commit_hash() + except CFSConfigurationError as err: + raise InputItemCreateError(str(err)) + + return layer.req_payload + class InputConfiguration(BaseInputItem): """A CFS Configuration from a bootprep input file.""" @@ -384,12 +347,14 @@ def __init__(self, data, instance, index, jinja_env, cfs_client, self.jinja_context = {} # The 'layers' property is required and must be a list, but it can be empty - self.layers = [InputConfigurationLayer.get_configuration_layer(layer_data, jinja_env, product_catalog) + self.layers = [InputConfigurationLayer.get_configuration_layer(layer_data, jinja_env, + cfs_client, product_catalog) for layer_data in self.data['layers']] self.additional_inventory = None if 'additional_inventory' in self.data: - self.additional_inventory = AdditionalInventory(self.data['additional_inventory'], jinja_env) + self.additional_inventory = AdditionalInventory(self.data['additional_inventory'], jinja_env, + cfs_client) def get_create_item_data(self): """Get the data to pass to the CFS API to create this configuration. @@ -442,6 +407,7 @@ def create(self, payload): payload (dict): the payload to pass to the CFS API to create the CFS configuration """ + # TODO: Consider using a method of the CFSConfiguration class from csm_api_client try: # request_body guaranteed to have 'name' key, so no need to catch ValueError self.cfs_client.put_configuration(self.name, payload) diff --git a/sat/cli/bootprep/main.py b/sat/cli/bootprep/main.py index e9869be9..ac7b54c9 100644 --- a/sat/cli/bootprep/main.py +++ b/sat/cli/bootprep/main.py @@ -210,7 +210,7 @@ def do_bootprep_run(schema_validator, args): LOGGER.info('Input file successfully validated against schema') session = SATSession() - cfs_client = CFSClientBase.get_cfs_client(session, 'v2') + cfs_client = CFSClientBase.get_cfs_client(session, get_config_value('cfs.api_version')) ims_client = IMSClient(session) # CASMTRIAGE-4288: IMS can be extremely slow to return DELETE requests for # large images, so this IMSClient will not use a timeout on HTTP requests diff --git a/sat/data/schema/bootprep_schema.yaml b/sat/data/schema/bootprep_schema.yaml index 5cc46057..84f46030 100644 --- a/sat/data/schema/bootprep_schema.yaml +++ b/sat/data/schema/bootprep_schema.yaml @@ -1,7 +1,7 @@ # # MIT License # -# (C) Copyright 2021-2023 Hewlett Packard Enterprise Development LP +# (C) Copyright 2021-2024 Hewlett Packard Enterprise Development LP # # Permission is hereby granted, free of charge, to any person obtaining a # copy of this software and associated documentation files (the "Software"), @@ -41,7 +41,7 @@ $schema: "https://json-schema.org/draft/2020-12/schema" # ... patch component when all input files that were valid under the old schema # are still valid under the new schema # -version: '1.0.7' +version: '1.1.0' title: Bootprep Input File description: > A description of the set of CFS configurations to create, the set of IMS @@ -113,7 +113,7 @@ properties: A layer of the CFS configuration defined using an explicit git repository URL and commit hash or branch name. type: object - required: ['git'] + required: ['git', 'playbook'] additionalProperties: false properties: name: @@ -150,7 +150,7 @@ properties: A layer of the CFS configuration defined using a product's configuration management repository. type: object - required: ['product'] + required: ['product', 'playbook'] additionalProperties: false properties: name: diff --git a/tests/cli/bootprep/input/test_configuration.py b/tests/cli/bootprep/input/test_configuration.py index e127a228..b5b343d5 100644 --- a/tests/cli/bootprep/input/test_configuration.py +++ b/tests/cli/bootprep/input/test_configuration.py @@ -28,13 +28,11 @@ import logging import unittest from unittest.mock import Mock, patch -from urllib.parse import urlparse -from cray_product_catalog.query import ProductCatalogError from csm_api_client.service.vcs import VCSError from jinja2.sandbox import SandboxedEnvironment -from csm_api_client.service.cfs import CFSClientBase +from csm_api_client.service.cfs import CFSClientBase, CFSConfigurationError from sat.cli.bootprep.errors import InputItemCreateError from sat.cli.bootprep.input.configuration import ( AdditionalInventory, @@ -42,7 +40,6 @@ GitInputConfigurationLayer, ProductInputConfigurationLayer, InputConfiguration, - LATEST_VERSION_VALUE ) from sat.cli.bootprep.input.instance import InputInstance @@ -59,6 +56,7 @@ def setUp(self): """Mock the constructors for the child classes""" self.mock_git_layer = patch_configuration('GitInputConfigurationLayer').start() self.mock_product_layer = patch_configuration('ProductInputConfigurationLayer').start() + self.mock_cfs_client = Mock() self.mock_product_catalog = Mock() self.mock_jinja_env = Mock() @@ -70,15 +68,21 @@ def test_get_configuration_layer_git(self): # Just needs a 'git' key; we're mocking the GitInputConfigurationLayer class layer_data = {'git': {}} layer = InputConfigurationLayer.get_configuration_layer(layer_data, self.mock_jinja_env, + self.mock_cfs_client, self.mock_product_catalog) self.assertEqual(self.mock_git_layer.return_value, layer) + self.mock_git_layer.assert_called_once_with(layer_data, self.mock_jinja_env, + self.mock_cfs_client) def test_get_configuration_layer_product(self): """Test the get_configuration_layer static method with a product layer""" layer_data = {'product': {}} layer = InputConfigurationLayer.get_configuration_layer(layer_data, self.mock_jinja_env, + self.mock_cfs_client, self.mock_product_catalog) self.assertEqual(self.mock_product_layer.return_value, layer) + self.mock_product_layer.assert_called_once_with(layer_data, self.mock_jinja_env, + self.mock_cfs_client, self.mock_product_catalog) def test_get_configuration_layer_unknown(self): """Test the get_configuration_layer static method with bad layer data""" @@ -89,6 +93,7 @@ def test_get_configuration_layer_unknown(self): expected_err = 'Unrecognized type of configuration layer' with self.assertRaisesRegex(ValueError, expected_err): InputConfigurationLayer.get_configuration_layer(layer_data, self.mock_jinja_env, + self.mock_cfs_client, self.mock_product_catalog) @@ -97,9 +102,14 @@ class TestInputConfigurationLayerBase(unittest.TestCase): module_path = 'sat.cli.bootprep.input.configuration' def setUp(self): - """Patch the resolve_branches class attribute on InputConfigurationLayer""" + """Mock needed functionality for both types of InputConfigurationLayer""" self.patch_resolve_branches(False).start() + self.mock_cfs_client = Mock() + self.mock_cfs_configuration_cls = self.mock_cfs_client.configuration_cls + self.mock_cfs_layer_cls = self.mock_cfs_configuration_cls.cfs_config_layer_cls + self.mock_add_inv_cls = self.mock_cfs_configuration_cls.cfs_additional_inventory_cls + def tearDown(self): patch.stopall() @@ -133,8 +143,6 @@ def setUp(self): } self.branch_head_commit = 'e6bfdb28d44669c4317d6dc021c22a75cebb3bfb' - self.mock_vcs_repo = patch(f'{self.module_path}.VCSRepo').start() - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.return_value = self.branch_head_commit self.mock_sat_version = '2.3.6' self.mock_network_type = 'cassini' @@ -148,124 +156,125 @@ def tearDown(self): patch.stopall() def test_playbook_property_present(self): - """Test the playbook property when a playbook is in the layer data""" - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + """Test the playbook property""" + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.playbook, layer.playbook) - def test_playbook_property_not_present(self): - """Test the playbook property when a playbook is not in the layer data""" - del self.branch_layer_data['playbook'] - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) - self.assertIsNone(layer.playbook) - def test_playbook_property_jinja_template(self): """Test the playbook property when the playbook uses Jinja2 templating""" self.branch_layer_data['playbook'] = 'shs_{{shs.network_type}}_install.yaml' - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(f'shs_{self.mock_network_type}_install.yaml', layer.playbook) def test_name_property_present(self): """Test the name property when the name is in the layer data""" - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.branch_layer_data['name'], layer.name) def test_name_property_not_present(self): """Test the name property when the name is not in the layer data""" del self.branch_layer_data['name'] - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertIsNone(layer.name) def test_name_property_jinja_template(self): """Test the name property when the name uses Jinja2 templating""" self.branch_layer_data['name'] = 'sat-ncn-{{sat.version}}' - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(f'sat-ncn-{self.mock_sat_version}', layer.name) def test_clone_url_property(self): """Test the clone_url property.""" - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.branch_layer_data['git']['url'], layer.clone_url) def test_branch_property_present(self): """Test the branch property when the branch is in the layer data""" - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.branch_layer_data['git']['branch'], layer.branch) def test_branch_property_not_present(self): """Test the branch property when the branch is not in the layer data""" - layer = GitInputConfigurationLayer(self.commit_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.commit_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertIsNone(layer.branch) def test_branch_property_jinja_template(self): """Test the branch property when the branch uses Jinja2 templating""" self.branch_layer_data['git']['branch'] = 'integration-{{sat.version}}' - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(f'integration-{self.mock_sat_version}', layer.branch) def test_commit_property_present(self): """Test the commit property when the commit is in the layer data""" - layer = GitInputConfigurationLayer(self.commit_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.commit_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.commit_layer_data['git']['commit'], layer.commit) def test_commit_property_not_present(self): """Test the commit property when the commit is not in the layer data""" - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) self.assertIsNone(layer.commit) - def test_get_cfs_api_data_optional_properties(self): - """Test get_create_item_data method with optional name and playbook properties present.""" - branch_layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) - commit_layer = GitInputConfigurationLayer(self.commit_layer_data, self.jinja_env) - subtests = (('branch', branch_layer), ('commit', commit_layer)) - for present_property, layer in subtests: - with self.subTest(present_property=present_property): - expected = deepcopy(getattr(self, f'{present_property}_layer_data')) - expected['cloneUrl'] = expected['git']['url'] - del expected['git']['url'] - # Move branch or commit to the top level - for key, value in expected['git'].items(): - expected[key] = value - del expected['git'] - self.assertEqual(expected, layer.get_cfs_api_data()) - - def test_get_cfs_api_data_special_parameters(self): - """Test get_create_item_data method with special_parameters present.""" - layer_data = deepcopy(self.branch_layer_data) - require_dkms = True - layer_data['special_parameters'] = {'ims_require_dkms': require_dkms} - layer = GitInputConfigurationLayer(layer_data, self.jinja_env) - - expected_api_data = deepcopy(layer_data) - # Move these values to where CFS expects them - expected_api_data['cloneUrl'] = expected_api_data['git']['url'] - expected_api_data['branch'] = expected_api_data['git']['branch'] - del expected_api_data['git'] - expected_api_data['specialParameters'] = {'imsRequireDkms': require_dkms} - del expected_api_data['special_parameters'] - - self.assertEqual(expected_api_data, layer.get_cfs_api_data()) - - def test_commit_property_branch_commit_lookup(self): - """Test looking up commit hash from branch in VCS when branch not supported in CSM""" - with self.patch_resolve_branches(True): - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) - self.assertEqual(layer.commit, self.branch_head_commit) + def test_get_cfs_api_data_no_resolve_branches(self): + """Test get_cfs_api_data method without branch resolution""" + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) - def test_commit_property_branch_commit_vcs_query_fails(self): - """Test looking up commit hash raises InputItemCreateError when VCS is inaccessible""" - with self.patch_resolve_branches(True): - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.side_effect = VCSError - with self.assertRaises(InputItemCreateError): - _ = layer.commit + cfs_api_data = layer.get_cfs_api_data() + + # All the work of getting CFS API request data is delegated to csm_api_client + self.mock_cfs_layer_cls.from_clone_url.assert_called_once_with( + clone_url=layer.clone_url, branch=layer.branch, commit=layer.commit, + name=layer.name, playbook=layer.playbook, ims_require_dkms=layer.ims_require_dkms + ) + cfs_layer_instance = self.mock_cfs_layer_cls.from_clone_url.return_value + self.assertEqual(cfs_layer_instance.req_payload, cfs_api_data) + cfs_layer_instance.resolve_branch_to_commit_hash.assert_not_called() + + def test_get_cfs_api_data_resolve_branches(self): + """Test get_cfs_api_data method with branch resolution""" + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) - def test_commit_property_branch_commit_lookup_fails(self): - """Test looking up commit hash for nonexistent branch when branch not supported in CSM""" with self.patch_resolve_branches(True): - layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env) - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.return_value = None - with self.assertRaises(InputItemCreateError): - _ = layer.commit + cfs_api_data = layer.get_cfs_api_data() + + # All the work of getting CFS API request data is delegated to csm_api_client + self.mock_cfs_layer_cls.from_clone_url.assert_called_once_with( + clone_url=layer.clone_url, branch=layer.branch, commit=layer.commit, + name=layer.name, playbook=layer.playbook, ims_require_dkms=layer.ims_require_dkms + ) + cfs_layer_instance = self.mock_cfs_layer_cls.from_clone_url.return_value + self.assertEqual(cfs_layer_instance.req_payload, cfs_api_data) + cfs_layer_instance.resolve_branch_to_commit_hash.assert_called_once_with() + + def test_get_cfs_api_data_value_error(self): + """Test get_cfs_api_data method when from_clone_url raises ValueError""" + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) + self.mock_cfs_layer_cls.from_clone_url.side_effect = ValueError('error') + with self.assertRaisesRegex(InputItemCreateError, 'error'): + _ = layer.get_cfs_api_data() + + def test_get_cfs_api_data_resolve_branch_error(self): + """Test get_cfs_api_data method when branch resolution fails""" + layer = GitInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client) + with self.patch_resolve_branches(True): + cfs_layer_instance = self.mock_cfs_layer_cls.from_clone_url.return_value + cfs_layer_instance.resolve_branch_to_commit_hash.side_effect = CFSConfigurationError('error') + with self.assertRaisesRegex(InputItemCreateError, 'error'): + _ = layer.get_cfs_api_data() class TestProductInputConfigurationLayer(TestInputConfigurationLayerBase): @@ -275,32 +284,12 @@ def setUp(self): """Mock K8s API to return fake product catalog data and set up layers""" super().setUp() - # Minimal set of product catalog data needed for these tests - self.old_url = 'https://vcs.local/vcs/cray/cos-config-management.git' - self.old_commit = '82537e59c24dd5607d5f5d6f92cdff971bd9c615' - self.new_url = 'https://vcs.local/vcs/cray/newcos-config-management.git' - self.new_commit = '6b0d9d55d399c92abae08002e75b9a1ce002f917' - self.product_name = 'cos' - self.product_version = '2.1.50' - - self.old_cos = Mock(clone_url=self.old_url, commit=self.old_commit) - self.new_cos = Mock(clone_url=self.new_url, commit=self.new_commit) - - def mock_get_product(product_name, product_version=None): - if product_name != self.product_name: - raise ProductCatalogError('Unknown product') - elif product_version == self.product_version: - return self.old_cos - elif not product_version: - return self.new_cos - else: - raise ProductCatalogError('Unknown version') - self.mock_product_catalog = Mock() - self.mock_product_catalog.get_product.side_effect = mock_get_product # Used to test variable substitution in Jinja2-templated fields self.jinja_env = SandboxedEnvironment() + self.product_name = 'cos' + self.product_version = '2.1.50' self.mock_network_type = 'cassini' self.jinja_env.globals = { self.product_name: {'version': self.product_version}, @@ -318,6 +307,7 @@ def mock_get_product(product_name, product_version=None): } self.version_layer = ProductInputConfigurationLayer(self.version_layer_data, self.jinja_env, + self.mock_cfs_client, self.mock_product_catalog) self.branch = 'integration' self.branch_layer_data = { @@ -330,6 +320,7 @@ def mock_get_product(product_name, product_version=None): } self.branch_layer = ProductInputConfigurationLayer(self.branch_layer_data, self.jinja_env, + self.mock_cfs_client, self.mock_product_catalog) self.commit = 'c07f317c4127d8667a4bd6c08d48e716b1d47da1' @@ -343,14 +334,12 @@ def mock_get_product(product_name, product_version=None): } self.commit_layer = ProductInputConfigurationLayer(self.commit_layer_data, self.jinja_env, + self.mock_cfs_client, self.mock_product_catalog) - self.branch_head_commit = 'e6bfdb28d44669c4317d6dc021c22a75cebb3bfb' - self.mock_vcs_repo = patch(f'{self.module_path}.VCSRepo').start() - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.return_value = self.branch_head_commit - + self.mock_api_gw = 'api_gateway.nmn' self.mock_sat_config = { - 'api_gateway.host': 'api_gateway.nmn' + 'api_gateway.host': self.mock_api_gw } patch_configuration('get_config_value', side_effect=self.mock_sat_config.get).start() @@ -361,7 +350,7 @@ def test_playbook_property_jinja_template(self): """Test the playbook property when the playbook uses Jinja2 templating""" self.branch_layer_data['playbook'] = 'shs_{{shs.network_type}}_install.yaml' layer = ProductInputConfigurationLayer(self.branch_layer_data, self.jinja_env, - self.mock_product_catalog) + self.mock_cfs_client, self.mock_product_catalog) self.assertEqual(f'shs_{self.mock_network_type}_install.yaml', layer.playbook) def test_product_name_property(self): @@ -374,7 +363,7 @@ def test_product_version_property_present(self): def test_product_version_property_not_present(self): """Test the product_version property when version is not in the layer data""" - self.assertEqual(LATEST_VERSION_VALUE, self.branch_layer.product_version) + self.assertIsNone(self.branch_layer.product_version) def test_product_version_jinja_template(self): """Test the product_version property when it uses Jinja2 templating""" @@ -382,60 +371,10 @@ def test_product_version_jinja_template(self): self.version_layer_data['product']['version'] = '{{' + f'{self.product_name}.version' + '}}' layer = ProductInputConfigurationLayer(self.version_layer_data, self.jinja_env, - self.mock_product_catalog) + self.mock_cfs_client, self.mock_product_catalog) self.assertEqual(self.product_version, layer.product_version) - def test_matching_product_explicit_version(self): - """Test getting the matching InstalledProductVersion for an explict version.""" - self.assertEqual(self.old_cos, self.version_layer.matching_product) - - def test_matching_product_no_version(self): - """Test getting the matching InstalledProductVersion for an assumed latest version.""" - self.assertEqual(self.new_cos, self.branch_layer.matching_product) - - def test_matching_product_latest_version(self): - """Test getting the matching InstalledProductVersion for an explict latest version.""" - latest_layer_data = deepcopy(self.branch_layer_data) - latest_layer_data['product']['version'] = LATEST_VERSION_VALUE - latest_layer = ProductInputConfigurationLayer(latest_layer_data, self.jinja_env, - self.mock_product_catalog) - self.assertEqual(self.new_cos, latest_layer.matching_product) - - def test_matching_product_no_product_catalog(self): - """Test getting the matching InstalledProductVersion when the product catalog is missing.""" - layer = ProductInputConfigurationLayer(self.version_layer_data, self.jinja_env, None) - err_regex = 'Product catalog data is not available' - with self.assertRaisesRegex(InputItemCreateError, err_regex): - _ = layer.matching_product - - def test_matching_product_software_inventory_error(self): - """Test getting the matching InstalledProductVersion when there is software inventory failure.""" - sw_inv_err_msg = 'unable find configmap' - self.mock_product_catalog.get_product.side_effect = ProductCatalogError(sw_inv_err_msg) - err_regex = f'Unable to get product data from product catalog: {sw_inv_err_msg}' - with self.assertRaisesRegex(InputItemCreateError, err_regex): - _ = self.version_layer.matching_product - - def test_clone_url_present(self): - """Test clone_url when present in product catalog data""" - old_url_parsed = urlparse(self.old_url) - new_url_parsed = urlparse(self.version_layer.clone_url) - # All parts of the URL should be unchanged, except for the 'netloc' which should - # be replaced with what is in the configuration. - for attr in ('scheme', 'path', 'params', 'query', 'fragment'): - self.assertEqual(getattr(old_url_parsed, attr), getattr(new_url_parsed, attr)) - self.assertEqual(new_url_parsed.netloc, self.mock_sat_config['api_gateway.host']) - - def test_clone_url_missing(self): - """Test clone_url when missing from product catalog data""" - self.old_cos.clone_url = None - err_regex = (f"No clone URL present for version {self.product_version} of " - f"product {self.product_name}") - - with self.assertRaisesRegex(InputItemCreateError, err_regex): - _ = self.version_layer.clone_url - def test_branch_property_present(self): """Test the branch property when branch is in the layer data""" self.assertEqual(self.branch, self.branch_layer.branch) @@ -450,46 +389,76 @@ def test_branch_property_jinja_template(self): self.branch_layer_data['product']['branch'] = 'integration-{{' + f'{self.product_name}.version' + '}}' layer = ProductInputConfigurationLayer(self.branch_layer_data, self.jinja_env, - self.mock_product_catalog) + self.mock_cfs_client, self.mock_product_catalog) self.assertEqual(f'integration-{self.product_version}', layer.branch) - def test_commit_property_branch_present(self): + def test_commit_property_present(self): + """Test the commit property when commit is in the layer data""" + self.assertEqual(self.commit, self.commit_layer.commit) + + def test_commit_property_not_present(self): """Test the commit property when a branch is in the layer data""" self.assertIsNone(self.branch_layer.commit) - def test_commit_property_branch_not_present(self): - """Test the commit property when a branch is not in the layer data""" - self.assertEqual(self.old_commit, self.version_layer.commit) - - def test_commit_property_branch_commit_lookup(self): - """Test looking up commit hash from branch in VCS when branch not supported in CSM""" - with self.patch_resolve_branches(True): - self.assertEqual(self.branch_layer.commit, self.branch_head_commit) - - def test_commit_property_branch_commit_vcs_query_fails(self): - """Test looking up commit hash raises InputItemCreateError when VCS is inaccessible""" - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.side_effect = VCSError - with self.patch_resolve_branches(True): - with self.assertRaises(InputItemCreateError): - _ = self.branch_layer.commit - - def test_commit_property_branch_commit_lookup_fails(self): - """Test looking up commit hash for nonexistent branch when branch not supported in CSM""" - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.return_value = None - with self.patch_resolve_branches(True): - with self.assertRaises(InputItemCreateError): - _ = self.branch_layer.commit - - def test_commit_property_when_commit_specified_in_input(self): - """Test the commit property when commit specified in the input file""" - self.assertEqual(self.commit_layer.commit, self.commit) - def test_commit_property_resolve_branches(self): """Test the commit property when resolving branches and commit specified in the input file""" with self.patch_resolve_branches(True): self.assertEqual(self.commit_layer.commit, self.commit) + def test_get_cfs_api_data_no_resolve_branches(self): + """Test get_cfs_api_data method without branch resolution""" + cfs_api_data = self.version_layer.get_cfs_api_data() + + # All the work of getting CFS API request data is delegated to csm_api_client + self.mock_cfs_layer_cls.from_product_catalog.assert_called_once_with( + product_name=self.product_name, api_gw_host=self.mock_api_gw, + product_version=self.product_version, commit=None, + branch=None, name=self.version_layer.name, + playbook=self.playbook, ims_require_dkms=None, + product_catalog=self.mock_product_catalog + ) + cfs_layer_instance = self.mock_cfs_layer_cls.from_product_catalog.return_value + self.assertEqual(cfs_layer_instance.req_payload, cfs_api_data) + cfs_layer_instance.resolve_branch_to_commit_hash.assert_not_called() + + def test_get_cfs_api_data_resolve_branches(self): + """Test get_cfs_api_data method with branch resolution""" + with self.patch_resolve_branches(True): + cfs_api_data = self.version_layer.get_cfs_api_data() + + # All the work of getting CFS API request data is delegated to csm_api_client + self.mock_cfs_layer_cls.from_product_catalog.assert_called_once_with( + product_name=self.product_name, api_gw_host=self.mock_api_gw, + product_version=self.product_version, commit=None, + branch=None, name=self.version_layer.name, + playbook=self.playbook, ims_require_dkms=None, + product_catalog=self.mock_product_catalog + ) + cfs_layer_instance = self.mock_cfs_layer_cls.from_product_catalog.return_value + self.assertEqual(cfs_layer_instance.req_payload, cfs_api_data) + cfs_layer_instance.resolve_branch_to_commit_hash.assert_called_once_with() + + def test_get_cfs_api_data_value_error(self): + """Test get_cfs_api_data method when from_product_catalog raises ValueError""" + self.mock_cfs_layer_cls.from_product_catalog.side_effect = ValueError('error') + with self.assertRaisesRegex(InputItemCreateError, 'error'): + _ = self.version_layer.get_cfs_api_data() + + def test_get_cfs_api_data_cfs_configuration_error(self): + """Test get_cfs_api_data method when from_product_catalog raises CFSConfigurationError""" + self.mock_cfs_layer_cls.from_product_catalog.side_effect = CFSConfigurationError('error') + with self.assertRaisesRegex(InputItemCreateError, 'error'): + _ = self.version_layer.get_cfs_api_data() + + def test_get_cfs_api_data_resolve_branch_error(self): + """Test get_cfs_api_data method when branch resolution raises CFSConfigurationError""" + with self.patch_resolve_branches(True): + cfs_layer_instance = self.mock_cfs_layer_cls.from_product_catalog.return_value + cfs_layer_instance.resolve_branch_to_commit_hash.side_effect = CFSConfigurationError('error') + with self.assertRaisesRegex(InputItemCreateError, 'error'): + _ = self.version_layer.get_cfs_api_data() + class TestAdditionalInventory(TestInputConfigurationLayerBase): """Tests for the AdditionalInventory class""" @@ -505,90 +474,97 @@ def setUp(self): self.data_with_branch = {'url': self.repo_url, 'branch': self.branch} self.data_with_name = {'url': self.repo_url, 'branch': self.branch, 'name': self.name} self.jinja_env = SandboxedEnvironment() - self.branch_head_commit = 'e64ef6c370166285e6a674724b74e912a3f4a21e' - self.mock_vcs_repo = patch(f'{self.module_path}.VCSRepo').start() - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.return_value = self.branch_head_commit def test_clone_url_property(self): """Test the clone_url property of AdditionalInventory""" - additional_inventory = AdditionalInventory(self.data_with_commit, self.jinja_env) + additional_inventory = AdditionalInventory(self.data_with_commit, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.repo_url, additional_inventory.clone_url) def test_commit_property_specified(self): """Test the commit property when specified""" - additional_inventory = AdditionalInventory(self.data_with_commit, self.jinja_env) + additional_inventory = AdditionalInventory(self.data_with_commit, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.commit_hash, additional_inventory.commit) def test_commit_property_unspecified(self): """Test the commit property when not specified""" - additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env) + additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env, + self.mock_cfs_client) self.assertIsNone(additional_inventory.commit) - def test_commit_property_branch_commit_lookup(self): - """Test looking up commit hash from branch in VCS when branch not supported in CSM""" - additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env) - with self.patch_resolve_branches(True): - self.assertEqual(additional_inventory.commit, self.branch_head_commit) - - def test_commit_property_no_resolve_branches(self): - """Test that commit is None when branch is specified with no_resolve_branches""" - additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env) - self.assertIsNone(additional_inventory.commit) - - def test_commit_property_vcs_query_fails(self): - """Test looking up commit hash raises InputItemCreateError when VCS is inaccessible""" - additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env) - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.side_effect = VCSError - with self.patch_resolve_branches(True): - with self.assertRaises(InputItemCreateError): - _ = additional_inventory.commit - - def test_commit_property_branch_lookup_fails(self): - """Test looking up commit hash for nonexistent branch""" - additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env) - self.mock_vcs_repo.return_value.get_commit_hash_for_branch.return_value = None - with self.patch_resolve_branches(True): - with self.assertRaises(InputItemCreateError): - _ = additional_inventory.commit - def test_branch_property_specified(self): """"Test the branch property when specified""" - additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env) + additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.branch, additional_inventory.branch) def test_branch_property_not_specified(self): """"Test the branch property when not specified""" - additional_inventory = AdditionalInventory(self.data_with_commit, self.jinja_env) + additional_inventory = AdditionalInventory(self.data_with_commit, self.jinja_env, + self.mock_cfs_client) self.assertIsNone(additional_inventory.branch) def test_name_property_specified(self): """Test the name property when specified""" - additional_inventory = AdditionalInventory(self.data_with_name, self.jinja_env) + additional_inventory = AdditionalInventory(self.data_with_name, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.name, additional_inventory.name) def test_name_property_not_specified(self): """Test the name property when not specified""" - additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env) + additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env, + self.mock_cfs_client) self.assertIsNone(additional_inventory.name) - def test_get_cfs_api_data_branch_no_resolve_branches(self): - """Test the get_cfs_api_data method with branch and name specified and no_resolve_branches""" - additional_inventory = AdditionalInventory(self.data_with_name, self.jinja_env) - expected = {'name': self.name, 'branch': self.branch, 'cloneUrl': self.repo_url} - self.assertEqual(expected, additional_inventory.get_cfs_api_data()) - - def test_get_cfs_api_data_branch_resolved(self): - """Test the get_cfs_api_data method with branch resolved to a commit hash""" - additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env) - expected = {'commit': self.branch_head_commit, 'cloneUrl': self.repo_url} + def test_get_cfs_api_data_no_resolve_branches(self): + """Test get_cfs_api_data method without branch resolution""" + additional_inventory = AdditionalInventory(self.data_with_name, self.jinja_env, + self.mock_cfs_client) + cfs_api_data = additional_inventory.get_cfs_api_data() + + # All the work of getting CFS API request data is delegated to csm_api_client + self.mock_add_inv_cls.from_clone_url.assert_called_once_with( + clone_url=self.repo_url, name=self.name, + branch=self.branch, commit=None + ) + add_inv_instance = self.mock_add_inv_cls.from_clone_url.return_value + self.assertEqual(add_inv_instance.req_payload, cfs_api_data) + add_inv_instance.resolve_branch_to_commit_hash.assert_not_called() + + def test_get_cfs_api_data_resolve_branches(self): + """Test get_cfs_api_data method with branch resolution""" + additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env, + self.mock_cfs_client) with self.patch_resolve_branches(True): - self.assertEqual(expected, additional_inventory.get_cfs_api_data()) - - def test_get_cfs_api_data_commit(self): - """Test the get_cfs_api_data method with commit specified""" - additional_inventory = AdditionalInventory(self.data_with_commit, self.jinja_env) - expected = {'commit': self.commit_hash, 'cloneUrl': self.repo_url} - self.assertEqual(expected, additional_inventory.get_cfs_api_data()) + cfs_api_data = additional_inventory.get_cfs_api_data() + + # All the work of getting CFS API request data is delegated to csm_api_client + self.mock_add_inv_cls.from_clone_url.assert_called_once_with( + clone_url=self.repo_url, name=None, + branch=self.branch, commit=None + ) + add_inv_instance = self.mock_add_inv_cls.from_clone_url.return_value + self.assertEqual(add_inv_instance.req_payload, cfs_api_data) + add_inv_instance.resolve_branch_to_commit_hash.assert_called_once_with() + + def test_get_cfs_api_data_value_error(self): + """Test get_cfs_api_data method when from_clone_url raises ValueError""" + additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env, + self.mock_cfs_client) + self.mock_add_inv_cls.from_clone_url.side_effect = ValueError('error') + with self.assertRaisesRegex(InputItemCreateError, 'error'): + _ = additional_inventory.get_cfs_api_data() + + def test_get_cfs_api_data_resolve_branch_error(self): + """Test get_cfs_api_data method when branch resolution raises CFSConfigurationError""" + additional_inventory = AdditionalInventory(self.data_with_branch, self.jinja_env, + self.mock_cfs_client) + with self.patch_resolve_branches(True): + add_inv_instance = self.mock_add_inv_cls.from_clone_url.return_value + add_inv_instance.resolve_branch_to_commit_hash.side_effect = CFSConfigurationError('error') + with self.assertRaisesRegex(InputItemCreateError, 'error'): + _ = additional_inventory.get_cfs_api_data() class TestInputConfiguration(unittest.TestCase): @@ -617,7 +593,7 @@ def setUp(self): self.mock_instance = Mock(spec=InputInstance) # Fake index of configuration data in an input file self.index = 0 - self.mock_cfs_client = Mock(spep=CFSClientBase) + self.mock_cfs_client = Mock(spec=CFSClientBase) def tearDown(self): patch.stopall() @@ -635,7 +611,8 @@ def test_init_with_additional_inventory(self): self.config_data['additional_inventory'] = additional_inventory_data config = InputConfiguration(self.config_data, self.mock_instance, self.index, self.jinja_env, self.mock_cfs_client, self.mock_product_catalog) - self.mock_additional_inventory_cls.assert_called_once_with(additional_inventory_data, self.jinja_env) + self.mock_additional_inventory_cls.assert_called_once_with(additional_inventory_data, self.jinja_env, + self.mock_cfs_client) self.assertEqual(self.mock_additional_inventory, config.additional_inventory) def test_name_property(self): @@ -673,7 +650,7 @@ def test_get_create_item_data_with_additional_inventory(self): config = InputConfiguration(self.config_data, self.mock_instance, self.index, self.jinja_env, self.mock_cfs_client, self.mock_product_catalog) self.mock_additional_inventory_cls.assert_called_once_with(additional_inventory_data, - self.jinja_env) + self.jinja_env, self.mock_cfs_client) self.assertEqual(expected, config.get_create_item_data()) def test_get_create_item_data_one_layer_failure(self): diff --git a/tests/cli/bootprep/test_validate.py b/tests/cli/bootprep/test_validate.py index 0120ded0..d28942d6 100644 --- a/tests/cli/bootprep/test_validate.py +++ b/tests/cli/bootprep/test_validate.py @@ -1,7 +1,7 @@ # # MIT License # -# (C) Copyright 2021-2023 Hewlett Packard Enterprise Development LP +# (C) Copyright 2021-2024 Hewlett Packard Enterprise Development LP # # Permission is hereby granted, free of charge, to any person obtaining a # copy of this software and associated documentation files (the "Software"), @@ -50,7 +50,8 @@ 'product': { 'name': 'sma', 'version': '1.4.2' - } + }, + 'playbook': 'sma-ldms-compute.yml' } VALID_CONFIG_LAYER_PRODUCT_BRANCH = { @@ -59,6 +60,7 @@ 'name': 'cos', 'branch': 'integration' }, + 'playbook': 'cos-compute.yml', 'special_parameters': { 'ims_require_dkms': True } @@ -69,7 +71,8 @@ 'git': { 'url': 'https://api-gw-service-nmn.local/vcs/cray/cpe-config-management.git', 'branch': 'integration' - } + }, + 'playbook': 'pe_deploy.yml' } VALID_CONFIG_LAYER_GIT_COMMIT = { @@ -80,7 +83,8 @@ }, 'special_parameters': { 'ims_require_dkms': True - } + }, + 'playbook': 'site.yml' } VALID_IMAGE_IMS_NAME_WITH_CONFIG_V1 = { @@ -781,6 +785,17 @@ def test_invalid_config_layer_missing_keys(self): ] self.assert_invalid_instance(instance, expected_errs) + def test_invalid_config_layer_missing_playbook(self): + """Invalid configuration missing 'playbook' key""" + layer = deepcopy(VALID_CONFIG_LAYER_PRODUCT_VERSION) + del layer['playbook'] + instance = self.get_instance_with_config_layer(layer) + expected_errs = [ + (('configurations', 0, 'layers', 0), NOT_VALID_ANY_OF_MESSAGE, 1), + (('configurations', 0, 'layers', 0), "'playbook' is a required property", 2) + ] + self.assert_invalid_instance(instance, expected_errs) + def test_invalid_config_layer_both_keys(self): """Invalid configuration layer with both 'git' and 'product' keys""" layer = {