A few worker-engine cleanups

- Remove usage of % with logs, just let the underlying
  LOG do the interpolation for us if the log level passes.
- Use LOG.exception instead of LOG.warn and then including
  the exception as a interpolated variable. Just let the
  LOG.exception output the exception trace and info.
- Capture the failure object before doing other work using
  a new context manager that can yield back a failure object
  before further code is called (to avoid a context switch
  happening in eventlet or elsewhere that would clear the
  exception state).

Change-Id: Ic1a504ba69c56ba226b1a067b5ade4590c245dd8

2 files changed, 9 insertions, 11 deletions

diff --git a/taskflow/tests/unit/worker_based/test_executor.py b/taskflow/tests/unit/worker_based/test_executor.py
index fbb6159..2544984 100644
--- a/taskflow/tests/unit/worker_based/test_executor.py
+++ b/taskflow/tests/unit/worker_based/test_executor.py
@@ -177,11 +177,11 @@ class TestWorkerTaskExecutor(test.MockTestCase):
         self.assertEqual(self.remote_task_mock.mock_calls, [])
         self.assertEqual(self.message_mock.mock_calls, [mock.call.ack()])
 
-    @mock.patch('taskflow.engines.worker_based.executor.LOG.warning')
-    def test_on_message_acknowledge_raises(self, mocked_warning):
+    @mock.patch('taskflow.engines.worker_based.executor.LOG.exception')
+    def test_on_message_acknowledge_raises(self, mocked_exception):
         self.message_mock.ack.side_effect = kombu_exc.MessageStateError()
         self.executor()._on_message({}, self.message_mock)
-        self.assertTrue(mocked_warning.called)
+        self.assertTrue(mocked_exception.called)
 
     @mock.patch('taskflow.engines.worker_based.remote_task.misc.wallclock')
     def test_on_wait_task_not_expired(self, mock_time):
diff --git a/taskflow/tests/unit/worker_based/test_server.py b/taskflow/tests/unit/worker_based/test_server.py
index 4a5326e..34af585 100644
--- a/taskflow/tests/unit/worker_based/test_server.py
+++ b/taskflow/tests/unit/worker_based/test_server.py
@@ -210,8 +210,8 @@ class TestServer(test.MockTestCase):
                   failures=dict((str(i), utils.FailureMatcher(f))
                                 for i, f in enumerate(failures)))))
 
-    @mock.patch("taskflow.engines.worker_based.server.LOG.error")
-    def test_reply_publish_failure(self, mocked_error):
+    @mock.patch("taskflow.engines.worker_based.server.LOG.exception")
+    def test_reply_publish_failure(self, mocked_exception):
         self.proxy_inst_mock.publish.side_effect = RuntimeError('Woot!')
 
         # create server and process request
@@ -222,9 +222,7 @@ class TestServer(test.MockTestCase):
             mock.call.proxy.publish({'state': 'FAILURE'}, self.task_uuid,
                                     self.reply_to)
         ])
-        self.assertEqual(mocked_error.mock_calls, [
-            mock.call("Failed to send reply: Woot!")
-        ])
+        self.assertTrue(mocked_exception.called)
 
     def test_on_update_progress(self):
         request = self.request(task='taskflow.tests.utils.ProgressingTask',
@@ -261,15 +259,15 @@ class TestServer(test.MockTestCase):
         ]
         self.assertEqual(self.master_mock.mock_calls, master_mock_calls)
 
-    @mock.patch("taskflow.engines.worker_based.server.LOG.error")
-    def test_process_request_parse_message_failure(self, mocked_error):
+    @mock.patch("taskflow.engines.worker_based.server.LOG.exception")
+    def test_process_request_parse_message_failure(self, mocked_exception):
         self.message_mock.properties = {}
         request = self.request()
         s = self.server(reset_master_mock=True)
         s._process_request(request, self.message_mock)
 
         self.assertEqual(self.master_mock.mock_calls, [])
-        self.assertTrue(mocked_error.called)
+        self.assertTrue(mocked_exception.called)
 
     @mock.patch('taskflow.engines.worker_based.server.pu')
     def test_process_request_parse_failure(self, pu_mock):