Add unit tests for coriolisclient.v1.* modules

coriolisclient/tests/v1/test_migrations.py

@@ -0,0 +1,206 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolisclient.tests import test_base
+from coriolisclient.v1 import migrations
+
+
+class MigrationTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis v1 Migration."""
+
+    def test_properties(self):
+        mock_client = mock.Mock()
+        self.migration = migrations.Migration(
+            mock_client,
+            {
+                "source_environment": {
+                    "source_environment1": mock.sentinel.source_environment},
+                "destination_environment": {
+                    "destination_environment1":
+                        mock.sentinel.destination_environment},
+                "transfer_result": {
+                    "transfer_result1": mock.sentinel.transfer_result},
+                "tasks": [{"task1": mock.sentinel.task1},
+                          {"task2": mock.sentinel.task2}]
+            }
+        )
+        self.assertEqual(
+            (
+                mock.sentinel.source_environment,
+                mock.sentinel.destination_environment,
+                mock.sentinel.transfer_result,
+                mock.sentinel.task1,
+                mock.sentinel.task2
+            ),
+            (
+                self.migration.source_environment.source_environment1,
+                (self.migration.destination_environment.
+                 destination_environment1),
+                self.migration.transfer_result.transfer_result1,
+                self.migration.tasks[0].task1,
+                self.migration.tasks[1].task2
+            )
+        )
+
+    @mock.patch.object(migrations.Migration, "get")
+    def test_properties_none(self, mock_get):
+        mock_client = mock.Mock()
+        self.migration = migrations.Migration(
+            mock_client,
+            {}
+        )
+        self.assertEqual(
+            (
+                None,
+                None,
+                None,
+                []
+            ),
+            (
+                self.migration.source_environment,
+                self.migration.destination_environment,
+                self.migration.transfer_result,
+                self.migration.tasks,
+            )
+        )
+        mock_get.assert_called_once()
+
+
+class MigrationManagerTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis v1 Migration Manager."""
+
+    def setUp(self):
+        mock_client = mock.Mock()
+        super(MigrationManagerTestCase, self).setUp()
+        self.migration = migrations.MigrationManager(mock_client)
+
+    @mock.patch.object(migrations.MigrationManager, "_list")
+    def test_list(self, mock_list):
+        result = self.migration.list(detail=True)
+
+        self.assertEqual(
+            mock_list.return_value,
+            result
+        )
+        mock_list.assert_called_once_with("/migrations/detail", "migrations")
+
+    @mock.patch.object(migrations.MigrationManager, "_get")
+    def test_get(self, mock_get):
+        result = self.migration.get(mock.sentinel.migration)
+
+        self.assertEqual(
+            mock_get.return_value,
+            result
+        )
+        mock_get.assert_called_once_with(
+            "/migrations/sentinel.migration", "migration")
+
+    @mock.patch.object(migrations.MigrationManager, "_post")
+    def test_create(self, mock_post):
+        expected_data = {
+            "origin_endpoint_id": mock.sentinel.origin_endpoint_id,
+            "destination_endpoint_id": mock.sentinel.destination_endpoint_id,
+            "source_environment": mock.sentinel.source_environment,
+            "destination_environment": {
+                "network_map": mock.sentinel.network_map,
+                "storage_mappings": mock.sentinel.storage_mappings
+            },
+            "instances": mock.sentinel.instances,
+            "network_map": mock.sentinel.network_map,
+            "notes": mock.sentinel.notes,
+            "storage_mappings": mock.sentinel.storage_mappings,
+            "skip_os_morphing": False,
+            "replication_count": mock.sentinel.replication_count,
+            "shutdown_instances": mock.sentinel.shutdown_instances,
+            "user_scripts": mock.sentinel.user_scripts,
+            "origin_minion_pool_id": mock.sentinel.origin_minion_pool_id,
+            "destination_minion_pool_id":
+                mock.sentinel.destination_minion_pool_id,
+            "instance_osmorphing_minion_pool_mappings":
+                mock.sentinel.instance_osmorphing_minion_pool_mappings,
+        }
+        expected_data = {"migration": expected_data}
+
+        result = self.migration.create(
+            mock.sentinel.origin_endpoint_id,
+            mock.sentinel.destination_endpoint_id,
+            mock.sentinel.source_environment,
+            {
+                "network_map": mock.sentinel.network_map,
+                "storage_mappings": mock.sentinel.storage_mappings
+            },
+            mock.sentinel.instances,
+            network_map=None,
+            notes=mock.sentinel.notes,
+            storage_mappings=None,
+            skip_os_morphing=False,
+            replication_count=mock.sentinel.replication_count,
+            shutdown_instances=mock.sentinel.shutdown_instances,
+            user_scripts=mock.sentinel.user_scripts,
+            origin_minion_pool_id=mock.sentinel.origin_minion_pool_id,
+            destination_minion_pool_id=
+            mock.sentinel.destination_minion_pool_id,
+            instance_osmorphing_minion_pool_mappings=
+            mock.sentinel.instance_osmorphing_minion_pool_mappings,
+        )
+
+        self.assertEqual(
+            mock_post.return_value,
+            result
+        )
+        mock_post.assert_called_once_with(
+            "/migrations", expected_data, "migration")
+
+    @mock.patch.object(migrations.MigrationManager, "_post")
+    def test_create_from_replica(self, mock_post):
+        expected_data = {
+            "replica_id": mock.sentinel.replica_id,
+            "clone_disks": False,
+            "force": False,
+            "skip_os_morphing": False,
+            "user_scripts": mock.sentinel.user_scripts,
+            "instance_osmorphing_minion_pool_mappings":
+                mock.sentinel.instance_osmorphing_minion_pool_mappings,
+        }
+        expected_data = {"migration": expected_data}
+
+        result = self.migration.create_from_replica(
+            mock.sentinel.replica_id,
+            clone_disks=False,
+            force=False,
+            skip_os_morphing=False,
+            user_scripts=mock.sentinel.user_scripts,
+            instance_osmorphing_minion_pool_mappings=
+            mock.sentinel.instance_osmorphing_minion_pool_mappings,
+        )
+
+        self.assertEqual(
+            mock_post.return_value,
+            result
+        )
+        mock_post.assert_called_once_with(
+            "/migrations", expected_data, "migration")
+
+    @mock.patch.object(migrations.MigrationManager, "_delete")
+    def test_delete(self, mock_delete):
+        result = self.migration.delete(mock.sentinel.migration)
+
+        self.assertEqual(
+            mock_delete.return_value,
+            result
+        )
+        mock_delete.assert_called_once_with(
+            "/migrations/%s" % mock.sentinel.migration)
+
+    def test_cancel(self):
+        result = self.migration.cancel(mock.sentinel.migration, force=False)
+
+        self.assertEqual(
+            self.migration.client.post.return_value,
+            result
+        )
+        self.migration.client.post.assert_called_once_with(
+            "/migrations/%s/actions" % mock.sentinel.migration,
+            json={'cancel': {'force': False}})

coriolisclient/tests/v1/test_minion_pools.py

@@ -0,0 +1,141 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolisclient.tests import test_base
+from coriolisclient.v1 import minion_pools
+
+
+class MinionPoolManagerTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis v1 Migration Pool Manager."""
+
+    def setUp(self):
+        mock_client = mock.Mock()
+        super(MinionPoolManagerTestCase, self).setUp()
+        self.minion_pool = minion_pools.MinionPoolManager(mock_client)
+
+    @mock.patch.object(minion_pools.MinionPoolManager, "_list")
+    def test_list(self, mock_list):
+        result = self.minion_pool.list()
+
+        self.assertEqual(
+            mock_list.return_value,
+            result
+        )
+        mock_list.assert_called_once_with(
+            "/minion_pools", response_key="minion_pools")
+
+    @mock.patch.object(minion_pools.MinionPoolManager, "_get")
+    def test_get(self, mock_get):
+        result = self.minion_pool.get(mock.sentinel.minion_pool)
+
+        self.assertEqual(
+            mock_get.return_value,
+            result
+        )
+        mock_get.assert_called_once_with(
+            "/minion_pools/sentinel.minion_pool", response_key="minion_pool")
+
+    @mock.patch.object(minion_pools.MinionPoolManager, "_post")
+    def test_create(self, mock_post):
+        expected_data = {
+            "name": mock.sentinel.name,
+            "endpoint_id": mock.sentinel.endpoint,
+            "platform": mock.sentinel.platform,
+            "os_type": mock.sentinel.os_type,
+            "environment_options": mock.sentinel.environment_options,
+            "minimum_minions": mock.sentinel.minimum_minions,
+            "maximum_minions": mock.sentinel.maximum_minions,
+            "minion_max_idle_time": mock.sentinel.minion_max_idle_time,
+            "minion_retention_strategy":
+                mock.sentinel.minion_retention_strategy,
+            "notes": mock.sentinel.notes,
+            "skip_allocation": False,
+        }
+        expected_data = {"minion_pool": expected_data}
+
+        result = self.minion_pool.create(
+            mock.sentinel.name,
+            mock.sentinel.endpoint,
+            mock.sentinel.platform,
+            mock.sentinel.os_type,
+            environment_options=mock.sentinel.environment_options,
+            minimum_minions=mock.sentinel.minimum_minions,
+            maximum_minions=mock.sentinel.maximum_minions,
+            minion_max_idle_time=mock.sentinel.minion_max_idle_time,
+            minion_retention_strategy=mock.sentinel.minion_retention_strategy,
+            notes=mock.sentinel.notes,
+            skip_allocation=False,
+        )
+
+        self.assertEqual(
+            mock_post.return_value,
+            result
+        )
+        mock_post.assert_called_once_with(
+            "/minion_pools", expected_data, response_key="minion_pool")
+
+    @mock.patch.object(minion_pools.MinionPoolManager, "_put")
+    def test_update(self, mock_put):
+        result = self.minion_pool.update(
+            mock.sentinel.minion_pool, mock.sentinel.updated_values)
+
+        self.assertEqual(
+            mock_put.return_value,
+            result
+        )
+        mock_put.assert_called_once_with(
+            "/minion_pools/%s" % mock.sentinel.minion_pool,
+            {"minion_pool": mock.sentinel.updated_values}, 'minion_pool')
+
+    @mock.patch.object(minion_pools.MinionPoolManager, "_delete")
+    def test_delete(self, mock_delete):
+        result = self.minion_pool.delete(mock.sentinel.minion_pool)
+
+        self.assertEqual(
+            mock_delete.return_value,
+            result
+        )
+        mock_delete.assert_called_once_with(
+            "/minion_pools/%s" % mock.sentinel.minion_pool)
+
+    @mock.patch.object(minion_pools.MinionPoolManager, "_post")
+    def test_allocate_minion_pool(self, mock_post):
+        result = self.minion_pool.allocate_minion_pool(
+            mock.sentinel.minion_pool)
+
+        self.assertEqual(
+            mock_post.return_value,
+            result
+        )
+        mock_post.assert_called_once_with(
+            "/minion_pools/%s/actions" % mock.sentinel.minion_pool,
+            {'allocate': None}, response_key='minion_pool')
+
+    @mock.patch.object(minion_pools.MinionPoolManager, "_post")
+    def test_refresh_minion_pool(self, mock_post):
+        result = self.minion_pool.refresh_minion_pool(
+            mock.sentinel.minion_pool)
+
+        self.assertEqual(
+            mock_post.return_value,
+            result
+        )
+        mock_post.assert_called_once_with(
+            "/minion_pools/%s/actions" % mock.sentinel.minion_pool,
+            {'refresh': None}, response_key='minion_pool')
+
+    @mock.patch.object(minion_pools.MinionPoolManager, "_post")
+    def test_deallocate_minion_pool(self, mock_post):
+        result = self.minion_pool.deallocate_minion_pool(
+            mock.sentinel.minion_pool, force=False)
+
+        self.assertEqual(
+            mock_post.return_value,
+            result
+        )
+        mock_post.assert_called_once_with(
+            "/minion_pools/%s/actions" % mock.sentinel.minion_pool,
+            {'deallocate': {'force': False}},
+            response_key='minion_pool')