diff --git a/coriolisclient/cli/deployments.py b/coriolisclient/cli/deployments.py index 31eb7ee..74c6139 100644 --- a/coriolisclient/cli/deployments.py +++ b/coriolisclient/cli/deployments.py @@ -200,7 +200,7 @@ def get_parser(self, prog_name): return parser def take_action(self, args): - m = self.app.client_manager.coriolis.deployments + d = self.app.client_manager.coriolis.deployments user_scripts = cli_utils.compose_user_scripts( args.global_scripts, args.instance_scripts) instance_osmorphing_minion_pool_mappings = None @@ -209,7 +209,7 @@ def take_action(self, args): mp['instance_id']: mp['pool_id'] for mp in args.instance_osmorphing_minion_pool_mappings} - deployment = m.create_from_transfer( + deployment = d.create_from_transfer( args.transfer, args.clone_disks, args.force, diff --git a/coriolisclient/tests/cli/test_deployments.py b/coriolisclient/tests/cli/test_deployments.py new file mode 100644 index 0000000..f00065f --- /dev/null +++ b/coriolisclient/tests/cli/test_deployments.py @@ -0,0 +1,340 @@ +# Copyright (c) 2024 Cloudbase Solutions Srl +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +# implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import argparse +import copy +import os +from unittest import mock + +from coriolisclient.cli import deployments +from coriolisclient.tests import test_base +from coriolisclient.v1 import common +from coriolisclient.v1 import deployments as v1_deployments + +DEPLOYMENT_ID = "depl_id" +DEPLOYMENT_DATA = { + "storage_mappings": None, + "id": DEPLOYMENT_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, +} +DEPLOYMENT_FORMATTED_DATA = [ + "depl_id", "COMPLETED", None, None, "transfer_1", "replica", "res1", + "instance1", "", "source1", None, "dest1", None, '{}', False, + '{\n "opt1": "env1"\n}', '{}', '{\n "net_source": "net_dest"\n}', "", "", + None, '{}', "", '{}'] + +DEPLOYMENT_LIST_DATA = { + "id": "1", + "transfer_id": "2", + "last_execution_status": "RUNNING", + "instances": ["instance1", "instance2"], + "notes": "test_notes", + "created_at": "2024-11-28T15:18:25.000000", +} +DEPLOYMENT_LIST_FORMATTED_DATA = ( + "1", "2", "RUNNING", "instance1\ninstance2", "test_notes", + "2024-11-28T15:18:25.000000") +APP_MOCK = mock.MagicMock() + + +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 = mock.Mock(**DEPLOYMENT_LIST_DATA) + self.assertEqual( + DEPLOYMENT_LIST_FORMATTED_DATA, + 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 = { + **DEPLOYMENT_DATA, + "info": {"depl": "info"}, + } + obj = mock.Mock(**obj_data) + obj.to_dict.return_value = obj_data + expected_result = copy.copy(DEPLOYMENT_FORMATTED_DATA) + expected_result.append(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.mock_app = APP_MOCK + self.cli = deployments.CreateDeployment(self.mock_app, '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(global_scripts=None, instance_scripts=None) + args.instance_osmorphing_minion_pool_mappings = [ + {"instance_id": "instance1", "pool_id": "pool1"}] + mock_fun = (self.mock_app.client_manager.coriolis.deployments. + create_from_transfer) + mock_fun.return_value = ( + v1_deployments.Deployment(mock.MagicMock(), DEPLOYMENT_DATA)) + expected_pool_mappings = {"instance1": "pool1"} + expected_user_scripts = {"global": {}, "instances": {}} + expected_data = copy.copy(DEPLOYMENT_FORMATTED_DATA) + + columns, data = self.cli.take_action(args) + + self.assertEqual( + deployments.DeploymentDetailFormatter().columns, columns) + self.assertEqual(expected_data, data) + mock_fun.assert_called_once_with( + args.transfer, args.clone_disks, args.force, args.skip_os_morphing, + user_scripts=expected_user_scripts, + instance_osmorphing_minion_pool_mappings=expected_pool_mappings) + + +class ShowDeploymentTestCase(test_base.CoriolisBaseTestCase): + + def setUp(self): + super(ShowDeploymentTestCase, self).setUp() + self.mock_app = APP_MOCK + self.cli = deployments.ShowDeployment(self.mock_app, 'app_args') + + def test_get_parser(self): + parser = self.cli.get_parser('coriolis') + args = parser.parse_args([DEPLOYMENT_ID, '--show-instances-data']) + self.assertEqual( + (DEPLOYMENT_ID, True), + (args.id, args.show_instances_data)) + + def test_take_action(self): + show_instances_data = False + args = mock.Mock( + id=DEPLOYMENT_ID, show_instances_data=show_instances_data) + mock_fun = self.mock_app.client_manager.coriolis.deployments.get + mock_fun.return_value = v1_deployments.Deployment( + mock.MagicMock(), DEPLOYMENT_DATA) + expected_data = DEPLOYMENT_FORMATTED_DATA + + columns, data = self.cli.take_action(args) + + self.assertEqual( + deployments.DeploymentDetailFormatter( + show_instances_data=show_instances_data).columns, columns) + self.assertEqual(expected_data, data) + mock_fun.assert_called_once_with(DEPLOYMENT_ID) + + +class CancelDeploymentTestCase(test_base.CoriolisBaseTestCase): + + def setUp(self): + super(CancelDeploymentTestCase, self).setUp() + self.mock_app = APP_MOCK + self.cli = deployments.CancelDeployment(self.mock_app, 'app_args') + + def test_get_parser(self): + parser = self.cli.get_parser('coriolis') + args = parser.parse_args([DEPLOYMENT_ID, '--force']) + self.assertEqual((DEPLOYMENT_ID, True), (args.id, args.force)) + + def test_take_action(self): + force = True + args = mock.Mock(id=DEPLOYMENT_ID, force=force) + mock_fun = self.mock_app.client_manager.coriolis.deployments.cancel + + self.cli.take_action(args) + + mock_fun.assert_called_once_with(DEPLOYMENT_ID, force) + + +class DeleteDeploymentTestCase(test_base.CoriolisBaseTestCase): + + def setUp(self): + super(DeleteDeploymentTestCase, self).setUp() + self.mock_app = APP_MOCK + self.cli = deployments.DeleteDeployment(self.mock_app, 'app_args') + + def test_get_parser(self): + parser = self.cli.get_parser('coriolis') + args = parser.parse_args([DEPLOYMENT_ID]) + self.assertEqual(DEPLOYMENT_ID, args.id) + + def test_take_action(self): + args = mock.Mock(id=DEPLOYMENT_ID) + mock_fun = self.mock_app.client_manager.coriolis.deployments.delete + + self.cli.take_action(args) + mock_fun.assert_called_once_with(DEPLOYMENT_ID) + + +class ListDeploymentTestCase(test_base.CoriolisBaseTestCase): + + def setUp(self): + super(ListDeploymentTestCase, self).setUp() + self.mock_app = APP_MOCK + self.cli = deployments.ListDeployment(self.mock_app, 'app_args') + + def test_get_parser(self): + parser = self.cli.get_parser('coriolis') + self.assertIsInstance(parser, argparse.ArgumentParser) + + def test_take_action(self): + mock_fun = self.mock_app.client_manager.coriolis.deployments.list + mock_fun.return_value = [ + v1_deployments.Deployment(mock.MagicMock(), DEPLOYMENT_LIST_DATA)] + + columns, data = self.cli.take_action(mock.ANY) + + self.assertEqual(deployments.DeploymentFormatter().columns, columns) + self.assertEqual([DEPLOYMENT_LIST_FORMATTED_DATA], list(data)) diff --git a/coriolisclient/tests/v1/test_deployments.py b/coriolisclient/tests/v1/test_deployments.py new file mode 100644 index 0000000..f92fe9a --- /dev/null +++ b/coriolisclient/tests/v1/test_deployments.py @@ -0,0 +1,124 @@ +# Copyright (c) 2024 Cloudbase Solutions Srl +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +# implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from unittest import mock + +from coriolisclient.tests import test_base +from coriolisclient.v1 import common +from coriolisclient.v1 import deployments + +DEPLOYMENT_ID = "1" + + +class DeploymentResourceTestCase(test_base.CoriolisBaseTestCase): + + def setUp(self): + super(DeploymentResourceTestCase, self).setUp() + self.source_env = {"source_opt": "env_value"} + self.dest_env = {"dest_opt": "env_value"} + self.transfer_result = {"result": "value"} + self.task = {"task": "type_1"} + self.info = { + "source_environment": self.source_env, + "destination_environment": self.dest_env, + "transfer_result": self.transfer_result, + "tasks": [self.task], + } + self.deployment = deployments.Deployment(None, self.info) + + def test_source_environment(self): + result = self.deployment.source_environment + + self.assertIsInstance(result, common.SourceEnvironment) + self.assertEqual(self.source_env, result._info) + + def test_destination_environment(self): + result = self.deployment.destination_environment + + self.assertIsInstance(result, common.DestinationEnvironment) + self.assertEqual(self.dest_env, result._info) + + def test_transfer_result(self): + result = self.deployment.transfer_result + + self.assertIsInstance(result, common.TransferResult) + self.assertEqual(self.transfer_result, result._info) + + def test_task(self): + result = self.deployment.tasks + + [self.assertIsInstance(t, common.Task) for t in result] + self.assertEqual(self.task, result[0]._info) + + +class DeploymentManagerTestCase(test_base.CoriolisBaseTestCase): + + def setUp(self): + super(DeploymentManagerTestCase, self).setUp() + mock_client = mock.Mock() + self.deployments = deployments.DeploymentManager(mock_client) + + def test_list(self): + with mock.patch.object(self.deployments, '_list') as mock_list: + result = self.deployments.list(detail=True) + self.assertEqual(mock_list.return_value, result) + mock_list.assert_called_once_with( + '/deployments/detail', 'deployments') + + def test_get(self): + deployment = mock.Mock(uuid=DEPLOYMENT_ID) + with mock.patch.object(self.deployments, '_get') as mock_get: + result = self.deployments.get(deployment) + self.assertEqual(mock_get.return_value, result) + mock_get.assert_called_once_with( + f'/deployments/{DEPLOYMENT_ID}', 'deployment') + + def test_create_from_transfer(self): + with mock.patch.object(self.deployments, '_post') as mock_post: + expected_data = { + "deployment": { + "transfer_id": mock.sentinel.transfer_id, + "clone_disks": True, + "force": False, + "skip_os_morphing": False, + "user_scripts": None, + "instance_osmorphing_minion_pool_mappings": + mock.sentinel.pool_mappings, + } + } + result = self.deployments.create_from_transfer( + mock.sentinel.transfer_id, clone_disks=True, force=False, + skip_os_morphing=False, user_scripts=None, + instance_osmorphing_minion_pool_mappings=( + mock.sentinel.pool_mappings)) + self.assertEqual(mock_post.return_value, result) + mock_post.assert_called_once_with( + "/deployments", expected_data, "deployment") + + def test_delete(self): + with mock.patch.object(self.deployments, '_delete') as mock_delete: + result = self.deployments.delete(DEPLOYMENT_ID) + self.assertEqual(mock_delete.return_value, result) + mock_delete.assert_called_once_with( + f'/deployments/{DEPLOYMENT_ID}') + + def test_cancel(self): + force = False + expected_data = {"cancel": {"force": force}} + with mock.patch.object(self.deployments.client, 'post') as mock_post: + result = self.deployments.cancel(DEPLOYMENT_ID, force=force) + self.assertEqual(mock_post.return_value, result) + mock_post.assert_called_once_with( + f"/deployments/{DEPLOYMENT_ID}/actions", json=expected_data)