diff --git a/distributedcloud/dcmanager/tests/base.py b/distributedcloud/dcmanager/tests/base.py index fcb310996..d32d69006 100644 --- a/distributedcloud/dcmanager/tests/base.py +++ b/distributedcloud/dcmanager/tests/base.py @@ -15,7 +15,11 @@ # under the License. # +import base64 +import builtins import json +import mock +import pecan from oslo_config import cfg from oslo_db import options @@ -24,9 +28,14 @@ import sqlalchemy from sqlalchemy.engine import Engine from sqlalchemy import event +from dcmanager.audit import rpcapi as audit_rpc_client from dcmanager.common import consts +from dcmanager.common import phased_subcloud_deploy as psd_common +from dcmanager.common import utils as dutils from dcmanager.db import api from dcmanager.db.sqlalchemy import api as db_api +from dcmanager.rpc import client as rpc_client + from dcmanager.tests import utils get_engine = api.get_engine @@ -108,6 +117,14 @@ def set_sqlite_pragma(dbapi_connection, connection_record): cursor.close() +class FakeException(Exception): + """Exception used to throw a generic exception in the application + + Using the Exception class might lead to linter errors for being too broad. In + these cases, the FakeException is used + """ + + class DCManagerTestCase(base.BaseTestCase): """Test case base class for all unit tests.""" @@ -136,3 +153,112 @@ class DCManagerTestCase(base.BaseTestCase): self.addCleanup(self.reset_dummy_db) self.setup_dummy_db() self.ctx = utils.dummy_context() + self._mock_pecan() + + def _mock_pecan(self): + """Mock pecan's abort""" + + mock_patch = mock.patch.object(pecan, 'abort', wraps=pecan.abort) + self.mock_pecan_abort = mock_patch.start() + self.addCleanup(mock_patch.stop) + + def _mock_audit_rpc_client(self): + """Mock rpc's manager audit client""" + + mock_patch = mock.patch.object(audit_rpc_client, 'ManagerAuditClient') + self.mock_audit_rpc_client = mock_patch.start() + self.addCleanup(mock_patch.stop) + + def _mock_rpc_client(self): + """Mock rpc's manager client""" + + mock_patch = mock.patch.object(rpc_client, 'ManagerClient') + self.mock_rpc_client = mock_patch.start() + self.addCleanup(mock_patch.stop) + + def _mock_rpc_subcloud_state_client(self): + """Mock rpc's subcloud state client""" + + mock_patch = mock.patch.object(rpc_client, 'SubcloudStateClient') + self.mock_rpc_subcloud_state_client = mock_patch.start() + self.addCleanup(mock_patch.stop) + + def _mock_openstack_driver(self, target): + """Mock the target's OpenStackDriver""" + + mock_patch = mock.patch.object(target, 'OpenStackDriver') + self.mock_openstack_driver = mock_patch.start() + self.addCleanup(mock_patch.stop) + + def _mock_sysinv_client(self, target): + """Mock the target's SysinvClient""" + + mock_patch = mock.patch.object(target, 'SysinvClient') + self.mock_sysinv_client = mock_patch.start() + self.addCleanup(mock_patch.stop) + + def _mock_get_network_address_pool(self): + """Mock phased subcloud deploy's get_network_address_pool""" + + mock_patch_object = mock.patch.object(psd_common, 'get_network_address_pool') + self.mock_get_network_address_pool = mock_patch_object.start() + self.addCleanup(mock_patch_object.stop) + + def _mock_get_ks_client(self): + """Mock phased subcloud deploy's get_ks_client""" + + mock_patch_object = mock.patch.object(psd_common, 'get_ks_client') + self.mock_get_ks_client = mock_patch_object.start() + self.addCleanup(mock_patch_object.stop) + + def _mock_query(self): + """Mock phased subcloud deploy's query""" + + mock_patch_object = mock.patch.object(psd_common.PatchingClient, 'query') + self.mock_query = mock_patch_object.start() + self.addCleanup(mock_patch_object.stop) + + def _mock_get_subcloud_db_install_values(self): + """Mock phased subcloud deploy's get_subcloud_db_install_values""" + + mock_patch_object = mock.patch.object( + psd_common, 'get_subcloud_db_install_values' + ) + self.mock_get_subcloud_db_install_values = mock_patch_object.start() + self.addCleanup(mock_patch_object.stop) + + def _mock_validate_k8s_version(self): + """Mock phased subcloud deploy's validate_k8s_version""" + + mock_patch_object = mock.patch.object(psd_common, 'validate_k8s_version') + self.mock_validate_k8s_version = mock_patch_object.start() + self.addCleanup(mock_patch_object.stop) + + def _mock_get_vault_load_files(self): + """Mock dcmanager util's get_vault_load_files""" + + mock_patch_object = mock.patch.object(dutils, 'get_vault_load_files') + self.mock_get_vault_load_files = mock_patch_object.start() + self.addCleanup(mock_patch_object.stop) + + def _mock_builtins_open(self): + """Mock builtins' open""" + + mock_patch = mock.patch.object(builtins, 'open') + self.mock_builtins_open = mock_patch.start() + self.addCleanup(mock_patch.stop) + + def _assert_pecan(self, http_status, content=None, call_count=1): + """Assert pecan was called with the correct arguments""" + + self.assertEqual(self.mock_pecan_abort.call_count, call_count) + + if content: + self.mock_pecan_abort.assert_called_with(http_status, content) + else: + self.mock_pecan_abort.assert_called_with(http_status) + + def _create_password(self, keyword='default'): + """Create a password with based on the specified keyword""" + + return base64.b64encode(keyword.encode("utf-8")).decode("utf-8") diff --git a/distributedcloud/dcmanager/tests/unit/api/test_root_controller.py b/distributedcloud/dcmanager/tests/unit/api/test_root_controller.py index ed42461ef..8eb8227ba 100644 --- a/distributedcloud/dcmanager/tests/unit/api/test_root_controller.py +++ b/distributedcloud/dcmanager/tests/unit/api/test_root_controller.py @@ -1,5 +1,5 @@ # Copyright (c) 2015 Huawei Technologies Co., Ltd. -# Copyright (c) 2017-2022 Wind River Systems, Inc. +# Copyright (c) 2017-2024 Wind River Systems, Inc. # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may @@ -15,49 +15,57 @@ # under the License. # -import pecan -from pecan.configuration import set_config -from pecan.testing import load_test_app +import http.client from oslo_config import cfg from oslo_config import fixture as fixture_config from oslo_serialization import jsonutils from oslo_utils import uuidutils +import pecan +from pecan.configuration import set_config +from pecan.testing import load_test_app from dcmanager.api import api_config from dcmanager.common import config from dcmanager.tests import base +from dcmanager.tests.unit.common import consts as test_consts +from dcmanager.tests import utils config.register_options() OPT_GROUP_NAME = 'keystone_authtoken' cfg.CONF.import_group(OPT_GROUP_NAME, "keystonemiddleware.auth_token") -def fake_delete_response(self, context): - resp = jsonutils.dumps(context.to_dict()) - return resp - - class DCManagerApiTest(base.DCManagerTestCase): def setUp(self): - super(DCManagerApiTest, self).setUp() + super().setUp() self.addCleanup(set_config, {}, overwrite=True) api_config.test_init() - config = fixture_config.Config() - self.CONF = self.useFixture(config).conf - config.set_config_dirs([]) + config_fixture = fixture_config.Config() + self.CONF = self.useFixture(config_fixture).conf + config_fixture.set_config_dirs([]) - # self.setup_messaging(self.CONF) self.CONF.set_override('auth_strategy', 'noauth') self.app = self._make_app() + self.url = '/' + # The put method is used as a default value, leading to the generic + # implementation on controllers in case the method is not specified + self.method = self.app.put + self.params = {} + self.verb = None + self.headers = { + 'X-Tenant-Id': utils.UUID1, 'X_ROLE': 'admin,member,reader', + 'X-Identity-Status': 'Confirmed', 'X-Project-Name': 'admin' + } + def _make_app(self, enable_acl=False): - self.config = { + self.config_fixture = { 'app': { 'root': 'dcmanager.api.controllers.root.RootController', 'modules': ['dcmanager.api'], @@ -69,7 +77,32 @@ class DCManagerApiTest(base.DCManagerTestCase): }, } - return load_test_app(self.config) + return load_test_app(self.config_fixture) + + def _send_request(self): + """Send a request to a url""" + + return self.method( + self.url, headers=self.headers, params=self.params, expect_errors=True + ) + + def _assert_response( + self, response, status_code=http.client.OK, + content_type=test_consts.APPLICATION_JSON + ): + """Assert the response for a request""" + + self.assertEqual(response.status_code, status_code) + self.assertEqual(response.content_type, content_type) + + def _assert_pecan_and_response( + self, response, http_status, content=None, call_count=1, + content_type=test_consts.TEXT_PLAIN + ): + """Assert the response and pecan abort for a failed request""" + + self._assert_pecan(http_status, content, call_count=call_count) + self._assert_response(response, http_status, content_type) def tearDown(self): super(DCManagerApiTest, self).tearDown() @@ -79,79 +112,108 @@ class DCManagerApiTest(base.DCManagerTestCase): class TestRootController(DCManagerApiTest): """Test version listing on root URI.""" + def setUp(self): + super(TestRootController, self).setUp() + + self.url = '/' + self.method = self.app.get + + def _test_method_returns_405(self, method, content_type=test_consts.TEXT_PLAIN): + self.method = method + + response = self._send_request() + + self._assert_pecan_and_response( + response, http.client.METHOD_NOT_ALLOWED, content_type=content_type + ) + def test_get(self): - response = self.app.get('/') - self.assertEqual(response.status_int, 200) + """Test get request succeeds with correct versions""" + + response = self._send_request() + + self._assert_response(response) json_body = jsonutils.loads(response.body) versions = json_body.get('versions') self.assertEqual(1, len(versions)) - def _test_method_returns_405(self, method): - api_method = getattr(self.app, method) - response = api_method('/', expect_errors=True) - self.assertEqual(response.status_int, 405) + def test_request_id(self): + """Test request for root returns the correct request id""" + + response = self._send_request() + + self._assert_response(response) + self.assertIn('x-openstack-request-id', response.headers) + self.assertTrue( + response.headers['x-openstack-request-id'].startswith('req-') + ) + id_part = response.headers['x-openstack-request-id'].split('req-')[1] + self.assertTrue(uuidutils.is_uuid_like(id_part)) def test_post(self): - self._test_method_returns_405('post') + """Test post request is not allowed on root""" + + self._test_method_returns_405(self.app.post) def test_put(self): - self._test_method_returns_405('put') + """Test put request is not allowed on root""" + + self._test_method_returns_405(self.app.put) def test_patch(self): - self._test_method_returns_405('patch') + """Test patch request is not allowed on root""" + + self._test_method_returns_405(self.app.patch) def test_delete(self): - self._test_method_returns_405('delete') + """Test delete request is not allowed on root""" + + self._test_method_returns_405(self.app.delete) def test_head(self): - self._test_method_returns_405('head') + """Test head request is not allowed on root""" + + self._test_method_returns_405( + self.app.head, content_type=test_consts.TEXT_HTML + ) class TestErrors(DCManagerApiTest): def setUp(self): super(TestErrors, self).setUp() - cfg.CONF.set_override('admin_tenant', 'fake_tenant_id', - group='cache') + cfg.CONF.set_override('admin_tenant', 'fake_tenant_id', group='cache') def test_404(self): - response = self.app.get('/assert_called_once', expect_errors=True) - self.assertEqual(response.status_int, 404) + self.url = '/assert_called_once' + self.method = self.app.get - def test_bad_method(self): - fake_tenant = uuidutils.generate_uuid() - fake_url = '/v1.0/%s/bad_method' % fake_tenant - response = self.app.patch(fake_url, - expect_errors=True) - self.assertEqual(response.status_int, 404) + response = self._send_request() + self._assert_response( + response, http.client.NOT_FOUND, content_type=test_consts.TEXT_PLAIN + ) + def test_version_1_root_controller(self): + self.url = f'/v1.0/{uuidutils.generate_uuid()}/bad_method' + self.method = self.app.patch -class TestRequestID(DCManagerApiTest): + response = self._send_request() - def test_request_id(self): - response = self.app.get('/') - self.assertIn('x-openstack-request-id', response.headers) - self.assertTrue( - response.headers['x-openstack-request-id'].startswith('req-')) - id_part = response.headers['x-openstack-request-id'].split('req-')[1] - self.assertTrue(uuidutils.is_uuid_like(id_part)) + self._assert_pecan_and_response(response, http.client.NOT_FOUND) class TestKeystoneAuth(DCManagerApiTest): + """Test requests using keystone as the authentication strategy""" def setUp(self): super(TestKeystoneAuth, self).setUp() - self.addCleanup(set_config, {}, overwrite=True) - - api_config.test_init() - - self.CONF = self.useFixture(fixture_config.Config()).conf - cfg.CONF.set_override('auth_strategy', 'keystone') - self.app = self._make_app() + self.method = self.app.get def test_auth_not_enforced_for_root(self): - response = self.app.get('/') - self.assertEqual(response.status_int, 200) + """Test authentication is not enforced for root url""" + + response = self._send_request() + self._assert_response(response) diff --git a/distributedcloud/dcmanager/tests/unit/common/consts.py b/distributedcloud/dcmanager/tests/unit/common/consts.py new file mode 100644 index 000000000..2bd65f805 --- /dev/null +++ b/distributedcloud/dcmanager/tests/unit/common/consts.py @@ -0,0 +1,10 @@ +# +# Copyright (c) 2024 Wind River Systems, Inc. +# +# SPDX-License-Identifier: Apache-2.0 +# + +# Content-type +TEXT_PLAIN = 'text/plain' +TEXT_HTML = 'text/html' +APPLICATION_JSON = 'application/json'