WiP - Add deployments unit tests

coriolisclient/tests/cli/test_deployments.py

@@ -0,0 +1,208 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+import copy
+import os
+from unittest import mock
+
+from cliff import command
+from cliff import lister
+from cliff import show
+
+from coriolisclient.cli import deployments
+from coriolisclient.tests import test_base
+from coriolisclient.v1 import common
+from coriolisclient.v1 import deployments as v1_deployments
+
+class DeploymentFormatterTestCase(test_base.CoriolisBaseTestCase):
+
+    def setUp(self):
+        super(DeploymentFormatterTestCase, self).setUp()
+        self.formatter = deployments.DeploymentFormatter()
+
+    def test__get_sorted_list(self):
+        obj1 = mock.Mock(created_at="2024-10-24T18:51:32.000000")
+        obj2 = mock.Mock(created_at="2024-11-18T15:18:25.000000")
+        obj3 = mock.Mock(created_at="2024-11-28T15:18:25.000000")
+        obj_list = [obj1, obj3, obj2]
+        self.assertEqual(
+            [obj1, obj2, obj3], self.formatter._get_sorted_list(obj_list))
+
+    def test__get_formatted_data(self):
+        obj_data = {
+            "id": "1",
+            "transfer_id": "2",
+            "last_execution_status": "RUNNING",
+            "instances": ["instance1", "instance2"],
+            "notes": "test_notes",
+            "created_at": "2024-11-28T15:18:25.000000",
+        }
+        obj = mock.Mock(**obj_data)
+        expected_instance_list = "instance1\ninstance2"
+        data = copy.deepcopy(obj_data)
+        data['instances'] = expected_instance_list
+        expected_result = tuple(data.values())
+        self.assertEqual(
+            expected_result, self.formatter._get_formatted_data(obj))
+
+
+class DeploymentDetailFormatterTestCase(test_base.CoriolisBaseTestCase):
+
+    def setUp(self):
+        super(DeploymentDetailFormatterTestCase, self).setUp()
+        self.formatter = deployments.DeploymentDetailFormatter(
+            show_instances_data=True)
+        self.progress_updates = [
+            {"created_at": "2024-10-24T18:51:32.000000",
+             "index": 0,
+             "message": "message 0"},
+            {"created_at": "2024-11-18T15:18:25.000000",
+             "index": 2,
+             "message": "message 2"},
+            {"created_at": "2024-11-28T15:18:25.000000",
+             "index": 1,
+             "message": "message 1"},
+        ]
+        self.formatted_progress_updates = (
+            f"2024-10-24T18:51:32.000000 message 0{os.linesep}"
+            f"2024-11-28T15:18:25.000000 message 1{os.linesep}"
+            f"2024-11-18T15:18:25.000000 message 2")
+        self.tasks_data = [
+            {
+                "id": "1",
+                "task_type": "type1",
+                "instance": "instance1",
+                "status": "COMPLETED",
+                "depends_on": ["task0"],
+                "exception_details": None,
+                "progress_updates": self.progress_updates,
+            },
+            {
+                "id": "2",
+                "task_type": "type2",
+                "instance": "instance1",
+                "status": "COMPLETED",
+                "depends_on": ["task0"],
+                "exception_details": None,
+                "progress_updates": self.progress_updates,
+            }
+        ]
+        self.formatted_tasks = [
+            f'id: 1{os.linesep}'
+            f'task_type: type1{os.linesep}'
+            f'instance: instance1{os.linesep}'
+            f'status: COMPLETED{os.linesep}'
+            f'depends_on: task0{os.linesep}'
+            f'exception_details: {os.linesep}'
+            f'progress_updates:{os.linesep}'
+            f'{self.formatted_progress_updates}',
+
+            f'id: 2{os.linesep}'
+            f'task_type: type2{os.linesep}'
+            f'instance: instance1{os.linesep}'
+            f'status: COMPLETED{os.linesep}'
+            f'depends_on: task0{os.linesep}'
+            f'exception_details: {os.linesep}'
+            f'progress_updates:{os.linesep}'
+            f'{self.formatted_progress_updates}'
+        ]
+        self.manager_mock = mock.MagicMock()
+
+    def test_init_no_instances_data(self):
+        formatter = deployments.DeploymentDetailFormatter(
+            show_instances_data=False)
+        self.assertNotIn('instances_data', formatter.columns)
+
+    def test__format_instances(self):
+        obj = mock.Mock(instances=["instance2", "instance3", "instance1"])
+        self.assertEqual(
+            f"instance1{os.linesep}instance2{os.linesep}instance3",
+            self.formatter._format_instances(obj))
+
+    def test__format_progress_updates(self):
+        task_dict = {
+            "progress_updates": self.progress_updates}
+        self.assertEqual(
+            self.formatted_progress_updates,
+            self.formatter._format_progress_updates(task_dict))
+
+    def test__format_task(self):
+        task_data = self.tasks_data[0]
+
+        task = common.Task(self.manager_mock, task_data)
+
+        self.assertEqual(
+            self.formatted_tasks[0], self.formatter._format_task(task))
+
+    def test__format_tasks(self):
+        obj = mock.Mock(
+            tasks=[common.Task(self.manager_mock, task_data)
+                   for task_data in self.tasks_data])
+        expected_result = (f'{self.formatted_tasks[0]}{os.linesep}{os.linesep}'
+                           f'{self.formatted_tasks[1]}')
+        self.assertEqual(expected_result, self.formatter._format_tasks(obj))
+
+    def test__get_formatted_data(self):
+        obj_data = {
+            "storage_mappings": None,
+            "id": "depl_id",
+            "last_execution_status": "COMPLETED",
+            "created_at": None,
+            "updated_at": None,
+            "transfer_id": "transfer_1",
+            "transfer_scenario_type": "replica",
+            "reservation_id": "res1",
+            "instances": ["instance1"],
+            "notes": "",
+            "origin_endpoint_id": "source1",
+            "origin_minion_pool_id": None,
+            "destination_endpoint_id": "dest1",
+            "destination_minion_pool_id": None,
+            "instance_osmorphing_minion_pool_mappings": None,
+            "shutdown_instances": False,
+            "destination_environment": {"opt1": "env1"},
+            "source_environment": None,
+            "network_map": {"net_source": "net_dest"},
+            "user_scripts": None,
+            "tasks": [],
+            "transfer_result": None,
+            "info": {"depl": "info"},
+        }
+        obj = mock.Mock(**obj_data)
+        obj.to_dict.return_value = obj_data
+        expected_dest_env = '{\n  "opt1": "env1"\n}'
+        expected_net_map = '{\n  "net_source": "net_dest"\n}'
+        expected_result = [
+            "depl_id", "COMPLETED", None, None, "transfer_1", "replica",
+            "res1", "instance1", "", "source1", None, "dest1", None, '{}',
+            False, expected_dest_env, '{}', expected_net_map, "", "", None,
+            '{}', "", '{}', obj_data['info']]
+        self.assertEqual(
+            expected_result, self.formatter._get_formatted_data(obj))
+
+
+class CreateDeploymentTestCase(test_base.CoriolisBaseTestCase):
+
+    def setUp(self):
+        super(CreateDeploymentTestCase, self).setUp()
+        self.cli = deployments.CreateDeployment(
+            'coriolis', 'app_arg')
+
+    def test_get_parser(self):
+        parser = self.cli.get_parser('coriolis')
+        global_script = "linux=/linux/path"
+        instance_script = "instance1=/instance1/path"
+        args = parser.parse_args([
+            'transfer_id', '--force', '--dont-clone-disks',
+            '--skip-os-morphing', '--user-script-global', global_script,
+            '--user-script-instance', instance_script])
+        self.assertEqual(
+            ('transfer_id', True, False, True, [global_script],
+             [instance_script]),
+            (args.transfer, args.force, args.clone_disks,
+             args.skip_os_morphing, args.global_scripts,
+             args.instance_scripts))
+
+    def test_take_action(self):
+        # args = mock.Mock()
+        # args.instance_osmorphing_minion_pool_mappings = []
+        pass