starlingx
/
distcloud
Code Issues Proposed changes

Improve unit test coverage for dcmanager's APIs (subclouds) 

Improves unit test coverage for dcmanager's subclouds API
from 55% to 90%.

Test plan:
    All of the tests were created taking into account the
output of 'tox -c tox.ini -e cover' command

Story: 2007082
Task: 49725

Change-Id: If31005a8f3420e94dd17f15bdac97af5253e8d5a
Signed-off-by: Raphael Lima <Raphael.Lima@windriver.com>
This commit is contained in:
Raphael Lima 2024-03-15 10:49:27 -03:00
parent b0bef0dbe8
commit e98e5443fc
11 changed files with 2633 additions and 2995 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
distributedcloud/dccommon/tests/unit/test_utils.py
View File

@ -1,5 +1,5 @@
#
# Copyright (c) 2022-2023 Wind River Systems, Inc.
# Copyright (c) 2022-2024 Wind River Systems, Inc.
#
# SPDX-License-Identifier: Apache-2.0
#
@ -29,11 +29,10 @@ class TestUtils(base.DCCommonTestCase):
def test_exec_playbook_timeout(self):
testscript = ['dccommon/tests/unit/test_utils_script.sh', '30']
ansible = utils.AnsiblePlaybook(FAKE_SUBCLOUD_NAME)
self.assertRaises(PlaybookExecutionTimeout,
ansible.run_playbook,
FAKE_LOG_FILE,
testscript,
timeout=2)
self.assertRaises(
PlaybookExecutionTimeout, ansible.run_playbook, FAKE_LOG_FILE,
testscript, timeout=2
)
def test_exec_playbook_timeout_requires_kill(self):
# This option ignores a regular TERM signal, and requires a
@ -41,8 +40,7 @@ class TestUtils(base.DCCommonTestCase):
# a hung process
script = ['dccommon/tests/unit/test_utils_script.sh', '30', 'TERM']
ansible = utils.AnsiblePlaybook(FAKE_SUBCLOUD_NAME)
self.assertRaises(PlaybookExecutionTimeout,
ansible.run_playbook,
FAKE_LOG_FILE,
script,
timeout=2)
self.assertRaises(
PlaybookExecutionTimeout, ansible.run_playbook, FAKE_LOG_FILE, script,
timeout=2
)

2
distributedcloud/dcmanager/common/phased_subcloud_deploy.py
View File

@ -601,7 +601,7 @@ def validate_install_values(payload, subcloud=None):
gateway_ip = netaddr.IPAddress(install_values['nexthop_gateway'])
except netaddr.AddrFormatError as e:
LOG.exception(e)
pecan.abort(400, _("nexthop_gateway address invalid: %s") % e)
pecan.abort(400, _("nexthop_gateway invalid: %s") % e)
if gateway_ip.version != ip_version:
pecan.abort(400, _("nexthop_gateway and bootstrap_address "
"must be the same IP version"))

2
distributedcloud/dcmanager/common/prestage.py
View File

@ -84,7 +84,7 @@ def global_prestage_validate(payload):
if is_system_controller_upgrading():
raise exceptions.PrestagePreCheckFailedException(
subcloud=dccommon_consts.SYSTEM_CONTROLLER_NAME,
details='Prestage operations not allowed while system'
details='Prestage operations are not allowed while system'
' controller upgrade is in progress.')
if ('sysadmin_password' not in payload

14
distributedcloud/dcmanager/tests/base.py
View File

@ -207,11 +207,25 @@ class DCManagerTestCase(base.BaseTestCase):
self.mock_sysinv_client = mock_patch_object.start()
self.addCleanup(mock_patch_object.stop)
def _mock_fm_client(self, target):
"""Mock the target's FmClient"""
mock_patch_object = mock.patch.object(target, 'FmClient')
self.mock_fm_client = mock_patch_object.start()
self.addCleanup(mock_patch_object.stop)
def _mock_read_from_cache(self, target):
mock_patch = mock.patch.object(target, '_read_from_cache')
self.mock_read_from_cache = mock_patch.start()
self.addCleanup(mock_patch.stop)
def _mock_vim_client(self, target):
"""Mock the target's VimClient"""
mock_patch_object = mock.patch.object(target, 'VimClient')
self.mock_vim_client = mock_patch_object.start()
self.addCleanup(mock_patch_object.stop)
def _mock_get_network_address_pool(self):
"""Mock phased subcloud deploy's get_network_address_pool"""

57
distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_phased_subcloud_deploy.py
View File

@ -21,8 +21,7 @@ from dcmanager.db import api as db_api
from dcmanager.tests.unit.api.test_root_controller import DCManagerApiTest
from dcmanager.tests.unit.api.v1.controllers.test_subclouds import \
FakeAddressPool
from dcmanager.tests.unit.api.v1.controllers.test_subclouds import \
TestSubcloudPost
from dcmanager.tests.unit.api.v1.controllers.test_subclouds import SubcloudAPIMixin
from dcmanager.tests.unit.common import fake_subcloud
from dcmanager.tests.unit.manager.test_system_peer_manager import \
TestSystemPeerManager
@ -85,24 +84,23 @@ class TestPhasedSubcloudDeployController(BaseTestPhasedSubcloudDeployController)
self.assertEqual(response.text, "null")
# Apply the TestSubcloudPost parameter validation tests to the subcloud deploy
# Apply the TestSubcloudsPost parameter validation tests to the subcloud deploy
# add endpoint as it uses the same parameter validation functions
class TestPhasedSubcloudDeployPost(
TestSubcloudPost, BaseTestPhasedSubcloudDeployController
):
class TestPhasedSubcloudDeployPost(BaseTestPhasedSubcloudDeployController):
"""Test class for post requests"""
API_PREFIX = FAKE_URL
RESULT_KEY = "phased-subcloud-deploy"
def setUp(self):
super().setUp()
self.method = self.app.post
self.params = self.get_post_params()
self.upload_files = self.get_post_upload_files()
self.params = copy.copy(fake_subcloud.FAKE_BOOTSTRAP_VALUE)
self.upload_files = SubcloudAPIMixin.get_post_upload_files(SubcloudAPIMixin)
self._mock_sysinv_client(psd_common)
self.mock_sysinv_client().get_management_address_pool.return_value = \
FakeAddressPool("192.168.204.0", 24, "192.168.204.2", "192.168.204.100")
self.mock_rpc_client().subcloud_deploy_create.side_effect = \
self.subcloud_deploy_create
@ -110,8 +108,29 @@ class TestPhasedSubcloudDeployPost(
subcloud = db_api.subcloud_get(context, subcloud_id)
return db_api.subcloud_db_model_to_dict(subcloud)
def test_post_create_fails_without_bootstrap_address(self):
"""Test post create fails without bootstrap address"""
def test_post_succeeds(self):
"""Test post succeeds"""
response = self._send_request()
self._assert_response(response)
self.mock_rpc_client().subcloud_deploy_create.assert_called_once()
def test_post_fails_without_payload(self):
"""Test post fails without payload"""
self.params = {}
self.upload_files = None
response = self._send_request()
self._assert_pecan_and_response(
response, http.client.BAD_REQUEST,
"Missing required parameter(s): bootstrap_values, bootstrap-address"
)
def test_post_fails_without_bootstrap_address(self):
"""Test post fails without bootstrap address"""
del self.params["bootstrap-address"]
@ -123,8 +142,8 @@ class TestPhasedSubcloudDeployPost(
)
self.mock_rpc_client().subcloud_deploy_create.assert_not_called()
def test_post_create_fails_without_bootstrap_values(self):
"""Test post create fails without bootstrap values"""
def test_post_fails_without_bootstrap_values(self):
"""Test post fails without bootstrap values"""
self.upload_files = None
@ -136,8 +155,8 @@ class TestPhasedSubcloudDeployPost(
)
self.mock_rpc_client().subcloud_deploy_create.assert_not_called()
def test_post_create_fails_with_rpc_client_remote_error(self):
"""Test post create fails with rpc client remote error"""
def test_post_fails_with_rpc_client_remote_error(self):
"""Test post fails with rpc client remote error"""
self.mock_rpc_client().subcloud_deploy_create.side_effect = \
RemoteError("msg", "value")
@ -149,8 +168,8 @@ class TestPhasedSubcloudDeployPost(
)
self.mock_rpc_client().subcloud_deploy_create.assert_called_once()
def test_post_create_fails_with_rpc_client_generic_exception(self):
"""Test post create fails with rpc client generic exception"""
def test_post_fails_with_rpc_client_generic_exception(self):
"""Test post fails with rpc client generic exception"""
self.mock_rpc_client().subcloud_deploy_create.side_effect = Exception()

23
distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_subcloud_deploy.py
View File

@ -21,10 +21,8 @@ import mock
from tsconfig.tsconfig import SW_VERSION
import webtest
from dccommon import consts as dccommon_consts
from dcmanager.api.controllers.v1 import subcloud_deploy
from dcmanager.common import consts
from dcmanager.common import phased_subcloud_deploy as psd_common
from dcmanager.common import utils as dutils
from dcmanager.tests.base import FakeException
from dcmanager.tests.unit.api.test_root_controller import DCManagerApiTest
@ -360,27 +358,6 @@ class TestSubcloudDeployGet(BaseTestSubcloudDeployController):
None, response.json["subcloud_deploy"][consts.DEPLOY_PRESTAGE]
)
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", consts.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"
)
class TestSubcloudDeployDelete(BaseTestSubcloudDeployController):
"""Test class for delete requests"""

4
distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_subcloud_peer_group.py
View File

@ -18,14 +18,14 @@ from dcmanager.tests.base import FakeException
from dcmanager.tests.unit.api.test_root_controller import DCManagerApiTest
from dcmanager.tests.unit.api.v1.controllers.mixins import APIMixin
from dcmanager.tests.unit.api.v1.controllers.mixins import PostJSONMixin
from dcmanager.tests.unit.api.v1.controllers.test_subclouds \
import FAKE_SUBCLOUD_DATA
from dcmanager.tests.unit.common import fake_subcloud
SAMPLE_SUBCLOUD_PEER_GROUP_NAME = 'GroupX'
SAMPLE_SUBCLOUD_PEER_GROUP_MAX_SUBCLOUDS_REHOMING = 50
SAMPLE_SUBCLOUD_PEER_GROUP_STATE = 'enabled'
SAMPLE_SUBCLOUD_PEER_GROUP_PIRORITY = 0
FAKE_UUID = '62c9592d-f799-4db9-8d40-6786a74d6021'
FAKE_SUBCLOUD_DATA = fake_subcloud.FAKE_SUBCLOUD_DATA
API_PREFIX = '/v1.0/subcloud-peer-groups'
RESULT_KEY = 'subcloud_peer_groups'

5277
distributedcloud/dcmanager/tests/unit/api/v1/controllers/test_subclouds.py
View File

File diff suppressed because it is too large Load Diff

158
distributedcloud/dcmanager/tests/unit/common/test_phased_subcloud_deploy.py
View File

@ -4,9 +4,45 @@
# SPDX-License-Identifier: Apache-2.0
#
import base64
import copy
import json
import os
import mock
from oslo_utils import timeutils
import six
from dccommon import consts as dccommon_consts
from dcmanager.common import consts
from dcmanager.common import phased_subcloud_deploy as psd_common
from dcmanager.tests.unit.common import fake_subcloud
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 = ""
@mock.patch.object(os, 'listdir')
@ -56,3 +92,125 @@ def test_check_deploy_files_in_alternate_location_with_deploy_playbook_not_exist
response = self.check_deploy_files_in_alternate_location(payload)
self.assertEqual(response, False)
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.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.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))
)
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
)
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
)

3
distributedcloud/dcmanager/tests/unit/manager/test_peer_monitor_manager.py
View File

@ -195,8 +195,7 @@ class TestPeerMonitor(base.DCManagerTestCase):
# Test the case where the association sync_status is unknown
pm = peer_monitor_manager.PeerMonitor(peer, self.ctx, mock.MagicMock())
pm._update_sync_status_when_secondary_site_becomes_reachable()
mock_dc_client().get_subcloud_peer_group.assert_called_once_with(
peer_group.peer_group_name)
mock_dc_client().get_subcloud_peer_group.assert_called_once()
mock_dc_client().get_system_peer.assert_called_once_with(
FAKE_SITE0_SYSTEM_UUID)
mock_dc_client().get_peer_group_association_with_peer_id_and_pg_id.\

68
distributedcloud/dcmanager/tests/unit/utils/test_utils.py
View File

@ -1,5 +1,5 @@
# Copyright 2016 Ericsson AB
# Copyright (c) 2017, 2019, 2021 Wind River Systems, Inc.
# Copyright (c) 2017, 2019, 2021, 2024 Wind River Systems, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -15,6 +15,7 @@
# under the License.
#
from dcmanager.common import utils
from dcmanager.tests import base
@ -22,4 +23,67 @@ class TestUtils(base.DCManagerTestCase):
def setUp(self):
super(TestUtils, self).setUp()
# Nothing to test yet
def test_get_management_subnet(self):
payload = {"management_subnet": "192.168.204.0/24"}
self.assertEqual(
utils.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(
utils.get_management_subnet(payload), payload["admin_subnet"]
)
def test_get_management_start_address(self):
payload = {"management_start_address": "192.168.204.2"}
self.assertEqual(
utils.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(
utils.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(
utils.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(
utils.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(
utils.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(
utils.get_management_gateway_address(payload),
payload["admin_gateway_address"],
)