# Copyright (c) 2017 Ericsson AB # Copyright (c) 2017-2024 Wind River Systems, Inc. # 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 base64 import copy import json import os import keyring import mock from oslo_utils import timeutils import six from six.moves import http_client from tsconfig.tsconfig import SW_VERSION import webtest import yaml from dccommon import consts as dccommon_consts from dcmanager.api.controllers.v1 import phased_subcloud_deploy as psd from dcmanager.api.controllers.v1 import subclouds from dcmanager.common import consts from dcmanager.common import exceptions from dcmanager.common import phased_subcloud_deploy as psd_common from dcmanager.common import prestage from dcmanager.common import utils as cutils from dcmanager.db.sqlalchemy import api as db_api from dcmanager.rpc import client as rpc_client from dcmanager.tests.unit.api import test_root_controller as testroot from dcmanager.tests.unit.api.v1.controllers.mixins import APIMixin from dcmanager.tests.unit.api.v1.controllers.mixins import PostMixin from dcmanager.tests.unit.common import fake_strategy from dcmanager.tests.unit.common import fake_subcloud from dcmanager.tests.unit.fakes import FakeVimStrategy from dcmanager.tests.unit.manager import test_system_peer_manager from dcmanager.tests import utils SAMPLE_SUBCLOUD_NAME = "SubcloudX" SAMPLE_SUBCLOUD_DESCRIPTION = "A Subcloud of mystery" FAKE_ID = fake_subcloud.FAKE_ID FAKE_URL = fake_subcloud.FAKE_URL WRONG_URL = fake_subcloud.WRONG_URL FAKE_HEADERS = fake_subcloud.FAKE_HEADERS FAKE_SUBCLOUD_DATA = fake_subcloud.FAKE_SUBCLOUD_DATA FAKE_BOOTSTRAP_VALUE = fake_subcloud.FAKE_BOOTSTRAP_VALUE FAKE_SUBCLOUD_INSTALL_VALUES = fake_subcloud.FAKE_SUBCLOUD_INSTALL_VALUES FAKE_SUBCLOUD_INSTALL_VALUES_WITH_PERSISTENT_SIZE = ( fake_subcloud.FAKE_SUBCLOUD_INSTALL_VALUES_WITH_PERSISTENT_SIZE ) FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD = fake_subcloud.FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD OAM_FLOATING_IP = "10.10.10.12" FAKE_PEER_GROUP_ID = 10 FAKE_PATCH = {"value": {"patchstate": "Partial-Apply"}} health_report_no_alarm = "System Health:\n \ All hosts are provisioned: [Fail]\n \ 1 Unprovisioned hosts\n \ All hosts are unlocked/enabled: [OK]\n \ All hosts have current configurations: [OK]\n \ All hosts are patch current: [OK]\n \ No alarms: [OK]\n \ All kubernetes nodes are ready: [OK]\n \ All kubernetes control plane pods are ready: [OK]" health_report_no_mgmt_alarm = ( "System Health:\n" "All hosts are provisioned: [OK]\n" "All hosts are unlocked/enabled: [OK]\n" "All hosts have current configurations: [OK]\n" "All hosts are patch current: [OK]\n" "Ceph Storage Healthy: [OK]\n" "No alarms: [Fail]\n" "[1] alarms found, [0] of which are management affecting\n" "All kubernetes nodes are ready: [OK]\n" "All kubernetes control plane pods are ready: [OK]" ) health_report_mgmt_alarm = ( "System Health:\n" "All hosts are provisioned: [OK]\n" "All hosts are unlocked/enabled: [OK]\n" "All hosts have current configurations: [OK]\n" "All hosts are patch current: [OK]\n" "Ceph Storage Healthy: [OK]\n" "No alarms: [Fail]\n" "[1] alarms found, [1] of which are management affecting\n" "All kubernetes nodes are ready: [OK]\n" "All kubernetes control plane pods are ready: [OK]" ) class Subcloud(object): def __init__(self, data, is_online): self.id = data["id"] self.name = data["name"] self.description = data["description"] self.location = data["location"] self.management_state = dccommon_consts.MANAGEMENT_UNMANAGED if is_online: self.availability_status = dccommon_consts.AVAILABILITY_ONLINE else: self.availability_status = dccommon_consts.AVAILABILITY_OFFLINE self.deploy_status = data["deploy_status"] self.management_subnet = data["management_subnet"] self.management_gateway_ip = data["management_gateway_address"] self.management_start_ip = data["management_start_address"] self.management_end_ip = data["management_end_address"] self.external_oam_subnet = data["external_oam_subnet"] self.external_oam_gateway_address = data["external_oam_gateway_address"] self.external_oam_floating_address = data["external_oam_floating_address"] self.systemcontroller_gateway_ip = data["systemcontroller_gateway_address"] self.created_at = timeutils.utcnow() self.updated_at = timeutils.utcnow() self.data_install = "" self.data_upgrade = "" class FakeAddressPool(object): def __init__(self, pool_network, pool_prefix, pool_start, pool_end): self.network = pool_network self.prefix = pool_prefix range = list() range.append(pool_start) range.append(pool_end) self.ranges = list() self.ranges.append(range) class FakeOAMAddressPool(object): def __init__( self, oam_subnet, oam_start_ip, oam_end_ip, oam_c1_ip, oam_c0_ip, oam_gateway_ip, oam_floating_ip, ): self.oam_start_ip = oam_start_ip self.oam_end_ip = oam_end_ip self.oam_c1_ip = oam_c1_ip self.oam_c0_ip = oam_c0_ip self.oam_subnet = oam_subnet self.oam_gateway_ip = oam_gateway_ip self.oam_floating_ip = oam_floating_ip class SubcloudAPIMixin(APIMixin): API_PREFIX = "/v1.0/subclouds" RESULT_KEY = "subclouds" # todo: populate the entire expected fields EXPECTED_FIELDS = [ "id", "name", "description", "location", "management-state", "created-at", "updated-at", ] FAKE_BOOTSTRAP_DATA = { "system_mode": "simplex", "name": "fake subcloud1", "management_subnet": "192.168.101.0/24", "management_start_address": "192.168.101.2", "management_end_address": "192.168.101.50", "management_gateway_address": "192.168.101.1", "external_oam_subnet": "10.10.10.0/24", "external_oam_gateway_address": "10.10.10.1", "external_oam_floating_address": "10.10.10.12", "systemcontroller_gateway_address": "192.168.204.101", } OPTIONAL_BOOTSTRAP_DATA = { "location": "fake location", "description": "fake description", } # based off MANDATORY_INSTALL_VALUES # bmc_password must be passed as a param FAKE_INSTALL_DATA = { "bootstrap_interface": "fake interface", "bootstrap_address": "10.10.10.12", "bootstrap_address_prefix": "10.10.10.12", "bmc_address": "128.224.64.1", "bmc_username": "fake bmc user", "install_type": 2, } list_of_post_files = psd.SUBCLOUD_BOOTSTRAP_GET_FILE_CONTENTS bootstrap_data = copy.copy(FAKE_BOOTSTRAP_DATA) install_data = copy.copy(FAKE_INSTALL_DATA) def setUp(self): super(SubcloudAPIMixin, self).setUp() def _get_test_subcloud_dict(self, **kw): # id should not be part of the structure subcloud = { "name": kw.get("name", SAMPLE_SUBCLOUD_NAME), "description": kw.get("description", SAMPLE_SUBCLOUD_DESCRIPTION), } return subcloud def _post_get_test_subcloud(self, **kw): post_body = self._get_test_subcloud_dict(**kw) return post_body # The following methods are required for subclasses of APIMixin def get_api_prefix(self): return self.API_PREFIX def get_result_key(self): return self.RESULT_KEY def get_expected_api_fields(self): return self.EXPECTED_FIELDS def get_omitted_api_fields(self): return [] def _create_db_object(self, context, **kw): creation_fields = self._get_test_subcloud_dict(**kw) return db_api.subcloud_create(context, **creation_fields) def get_post_params(self): return copy.copy(FAKE_BOOTSTRAP_VALUE) def set_list_of_post_files(self, value): self.list_of_post_files = value def get_post_upload_files(self): fields = list() for f in self.list_of_post_files: fake_name = f + "_fake" # The data in the bootstrap file needs to be dictionary syntax if f == consts.BOOTSTRAP_VALUES: fake_content = json.dumps(self.bootstrap_data).encode("utf-8") elif f == consts.INSTALL_VALUES: fake_content = json.dumps(self.install_data).encode("utf-8") else: fake_content = "fake content".encode("utf-8") fields.append((f, fake_name, fake_content)) return fields def get_post_object(self): return self._post_get_test_subcloud() def get_update_object(self): update_object = {"description": "Updated description"} return update_object # Combine Subcloud Group API with mixins to test post, get, update and delete class TestSubcloudPost(testroot.DCManagerApiTest, SubcloudAPIMixin, PostMixin): def setUp(self): super(TestSubcloudPost, self).setUp() self.list_of_post_files = psd.SUBCLOUD_BOOTSTRAP_GET_FILE_CONTENTS self.bootstrap_data = copy.copy(self.FAKE_BOOTSTRAP_DATA) self.install_data = copy.copy(self.FAKE_INSTALL_DATA) self.management_address_pool = FakeAddressPool( "192.168.204.0", 24, "192.168.204.2", "192.168.204.100" ) p = mock.patch.object(psd_common, "get_network_address_pool") self.mock_get_network_address_pool = p.start() self.mock_get_network_address_pool.return_value = ( self.management_address_pool ) self.addCleanup(p.stop) p = mock.patch.object(rpc_client, "ManagerClient") self.mock_rpc_client = p.start() self.addCleanup(p.stop) p = mock.patch.object(psd_common, "get_ks_client") self.mock_get_ks_client = p.start() self.addCleanup(p.stop) p = mock.patch.object(psd_common.PatchingClient, "query") self.mock_query = p.start() self.addCleanup(p.stop) p = mock.patch.object(rpc_client, "SubcloudStateClient") self.mock_rpc_state_client = p.start() self.addCleanup(p.stop) def _verify_post_failure(self, response, param, value): self.assertEqual( http_client.BAD_REQUEST, response.status_code, message=( "%s=%s returned %s instead of %s" % (param, value, response.status_code, http_client.BAD_REQUEST) ), ) # Note: response failures return 'text' rather than json self.assertEqual("text/plain", response.content_type) def _verify_post_success(self, response): self.assertEqual(http_client.OK, response.status_code) self.assertEqual("application/json", response.content_type) self.assert_fields(response.json) def test_post_subcloud_wrong_url(self): """Test POST operation rejected when going to the wrong URL.""" params = self.get_post_params() upload_files = self.get_post_upload_files() six.assertRaisesRegex( self, webtest.app.AppError, "404 *", self.app.post, WRONG_URL, params=params, upload_files=upload_files, headers=self.get_api_headers(), ) def test_post_no_body(self): """Test POST operation with nearly everything wrong with it.""" six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.post, self.get_api_prefix(), params={}, headers=self.get_api_headers(), ) @mock.patch.object(cutils, 'LOG') def test_post_subcloud_boostrap_file_malformed(self, mock_logging): """Test POST operation with malformed bootstrap file contents.""" params = self.get_post_params() valid_keyval = "key: val" invalid_keyval = "key:val" # A space is required after the colon fake_name = consts.BOOTSTRAP_VALUES + "_fake" corrupt_fake_content = \ (yaml.dump(self.FAKE_BOOTSTRAP_DATA) + invalid_keyval).encode("utf-8") upload_files = list() upload_files.append( (consts.BOOTSTRAP_VALUES, fake_name, corrupt_fake_content) ) response = self.app.post(self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True) self.assertEqual(mock_logging.error.call_count, 1) log_string = mock_logging.error.call_args[0][0] self.assertIn( 'Error: Unable to load bootstrap_values file contents', log_string ) self.assertEqual(http_client.BAD_REQUEST, response.status_code) # try with valid new entry and verify it works valid_content = \ (yaml.dump(self.FAKE_BOOTSTRAP_DATA) + valid_keyval).encode("utf-8") upload_files = list() upload_files.append((consts.BOOTSTRAP_VALUES, fake_name, valid_content)) response = self.app.post(self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers()) self._verify_post_success(response) def test_post_subcloud_boostrap_entries_missing(self): """Test POST operation with some mandatory boostrap fields missing. Example: name is a required field """ self.list_of_post_files = psd.SUBCLOUD_BOOTSTRAP_GET_FILE_CONTENTS params = self.get_post_params() for key in self.FAKE_BOOTSTRAP_DATA: self.bootstrap_data = copy.copy(self.FAKE_BOOTSTRAP_DATA) del self.bootstrap_data[key] upload_files = self.get_post_upload_files() response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True, ) self._verify_post_failure(response, key, None) # try with nothing removed and verify it works self.bootstrap_data = copy.copy(self.FAKE_BOOTSTRAP_DATA) upload_files = self.get_post_upload_files() response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), ) self._verify_post_success(response) def _test_post_param_inputs(self, param_key, bad_values, good_value): upload_files = self.get_post_upload_files() params = self.get_post_params() # Test all the bad param values for bad_value in bad_values: params[param_key] = bad_value response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True, ) self._verify_post_failure(response, param_key, bad_value) # Test that a good value will work params[param_key] = good_value response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), ) self._verify_post_success(response) def test_post_subcloud_bad_bootstrap_address(self): """Test POST operation with a bad bootstrap-address""" param_key = "bootstrap-address" # bootstrap-address must be valid IP address bad_values = [ "10.10.10.wut", # including letters in the IP "10.10.10.276", # 276 is invalid ] good_values = "10.10.10.3" self._test_post_param_inputs(param_key, bad_values, good_values) def test_post_subcloud_bad_IPv6_bootstrap_address(self): """Test POST operation with a bad bootstrap-address""" param_key = "bootstrap-address" # bootstrap-address must be valid IP address bad_values = [ "2620::10a:a103::1135", # more than one double colons "2620:10a:a001:a103::wut", # invalid letter "2620:10a:a001:a103:1135", # Incomplete IP ] good_values = "2620:10a:a001:a103::1135" self._test_post_param_inputs(param_key, bad_values, good_values) def test_post_subcloud_bad_gateway(self): """Test POST with an invalid gateway.""" param_key = "systemcontroller_gateway_address" # systemcontroller_gateway_address must be appropriate address within # the management address pool which is # 192.168.204.0/24 greater than 100 bad_values = [ "192.168.205.101", # 205.xx not in the pool "192.168.204.99", # 99 is reserved in the pool "192.168.276.276", # 276 is not a valid IP address "192.168.206.wut", # including letters in the IP "192.168.204", # incomplete IP ] good_value = "192.168.204.101" self._test_post_param_inputs(param_key, bad_values, good_value) def test_post_subcloud_bad_subnet(self): """Test POST with an invalid subnet.""" param_key = "management_subnet" bad_values = [ "192.168.101.0/32", # /32 would be just one IP "192.168.101.0/33", # /33 is an invalid CIDR "192.168.276.0/24", # 276 makes no sense as an IP "192.168.206.wut/24", # including letters in the IP "192.168.204/24", # incomplete CIDR ] good_value = "192.168.101.0/24" self._test_post_param_inputs(param_key, bad_values, good_value) def test_post_subcloud_bad_start_ip(self): """Test POST with an invalid management_start_address. The management_start_address cannot be after the end or too close since there must be enough range to allocate the IPs. """ param_key = "management_start_address" # subnet is 192.168.101.0/24 # end address is 192.168.101.50 bad_values = [ "192.168.100.2", # xx.xx.100.xx is not in the subnet "192.168.101.51", # start is higher than end "192.168.101.48", # start is too close to end "192.168.276.0", # 276 makes no sense as an IP "192.168.206.wut", # including letters in the IP "192.168.204", # incomplete IP ] good_value = "192.168.101.2" self._test_post_param_inputs(param_key, bad_values, good_value) def test_post_subcloud_bad_end_ip(self): """Test POST with an invalid management_end_address. The management_end_address cannot be less than the start or too close since there must be enough range to allocate the IPs. """ param_key = "management_end_address" # subnet is 192.168.101.0/24 # start address is 192.168.101.2 bad_values = [ "192.168.100.50", # xx.xx.100.xx is not in the subnet "192.168.101.1", # end is less than start "192.168.101.4", # end is too close to start "192.168.276.50", # 276 makes no sense as an IP "192.168.206.wut", # including letters in the IP "192.168.204", # incomplete IP ] good_value = "192.168.101.50" self._test_post_param_inputs(param_key, bad_values, good_value) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_values(self, mock_vault_files): """Test POST operation with install values is supported by the API.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") # pass a different "install" list of files for this POST self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) upload_files = self.get_post_upload_files() params = self.get_post_params() # add bmc_password to params params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ) } ) response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), ) self._verify_post_success(response) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_without_release_parameter(self, mock_vault_files): """Test POST operation without release parameter.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) upload_files = self.get_post_upload_files() params = self.get_post_params() # add bmc_password to params params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ) } ) response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), ) self._verify_post_success(response) # Verify that the subcloud installed with the active release # when no release parameter provided. self.assertEqual(SW_VERSION, response.json["software-version"]) def test_post_subcloud_release_not_match_install_values_sw(self): """Release parameter not match software_version in the install_values.""" self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) upload_files = self.get_post_upload_files() params = self.get_post_params() # add bmc_password and release to params params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ), "release": "21.12", } ) with mock.patch('builtins.open', mock.mock_open( read_data=fake_subcloud.FAKE_UPGRADES_METADATA )): response = self.app.post(self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True) # Verify the request was rejected self.assertEqual(response.status_code, http_client.BAD_REQUEST) @mock.patch.object(psd_common, "validate_k8s_version") @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_with_release_parameter( self, mock_vault_files, mock_validate_k8s_version ): """Test POST operation with release parameter.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") software_version = "21.12" # Update the software_version value to match the release parameter value, # otherwise, the request will be rejected self.install_data["software_version"] = software_version self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) upload_files = self.get_post_upload_files() params = self.get_post_params() # add bmc_password and release to params params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ), "release": software_version, } ) with mock.patch('builtins.open', mock.mock_open( read_data=fake_subcloud.FAKE_UPGRADES_METADATA )): response = self.app.post(self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True) self.assertEqual(response.status_code, http_client.OK) self.assertEqual(software_version, response.json["software-version"]) # Revert the software_version value self.install_data["software_version"] = SW_VERSION @mock.patch.object(psd_common.PatchingClient, "query") def test_post_subcloud_when_partial_applied_patch(self, mock_query): """Test POST operation when there is a partial-applied patch.""" upload_files = self.get_post_upload_files() params = self.get_post_params() mock_query.return_value = FAKE_PATCH response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True, ) self.assertEqual(http_client.UNPROCESSABLE_ENTITY, response.status_code) self.assertEqual("text/plain", response.content_type) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_values_no_bmc_password(self, mock_vault_files): """Test POST operation with install values is supported by the API.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") # pass a different "install" list of files for this POST self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) upload_files = self.get_post_upload_files() params = self.get_post_params() # for this unit test, omit adding bmc_password to params response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True, ) self._verify_post_failure(response, "bmc_password", None) # add the bmc_password and verify that now it works params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ) } ) response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), ) self._verify_post_success(response) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_missing_image(self, mock_vault_files): """Test POST operation without image in install values and vault files.""" mock_vault_files.return_value = (None, None) params = self.get_post_params() # add bmc_password to params params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ) } ) self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) self.install_data = copy.copy(self.FAKE_INSTALL_DATA) upload_files = self.get_post_upload_files() response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True, ) self.assertEqual(response.status_code, http_client.BAD_REQUEST) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_values_missing(self, mock_vault_files): """Test POST operation with install values fails if data missing.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") params = self.get_post_params() # add bmc_password to params params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ) } ) self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) # for each entry in install content, try with one key missing for key in self.FAKE_INSTALL_DATA: self.install_data = copy.copy(self.FAKE_INSTALL_DATA) del self.install_data[key] upload_files = self.get_post_upload_files() response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True, ) self._verify_post_failure(response, key, None) @mock.patch("dcmanager.common.utils.get_vault_load_files") @mock.patch.object(cutils, "get_playbook_for_software_version") @mock.patch.object(cutils, "get_value_from_yaml_file") def test_post_subcloud_bad_kubernetes_version( self, mock_get_value_from_yaml_file, mock_get_playbook_for_software_version, mock_vault_files, ): """Test POST operation with bad kubernetes_version.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") software_version = "21.12" # Update the software_version value to match the release parameter value, # otherwise, the request will be rejected self.install_data["software_version"] = software_version params = self.get_post_params() # add bmc_password to params params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ), "release": software_version, } ) # Add kubernetes version to bootstrap_data self.bootstrap_data["kubernetes_version"] = "1.21.8" mock_get_value_from_yaml_file.return_value = "1.23.1" self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) self.install_data = copy.copy(self.FAKE_INSTALL_DATA) upload_files = self.get_post_upload_files() with mock.patch('builtins.open', mock.mock_open( read_data=fake_subcloud.FAKE_UPGRADES_METADATA )): response = self.app.post(self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True) self.assertEqual(response.status_code, http_client.BAD_REQUEST) # Revert the change of bootstrap_data del self.bootstrap_data["kubernetes_version"] def _test_post_input_value_inputs( self, setup_overrides, required_overrides, param_key, bad_values, good_value ): """This utility checks for test permutions. The setup_overrides are the initial modifications to the install data The required_overrides are all tested to see that if any of them are missing, the 'good' value will not work. The param_key is tested with the list of bad_values to ensure they fail The param_key is tested with the good value to ensure it passes. """ params = self.get_post_params() params.update( { "bmc_password": base64.b64encode("fake pass".encode("utf-8")).decode( "utf-8" ) } ) self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS) # Setup starting install data # Note: upload_files are populated based on the install values data. starting_data = copy.copy(self.FAKE_INSTALL_DATA) for key, val in setup_overrides.items(): starting_data[key] = val starting_data["image"] = "fake image" # Test all the bad param values for bad_value in bad_values: self.install_data = copy.copy(starting_data) # Apply all required_overrides for key, val in required_overrides.items(): self.install_data[key] = val # Apply the bad value self.install_data[param_key] = bad_value upload_files = self.get_post_upload_files() response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True, ) self._verify_post_failure(response, param_key, bad_value) # Test that any missing override required to use with the good value # will cause a failure for missing_override in required_overrides: self.install_data = copy.copy(starting_data) # We cannot simply delete the missing override, but we can skip it for key, val in required_overrides.items(): if key != missing_override: self.install_data[key] = val # The 'good' value should still fail if a required override missing self.install_data[param_key] = good_value upload_files = self.get_post_upload_files() response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), expect_errors=True, ) self._verify_post_failure(response, param_key, bad_value) # Test that a good value and all required overrides works self.install_data = copy.copy(starting_data) for key, val in required_overrides.items(): self.install_data[key] = val self.install_data[param_key] = good_value upload_files = self.get_post_upload_files() response = self.app.post( self.get_api_prefix(), params=params, upload_files=upload_files, headers=self.get_api_headers(), ) self._verify_post_success(response) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_values_invalid_type(self, mock_vault_files): """Test POST with an invalid type specified in install values.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") setup_overrides = {} required_overrides = {} # the install_type must a number 0 <= X <=5 install_key = "install_type" bad_values = [ -1, # negative 6, # too big "3", # alphbetical "w", # really alphbetical "", # empty None, # None ] good_value = 3 self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_bad_bootstrap_ip(self, mock_vault_files): """Test POST with invalid boostrap ip specified in install values.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") setup_overrides = {} required_overrides = {} install_key = "bootstrap_address" bad_values = [ "192.168.1.256", # 256 is not valid "192.168.206.wut", # including letters in the IP None, # None ] # Note: an incomplete IP address is 10.10.10 is considered valid good_value = "10.10.10.12" self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_bad_bmc_ip(self, mock_vault_files): """Test POST with invalid bmc ip specified in install values.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") setup_overrides = {} required_overrides = {} install_key = "bmc_address" bad_values = [ "128.224.64.256", # 256 is not valid "128.224.64.wut", # including letters in the IP None, # None ] good_value = "128.224.64.1" self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_bad_persistent_size(self, mock_vault_files): """Test POST with invalid persistent_size specified in install values.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") setup_overrides = {} required_overrides = {} install_key = "persistent_size" bad_values = [ "4000o", # not an integer "20000", # less than 30000 40000.1, # fraction None, # None ] good_value = 40000 self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_bad_nexthop_gateway(self, mock_vault_files): """Test POST with invalid nexthop_gateway in install values.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") setup_overrides = {} required_overrides = {} # nexthop_gateway is not required. but if provided, it must be valid install_key = "nexthop_gateway" bad_values = [ "128.224.64.256", # 256 is not valid "128.224.64.wut", # including letters in the IP None, # None ] good_value = "192.168.1.2" self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_bad_network_address(self, mock_vault_files): """Test POST with invalid network_address in install values.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") setup_overrides = {} # The nexthop_gateway is required when network_address is present # The network mask is required when network address is present required_overrides = { "nexthop_gateway": "192.168.1.2", "network_mask": 32, # Note: this netmask is validated when used } # network_address is not required. but if provided, it must be valid install_key = "network_address" # todo(abailey): None will cause the API to fail bad_values = [ "fd01:6::0", # mis-match ipv6 vs ipv4 ] good_value = "192.168.101.10" # ipv4 self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_bad_network_mask(self, mock_vault_files): """Test POST with invalid network_mask in install values.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") # network_address is not required. but if provided a valid network_mask # is needed setup_overrides = { "nexthop_gateway": "192.168.1.2", "network_address": "192.168.101.10", } required_overrides = {} install_key = "network_mask" bad_values = [ None, # None 64, # network_mask cannot really be greater than 32 -1, # network_mask cannot really be negative "junk", # network_mask cannot be a junk string ] good_value = 32 self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_diff_bmc_ip_version(self, mock_vault_files): """Test POST install values with mismatched(ipv4/ipv6) bmc ip.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") setup_overrides = {"bootstrap_address": "192.168.1.2"} required_overrides = {} # bootstrap address ip version must match bmc_address. default ipv4 install_key = "bmc_address" bad_values = [ "fd01:6::7", # ipv6 None, # None "192.168.-1.1", # bad ipv4 ] good_value = "192.168.1.7" # ipv4 self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_diff_bmc_ip_version_ipv6(self, mock_vault_files): """Test POST install values with mismatched(ipv6/ipv4) bmc ip.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") # version of bootstrap address must be same as bmc_address setup_overrides = {"bootstrap_address": "fd01:6::7"} required_overrides = {} install_key = "bmc_address" bad_values = [ "192.168.1.7", # ipv4 None, # None "fd01:6:-1", # bad ipv6 ] good_value = "fd01:6::7" # ipv6 self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_diff_nexthop_ip_version(self, mock_vault_files): """Test POST install values mismatched(ipv4/ipv6) nexthop_gateway.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") # ip version of bootstrap address must be same as nexthop_gateway # All required addresses (like bmc address) much match bootstrap # default bmc address is ipv4 setup_overrides = {"bootstrap_address": "192.168.1.5"} required_overrides = {} install_key = "nexthop_gateway" bad_values = [ "fd01:6::7", ] # ipv6 good_value = "192.168.1.7" # ipv4 self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) @mock.patch("dcmanager.common.utils.get_vault_load_files") def test_post_subcloud_install_diff_nexthop_ip_version_ipv6( self, mock_vault_files ): """Test POST install values with mismatched(ipv6/ipv4) bmc ip.""" mock_vault_files.return_value = ("fake_iso", "fake_sig") # version of bootstrap address must be same as nexthop_gateway # All required addresses must also be setup ipv6 such as bmc_address # default bmc address is ipv4 setup_overrides = {"bootstrap_address": "fd01:6::6"} required_overrides = {"bmc_address": "fd01:6::7"} install_key = "nexthop_gateway" bad_values = [ "192.168.1.7", ] # ipv4 good_value = "fd01:6::8" # ipv6 self._test_post_input_value_inputs( setup_overrides, required_overrides, install_key, bad_values, good_value ) class TestSubcloudAPIOther(testroot.DCManagerApiTest): """Test GET, delete and patch API calls""" def setUp(self): super(TestSubcloudAPIOther, self).setUp() self.ctx = utils.dummy_context() p = mock.patch.object(rpc_client, "SubcloudStateClient") self.mock_rpc_state_client = p.start() self.addCleanup(p.stop) p = mock.patch.object(rpc_client, "ManagerClient") self.mock_rpc_client = p.start() self.addCleanup(p.stop) p = mock.patch.object(psd_common, "get_ks_client") self.mock_get_ks_client = p.start() self.addCleanup(p.stop) def test_delete_subcloud(self): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) delete_url = FAKE_URL + "/" + str(subcloud.id) self.mock_rpc_client().delete_subcloud.return_value = True response = self.app.delete_json(delete_url, headers=FAKE_HEADERS) self.mock_rpc_client().delete_subcloud.assert_called_once_with( mock.ANY, mock.ANY ) self.assertEqual(response.status_int, 200) def test_delete_wrong_request(self): delete_url = WRONG_URL + "/" + FAKE_ID six.assertRaisesRegex( self, webtest.app.AppError, "404 *", self.app.delete_json, delete_url, headers=FAKE_HEADERS, ) @mock.patch.object(subclouds.SubcloudsController, "_get_oam_addresses") def test_get_subcloud(self, mock_get_oam_addresses): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) get_url = FAKE_URL + "/" + str(subcloud.id) response = self.app.get(get_url, headers=FAKE_HEADERS) self.assertEqual(response.content_type, "application/json") self.assertEqual(response.status_code, http_client.OK) self.assertEqual(response.json.get("oam_floating_ip", None), None) self.assertEqual(response.json["name"], subcloud.name) @mock.patch.object( subclouds.SubcloudsController, "_get_deploy_config_sync_status" ) @mock.patch.object(subclouds.SubcloudsController, "_get_oam_addresses") def test_get_online_subcloud_with_additional_detail( self, mock_get_oam_addresses, mock_get_deploy_config_sync_status ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) updated_subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, ) get_url = FAKE_URL + "/" + str(updated_subcloud.id) + "/detail" oam_addresses = FakeOAMAddressPool( "10.10.10.254", "10.10.10.1", "10.10.10.254", "10.10.10.4", "10.10.10.3", "10.10.10.1", "10.10.10.2", ) mock_get_oam_addresses.return_value = oam_addresses mock_get_deploy_config_sync_status.return_value = ( dccommon_consts.DEPLOY_CONFIG_UP_TO_DATE ) response = self.app.get(get_url, headers=FAKE_HEADERS) self.assertEqual(response.content_type, "application/json") self.assertEqual(response.status_code, http_client.OK) self.assertEqual("10.10.10.2", response.json["oam_floating_ip"]) self.assertEqual( "Deployment: configurations up-to-date", response.json["deploy_config_sync_status"], ) def test_get_offline_subcloud_with_additional_detail(self): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) get_url = FAKE_URL + "/" + str(subcloud.id) + "/detail" response = self.app.get(get_url, headers=FAKE_HEADERS) self.assertEqual(response.content_type, "application/json") self.assertEqual(response.status_code, http_client.OK) self.assertEqual("unavailable", response.json["oam_floating_ip"]) self.assertEqual("unknown", response.json["deploy_config_sync_status"]) @mock.patch.object( subclouds.SubcloudsController, "_get_deploy_config_sync_status" ) @mock.patch.object(subclouds.SubcloudsController, "_get_oam_addresses") def test_get_subcloud_deploy_config_status_unknown( self, mock_get_oam_addresses, mock_get_deploy_config_sync_status ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) updated_subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, ) get_url = FAKE_URL + "/" + str(updated_subcloud.id) + "/detail" mock_get_oam_addresses.return_value = None mock_get_deploy_config_sync_status.return_value = None response = self.app.get(get_url, headers=FAKE_HEADERS) self.assertEqual(response.content_type, "application/json") self.assertEqual(response.status_code, http_client.OK) self.assertEqual("unknown", response.json["deploy_config_sync_status"]) @mock.patch.object(subclouds.SubcloudsController, "_get_oam_addresses") def test_get_subcloud_oam_ip_unavailable(self, mock_get_oam_addresses): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) updated_subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, ) get_url = FAKE_URL + "/" + str(updated_subcloud.id) + "/detail" self.mock_get_ks_client.return_value = "ks_client" mock_get_oam_addresses.return_value = None response = self.app.get(get_url, headers=FAKE_HEADERS) self.assertEqual(response.content_type, "application/json") self.assertEqual(response.status_code, http_client.OK) self.assertEqual("unavailable", response.json["oam_floating_ip"]) def test_get_wrong_request(self): get_url = WRONG_URL + "/" + FAKE_ID six.assertRaisesRegex( self, webtest.app.AppError, "404 *", self.app.get, get_url, headers=FAKE_HEADERS, ) def test_get_subcloud_all(self): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) get_url = FAKE_URL response = self.app.get(get_url, headers=FAKE_HEADERS) self.assertEqual(response.json["subclouds"][0]["name"], subcloud.name) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_patch_subcloud(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = {"management-state": dccommon_consts.MANAGEMENT_UNMANAGED} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data ) self.assertEqual(response.status_int, 200) # Verify subcloud was updated with correct values updated_subcloud = db_api.subcloud_get_by_name(self.ctx, subcloud.name) self.assertEqual( dccommon_consts.MANAGEMENT_UNMANAGED, updated_subcloud.management_state ) # Verify that the update PGA sync_status is not called since subcloud # isn't associated with an SPG. self.mock_rpc_client().update_association_sync_status.assert_not_called() @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_update_subcloud_group_value(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) good_values = [1, "1"] expected_group_id = 1 for x in good_values: data = {"group_id": x} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data ) self.assertEqual(response.status_int, 200) # Verify subcloud was updated with correct values updated_subcloud = db_api.subcloud_get_by_name(self.ctx, subcloud.name) self.assertEqual(expected_group_id, updated_subcloud.group_id) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_update_subcloud_group_value_by_name(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) expected_group_id = 1 data = {"group_id": "Default"} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data ) self.assertEqual(response.status_int, 200) # Verify subcloud was updated with correct values updated_subcloud = db_api.subcloud_get_by_name(self.ctx, subcloud.name) self.assertEqual(expected_group_id, updated_subcloud.group_id) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_update_subcloud_group_bad_value(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) # There is only 1 subcloud group 'Default' which has id '1' # This should test that boolean, zero, negative, float and bad values # all get rejected bad_values = [0, -1, 2, "0", "-1", 0.5, "BadName", "False", "True"] for x in bad_values: data = {"group_id": x} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, expect_errors=True, ) self.assertEqual(response.status_int, 400) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, "get_vault_load_files") def test_update_subcloud_install_values_persistent_size( self, mock_vault_files, mock_get_patch_data ): mock_vault_files.return_value = ("fake_iso", "fake_sig") subcloud = fake_subcloud.create_fake_subcloud(self.ctx, data_install=None) payload = {} install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES_WITH_PERSISTENT_SIZE) encoded_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) data = {"bmc_password": encoded_password} payload.update({"install_values": install_data}) payload.update(data) self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = payload fake_content = "fake content".encode("utf-8") response = self.app.patch( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, upload_files=[("install_values", "fake_name", fake_content)], ) install_data.update({"bmc_password": encoded_password}) self.mock_rpc_client().update_subcloud.assert_called_once_with( mock.ANY, subcloud.id, management_state=None, description=None, location=None, group_id=None, data_install=json.dumps(install_data), force=None, peer_group_id=None, bootstrap_values=None, bootstrap_address=None, deploy_status=None) self.assertEqual(response.status_int, 200) # Verify that the update PGA sync_status is not called since subcloud # isn't associated with an SPG. self.mock_rpc_client().update_association_sync_status.assert_not_called() @mock.patch.object(subclouds.SubcloudsController, "_check_existing_vim_strategy") @mock.patch.object(psd_common, "get_network_address_pool") @mock.patch.object(subclouds.SubcloudsController, "_validate_network_reconfiguration") @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_patch_subcloud_network_values( self, mock_get_patch_data, mock_validate_network_reconfiguration, mock_mgmt_address_pool, mock_check_existing_vim_strategy): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, ) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) payload = { "sysadmin_password": fake_password, "bootstrap_address": "192.168.102.2", "management_subnet": "192.168.102.0/24", "management_start_ip": "192.168.102.5", "management_end_ip": "192.168.102.49", "management_gateway_ip": "192.168.102.1", } fake_management_address_pool = FakeAddressPool( "192.168.204.0", 24, "192.168.204.2", "192.168.204.100" ) mock_mgmt_address_pool.return_value = fake_management_address_pool mock_check_existing_vim_strategy.return_value = False self.mock_rpc_client().update_subcloud_with_network_reconfig.return_value = ( True ) mock_get_patch_data.return_value = payload response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=payload ) self.assertEqual(response.status_int, 200) mock_validate_network_reconfiguration.assert_called_once() self.mock_rpc_client().update_subcloud_with_network_reconfig.\ assert_called_once_with( mock.ANY, subcloud.id, payload ) self.assertEqual(response.status_int, 200) @mock.patch.object(subclouds.SubcloudsController, "_check_existing_vim_strategy") @mock.patch.object(psd_common, "get_network_address_pool") @mock.patch.object(subclouds.SubcloudsController, "_validate_network_reconfiguration") @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_patch_subcloud_network_values_with_onging_strategy( self, mock_get_patch_data, mock_validate_network_reconfiguration, mock_mgmt_address_pool, mock_check_existing_vim_strategy): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE) fake_password = ( base64.b64encode('testpass'.encode("utf-8"))).decode('ascii') payload = {'sysadmin_password': fake_password, 'bootstrap_address': "192.168.102.2", 'management_subnet': "192.168.102.0/24", 'management_start_ip': "192.168.102.5", 'management_end_ip': "192.168.102.49", 'management_gateway_ip': "192.168.102.1"} fake_management_address_pool = FakeAddressPool('192.168.204.0', 24, '192.168.204.2', '192.168.204.100') mock_mgmt_address_pool.return_value = fake_management_address_pool mock_check_existing_vim_strategy.return_value = True self.mock_rpc_client().update_subcloud_with_network_reconfig.\ return_value = True mock_get_patch_data.return_value = payload six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, f"{ FAKE_URL }/{ subcloud.id }", headers=FAKE_HEADERS, params=payload, ) mock_validate_network_reconfiguration.assert_called_once() self.mock_rpc_client().update_subcloud_with_network_reconfig.\ assert_not_called() @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, "get_vault_load_files") def test_patch_subcloud_install_values(self, mock_vault_files, mock_get_patch_data): mock_vault_files.return_value = ("fake_iso", "fake_sig") subcloud = fake_subcloud.create_fake_subcloud(self.ctx, data_install=None) payload = {} install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) encoded_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) data = {"bmc_password": encoded_password} payload.update({"install_values": install_data}) payload.update(data) self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = payload fake_content = "fake content".encode("utf-8") response = self.app.patch( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, upload_files=[("install_values", "fake_name", fake_content)], ) install_data.update({"bmc_password": encoded_password}) self.mock_rpc_client().update_subcloud.assert_called_once_with( mock.ANY, subcloud.id, management_state=None, description=None, location=None, group_id=None, data_install=json.dumps(install_data), force=None, peer_group_id=None, bootstrap_values=None, bootstrap_address=None, deploy_status=None) self.assertEqual(response.status_int, 200) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, "get_vault_load_files") def test_patch_subcloud_install_values_with_existing_data_install( self, mock_vault_files, mock_get_patch_data ): mock_vault_files.return_value = ("fake_iso", "fake_sig") install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) subcloud = fake_subcloud.create_fake_subcloud( self.ctx, data_install=json.dumps(install_data) ) install_data.update({"install_type": 2}) payload = {} encoded_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) data = {"bmc_password": encoded_password} payload.update({"install_values": install_data}) payload.update(data) self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = payload fake_content = "fake content".encode("utf-8") response = self.app.patch( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, upload_files=[("install_values", "fake_name", fake_content)], ) install_data.update({"bmc_password": encoded_password}) self.mock_rpc_client().update_subcloud.assert_called_once_with( mock.ANY, subcloud.id, management_state=None, description=None, location=None, group_id=None, data_install=json.dumps(install_data), force=None, peer_group_id=None, bootstrap_values=None, bootstrap_address=None, deploy_status=None) self.assertEqual(response.status_int, 200) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_patch_subcloud_no_body(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = {} mock_get_patch_data.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, ) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_patch_subcloud_bad_status(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = {"management-state": "bad-status"} mock_get_patch_data.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, ) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_patch_subcloud_bad_force_value(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = { "management-state": dccommon_consts.MANAGEMENT_MANAGED, "force": "bad-value", } mock_get_patch_data.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, ) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_patch_subcloud_forced_unmanaged(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = { "management-state": dccommon_consts.MANAGEMENT_UNMANAGED, "force": True, } mock_get_patch_data.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, ) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") def test_patch_subcloud_forced_manage(self, mock_get_patch_data): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) payload = { "management-state": dccommon_consts.MANAGEMENT_MANAGED, "force": True, } self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = payload response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=payload ) self.mock_rpc_client().update_subcloud.assert_called_once_with( mock.ANY, mock.ANY, management_state=dccommon_consts.MANAGEMENT_MANAGED, description=None, location=None, group_id=None, data_install=None, force=True, peer_group_id=None, bootstrap_values=None, bootstrap_address=None, deploy_status=None) self.assertEqual(response.status_int, 200) @mock.patch.object(subclouds.SubcloudsController, '_get_updatestatus_payload') def test_subcloud_updatestatus(self, mock_get_updatestatus_payload): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = {"endpoint": "dc-cert", "status": "in-sync"} mock_get_updatestatus_payload.return_value = data self.mock_rpc_state_client().update_subcloud_endpoint_status.return_value = ( True ) response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id) + "/update_status", data, headers=FAKE_HEADERS, ) self.mock_rpc_state_client().update_subcloud_endpoint_status.\ assert_called_once_with(mock.ANY, subcloud.name, subcloud.region_name, 'dc-cert', 'in-sync') self.assertEqual(response.status_int, 200) @mock.patch.object(subclouds.SubcloudsController, "_get_updatestatus_payload") def test_subcloud_updatestatus_invalid_endpoint( self, mock_get_updatestatus_payload ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = {"endpoint": "any-other-endpoint", "status": "in-sync"} mock_get_updatestatus_payload.return_value = data self.mock_rpc_client().update_subcloud_endpoint_status.return_value = True six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/update_status", headers=FAKE_HEADERS, params=data, ) self.mock_rpc_client().update_subcloud_endpoint_status.assert_not_called() @mock.patch.object(subclouds.SubcloudsController, "_get_updatestatus_payload") def test_subcloud_updatestatus_invalid_status( self, mock_get_updatestatus_payload ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = {"endpoint": "dc-cert", "status": "not-sure"} mock_get_updatestatus_payload.return_value = data self.mock_rpc_client().update_subcloud_endpoint_status.return_value = True six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/update_status", headers=FAKE_HEADERS, params=data, ) self.mock_rpc_client().update_subcloud_endpoint_status.assert_not_called() @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, 'is_leader_on_local_site') @mock.patch.object(cutils, 'subcloud_peer_group_get_by_ref') def test_add_subcloud_to_peer_group(self, mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data): peer_group = test_system_peer_manager.TestSystemPeerManager. \ create_subcloud_peer_group_static( self.ctx, peer_group_name='SubcloudPeerGroup1') mock_get_peer_group.return_value = peer_group subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, deploy_status=consts.DEPLOY_STATE_DONE, management_state=dccommon_consts.MANAGEMENT_MANAGED, prestage_status=consts.PRESTAGE_STATE_COMPLETE, rehome_data="{\"saved_payload\": " "{\"system_mode\": \"simplex\"," "\"bootstrap-address\": \"192.168.100.100\"}}" ) mock_is_leader_on_local_site.return_value = True data = {"peer_group": peer_group.id} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data ) self.assertEqual(response.status_int, 200) self.mock_rpc_client().update_subcloud.assert_called_once() self.mock_rpc_client().update_association_sync_status. \ assert_called_once() @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, 'is_leader_on_local_site') @mock.patch.object(cutils, 'subcloud_peer_group_get_by_ref') def test_add_subcloud_to_peer_group_without_rehome_data( self, mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data ): peer_group = test_system_peer_manager.TestSystemPeerManager. \ create_subcloud_peer_group_static( self.ctx, peer_group_name='SubcloudPeerGroup1') mock_get_peer_group.return_value = peer_group subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, deploy_status=consts.DEPLOY_STATE_DONE, management_state=dccommon_consts.MANAGEMENT_MANAGED, prestage_status=consts.PRESTAGE_STATE_COMPLETE ) mock_is_leader_on_local_site.return_value = True data = {"peer_group": peer_group.id} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, ) self.mock_rpc_client().update_subcloud.assert_not_called() self.mock_rpc_client().update_association_sync_status. \ assert_not_called() @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, 'is_leader_on_local_site') @mock.patch.object(cutils, 'subcloud_peer_group_get_by_ref') def test_remove_subcloud_from_peer_group(self, mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data): peer_group = test_system_peer_manager.TestSystemPeerManager. \ create_subcloud_peer_group_static( self.ctx, peer_group_name='SubcloudPeerGroup1') mock_get_peer_group.return_value = peer_group subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, deploy_status=consts.DEPLOY_STATE_DONE, management_state=dccommon_consts.MANAGEMENT_MANAGED, prestage_status=consts.PRESTAGE_STATE_COMPLETE, peer_group_id=peer_group.id ) mock_is_leader_on_local_site.return_value = True data = {"peer_group": "none"} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data ) self.assertEqual(response.status_int, 200) self.mock_rpc_client().update_subcloud.assert_called_once() self.mock_rpc_client().update_association_sync_status. \ assert_called_once() @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, 'is_leader_on_local_site') @mock.patch.object(cutils, 'subcloud_peer_group_get_by_ref') def test_update_subcloud_on_non_primary_site(self, mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data): peer_group = test_system_peer_manager.TestSystemPeerManager. \ create_subcloud_peer_group_static( self.ctx, peer_group_name='SubcloudPeerGroup1', group_priority=1) mock_get_peer_group.return_value = peer_group subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, deploy_status=consts.DEPLOY_STATE_DONE, management_state=dccommon_consts.MANAGEMENT_MANAGED, prestage_status=consts.PRESTAGE_STATE_COMPLETE, peer_group_id=peer_group.id ) mock_is_leader_on_local_site.return_value = True data = {"description": "test subcloud"} mock_get_patch_data.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=data, ) self.mock_rpc_client().update_subcloud.assert_not_called() self.mock_rpc_client().update_association_sync_status.assert_not_called() @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, 'is_leader_on_local_site') @mock.patch.object(cutils, 'subcloud_peer_group_get_by_ref') def test_update_subcloud_bootstrap_address_on_primary_site( self, mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data): peer_group = test_system_peer_manager.TestSystemPeerManager. \ create_subcloud_peer_group_static( self.ctx, peer_group_name='SubcloudPeerGroup1', group_priority=consts.PEER_GROUP_PRIMARY_PRIORITY) mock_get_peer_group.return_value = peer_group subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, deploy_status=consts.DEPLOY_STATE_DONE, management_state=dccommon_consts.MANAGEMENT_MANAGED, prestage_status=consts.PRESTAGE_STATE_COMPLETE, peer_group_id=peer_group.id ) mock_is_leader_on_local_site.return_value = True data = {"bootstrap_address": "192.168.10.22"} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = data response = self.app.patch_json( f"{FAKE_URL}/{subcloud.id}", headers=FAKE_HEADERS, params=data ) self.assertEqual(response.status_int, 200) self.mock_rpc_client().update_subcloud.assert_called_once_with( mock.ANY, subcloud.id, management_state=None, description=None, location=None, group_id=None, data_install=None, force=None, peer_group_id=None, bootstrap_values=None, bootstrap_address=data["bootstrap_address"], deploy_status=None) self.mock_rpc_client().update_association_sync_status. \ assert_called_once() def update_subcloud_bootstrap_address_on_non_primary_site( self, mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data, primary_site_availability_state): peer = test_system_peer_manager.TestSystemPeerManager. \ create_system_peer_static( self.ctx, peer_name='SystemPeer1') peer = db_api.system_peer_update( self.ctx, peer.id, availability_state=primary_site_availability_state) peer_group = test_system_peer_manager.TestSystemPeerManager. \ create_subcloud_peer_group_static( self.ctx, peer_group_name='SubcloudPeerGroup1', group_priority=1) association = test_system_peer_manager.TestSystemPeerManager. \ create_peer_group_association_static( self.ctx, system_peer_id=peer.id, peer_group_id=peer_group.id) mock_get_peer_group.return_value = peer_group subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, deploy_status=consts.DEPLOY_STATE_REHOME_FAILED, management_state=dccommon_consts.MANAGEMENT_MANAGED, prestage_status=consts.PRESTAGE_STATE_COMPLETE, peer_group_id=peer_group.id ) mock_is_leader_on_local_site.return_value = True patch_data = {"bootstrap_address": "192.168.10.22"} self.mock_rpc_client().update_subcloud.return_value = True mock_get_patch_data.return_value = patch_data return association, subcloud, patch_data @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, 'is_leader_on_local_site') @mock.patch.object(cutils, 'subcloud_peer_group_get_by_ref') def test_update_subcloud_bootstrap_address_when_primary_site_is_available( self, mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data): association, subcloud, patch_data = \ self.update_subcloud_bootstrap_address_on_non_primary_site( mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data, consts.SYSTEM_PEER_AVAILABILITY_STATE_AVAILABLE) six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id), headers=FAKE_HEADERS, params=patch_data, ) self.mock_rpc_client().update_subcloud.assert_not_called() self.mock_rpc_client().update_association_sync_status.assert_not_called() self.assertEqual(consts.ASSOCIATION_SYNC_STATUS_IN_SYNC, db_api.peer_group_association_get( self.ctx, association.id).sync_status) @mock.patch.object(subclouds.SubcloudsController, "_get_patch_data") @mock.patch.object(cutils, 'is_leader_on_local_site') @mock.patch.object(cutils, 'subcloud_peer_group_get_by_ref') def test_update_subcloud_bootstrap_address_when_primary_site_is_unavailable( self, mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data): association, subcloud, patch_data = \ self.update_subcloud_bootstrap_address_on_non_primary_site( mock_get_peer_group, mock_is_leader_on_local_site, mock_get_patch_data, consts.SYSTEM_PEER_AVAILABILITY_STATE_UNAVAILABLE) response = self.app.patch_json( f"{FAKE_URL}/{subcloud.id}", headers=FAKE_HEADERS, params=patch_data ) self.assertEqual(response.status_int, 200) self.mock_rpc_client().update_subcloud.assert_called_once_with( mock.ANY, subcloud.id, management_state=None, description=None, location=None, group_id=None, data_install=None, force=None, peer_group_id=None, bootstrap_values=None, bootstrap_address=patch_data["bootstrap_address"], deploy_status=None) self.assertEqual(consts.ASSOCIATION_SYNC_STATUS_OUT_OF_SYNC, db_api.peer_group_association_get( self.ctx, association.id).sync_status) def test_get_config_file_path(self): bootstrap_file = psd_common.get_config_file_path("subcloud1") install_values = psd_common.get_config_file_path( "subcloud1", "install_values" ) deploy_config = psd_common.get_config_file_path( "subcloud1", consts.DEPLOY_CONFIG ) self.assertEqual( bootstrap_file, f"{dccommon_consts.ANSIBLE_OVERRIDES_PATH}/subcloud1.yml" ) self.assertEqual( install_values, f"{dccommon_consts.ANSIBLE_OVERRIDES_PATH}/subcloud1/install_values.yml", ) self.assertEqual( deploy_config, f"{dccommon_consts.ANSIBLE_OVERRIDES_PATH}/subcloud1_deploy_config.yml", ) def test_format_ip_address(self): fake_payload = {} good_values = { "10.10.10.3": "10.10.10.3", "2620:10a:a001:a103::1135": "2620:10a:a001:a103::1135", "2620:10A:A001:A103::1135": "2620:10a:a001:a103::1135", "2620:010a:a001:a103::1135": "2620:10a:a001:a103::1135", "2620:10a:a001:a103:0000::1135": "2620:10a:a001:a103::1135", } for k, v in good_values.items(): fake_payload["bootstrap-address"] = k psd_common.format_ip_address(fake_payload) self.assertEqual(fake_payload["bootstrap-address"], v) fake_payload[consts.INSTALL_VALUES] = {} for k, v in good_values.items(): fake_payload[consts.INSTALL_VALUES]['bmc_address'] = k psd_common.format_ip_address(fake_payload) self.assertEqual(fake_payload[consts.INSTALL_VALUES]['bmc_address'], v) fake_payload['othervalues1'] = 'othervalues1' fake_payload[consts.INSTALL_VALUES]['othervalues2'] = 'othervalues2' psd_common.format_ip_address(fake_payload) self.assertEqual(fake_payload['othervalues1'], 'othervalues1') self.assertEqual(fake_payload[consts.INSTALL_VALUES]['othervalues2'], 'othervalues2') def test_get_subcloud_db_install_values(self): install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) encoded_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) install_data["bmc_password"] = encoded_password test_subcloud = copy.copy(FAKE_SUBCLOUD_DATA) subcloud_info = Subcloud(test_subcloud, False) subcloud_info.data_install = json.dumps(install_data) actual_result = psd_common.get_subcloud_db_install_values(subcloud_info) self.assertEqual( json.loads(json.dumps(install_data)), json.loads(json.dumps(actual_result)), ) @mock.patch.object(keyring, "get_password") def test_get_subcloud_db_install_values_without_bmc_password(self, mock_keyring): install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) subcloud = fake_subcloud.create_fake_subcloud( self.ctx, data_install=json.dumps(install_data)) six.assertRaisesRegex(self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + '/' + str(subcloud.id) + '/redeploy', headers=FAKE_HEADERS) @mock.patch.object(psd_common, 'upload_config_file') @mock.patch.object(psd_common.PatchingClient, 'query') @mock.patch.object(os.path, 'isdir') @mock.patch.object(os, 'listdir') @mock.patch.object(cutils, 'get_vault_load_files') @mock.patch.object(psd_common, 'validate_k8s_version') @mock.patch.object(psd_common, 'validate_subcloud_config') @mock.patch.object(psd_common, 'validate_bootstrap_values') def test_redeploy_subcloud( self, mock_validate_bootstrap_values, mock_validate_subcloud_config, mock_validate_k8s_version, mock_get_vault_load_files, mock_os_listdir, mock_os_isdir, mock_query, mock_upload_config_file, ): fake_bmc_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) fake_sysadmin_password = base64.b64encode( "sysadmin_password".encode("utf-8") ).decode("utf-8") install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) install_data.pop("software_version") bootstrap_data = copy.copy(fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA) config_data = {"deploy_config": "deploy config values"} redeploy_data = { **install_data, **bootstrap_data, **config_data, "sysadmin_password": fake_sysadmin_password, "bmc_password": fake_bmc_password, } subcloud = fake_subcloud.create_fake_subcloud( self.ctx, name=bootstrap_data["name"] ) mock_query.return_value = {} mock_get_vault_load_files.return_value = ("iso_file_path", "sig_file_path") mock_os_isdir.return_value = True mock_upload_config_file.return_value = True mock_os_listdir.return_value = [ "deploy_chart_fake.tgz", "deploy_overrides_fake.yaml", "deploy_playbook_fake.yaml", ] upload_files = [ ( "install_values", "install_fake_filename", json.dumps(install_data).encode("utf-8"), ), ( "bootstrap_values", "bootstrap_fake_filename", json.dumps(bootstrap_data).encode("utf-8"), ), ( "deploy_config", "config_fake_filename", json.dumps(config_data).encode("utf-8"), ), ] response = self.app.patch( FAKE_URL + "/" + str(subcloud.id) + "/redeploy", headers=FAKE_HEADERS, params=redeploy_data, upload_files=upload_files, ) mock_validate_bootstrap_values.assert_called_once() mock_validate_subcloud_config.assert_called_once() mock_validate_k8s_version.assert_called_once() self.mock_rpc_client().redeploy_subcloud.assert_called_once_with( mock.ANY, subcloud.id, mock.ANY ) self.assertEqual(response.status_int, 200) self.assertEqual(SW_VERSION, response.json["software-version"]) @mock.patch.object(cutils, "load_yaml_file") @mock.patch.object(psd_common.PatchingClient, "query") @mock.patch.object(os.path, "exists") @mock.patch.object(os.path, "isdir") @mock.patch.object(os, "listdir") @mock.patch.object(cutils, "get_vault_load_files") @mock.patch.object(psd_common, "validate_k8s_version") def test_redeploy_subcloud_no_request_data( self, mock_validate_k8s_version, mock_get_vault_load_files, mock_os_listdir, mock_os_isdir, mock_path_exists, mock_query, mock_load_yaml, ): fake_bmc_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) fake_sysadmin_password = base64.b64encode( "sysadmin_password".encode("utf-8") ).decode("utf-8") install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) install_data.pop("software_version") install_data["bmc_password"] = fake_bmc_password redeploy_data = {"sysadmin_password": fake_sysadmin_password} subcloud = fake_subcloud.create_fake_subcloud( self.ctx, name=fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA["name"], data_install=json.dumps(install_data), ) config_file = psd_common.get_config_file_path( subcloud.name, consts.DEPLOY_CONFIG ) mock_query.return_value = {} mock_get_vault_load_files.return_value = ("iso_file_path", "sig_file_path") mock_os_isdir.return_value = True mock_os_listdir.return_value = [ "deploy_chart_fake.tgz", "deploy_overrides_fake.yaml", "deploy_playbook_fake.yaml", ] mock_path_exists.side_effect = lambda x: True if x == config_file else False mock_load_yaml.return_value = {"software_version": SW_VERSION} response = self.app.patch( FAKE_URL + "/" + str(subcloud.id) + "/redeploy", headers=FAKE_HEADERS, params=redeploy_data, ) mock_validate_k8s_version.assert_called_once() self.mock_rpc_client().redeploy_subcloud.assert_called_once_with( mock.ANY, subcloud.id, mock.ANY ) self.assertEqual(response.status_int, 200) self.assertEqual(SW_VERSION, response.json["software-version"]) @mock.patch.object(psd_common, "upload_config_file") @mock.patch.object(psd_common.PatchingClient, "query") @mock.patch.object(os.path, "isdir") @mock.patch.object(os, "listdir") @mock.patch.object(cutils, "get_vault_load_files") @mock.patch.object(psd_common, "validate_k8s_version") @mock.patch.object(psd_common, "validate_subcloud_config") @mock.patch.object(psd_common, "validate_bootstrap_values") def test_redeploy_subcloud_with_release_version( self, mock_validate_bootstrap_values, mock_validate_subcloud_config, mock_validate_k8s_version, mock_get_vault_load_files, mock_os_listdir, mock_os_isdir, mock_query, mock_upload_config_file, ): fake_bmc_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) fake_sysadmin_password = base64.b64encode( "sysadmin_password".encode("utf-8") ).decode("utf-8") install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) install_data.pop("software_version") bootstrap_data = copy.copy(fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA) config_data = {"deploy_config": "deploy config values"} redeploy_data = { **install_data, **bootstrap_data, **config_data, "sysadmin_password": fake_sysadmin_password, "bmc_password": fake_bmc_password, "release": fake_subcloud.FAKE_SOFTWARE_VERSION, } subcloud = fake_subcloud.create_fake_subcloud( self.ctx, name=bootstrap_data["name"], software_version=SW_VERSION ) mock_query.return_value = {} mock_get_vault_load_files.return_value = ("iso_file_path", "sig_file_path") mock_os_isdir.return_value = True mock_upload_config_file.return_value = True mock_os_listdir.return_value = [ "deploy_chart_fake.tgz", "deploy_overrides_fake.yaml", "deploy_playbook_fake.yaml", ] upload_files = [ ( "install_values", "install_fake_filename", json.dumps(install_data).encode("utf-8"), ), ( "bootstrap_values", "bootstrap_fake_filename", json.dumps(bootstrap_data).encode("utf-8"), ), ( "deploy_config", "config_fake_filename", json.dumps(config_data).encode("utf-8"), ), ] with mock.patch('builtins.open', mock.mock_open( read_data=fake_subcloud.FAKE_UPGRADES_METADATA )): response = self.app.patch( FAKE_URL + '/' + str(subcloud.id) + '/redeploy', headers=FAKE_HEADERS, params=redeploy_data, upload_files=upload_files) mock_validate_bootstrap_values.assert_called_once() mock_validate_subcloud_config.assert_called_once() mock_validate_k8s_version.assert_called_once() self.mock_rpc_client().redeploy_subcloud.assert_called_once_with( mock.ANY, subcloud.id, mock.ANY ) self.assertEqual(response.status_int, 200) self.assertEqual( fake_subcloud.FAKE_SOFTWARE_VERSION, response.json["software-version"] ) @mock.patch.object(cutils, "load_yaml_file") @mock.patch.object(psd_common.PatchingClient, "query") @mock.patch.object(os.path, "exists") @mock.patch.object(os.path, "isdir") @mock.patch.object(os, "listdir") @mock.patch.object(cutils, "get_vault_load_files") def test_redeploy_subcloud_no_request_body( self, mock_get_vault_load_files, mock_os_listdir, mock_os_isdir, mock_path_exists, mock_query, mock_load_yaml, ): fake_bmc_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) install_data.pop("software_version") install_data["bmc_password"] = fake_bmc_password subcloud = fake_subcloud.create_fake_subcloud( self.ctx, name=fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA["name"], data_install=json.dumps(install_data), ) config_file = psd_common.get_config_file_path( subcloud.name, consts.DEPLOY_CONFIG ) mock_query.return_value = {} mock_get_vault_load_files.return_value = ("iso_file_path", "sig_file_path") mock_os_isdir.return_value = True mock_os_listdir.return_value = [ "deploy_chart_fake.tgz", "deploy_overrides_fake.yaml", "deploy_playbook_fake.yaml", ] mock_path_exists.side_effect = lambda x: True if x == config_file else False mock_load_yaml.return_value = {"software_version": SW_VERSION} six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/redeploy", headers=FAKE_HEADERS, params={}, ) def test_redeploy_online_subcloud(self): subcloud = fake_subcloud.create_fake_subcloud( self.ctx, name=fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA["name"] ) db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, ) six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/redeploy", headers=FAKE_HEADERS, params={}, ) self.mock_rpc_client().redeploy_subcloud.assert_not_called() def test_redeploy_managed_subcloud(self): subcloud = fake_subcloud.create_fake_subcloud( self.ctx, name=fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA["name"] ) db_api.subcloud_update( self.ctx, subcloud.id, management_state=dccommon_consts.MANAGEMENT_MANAGED, ) six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/redeploy", headers=FAKE_HEADERS, params={}, ) self.mock_rpc_client().redeploy_subcloud.assert_not_called() @mock.patch.object(cutils, "load_yaml_file") @mock.patch.object(psd_common.PatchingClient, "query") @mock.patch.object(os.path, "exists") @mock.patch.object(os.path, "isdir") @mock.patch.object(os, "listdir") @mock.patch.object(cutils, "get_vault_load_files") @mock.patch.object(psd_common, "validate_k8s_version") def test_redeploy_subcloud_missing_required_value( self, mock_validate_k8s_version, mock_get_vault_load_files, mock_os_listdir, mock_os_isdir, mock_path_exists, mock_query, mock_load_yaml, ): fake_bmc_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) fake_sysadmin_password = base64.b64encode( "sysadmin_password".encode("utf-8") ).decode("utf-8") install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) install_data.pop("software_version") install_data["bmc_password"] = fake_bmc_password subcloud = fake_subcloud.create_fake_subcloud( self.ctx, name=fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA["name"], data_install=json.dumps(install_data), ) config_file = psd_common.get_config_file_path( subcloud.name, consts.DEPLOY_CONFIG ) mock_query.return_value = {} mock_get_vault_load_files.return_value = ("iso_file_path", "sig_file_path") mock_os_isdir.return_value = True mock_os_listdir.return_value = [ "deploy_chart_fake.tgz", "deploy_overrides_fake.yaml", "deploy_playbook_fake.yaml", ] mock_path_exists.side_effect = lambda x: True if x == config_file else False mock_load_yaml.return_value = {"software_version": SW_VERSION} for k in [ "name", "system_mode", "external_oam_subnet", "external_oam_gateway_address", "external_oam_floating_address", "sysadmin_password", ]: bootstrap_values = copy.copy(fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA) redeploy_data = { **bootstrap_values, "sysadmin_password": fake_sysadmin_password, } del redeploy_data[k] upload_files = [ ( "bootstrap_values", "bootstrap_fake_filename", json.dumps(redeploy_data).encode("utf-8"), ) ] six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/redeploy", headers=FAKE_HEADERS, params=redeploy_data, upload_files=upload_files, ) @mock.patch.object(psd_common, "upload_config_file") @mock.patch.object(psd_common.PatchingClient, "query") @mock.patch.object(os.path, "isdir") @mock.patch.object(os, "listdir") @mock.patch.object(cutils, "get_vault_load_files") @mock.patch.object(psd_common, "validate_k8s_version") @mock.patch.object(psd_common, "validate_subcloud_config") @mock.patch.object(psd_common, "validate_bootstrap_values") def test_redeploy_subcloud_missing_stored_values( self, mock_validate_bootstrap_values, mock_validate_subcloud_config, mock_validate_k8s_version, mock_get_vault_load_files, mock_os_listdir, mock_os_isdir, mock_query, mock_upload_config_values, ): fake_bmc_password = base64.b64encode("bmc_password".encode("utf-8")).decode( "utf-8" ) fake_sysadmin_password = base64.b64encode( "sysadmin_password".encode("utf-8") ).decode("utf-8") install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES) install_data.pop("software_version") bootstrap_data = copy.copy(fake_subcloud.FAKE_BOOTSTRAP_FILE_DATA) config_data = {"deploy_config": "deploy config values"} for k in [ "management_subnet", "management_start_address", "management_end_address", "management_gateway_address", "systemcontroller_gateway_address", ]: del bootstrap_data[k] redeploy_data = { **install_data, **bootstrap_data, **config_data, "sysadmin_password": fake_sysadmin_password, "bmc_password": fake_bmc_password, } subcloud = fake_subcloud.create_fake_subcloud( self.ctx, name=bootstrap_data["name"] ) mock_query.return_value = {} mock_get_vault_load_files.return_value = ("iso_file_path", "sig_file_path") mock_os_isdir.return_value = True mock_upload_config_values.return_value = True mock_os_listdir.return_value = [ "deploy_chart_fake.tgz", "deploy_overrides_fake.yaml", "deploy_playbook_fake.yaml", ] upload_files = [ ( "install_values", "install_fake_filename", json.dumps(install_data).encode("utf-8"), ), ( "bootstrap_values", "bootstrap_fake_filename", json.dumps(bootstrap_data).encode("utf-8"), ), ( "deploy_config", "config_fake_filename", json.dumps(config_data).encode("utf-8"), ), ] response = self.app.patch( FAKE_URL + "/" + str(subcloud.id) + "/redeploy", headers=FAKE_HEADERS, params=redeploy_data, upload_files=upload_files, ) mock_validate_bootstrap_values.assert_called_once() mock_validate_subcloud_config.assert_called_once() mock_validate_k8s_version.assert_called_once() self.mock_rpc_client().redeploy_subcloud.assert_called_once_with( mock.ANY, subcloud.id, mock.ANY ) self.assertEqual(response.status_int, 200) self.assertEqual(SW_VERSION, response.json["software-version"]) @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(prestage, "_get_prestage_subcloud_info") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_validate_detailed( self, mock_get_prestage_payload, mock_prestage_subcloud_info, mock_controller_upgrade, ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, management_state=dccommon_consts.MANAGEMENT_MANAGED, ) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password, "force": False} mock_controller_upgrade.return_value = list() mock_prestage_subcloud_info.return_value = ( consts.SYSTEM_MODE_SIMPLEX, health_report_no_alarm, OAM_FLOATING_IP, ) self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) self.mock_rpc_client().prestage_subcloud.assert_called_once_with( mock.ANY, mock.ANY ) self.assertEqual(response.status_int, 200) @mock.patch.object(cutils, "get_systemcontroller_installed_loads") @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_invalid_release( self, mock_get_prestage_payload, mock_controller_upgrade, mock_installed_loads, ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, management_state=dccommon_consts.MANAGEMENT_MANAGED, ) fake_release = "21.12" mock_installed_loads.return_value = ["22.12"] fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = { "sysadmin_password": fake_password, "force": False, "release": fake_release, } mock_controller_upgrade.return_value = list() self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") @mock.patch.object(prestage, "_get_system_controller_upgrades") def test_prestage_subcloud_unmanaged( self, mock_controller_upgrade, mock_get_prestage_payload ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, management_state=dccommon_consts.MANAGEMENT_UNMANAGED, ) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password} mock_controller_upgrade.return_value = list() self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") @mock.patch.object(prestage, "_get_system_controller_upgrades") def test_prestage_subcloud_offline( self, mock_controller_upgrade, mock_get_prestage_payload ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_OFFLINE, management_state=dccommon_consts.MANAGEMENT_MANAGED, ) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password} mock_controller_upgrade.return_value = list() self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) def test_prestage_subcloud_backup_in_progress(self): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, deploy_status=consts.DEPLOY_STATE_DONE, management_state=dccommon_consts.MANAGEMENT_MANAGED, backup_status=consts.BACKUP_STATE_IN_PROGRESS, ) self.assertRaises( exceptions.PrestagePreCheckFailedException, prestage.initial_subcloud_validate, subcloud, [fake_subcloud.FAKE_SOFTWARE_VERSION], fake_subcloud.FAKE_SOFTWARE_VERSION, ) @mock.patch.object(cutils, "get_systemcontroller_installed_loads") @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(prestage, "_get_prestage_subcloud_info") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_duplex( self, mock_get_prestage_payload, mock_prestage_subcloud_info, mock_controller_upgrade, mock_installed_loads, ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, management_state=dccommon_consts.MANAGEMENT_MANAGED, ) fake_release = "21.12" mock_installed_loads.return_value = [fake_release] fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password, "force": False} mock_controller_upgrade.return_value = list() mock_prestage_subcloud_info.return_value = ( consts.SYSTEM_MODE_DUPLEX, health_report_no_alarm, OAM_FLOATING_IP, ) self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(prestage, "_get_prestage_subcloud_info") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_non_mgmt_alarm( self, mock_get_prestage_payload, mock_prestage_subcloud_info, mock_controller_upgrade, ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, management_state=dccommon_consts.MANAGEMENT_MANAGED, ) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password, "force": False} mock_controller_upgrade.return_value = list() mock_prestage_subcloud_info.return_value = ( consts.SYSTEM_MODE_SIMPLEX, health_report_no_mgmt_alarm, OAM_FLOATING_IP, ) self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) self.mock_rpc_client().prestage_subcloud.assert_called_once_with( mock.ANY, mock.ANY ) self.assertEqual(response.status_int, 200) @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(prestage, "_get_prestage_subcloud_info") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_mgmt_alarm( self, mock_get_prestage_payload, mock_prestage_subcloud_info, mock_controller_upgrade, ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, management_state=dccommon_consts.MANAGEMENT_MANAGED, ) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password, "force": False} mock_controller_upgrade.return_value = list() mock_prestage_subcloud_info.return_value = ( consts.SYSTEM_MODE_SIMPLEX, health_report_mgmt_alarm, OAM_FLOATING_IP, ) self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(prestage, "_get_prestage_subcloud_info") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_mgmt_alarm_force( self, mock_get_prestage_payload, mock_prestage_subcloud_info, mock_controller_upgrade, ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, management_state=dccommon_consts.MANAGEMENT_MANAGED, ) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password, "force": True} mock_controller_upgrade.return_value = list() mock_prestage_subcloud_info.return_value = ( consts.SYSTEM_MODE_SIMPLEX, health_report_mgmt_alarm, OAM_FLOATING_IP, ) self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data response = self.app.patch_json( FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) self.mock_rpc_client().prestage_subcloud.assert_called_once_with( mock.ANY, mock.ANY ) self.assertEqual(response.status_int, 200) @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(prestage, "_get_prestage_subcloud_info") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_not_allowed_state( self, mock_get_prestage_payload, mock_prestage_subcloud_info, mock_controller_upgrade, ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) subcloud = db_api.subcloud_update( self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE, management_state=dccommon_consts.MANAGEMENT_MANAGED, deploy_status="NotAllowedState", ) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password, "force": False} mock_controller_upgrade.return_value = list() mock_prestage_subcloud_info.return_value = ( consts.SYSTEM_MODE_SIMPLEX, health_report_no_alarm, OAM_FLOATING_IP, ) self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_controller_upgrading( self, mock_get_prestage_payload, mock_controller_upgrade ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) fake_password = (base64.b64encode("testpass".encode("utf-8"))).decode( "ascii" ) data = {"sysadmin_password": fake_password} mock_controller_upgrade.return_value = list("upgrade") self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_no_password( self, mock_get_prestage_payload, mock_controller_upgrade ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = {} mock_controller_upgrade.return_value = list() self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) @mock.patch.object(prestage, "_get_system_controller_upgrades") @mock.patch.object(subclouds.SubcloudsController, "_get_prestage_payload") def test_prestage_subcloud_password_not_encoded( self, mock_get_prestage_payload, mock_controller_upgrade ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) data = {"sysadmin_password": "notencoded"} mock_controller_upgrade.return_value = list() self.mock_rpc_client().prestage_subcloud.return_value = True mock_get_prestage_payload.return_value = data six.assertRaisesRegex( self, webtest.app.AppError, "400 *", self.app.patch_json, FAKE_URL + "/" + str(subcloud.id) + "/prestage", headers=FAKE_HEADERS, params=data, ) def test_get_management_subnet(self): payload = {"management_subnet": "192.168.204.0/24"} self.assertEqual( cutils.get_management_subnet(payload), payload["management_subnet"] ) def test_get_management_subnet_return_admin(self): payload = { "admin_subnet": "192.168.205.0/24", "management_subnet": "192.168.204.0/24", } self.assertEqual( cutils.get_management_subnet(payload), payload["admin_subnet"] ) def test_get_management_start_address(self): payload = {"management_start_address": "192.168.204.2"} self.assertEqual( cutils.get_management_start_address(payload), payload["management_start_address"], ) def test_get_management_start_address_return_admin(self): payload = { "admin_start_address": "192.168.205.2", "management_start_address": "192.168.204.2", } self.assertEqual( cutils.get_management_start_address(payload), payload["admin_start_address"], ) def test_get_management_end_address(self): payload = {"management_end_address": "192.168.204.50"} self.assertEqual( cutils.get_management_end_address(payload), payload["management_end_address"], ) def test_get_management_end_address_return_admin(self): payload = { "admin_end_address": "192.168.205.50", "management_end_address": "192.168.204.50", } self.assertEqual( cutils.get_management_end_address(payload), payload["admin_end_address"] ) def test_get_management_gateway_address(self): payload = {"management_gateway_address": "192.168.204.1"} self.assertEqual( cutils.get_management_gateway_address(payload), payload["management_gateway_address"], ) def test_get_management_gateway_address_return_admin(self): payload = { "admin_gateway_address": "192.168.205.1", "management_gateway_address": "192.168.204.1", } self.assertEqual( cutils.get_management_gateway_address(payload), payload["admin_gateway_address"], ) def test_validate_admin_config_subnet_small(self): admin_subnet = "192.168.205.0/32" admin_start_address = "192.168.205.2" admin_end_address = "192.168.205.50" admin_gateway_address = "192.168.205.1" six.assertRaisesRegex( self, Exception, "Subnet too small*", psd_common.validate_admin_network_config, admin_subnet, admin_start_address, admin_end_address, admin_gateway_address, existing_networks=None, operation=None, ) def test_validate_admin_config_start_address_outOfSubnet(self): admin_subnet = "192.168.205.0/28" admin_start_address = "192.168.205.200" admin_end_address = "192.168.205.50" admin_gateway_address = "192.168.205.1" six.assertRaisesRegex( self, Exception, "Address must be in subnet*", psd_common.validate_admin_network_config, admin_subnet, admin_start_address, admin_end_address, admin_gateway_address, existing_networks=None, operation=None, ) def test_validate_admin_config_end_address_outOfSubnet(self): admin_subnet = "192.168.205.0/28" admin_start_address = "192.168.205.1" admin_end_address = "192.168.205.50" admin_gateway_address = "192.168.205.1" six.assertRaisesRegex( self, Exception, "Address must be in subnet*", psd_common.validate_admin_network_config, admin_subnet, admin_start_address, admin_end_address, admin_gateway_address, existing_networks=None, operation=None, ) def test_validate_admin_config_gateway_address_outOfSubnet(self): admin_subnet = "192.168.205.0/28" admin_start_address = "192.168.205.1" admin_end_address = "192.168.205.12" admin_gateway_address = "192.168.205.50" six.assertRaisesRegex( self, Exception, "Address must be in subnet*", psd_common.validate_admin_network_config, admin_subnet, admin_start_address, admin_end_address, admin_gateway_address, existing_networks=None, operation=None, ) @mock.patch.object(subclouds, "OpenStackDriver") @mock.patch.object(subclouds.vim, "VimClient") def test_check_other_strategy_exists(self, mock_vim_client, mock_os_driver): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) fake_strategy.create_fake_strategy_step( self.ctx, subcloud_id=subcloud.id, state=consts.STRATEGY_STATE_INITIAL) sc = subclouds.SubcloudsController() result = sc._check_existing_vim_strategy(self.ctx, subcloud) self.assertTrue(result) mock_os_driver.assert_not_called() mock_vim_client.assert_not_called() @mock.patch.object(subclouds, "db_api") @mock.patch.object(subclouds, "OpenStackDriver") @mock.patch.object(subclouds.vim, "VimClient") def test_no_vim_strategy(self, mock_vim_client, mock_os_driver, mock_db_api): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) sc = subclouds.SubcloudsController() result = sc._check_existing_vim_strategy(self.ctx, subcloud) mock_vim_client_instance = mock_vim_client.return_value mock_vim_client_instance.get_strategy.return_value = None mock_db_api.strategy_step_get.side_effect = \ exceptions.StrategyStepNotFound sc = subclouds.SubcloudsController() result = sc._check_existing_vim_strategy(self.ctx, subcloud) mock_os_driver.assert_called_with(region_name=subcloud.region_name, region_clients=None) mock_vim_client.assert_called_with( subcloud.region_name, mock_os_driver.return_value.keystone_client.session) self.assertFalse(result) @mock.patch.object(subclouds, "db_api") @mock.patch.object(subclouds, "OpenStackDriver") @mock.patch.object(subclouds.vim, "VimClient") def test_check_system_config_update_strategy_exists( self, mock_vim_client, mock_os_driver, mock_db_api ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) fake_strategy = FakeVimStrategy(state=subclouds.vim.STATE_APPLYING) mock_db_api.strategy_step_get.side_effect = \ exceptions.StrategyStepNotFound mock_vim_client_instance = mock_vim_client.return_value mock_vim_client_instance.get_strategy.return_value = fake_strategy sc = subclouds.SubcloudsController() result = sc._check_existing_vim_strategy(self.ctx, subcloud) mock_os_driver.assert_called_with(region_name=subcloud.region_name, region_clients=None) mock_vim_client.assert_called_with( subcloud.region_name, mock_os_driver.return_value.keystone_client.session) self.assertTrue(result) @mock.patch.object(subclouds, "db_api") @mock.patch.object(subclouds, "OpenStackDriver") @mock.patch.object(subclouds.vim, "VimClient") def test_check_system_config_update_strategy_applied( self, mock_vim_client, mock_os_driver, mock_db_api ): subcloud = fake_subcloud.create_fake_subcloud(self.ctx) fake_strategy = FakeVimStrategy(state=subclouds.vim.STATE_APPLIED) mock_db_api.strategy_step_get.side_effect = \ exceptions.StrategyStepNotFound mock_vim_client_instance = mock_vim_client.return_value mock_vim_client_instance.get_strategy.return_value = fake_strategy sc = subclouds.SubcloudsController() result = sc._check_existing_vim_strategy(self.ctx, subcloud) mock_os_driver.assert_called_with(region_name=subcloud.region_name, region_clients=None) mock_vim_client.assert_called_with( subcloud.region_name, mock_os_driver.return_value.keystone_client.session) self.assertFalse(result)