Add unit tests for coriolis.cmd.* modules

coriolis/tests/cmd/test_api.py

@@ -0,0 +1,80 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import sys
+from unittest import mock
+
+from coriolis.cmd import api
+from coriolis import service
+from coriolis.tests import test_base
+from coriolis import utils
+
+
+class ApiTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis api CMD"""
+
+    @mock.patch.object(service, 'service')
+    @mock.patch.object(service, 'WSGIService')
+    @mock.patch.object(utils, 'setup_logging')
+    @mock.patch('coriolis.cmd.api.CONF')
+    @mock.patch.object(service, 'get_worker_count_from_args')
+    @mock.patch.object(sys, 'argv')
+    def test_main(
+        self,
+        mock_argv,
+        mock_get_worker_count_from_args,
+        mock_conf,
+        mock_setup_logging,
+        mock_WSGIService,
+        mock_service
+    ):
+        worker_count = mock.sentinel.worker_count
+        args = ['mock_arg_1', 'mock_arg_2']
+        mock_get_worker_count_from_args.return_value = (worker_count, args)
+
+        api.main()
+
+        mock_get_worker_count_from_args.assert_called_once_with(mock_argv)
+        mock_conf.assert_called_once_with(
+            ['mock_arg_2'], project='coriolis', version="1.0.0")
+        mock_setup_logging.assert_called_once()
+        mock_WSGIService.assert_called_once_with(
+            'coriolis-api', worker_count=worker_count)
+        mock_service.launch.assert_called_once_with(
+            mock_conf, mock_WSGIService.return_value,
+            workers=mock_WSGIService.return_value.
+            get_workers_count.return_value)
+        mock_service.launch.return_value.wait.assert_called_once()
+
+    @mock.patch.object(service, 'service')
+    @mock.patch.object(service, 'WSGIService')
+    @mock.patch.object(utils, 'setup_logging')
+    @mock.patch('coriolis.cmd.api.CONF')
+    @mock.patch.object(service, 'get_worker_count_from_args')
+    @mock.patch.object(sys, 'argv')
+    def test_main_no_worker_count(
+        self,
+        mock_argv,
+        mock_get_worker_count_from_args,
+        mock_conf,
+        mock_setup_logging,
+        mock_WSGIService,
+        mock_service
+    ):
+        worker_count = None
+        args = ['mock_arg_1', 'mock_arg_2']
+        mock_get_worker_count_from_args.return_value = (worker_count, args)
+
+        api.main()
+
+        mock_get_worker_count_from_args.assert_called_once_with(mock_argv)
+        mock_conf.assert_called_once_with(
+            ['mock_arg_2'], project='coriolis', version="1.0.0")
+        mock_setup_logging.assert_called_once()
+        mock_WSGIService.assert_called_once_with(
+            'coriolis-api', worker_count=mock_conf.api.worker_count)
+        mock_service.launch.assert_called_once_with(
+            mock_conf, mock_WSGIService.return_value,
+            workers=mock_WSGIService.return_value.
+            get_workers_count.return_value)
+        mock_service.launch.return_value.wait.assert_called_once()

coriolis/tests/cmd/test_conductor.py

@@ -0,0 +1,96 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import sys
+from unittest import mock
+
+from coriolis.cmd import conductor
+from coriolis.conductor.rpc import server as rpc_server
+from coriolis import constants
+from coriolis import service
+from coriolis.tests import test_base
+from coriolis import utils
+
+
+class ConductorTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis conductor CMD"""
+
+    @mock.patch.object(service, 'service')
+    @mock.patch.object(rpc_server, 'ConductorServerEndpoint')
+    @mock.patch.object(service, 'MessagingService')
+    @mock.patch.object(service, 'check_locks_dir_empty')
+    @mock.patch.object(utils, 'setup_logging')
+    @mock.patch('coriolis.cmd.conductor.CONF')
+    @mock.patch.object(service, 'get_worker_count_from_args')
+    @mock.patch.object(sys, 'argv')
+    def test_main(
+        self,
+        mock_argv,
+        mock_get_worker_count_from_args,
+        mock_conf,
+        mock_setup_logging,
+        mock_check_locks_dir_empty,
+        mock_MessagingService,
+        mock_ConductorServerEndpoint,
+        mock_service
+    ):
+        worker_count = mock.sentinel.worker_count
+        args = ['mock_arg_1', 'mock_arg_2']
+        mock_get_worker_count_from_args.return_value = (worker_count, args)
+
+        conductor.main()
+
+        mock_get_worker_count_from_args.assert_called_once_with(mock_argv)
+        mock_conf.assert_called_once_with(
+            ['mock_arg_2'], project='coriolis', version="1.0.0")
+        mock_setup_logging.assert_called_once()
+        mock_check_locks_dir_empty.assert_called_once()
+        mock_MessagingService.assert_called_once_with(
+            constants.CONDUCTOR_MAIN_MESSAGING_TOPIC,
+            [mock_ConductorServerEndpoint.return_value],
+            rpc_server.VERSION, worker_count=worker_count)
+        mock_service.launch.assert_called_once_with(
+            mock_conf, mock_MessagingService.return_value,
+            workers=mock_MessagingService.return_value.
+            get_workers_count.return_value)
+        mock_service.launch.return_value.wait.assert_called_once()
+
+    @mock.patch.object(service, 'service')
+    @mock.patch.object(rpc_server, 'ConductorServerEndpoint')
+    @mock.patch.object(service, 'MessagingService')
+    @mock.patch.object(service, 'check_locks_dir_empty')
+    @mock.patch.object(utils, 'setup_logging')
+    @mock.patch('coriolis.cmd.conductor.CONF')
+    @mock.patch.object(service, 'get_worker_count_from_args')
+    @mock.patch.object(sys, 'argv')
+    def test_main_no_worker_count(
+        self,
+        mock_argv,
+        mock_get_worker_count_from_args,
+        mock_conf,
+        mock_setup_logging,
+        mock_check_locks_dir_empty,
+        mock_MessagingService,
+        mock_ConductorServerEndpoint,
+        mock_service
+    ):
+        worker_count = None
+        args = ['mock_arg_1', 'mock_arg_2']
+        mock_get_worker_count_from_args.return_value = (worker_count, args)
+
+        conductor.main()
+
+        mock_get_worker_count_from_args.assert_called_once_with(mock_argv)
+        mock_conf.assert_called_once_with(
+            ['mock_arg_2'], project='coriolis', version="1.0.0")
+        mock_setup_logging.assert_called_once()
+        mock_check_locks_dir_empty.assert_called_once()
+        mock_MessagingService.assert_called_once_with(
+            constants.CONDUCTOR_MAIN_MESSAGING_TOPIC,
+            [mock_ConductorServerEndpoint.return_value],
+            rpc_server.VERSION, worker_count=mock_conf.conductor.worker_count)
+        mock_service.launch.assert_called_once_with(
+            mock_conf, mock_MessagingService.return_value,
+            workers=mock_MessagingService.return_value.
+            get_workers_count.return_value)
+        mock_service.launch.return_value.wait.assert_called_once()

coriolis/tests/cmd/test_db_sync.py

@@ -0,0 +1,35 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import sys
+from unittest import mock
+
+from coriolis.cmd import db_sync
+from coriolis.db import api as db_api
+from coriolis.tests import test_base
+from coriolis import utils
+
+
+class DBSyncTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis db_sync CMD"""
+
+    @mock.patch.object(db_api, 'db_sync')
+    @mock.patch.object(db_api, 'get_engine')
+    @mock.patch.object(utils, 'setup_logging')
+    @mock.patch('coriolis.cmd.db_sync.CONF')
+    @mock.patch.object(sys, 'argv')
+    def test_main(
+        self,
+        mock_argv,
+        mock_conf,
+        mock_setup_logging,
+        mock_get_engine,
+        mock_db_sync
+    ):
+        db_sync.main()
+
+        mock_conf.assert_called_once_with(
+            mock_argv[1:], project='coriolis', version="1.0.0")
+        mock_setup_logging.assert_called_once()
+        mock_get_engine.assert_called_once()
+        mock_db_sync.assert_called_once_with(mock_get_engine.return_value)

coriolis/tests/cmd/test_minion_manager.py

@@ -0,0 +1,48 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import sys
+from unittest import mock
+
+from coriolis.cmd import minion_manager
+from coriolis import constants
+from coriolis.minion_manager.rpc import server as rpc_server
+from coriolis import service
+from coriolis.tests import test_base
+from coriolis import utils
+
+
+class MinionManagerTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis minion_manager CMD"""
+
+    @mock.patch.object(service, 'service')
+    @mock.patch.object(service, 'MessagingService')
+    @mock.patch.object(rpc_server, 'MinionManagerServerEndpoint')
+    @mock.patch.object(utils, 'setup_logging')
+    @mock.patch('coriolis.cmd.minion_manager.CONF')
+    @mock.patch.object(sys, 'argv')
+    def test_main(
+        self,
+        mock_argv,
+        mock_conf,
+        mock_setup_logging,
+        mock_MinionManagerServerEndpoint,
+        mock_MessagingService,
+        mock_service
+    ):
+        minion_manager.main()
+
+        mock_conf.assert_called_once_with(
+            mock_argv[1:], project='coriolis', version="1.0.0")
+        mock_setup_logging.assert_called_once()
+        mock_MinionManagerServerEndpoint.assert_called_once()
+        mock_MessagingService.assert_called_once_with(
+            constants.MINION_MANAGER_MAIN_MESSAGING_TOPIC,
+            [mock_MinionManagerServerEndpoint.return_value],
+            rpc_server.VERSION,
+            worker_count=mock_conf.minion_manager.worker_count)
+        mock_service.launch.assert_called_once_with(
+            mock_conf, mock_MessagingService.return_value,
+            workers=mock_MessagingService.return_value.
+            get_workers_count.return_value)
+        mock_service.launch.return_value.wait.assert_called_once()

coriolis/tests/cmd/test_replica_cron.py

@@ -0,0 +1,48 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import sys
+from unittest import mock
+
+from coriolis.cmd import replica_cron
+from coriolis import constants
+from coriolis.replica_cron.rpc import server as rpc_server
+from coriolis import service
+from coriolis.tests import test_base
+from coriolis import utils
+
+
+class ReplicaCronTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis replica_cron CMD"""
+
+    @mock.patch.object(service, 'service')
+    @mock.patch.object(service, 'MessagingService')
+    @mock.patch.object(rpc_server, 'ReplicaCronServerEndpoint')
+    @mock.patch.object(utils, 'setup_logging')
+    @mock.patch('coriolis.cmd.replica_cron.CONF')
+    @mock.patch.object(sys, 'argv')
+    def test_main(
+        self,
+        mock_argv,
+        mock_conf,
+        mock_setup_logging,
+        mock_ReplicaCronServerEndpoint,
+        mock_MessagingService,
+        mock_service
+    ):
+        replica_cron.main()
+
+        mock_conf.assert_called_once_with(
+            mock_argv[1:], project='coriolis', version="1.0.0")
+        mock_setup_logging.assert_called_once()
+        mock_ReplicaCronServerEndpoint.assert_called_once()
+        mock_MessagingService.assert_called_once_with(
+            constants.REPLICA_CRON_MAIN_MESSAGING_TOPIC,
+            [mock_ReplicaCronServerEndpoint.return_value],
+            rpc_server.VERSION,
+            worker_count=1)
+        mock_service.launch.assert_called_once_with(
+            mock_conf, mock_MessagingService.return_value,
+            workers=mock_MessagingService.return_value.
+            get_workers_count.return_value)
+        mock_service.launch.return_value.wait.assert_called_once()

coriolis/tests/cmd/test_scheduler.py

@@ -0,0 +1,48 @@
+# Copyright 2024 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import sys
+from unittest import mock
+
+from coriolis.cmd import scheduler
+from coriolis import constants
+from coriolis.scheduler.rpc import server as rpc_server
+from coriolis import service
+from coriolis.tests import test_base
+from coriolis import utils
+
+
+class SchedulerTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis scheduler CMD"""
+
+    @mock.patch.object(service, 'service')
+    @mock.patch.object(service, 'MessagingService')
+    @mock.patch.object(rpc_server, 'SchedulerServerEndpoint')
+    @mock.patch.object(utils, 'setup_logging')
+    @mock.patch('coriolis.cmd.scheduler.CONF')
+    @mock.patch.object(sys, 'argv')
+    def test_main(
+        self,
+        mock_argv,
+        mock_conf,
+        mock_setup_logging,
+        mock_SchedulerServerEndpoint,
+        mock_MessagingService,
+        mock_service
+    ):
+        scheduler.main()
+
+        mock_conf.assert_called_once_with(
+            mock_argv[1:], project='coriolis', version="1.0.0")
+        mock_setup_logging.assert_called_once()
+        mock_SchedulerServerEndpoint.assert_called_once()
+        mock_MessagingService.assert_called_once_with(
+            constants.SCHEDULER_MAIN_MESSAGING_TOPIC,
+            [mock_SchedulerServerEndpoint.return_value],
+            rpc_server.VERSION,
+            worker_count=mock_conf.scheduler.worker_count)
+        mock_service.launch.assert_called_once_with(
+            mock_conf, mock_MessagingService.return_value,
+            workers=mock_MessagingService.return_value.
+            get_workers_count.return_value)
+        mock_service.launch.return_value.wait.assert_called_once()