starlingx/distcloud
Code Issues Proposed changes

Merge "Create a base structure for API tests"

Browse Source
This commit is contained in:
Zuul 2024-02-19 12:25:15 +00:00 committed by Gerrit Code Review
commit 07232ba937
3 changed files with 252 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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")

170
distributedcloud/dcmanager/tests/unit/api/test_root_controller.py
View File

@ -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)

10
distributedcloud/dcmanager/tests/unit/common/consts.py Normal file
View File

@ -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'