starlingx/openstack-armada-app
Code Issues Proposed changes

Merge "Improve tests coverage of OpenstackBaseHelm class"

Browse Source
This commit is contained in:
Zuul
2025-10-03 13:10:41 +00:00
committed by Gerrit Code Review
parent 2750a9a412 b3e3a3ed6e
commit 9a98291917
1 changed files with 152 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

153
python3-k8sapp-openstack/k8sapp_openstack/k8sapp_openstack/tests/helm/test_openstack.py
View File

@@ -6,6 +6,7 @@
import base64
import mock
from sqlalchemy.orm.exc import NoResultFound
from sysinv.helm import helm
from sysinv.tests.db import base as dbbase
from sysinv.tests.db import utils as dbutils
@@ -38,13 +39,75 @@ class OpenstackHelmUnitTests(OpenstackBaseHelmTestCase,
dbbase.ControllerHostTestCase):
def setUp(self):
super(OpenstackHelmUnitTests, self).setUp()
self.operator = helm.HelmOperator(self.dbapi)
self.operator = helm.HelmOperator()
self.helm = MockOpenstackHelm(self.operator)
def test_get_namespaces(self):
"""Tests that namespaces match the list of supported namespaces."""
self.assertEqual(self.helm.get_namespaces(), self.helm.SUPPORTED_NAMESPACES)
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.dbapi', new=None)
def test_get_service_parameters_dbapi_none(self, *_):
"""Asserts on empty service parameters when dbapi is unavailable."""
result = self.helm._get_service_parameters(app_constants.HELM_CHART_KEYSTONE)
self.assertEqual(result, [])
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.dbapi.service_parameter_get_all',
return_value=[{'section': "test"}])
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.dbapi', new=mock.Mock())
def test_get_service_parameters_with_result(self, service_parameter_get_all, *_):
"""Tests retrieval of services current configuration parameters."""
result = self.helm._get_service_parameters(app_constants.HELM_CHART_KEYSTONE)
self.assertEqual(result, [{'section': "test"}])
service_parameter_get_all.assert_called_once_with(service=app_constants.HELM_CHART_KEYSTONE)
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.dbapi.service_parameter_get_all',
side_effect=NoResultFound)
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.dbapi', new=mock.Mock())
def test_get_service_parameters_no_result(self, service_parameter_get_all, *_):
"""Tests handling of missing service configuration parameters."""
result = self.helm._get_service_parameters(app_constants.HELM_CHART_KEYSTONE)
self.assertEqual(result, [])
service_parameter_get_all.assert_called_once_with(service=app_constants.HELM_CHART_KEYSTONE)
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.context', new={
'_service_params': {
app_constants.HELM_CHART_KEYSTONE: [1, 2, 3]
}
})
def test_get_service_parameter_configs_cache(self, *_):
"""Tests retrieval of cached service configuration values."""
result = self.helm._get_service_parameter_configs(app_constants.HELM_CHART_KEYSTONE)
self.assertEqual(result, [1, 2, 3])
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._get_service_parameters',
return_value=[{"section": "test"}])
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.context', new={})
def test_get_service_parameter_configs_miss(self, *_):
"""Tests post retrieval caching of service configuration values."""
result = self.helm._get_service_parameter_configs(app_constants.HELM_CHART_KEYSTONE)
self.assertEqual(result, [{"section": "test"}])
self.assertIn(
app_constants.HELM_CHART_KEYSTONE,
self.helm.context['_service_params'])
def test_service_parameter_lookup_one_found(self, *_):
"""Asserts successful lookup of a configuration parameter."""
params = [
{"section": "test1", "name": "name1", "value": "value1"},
{"section": "test2", "name": "name2", "value": "v2"}
]
result = self.helm._service_parameter_lookup_one(params, "test2", "name2", "default")
self.assertEqual(result, "v2")
def test_service_parameter_lookup_one_not_found(self, *_):
"""Asserts an unsuccessful lookup of a configuration parameter."""
params = [
{"section": "test1", "name": "name1", "value": "value1"}
]
result = self.helm._service_parameter_lookup_one(params, "testX", "nameX", "default")
self.assertEqual(result, "default")
@mock.patch('k8sapp_openstack.helm.keystone.KeystoneHelm.get_admin_project_name',
return_value="admin_project")
def test_get_admin_project_name(self, *_):
@@ -69,6 +132,59 @@ class OpenstackHelmUnitTests(OpenstackBaseHelmTestCase,
self.helm._get_database_username(app_constants.HELM_CHART_NOVA),
"admin-nova")
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._generate_random_password', return_value="pw")
@mock.patch('k8sapp_openstack.helm.openstack.keyring.get_password', return_value=None)
@mock.patch('k8sapp_openstack.helm.openstack.keyring.set_password')
def test_get_keyring_password_generates(self, set_password, *_):
"""Tests keyring password generation associated with a service."""
pw = self.helm._get_keyring_password(app_constants.HELM_CHART_KEYSTONE, "usr")
self.assertEqual(pw, b"pw")
set_password.assert_called_once_with(app_constants.HELM_CHART_KEYSTONE, "usr", "pw")
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._region_config',
return_value=False)
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._region_name',
return_value="RegionOne")
def test_get_service_region_name_default(self, *_):
"""Asserts on the default region name of a service."""
self.assertEqual(
self.helm._get_service_region_name(app_constants.HELM_CHART_KEYSTONE),
"RegionOne")
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._region_config',
return_value=True)
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._get_service_config',
return_value=mock.Mock(region_name="RegionTwo"))
def test_get_service_region_name_with_config(self, *_):
"""Tests the retrieval of region name of a service."""
self.assertEqual(
self.helm._get_service_region_name(app_constants.HELM_CHART_KEYSTONE),
b"RegionTwo")
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._get_service_config',
return_value=mock.Mock(capabilities={'service_type': "stype"}))
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._region_config',
return_value=True)
def test_get_configured_service_type_with_region_config(self, *_):
"""Asserts on configured service type in a region."""
self.assertEqual(self.helm._get_configured_service_type(app_constants.HELM_CHART_KEYSTONE), 'stype')
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._get_service_config',
return_value=mock.Mock(capabilities={'v1_service_type': "stypev1"}))
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._region_config', return_value=True)
def test_get_configured_service_type_with_version(self, *_):
"""Asserts on specific version of a service type."""
service_type = self.helm._get_configured_service_type(
app_constants.HELM_CHART_KEYSTONE,
version='v1')
self.assertEqual(service_type, "stypev1")
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._region_config', return_value=False)
def test_get_configured_service_type_no_region_config(self, *_):
"""Asserts on default service type configuration."""
self.helm._region_config = mock.Mock(return_value=False)
self.assertIsNone(self.helm._get_configured_service_type(app_constants.HELM_CHART_KEYSTONE))
@mock.patch('k8sapp_openstack.utils.get_services_fqdn_pattern',
return_value="{service_name}.{endpoint_domain}")
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._get_service_parameter',
@@ -95,6 +211,32 @@ class OpenstackHelmUnitTests(OpenstackBaseHelmTestCase,
result = self.helm._get_service_default_dns_name(app_constants.HELM_CHART_KEYSTONE)
self.assertIn(app_constants.HELM_CHART_KEYSTONE, result)
@mock.patch("k8sapp_openstack.helm.openstack.app_utils.is_ceph_backend_available", return_value=True)
def test_get_ceph_client_overrides_rook(self, *_):
"""Matches the default backend user secret name."""
result = self.helm._get_ceph_client_overrides()
self.assertIn("user_secret_name", result)
@mock.patch("k8sapp_openstack.helm.openstack.app_utils.is_ceph_backend_available", return_value=False)
@mock.patch("sysinv.common.storage_backend_conf.K8RbdProvisioner.get_user_secret_name", return_value="secret")
def test_get_ceph_client_overrides_rbd(self, *_):
"""Matches the stored backend user secret name."""
result = self.helm._get_ceph_client_overrides()
self.assertEqual(result["user_secret_name"], "secret")
@mock.patch("k8sapp_openstack.helm.openstack.OpenstackBaseHelm._get_interface_networks", return_value={})
def test_get_interface_network_query_not_found(self, *_):
"""Asserts on exception when no matching network interfaces are found."""
self.assertRaises(openstack.exception.InterfaceNetworkNotFoundByHostInterfaceNetwork,
self.helm._get_interface_network_query, 1, 3)
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm._is_enabled', return_value=False)
def test_execute_kustomize_updates(self, *_):
"""Asserts on the deletion of disabled cluster resources."""
operator = mock.Mock(APP="app")
self.helm.execute_kustomize_updates(operator)
operator.helm_release_resource_delete.assert_called_once_with(self.helm.HELM_RELEASE)
@mock.patch('sysinv.common.kubernetes.KubeOperator.kube_get_secret',
return_value=mock.Mock(data={'key': base64.b64encode(b'secret\n')}))
def test_get_rook_ceph_admin_keyring(self, *_):
@@ -105,3 +247,12 @@ class OpenstackHelmUnitTests(OpenstackBaseHelmTestCase,
def test_get_ca_file(self):
"""Matches the default certificate authority file path."""
self.assertEqual(self.helm.get_ca_file(), "/etc/ssl/certs/openstack-helm.crt")
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.dbapi.helm_override_get',
return_value=mock.Mock(system_overrides={'test': "name"}))
@mock.patch('k8sapp_openstack.helm.openstack.OpenstackBaseHelm.dbapi', new=mock.Mock())
@mock.patch('sysinv.common.utils.find_openstack_app', return_value=mock.Mock(id=1))
def test_get_or_generate_password(self, *_):
"""Tests on general purpose password retrieval or generation."""
pw = self.helm._get_or_generate_password('chart', 'ns', 'test')
self.assertEqual(pw, b'name')