pep8: fix coriolis/ tests, providers, scheduler, taskflow

cloudbase · Oct 5, 2023 · 86f83d9 · 86f83d9
1 parent ce640b3
commit 86f83d9
Show file tree

Hide file tree

Showing 6 changed files with 18 additions and 11 deletions.
diff --git a/coriolis/providers/replicator.py b/coriolis/providers/replicator.py
@@ -438,8 +438,8 @@ def wait_for_chunks(self):
                 perc_step = perc_steps.get(devName)
                 if perc_step is None:
                     perc_step = self._event_manager.add_percentage_step(
-                        "Performing chunking for disk %s (total size %.2f MB)" % (
-                            devName, dev_size), 100)
+                        "Performing chunking for disk %s (total size %.2f MB)"
+                        % (devName, dev_size), 100)
                     perc_steps[devName] = perc_step
                 perc_done = vol["checksum-status"]["percentage"]
                 self._event_manager.set_percentage_step(
@@ -837,7 +837,8 @@ def _download_full_disk(self, disk, path):
 
         total = 0
         with self._cli._cli.get(diskUri, stream=True,
-                                timeout=CONF.replicator.default_requests_timeout) as dw:
+                                timeout=(CONF.replicator.
+                                         default_requests_timeout)) as dw:
             with open(path, 'wb') as dsk:
                 for chunk in dw.iter_content(chunk_size=self._chunk_size):
                     if chunk:

diff --git a/coriolis/scheduler/filters/trivial_filters.py b/coriolis/scheduler/filters/trivial_filters.py
@@ -97,8 +97,8 @@ def rate_service(self, service):
         for platform_type in self._provider_requirements:
             if platform_type not in service.providers:
                 LOG.debug(
-                    "Service with ID '%s' does not have a provider for platform "
-                    "type '%s'", service.id, platform_type)
+                    "Service with ID '%s' does not have a provider for "
+                    "platform type '%s'", service.id, platform_type)
                 return 0
 
             available_types = service.providers[

diff --git a/coriolis/taskflow/base.py b/coriolis/taskflow/base.py
@@ -190,8 +190,8 @@ def _get_worker_service_rpc_for_task(
             ctxt, task_info, origin, destination, retry_count=retry_count,
             retry_period=retry_period, random_choice=True)
         LOG.debug(
-            "[Task '%s'] Was offered the following worker service for executing "
-            "Taskflow worker task '%s': %s",
+            "[Task '%s'] Was offered the following worker service for "
+            "executing Taskflow worker task '%s': %s",
             self._task_name, task_id, worker_service['id'])
 
         return rpc_worker_client.WorkerClient.from_service_definition(

diff --git a/coriolis/tasks/osmorphing_tasks.py b/coriolis/tasks/osmorphing_tasks.py
@@ -108,7 +108,8 @@ def _run(self, ctxt, instance, origin, destination, task_info,
         instance_deployment_info = task_info["instance_deployment_info"]
 
         import_info = provider.deploy_os_morphing_resources(
-            ctxt, connection_info, target_environment, instance_deployment_info)
+            ctxt, connection_info, target_environment,
+            instance_deployment_info)
 
         schemas.validate_value(
             import_info, schemas.CORIOLIS_OS_MORPHING_RESOURCES_SCHEMA,

diff --git a/coriolis/tests/db/test_api.py b/coriolis/tests/db/test_api.py
@@ -5,7 +5,8 @@
 
 from coriolis.db import api
 from coriolis import exception
-from coriolis.tests import test_base, testutils
+from coriolis.tests import test_base
+from coriolis.tests import testutils
 
 
 class DBAPITestCase(test_base.CoriolisBaseTestCase):

diff --git a/coriolis/tests/worker/rpc/test_server.py b/coriolis/tests/worker/rpc/test_server.py
@@ -10,9 +10,12 @@
 import eventlet
 import psutil
 
-from coriolis import schemas, constants, utils, exception
+from coriolis import constants
+from coriolis import exception
 from coriolis.providers import factory as providers_factory
+from coriolis import schemas
 from coriolis.tests import test_base
+from coriolis import utils
 from coriolis.worker.rpc import server
 
 
@@ -298,7 +301,8 @@ def call_exec_task(report_to_conductor=True):
         mock_client.reset_mock()
         mock_exec.reset_mock()
         mock_exec.side_effect = Exception("mock_message")
-        self.assertRaises(Exception, lambda: call_exec_task(False))
+        self.assertRaises(Exception,  # noqa: H202
+                          lambda: call_exec_task(False))
 
     @mock.patch.object(schemas, "validate_value")
     @mock.patch.object(utils, "get_secret_connection_info")