Add more unit tests for coriolis main modules

coriolis/tests/test_rpc.py

@@ -0,0 +1,224 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis import context
+from coriolis import rpc
+from coriolis.tests import test_base
+
+
+class RequestContextSerializerTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis RequestContextSerializer class."""
+
+    def setUp(self):
+        super(RequestContextSerializerTestCase, self).setUp()
+        self.base = mock.Mock()
+        self.ctxt = mock.Mock()
+        self.entity = {'foo': 'bar'}
+        self.serializer = rpc.RequestContextSerializer(self.base)
+
+    def test_serialize_entity_base_none(self):
+        serializer = rpc.RequestContextSerializer(None)
+        result = serializer.serialize_entity(self.ctxt, self.entity)
+        self.assertEqual(result, self.entity)
+
+    def test_serialize_entity_base_not_none(self):
+        result = self.serializer.serialize_entity(self.ctxt, self.entity)
+        self.base.serialize_entity.assert_called_once_with(self.ctxt,
+                                                           self.entity)
+        self.assertEqual(result, self.base.serialize_entity.return_value)
+
+    def test_deserialize_entity_base_none(self):
+        serializer = rpc.RequestContextSerializer(None)
+        result = serializer.deserialize_entity(self.ctxt, self.entity)
+        self.assertEqual(result, self.entity)
+
+    def test_deserialize_entity_base_not_none(self):
+        result = self.serializer.deserialize_entity(self.ctxt, self.entity)
+        self.base.deserialize_entity.assert_called_once_with(self.ctxt,
+                                                             self.entity)
+        self.assertEqual(result, self.base.deserialize_entity.return_value)
+
+    def test_serialize_context(self):
+        result = self.serializer.serialize_context(self.ctxt)
+        self.ctxt.to_dict.assert_called_once_with()
+        self.assertEqual(result, self.ctxt.to_dict.return_value)
+
+    def test_deserialize_context(self):
+        ctxt_dict = {'foo': 'bar'}
+        with mock.patch.object(
+            context.RequestContext, 'from_dict') as mock_from_dict:
+            result = self.serializer.deserialize_context(ctxt_dict)
+
+            mock_from_dict.assert_called_once_with(ctxt_dict)
+            self.assertEqual(result, mock_from_dict.return_value)
+
+
+class RpcTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis RPC module."""
+
+    @mock.patch('coriolis.rpc.messaging.get_transport')
+    def test_get_transport(self, mock_get_transport):
+        result = rpc._get_transport()
+
+        mock_get_transport.assert_called_once_with(
+            mock.ANY, mock.ANY, allowed_remote_exmods=rpc.ALLOWED_EXMODS)
+        self.assertEqual(result, mock_get_transport.return_value)
+
+    @mock.patch('coriolis.rpc.messaging.get_rpc_server')
+    @mock.patch('coriolis.rpc._get_transport')
+    @mock.patch('coriolis.rpc.RequestContextSerializer')
+    def test_get_server(self, mock_context_serializer, mock_get_transport,
+                        mock_get_rpc_server):
+        target = mock.Mock()
+        endpoints = mock.Mock()
+        serializer = mock.Mock()
+        mock_context_serializer.return_value = serializer
+
+        result = rpc.get_server(target, endpoints, serializer)
+
+        mock_context_serializer.assert_called_once_with(serializer)
+        mock_get_rpc_server.assert_called_once_with(
+            mock_get_transport.return_value, target, endpoints,
+            executor='eventlet', serializer=serializer)
+        self.assertEqual(result, mock_get_rpc_server.return_value)
+
+    @mock.patch('coriolis.rpc.messaging.get_transport')
+    def test_init(self, mock_get_transport):
+        rpc._TRANSPORT = None
+        result = rpc.init()
+
+        mock_get_transport.assert_called_once()
+        self.assertEqual(result, mock_get_transport.return_value)
+        self.assertEqual(rpc._TRANSPORT, mock_get_transport.return_value)
+
+    @mock.patch('coriolis.rpc.messaging.get_transport')
+    def test_init_already_initialized(self, mock_get_transport):
+        rpc._TRANSPORT = mock.Mock()
+        result = rpc.init()
+        self.assertEqual(result, rpc._TRANSPORT)
+        mock_get_transport.assert_not_called()
+
+
+class BaseRPCClientTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis BaseRPCClient class."""
+
+    def setUp(self):
+        super(BaseRPCClientTestCase, self).setUp()
+        self.target = mock.Mock()
+        self.timeout = 60
+        self.serializer = mock.Mock()
+        self.method = mock.Mock()
+        self.host = mock.Mock()
+        self.args = {'foo': 'bar'}
+        self.client = rpc.BaseRPCClient(self.target, timeout=self.timeout)
+
+    def test_init(self):
+        with mock.patch.object(rpc, 'RequestContextSerializer') as mock_ser:
+            client = rpc.BaseRPCClient(self.target, timeout=self.timeout,
+                                       serializer=self.serializer)
+            mock_ser.assert_called_once_with(self.serializer)
+            self.assertEqual(client._serializer, mock_ser.return_value)
+        self.assertEqual(client._target, self.target)
+        self.assertEqual(client._timeout, self.timeout)
+        self.assertEqual(client._transport_conn, None)
+
+    def test_init_timeout_is_None(self):
+        with mock.patch.object(rpc, 'RequestContextSerializer') as mock_ser:
+            client = rpc.BaseRPCClient(self.target, timeout=None,
+                                       serializer=self.serializer)
+            mock_ser.assert_called_once_with(self.serializer)
+            self.assertEqual(client._serializer, mock_ser.return_value)
+        self.assertEqual(client._target, self.target)
+        self.assertEqual(client._timeout,
+                         rpc.CONF.default_messaging_timeout)
+        self.assertEqual(client._transport_conn, None)
+
+    def test_repr(self):
+        result = self.client.__repr__()
+        self.assertEqual(result, "<RPCClient(target=%s, timeout=%s)>" % (
+            self.target, self.timeout))
+
+    @mock.patch.object(rpc, '_get_transport')
+    def test_transport_property_when_transport_is_None(self,
+                                                       mock_get_transport):
+        with mock.patch.object(rpc, '_TRANSPORT', None):
+            result = self.client._transport
+
+        mock_get_transport.assert_called_once()
+        self.assertEqual(result, mock_get_transport.return_value)
+
+    @mock.patch.object(rpc, '_get_transport')
+    def test_transport_property_when_transport_conn_is_not_None(
+            self, mock_get_transport):
+        with mock.patch.object(rpc, '_TRANSPORT', None):
+            self.client._transport_conn = mock.Mock()
+            result = self.client._transport
+
+        mock_get_transport.assert_not_called()
+        self.assertEqual(result, self.client._transport_conn)
+
+    @mock.patch.object(rpc, '_TRANSPORT')
+    def test_transport_property_when_transport_is_not_None(self,
+                                                           mock_transport):
+        result = self.client._transport
+
+        self.assertEqual(result, mock_transport)
+        self.assertEqual(self.client._transport_conn, None)
+
+    @mock.patch('oslo_messaging.RPCClient')
+    @mock.patch('coriolis.rpc.messaging.get_transport')
+    def test_rpc_client(self, mock_get_transport, mock_rpc_client):
+        mock_get_transport.return_value = mock.MagicMock()
+        result = self.client._rpc_client()
+
+        mock_rpc_client.assert_called_once_with(
+            self.client._transport, self.client._target,
+            serializer=self.client._serializer, timeout=self.client._timeout)
+        self.assertEqual(result, mock_rpc_client.return_value)
+
+    def test_call(self):
+        with mock.patch('coriolis.rpc.BaseRPCClient._rpc_client') as rpc_mock:
+            result = self.client._call(mock.sentinel.ctxt, self.method,
+                                       **self.args)
+
+            rpc_mock.assert_called_once()
+            rpc_mock.return_value.call.assert_called_once_with(
+                mock.sentinel.ctxt, self.method, **self.args)
+            self.assertEqual(
+                result, rpc_mock.return_value.call.return_value)
+
+    def test_call_on_host(self):
+        with mock.patch('coriolis.rpc.BaseRPCClient._rpc_client') as rpc_mock:
+            result = self.client._call_on_host(self.host, mock.sentinel.ctxt,
+                                               self.method, **self.args)
+
+            rpc_mock.assert_called_once()
+            rpc_mock.return_value.prepare.assert_called_once_with(
+                server=self.host)
+            rpc_mock.return_value.prepare.return_value.call.\
+                assert_called_once_with(mock.sentinel.ctxt, self.method,
+                                        **self.args)
+            self.assertEqual(result, rpc_mock.return_value.prepare.
+                             return_value.call.return_value)
+
+    def test_cast(self):
+        with mock.patch('coriolis.rpc.BaseRPCClient._rpc_client') as rpc_mock:
+            self.client._cast(mock.sentinel.ctxt, self.method, **self.args)
+
+            rpc_mock.assert_called_once()
+            rpc_mock.return_value.cast.assert_called_once_with(
+                mock.sentinel.ctxt, self.method, **self.args)
+
+    def test_cast_for_host(self):
+        with mock.patch('coriolis.rpc.BaseRPCClient._rpc_client') as rpc_mock:
+            self.client._cast_for_host(self.host, mock.sentinel.ctxt,
+                                       self.method, **self.args)
+
+            rpc_mock.assert_called_once()
+            rpc_mock.return_value.prepare.assert_called_once_with(
+                server=self.host)
+            rpc_mock.return_value.prepare.return_value.cast.\
+                assert_called_once_with(mock.sentinel.ctxt, self.method,
+                                        **self.args)

coriolis/tests/test_schemas.py

@@ -2,11 +2,13 @@
 # All Rights Reserved.
 
 import json
+import logging
+from unittest import mock
 
 import jinja2
 import jsonschema
-from unittest import mock
 
+from coriolis import exception
 from coriolis import schemas
 from coriolis.tests import test_base
 
@@ -66,6 +68,40 @@ def test_get_schema(self, mock_loads, mock_loader, mock_environ):
 
         self.assertEqual(res, test_loaded_schema)
 
+    @mock.patch.object(jinja2, 'Environment')
+    @mock.patch.object(jinja2, 'PackageLoader')
+    def test_get_schema_parent_package_exists(self, mock_loader, mock_environ):
+        test_schema_name = 'test.schema.name'
+        test_package_name = 'test.package.name'
+
+        mock_template = mock.MagicMock()
+        mock_template.render.return_value = '{}'
+
+        mock_environ.return_value.get_template.return_value = mock_template
+
+        mock_loader.side_effect = [ValueError(), mock.MagicMock()]
+
+        schemas.get_schema(test_package_name, test_schema_name)
+
+        parent_package = "test.package"
+        mock_loader.assert_has_calls([
+            mock.call(test_package_name, schemas.DEFAULT_SCHEMAS_DIRECTORY),
+            mock.call(parent_package, schemas.DEFAULT_SCHEMAS_DIRECTORY)])
+
+    @mock.patch.object(jinja2, 'PackageLoader')
+    def test_get_schema_parent_package_not_exists(self, mock_loader):
+        test_schema_name = 'test.schema.name'
+        test_package_name = 'testpackage'
+
+        mock_loader.side_effect = ValueError()
+
+        self.assertRaises(
+            ValueError, schemas.get_schema, test_package_name,
+            test_schema_name)
+
+        mock_loader.assert_called_once_with(
+            test_package_name, schemas.DEFAULT_SCHEMAS_DIRECTORY)
+
     @mock.patch.object(jsonschema, 'validate')
     def test_validate_value(self, mock_validate):
         test_value = mock.sentinel.test_value
@@ -90,3 +126,28 @@ def test_validate_string(self, mock_validate, mock_loads):
         mock_loads.assert_called_once_with(test_string)
         mock_validate.assert_called_once_with(
             test_value, test_schema, format_checker=None)
+
+    @mock.patch.object(jsonschema, 'validate')
+    def test_validate_value_raises_on_error(self, mock_validate):
+        test_value = mock.sentinel.test_value
+        test_schema = mock.sentinel.test_schema
+
+        mock_validate.side_effect = jsonschema.exceptions.ValidationError(
+            'Test Error')
+
+        self.assertRaises(
+            exception.SchemaValidationException, schemas.validate_value,
+            test_value, test_schema, raise_on_error=True)
+
+    @mock.patch.object(jsonschema, 'validate')
+    def test_validate_value_no_raise_on_error(self, mock_validate):
+        test_value = mock.sentinel.test_value
+        test_schema = mock.sentinel.test_schema
+
+        mock_validate.side_effect = jsonschema.exceptions.ValidationError(
+            'Test Error')
+
+        with self.assertLogs('coriolis.schemas', level=logging.WARN):
+            result = schemas.validate_value(test_value, test_schema,
+                                            raise_on_error=False)
+        self.assertEqual(result, False)

coriolis/tests/test_secrets.py

@@ -0,0 +1,72 @@
+# Copyright 2023 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import json
+from unittest import mock
+
+from barbicanclient import client as barbican_client
+import keystoneauth1
+
+from coriolis import keystone
+from coriolis import secrets
+from coriolis.tests import test_base
+
+
+class SecretsTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis secrets module."""
+
+    @mock.patch.object(keystone, 'create_keystone_session')
+    @mock.patch.object(barbican_client, 'Client')
+    def test_get_barbican_secret_payload(self, mock_client, mock_session):
+        mock_session.return_value = mock.sentinel.session
+        mock_secret = mock.MagicMock()
+        mock_secret.payload = mock.sentinel.payload
+        mock_barbican = mock.MagicMock()
+        mock_barbican.secrets.get.return_value = mock_secret
+        mock_client.return_value = mock_barbican
+
+        result = secrets._get_barbican_secret_payload(
+            mock.sentinel.ctxt, mock.sentinel.secret_ref)
+
+        self.assertEqual(mock.sentinel.payload, result)
+        mock_session.assert_called_once_with(mock.sentinel.ctxt)
+        mock_client.assert_called_once_with(session=mock.sentinel.session)
+        mock_barbican.secrets.get.assert_called_once_with(
+            mock.sentinel.secret_ref)
+
+    @mock.patch.object(secrets, '_get_barbican_secret_payload')
+    def test_get_secret_success(self, mock_get_payload):
+        mock_get_payload.return_value = json.dumps({'key': 'value'})
+
+        result = secrets.get_secret(mock.sentinel.ctxt,
+                                    mock.sentinel.secret_ref)
+
+        self.assertEqual({'key': 'value'}, result)
+        mock_get_payload.assert_called_once_with(
+            mock.sentinel.ctxt, mock.sentinel.secret_ref)
+
+    @mock.patch.object(secrets, '_get_barbican_secret_payload')
+    def test_get_secret_unauthorized(self, mock_get_payload):
+        mock_ctxt = mock.MagicMock()
+        mock_get_payload.side_effect = [
+            keystoneauth1.exceptions.http.Unauthorized,
+            json.dumps({'key': 'value'})]
+
+        with mock.patch('copy.deepcopy', side_effect=lambda func: func):
+            result = secrets.get_secret(mock_ctxt, mock.sentinel.secret_ref)
+
+        self.assertEqual({'key': 'value'}, result)
+        self.assertEqual(2, mock_get_payload.call_count)
+        self.assertEqual(mock_ctxt.trust_id, None)
+        mock_get_payload.assert_called_with(
+            mock_ctxt, mock.sentinel.secret_ref)
+
+    @mock.patch.object(secrets, '_get_barbican_secret_payload')
+    def test_get_secret_raises_value_error(self, mock_get_payload):
+        mock_get_payload.side_effect = ValueError("Test exception")
+
+        self.assertRaises(ValueError, secrets.get_secret, mock.sentinel.ctxt,
+                          mock.sentinel.secret_ref)
+
+        mock_get_payload.assert_called_once_with(
+            mock.sentinel.ctxt, mock.sentinel.secret_ref)