chore: Clean up pipeline fixtures (#350)

* Clean up pipeline fixtures

* CR feedback

Author: BertKleewein

azure-iot-device/tests/common/pipeline/conftest.py

@@ -5,12 +5,8 @@
 # --------------------------------------------------------------------------
 
 from tests.common.pipeline.fixtures import (
-    callback,
-    event,
-    op,
-    op2,
-    op3,
-    new_op,
+    arbitrary_event,
+    arbitrary_op,
     fake_pipeline_thread,
     fake_non_pipeline_thread,
     unhandled_error_handler,

azure-iot-device/tests/common/pipeline/fixtures.py

@@ -14,25 +14,13 @@
 )
 
 
-# TODO: remove this fixture
-# Using it is dangerous if multiple ops use this same callback
-# within the scope of the same test
-# (i.e. an op under test, and an op run in test setup)
-# What happens is that those operations all are tied together with this
-# same callback mock, and the same callback is called multiple times
-# leading to unexpected behavior
-@pytest.fixture
-def callback(mocker):
-    return mocker.MagicMock()
-
-
 class FakeEvent(pipeline_events_base.PipelineEvent):
     def __init__(self):
         super(FakeEvent, self).__init__()
 
 
 @pytest.fixture
-def event():
+def arbitrary_event():
     return FakeEvent()
 
 
@@ -42,30 +30,9 @@ def __init__(self, callback=None):
 
 
 @pytest.fixture
-def op(mocker):
-    op = FakeOperation(callback=mocker.MagicMock())
-    op.name = "op"
-    return op
-
-
-@pytest.fixture
-def op2(mocker):
-    op = FakeOperation(callback=mocker.MagicMock())
-    op.name = "op2"
-    return op
-
-
-@pytest.fixture
-def op3(mocker):
-    op = FakeOperation(callback=mocker.MagicMock())
-    op.name = "op3"
-    return op
-
-
-@pytest.fixture
-def new_op(mocker):
+def arbitrary_op(mocker):
     op = FakeOperation(callback=mocker.MagicMock())
-    op.name = "new_op"
+    op.name = "arbitrary_op"
     return op
 
 

azure-iot-device/tests/common/pipeline/pipeline_stage_test.py

@@ -72,9 +72,9 @@ def add_unknown_ops_tests(cls, module, all_ops, handled_ops):
     @pytest.mark.describe("{} - .run_op() -- unknown and unhandled operations".format(cls.__name__))
     class LocalTestObject(object):
         @pytest.fixture
-        def op(self, op_cls, callback):
+        def op(self, op_cls, mocker):
             op = make_mock_op_or_event(op_cls)
-            op.callback = callback
+            op.callback = mocker.MagicMock()
             op.action = "pend"
             add_mock_method_waiter(op, "callback")
             return op

azure-iot-device/tests/common/pipeline/test_pipeline_flow.py

@@ -18,6 +18,7 @@
     assert_callback_failed,
     assert_callback_succeeded,
 )
+from .fixtures import FakeOperation
 
 logging.basicConfig(level=logging.DEBUG)
 
@@ -49,188 +50,206 @@ def stage(mocker, arbitrary_exception, arbitrary_base_exception):
 
 @pytest.mark.describe("PipelineFlow - ._send_worker_op_down()")
 class TestSendWorkerOpDown(object):
-    @pytest.mark.it("Runs the new op and does not continue running the original op")
-    def test_runs_new_op(self, mocker, stage, op, new_op):
-        stage._send_worker_op_down(worker_op=new_op, op=op)
+    @pytest.fixture
+    def arbitrary_worker_op(self, mocker):
+        op = FakeOperation(callback=mocker.MagicMock())
+        op.name = "arbitrary_worker_op"
+        return op
+
+    @pytest.mark.it("Runs the worker op and does not continue running the original op")
+    def test_runs_worker_op(self, mocker, stage, arbitrary_op, arbitrary_worker_op):
+        stage._send_worker_op_down(worker_op=arbitrary_worker_op, op=arbitrary_op)
         assert stage.next.run_op.call_count == 1
-        assert stage.next.run_op.call_args == mocker.call(new_op)
+        assert stage.next.run_op.call_args == mocker.call(arbitrary_worker_op)
 
-    @pytest.mark.it("Completes the original op after the new op completes")
-    def test_completes_original_op_after_new_op_completes(self, stage, op, new_op):
-        callback = op.callback
-        new_op.action = "pend"
+    @pytest.mark.it("Completes the original op after the worker op completes")
+    def test_completes_original_op_after_worker_op_completes(
+        self, stage, arbitrary_op, arbitrary_worker_op
+    ):
+        callback = arbitrary_op.callback
+        arbitrary_worker_op.action = "pend"
 
-        stage._send_worker_op_down(worker_op=new_op, op=op)
-        assert callback.call_count == 0  # because new_op is pending
+        stage._send_worker_op_down(worker_op=arbitrary_worker_op, op=arbitrary_op)
+        assert callback.call_count == 0  # because arbitrary_worker_op is pending
 
-        stage.next._complete_op(new_op)
-        assert_callback_succeeded(op=op)
+        stage.next._complete_op(arbitrary_worker_op)
+        assert_callback_succeeded(op=arbitrary_op)
 
-    @pytest.mark.it("Returns the new op failure in the original op if new op fails")
-    def test_returns_new_op_failure_in_original_op(self, stage, op, new_op, arbitrary_exception):
-        new_op.action = "fail"
-        stage._send_worker_op_down(worker_op=new_op, op=op)
-        assert_callback_failed(op, error=arbitrary_exception)
+    @pytest.mark.it("Returns the worker op failure in the original op if worker op fails")
+    def test_returns_worker_op_failure_in_original_op(
+        self, stage, arbitrary_op, arbitrary_worker_op, arbitrary_exception
+    ):
+        arbitrary_worker_op.action = "fail"
+        stage._send_worker_op_down(worker_op=arbitrary_worker_op, op=arbitrary_op)
+        assert_callback_failed(arbitrary_op, error=arbitrary_exception)
 
 
 @pytest.mark.describe("PipelineFlow - ._send_op_down()")
 class TestSendOpDown(object):
     @pytest.mark.it("Fails the op if there is no next stage")
-    def test_fails_op_when_no_next_stage(self, stage, op):
+    def test_fails_op_when_no_next_stage(self, stage, arbitrary_op):
         stage.next = None
-        stage._send_op_down(op)
-        assert_callback_failed(op=op, error=pipeline_exceptions.PipelineError)
+        stage._send_op_down(arbitrary_op)
+        assert_callback_failed(op=arbitrary_op, error=pipeline_exceptions.PipelineError)
 
     @pytest.mark.it("Passes the op to the next stage")
-    def test_passes_op_to_next_stage(self, mocker, stage, op):
-        stage._send_op_down(op)
+    def test_passes_op_to_next_stage(self, mocker, stage, arbitrary_op):
+        stage._send_op_down(arbitrary_op)
         assert stage.next.run_op.call_count == 1
-        assert stage.next.run_op.call_args == mocker.call(op)
+        assert stage.next.run_op.call_args == mocker.call(arbitrary_op)
 
 
 @pytest.mark.describe("PipelineFlow - ._complete_op()")
 class TestCompleteOp(object):
     @pytest.mark.it("Calls the op callback on success")
-    def test_calls_callback_on_success(self, stage, op):
-        stage._complete_op(op)
-        assert_callback_succeeded(op)
+    def test_calls_callback_on_success(self, stage, arbitrary_op):
+        stage._complete_op(arbitrary_op)
+        assert_callback_succeeded(arbitrary_op)
 
     @pytest.mark.it("Calls the op callback on failure")
-    def test_calls_callback_on_error(self, stage, op, arbitrary_exception):
-        stage._complete_op(op, error=arbitrary_exception)
-        assert_callback_failed(op=op, error=arbitrary_exception)
+    def test_calls_callback_on_error(self, stage, arbitrary_op, arbitrary_exception):
+        stage._complete_op(arbitrary_op, error=arbitrary_exception)
+        assert_callback_failed(op=arbitrary_op, error=arbitrary_exception)
 
     @pytest.mark.it(
         "Calls handle_background_exception with a PipelineError if the op has previously been completed"
     )
-    def test_background_exception_if_called_twice(self, stage, op, mocker):
+    def test_background_exception_if_called_twice(self, stage, arbitrary_op, mocker):
         mocker.spy(handle_exceptions, "handle_background_exception")
-        stage._complete_op(op)
-        stage._complete_op(op)
+        stage._complete_op(arbitrary_op)
+        stage._complete_op(arbitrary_op)
         assert handle_exceptions.handle_background_exception.call_count == 1
         assert (
             handle_exceptions.handle_background_exception.call_args[0][0].__class__
             == pipeline_exceptions.PipelineError
         )
 
     @pytest.mark.it("Does not call the callback if the op has previously been completed")
-    def test_no_callback_if_called_twice(self, stage, op):
-        stage._complete_op(op)
-        op.callback.reset_mock()
-        stage._complete_op(op)
-        assert op.callback.call_count == 0
+    def test_no_callback_if_called_twice(self, stage, arbitrary_op):
+        stage._complete_op(arbitrary_op)
+        arbitrary_op.callback.reset_mock()
+        stage._complete_op(arbitrary_op)
+        assert arbitrary_op.callback.call_count == 0
 
     @pytest.mark.it(
         "Handles Exceptions raised in operation callback and passes them to the unhandled error handler"
     )
     def test_op_callback_raises_exception(
-        self, stage, op, arbitrary_exception, mocker, unhandled_error_handler
+        self, stage, arbitrary_op, arbitrary_exception, mocker, unhandled_error_handler
     ):
-        op.callback = mocker.Mock(side_effect=arbitrary_exception)
-        stage._complete_op(op)
-        assert op.callback.call_count == 1
-        assert op.callback.call_args == mocker.call(op, error=None)
+        arbitrary_op.callback = mocker.Mock(side_effect=arbitrary_exception)
+        stage._complete_op(arbitrary_op)
+        assert arbitrary_op.callback.call_count == 1
+        assert arbitrary_op.callback.call_args == mocker.call(arbitrary_op, error=None)
         assert unhandled_error_handler.call_count == 1
         assert unhandled_error_handler.call_args == mocker.call(arbitrary_exception)
 
     @pytest.mark.it("Allows any BaseExceptions raised in operation callback to propagate")
-    def test_op_callback_raises_base_exception(self, stage, op, arbitrary_base_exception, mocker):
-        op.callback = mocker.Mock(side_effect=arbitrary_base_exception)
+    def test_op_callback_raises_base_exception(
+        self, stage, arbitrary_op, arbitrary_base_exception, mocker
+    ):
+        arbitrary_op.callback = mocker.Mock(side_effect=arbitrary_base_exception)
         with pytest.raises(arbitrary_base_exception.__class__):
-            stage._complete_op(op)
+            stage._complete_op(arbitrary_op)
 
 
 @pytest.mark.describe("PipelineFlow - ._send_completed_op_up()")
 class TestSendCompletedOpUp(object):
     @pytest.mark.it("Calls the op callback on success")
-    def test_calls_callback_on_success(self, stage, op):
-        op.completed = True
-        stage._send_completed_op_up(op)
-        assert_callback_succeeded(op)
+    def test_calls_callback_on_success(self, stage, arbitrary_op):
+        arbitrary_op.completed = True
+        stage._send_completed_op_up(arbitrary_op)
+        assert_callback_succeeded(arbitrary_op)
 
     @pytest.mark.it("Calls the op callback on failure")
-    def test_calls_callback_on_error(self, stage, op, arbitrary_exception):
-        op.completed = True
-        stage._send_completed_op_up(op, error=arbitrary_exception)
-        assert_callback_failed(op=op, error=arbitrary_exception)
+    def test_calls_callback_on_error(self, stage, arbitrary_op, arbitrary_exception):
+        arbitrary_op.completed = True
+        stage._send_completed_op_up(arbitrary_op, error=arbitrary_exception)
+        assert_callback_failed(op=arbitrary_op, error=arbitrary_exception)
 
     @pytest.mark.it(
         "Calls the callback with a PipelineError if the operation has not been completed"
     )
-    def test_error_if_not_completed(self, stage, op):
+    def test_error_if_not_completed(self, stage, arbitrary_op):
         with pytest.raises(pipeline_exceptions.PipelineError):
-            stage._send_completed_op_up(op)
+            stage._send_completed_op_up(arbitrary_op)
 
     @pytest.mark.it(
         "Handles Exceptions raised in operation callback and passes them to the unhandled error handler"
     )
     def test_op_callback_raises_exception(
-        self, stage, op, arbitrary_exception, mocker, unhandled_error_handler
+        self, stage, arbitrary_op, arbitrary_exception, mocker, unhandled_error_handler
     ):
-        op.callback = mocker.Mock(side_effect=arbitrary_exception)
-        op.completed = True
-        stage._send_completed_op_up(op)
-        assert op.callback.call_count == 1
-        assert op.callback.call_args == mocker.call(op, error=None)
+        arbitrary_op.callback = mocker.Mock(side_effect=arbitrary_exception)
+        arbitrary_op.completed = True
+        stage._send_completed_op_up(arbitrary_op)
+        assert arbitrary_op.callback.call_count == 1
+        assert arbitrary_op.callback.call_args == mocker.call(arbitrary_op, error=None)
         assert unhandled_error_handler.call_count == 1
         assert unhandled_error_handler.call_args == mocker.call(arbitrary_exception)
 
     @pytest.mark.it("Allows any BaseExceptions raised in operation callback to propagate")
-    def test_op_callback_raises_base_exception(self, stage, op, arbitrary_base_exception, mocker):
-        op.callback = mocker.Mock(side_effect=arbitrary_base_exception)
-        op.completed = True
+    def test_op_callback_raises_base_exception(
+        self, stage, arbitrary_op, arbitrary_base_exception, mocker
+    ):
+        arbitrary_op.callback = mocker.Mock(side_effect=arbitrary_base_exception)
+        arbitrary_op.completed = True
         with pytest.raises(arbitrary_base_exception.__class__):
-            stage._send_completed_op_up(op)
+            stage._send_completed_op_up(arbitrary_op)
 
 
 @pytest.mark.describe("PipelineFlow - ._send_op_down_and_intercept_return()")
 class TestSendOpDownAndInterceptReturn(object):
     @pytest.mark.it("Calls _send_op_down to send the op down")
-    def test_sends_op_down(self, stage, op, mocker):
+    def test_sends_op_down(self, stage, arbitrary_op, mocker):
         intercepted_return = mocker.MagicMock()
         mocker.spy(stage, "_send_op_down")
-        stage._send_op_down_and_intercept_return(op, intercepted_return)
+        stage._send_op_down_and_intercept_return(arbitrary_op, intercepted_return)
         assert stage._send_op_down.call_count == 1
-        assert stage._send_op_down.call_args == mocker.call(op)
+        assert stage._send_op_down.call_args == mocker.call(arbitrary_op)
 
     @pytest.mark.it("Calls the intercepted_return function when the op succeeds")
-    def test_calls_intercepted_return_on_op_success(self, stage, op, mocker):
+    def test_calls_intercepted_return_on_op_success(self, stage, arbitrary_op, mocker):
         intercepted_return = mocker.MagicMock()
-        stage._send_op_down_and_intercept_return(op, intercepted_return)
-        assert intercepted_return.call_args == mocker.call(op=op, error=None)
+        stage._send_op_down_and_intercept_return(arbitrary_op, intercepted_return)
+        assert intercepted_return.call_args == mocker.call(op=arbitrary_op, error=None)
 
     @pytest.mark.it("Calls the intercepted_return function when the op fails")
-    def test_calls_intercepted_return_on_op_failure(self, stage, op, mocker, arbitrary_exception):
-        op.action = "fail"
+    def test_calls_intercepted_return_on_op_failure(
+        self, stage, arbitrary_op, mocker, arbitrary_exception
+    ):
+        arbitrary_op.action = "fail"
         intercepted_return = mocker.MagicMock()
-        stage._send_op_down_and_intercept_return(op, intercepted_return)
-        assert intercepted_return.call_args == mocker.call(op=op, error=arbitrary_exception)
+        stage._send_op_down_and_intercept_return(arbitrary_op, intercepted_return)
+        assert intercepted_return.call_args == mocker.call(
+            op=arbitrary_op, error=arbitrary_exception
+        )
 
     @pytest.mark.it(
         "Ensures that the op callback is set to its original value when the intercepted_return function is called"
     )
-    def test_restores_callback_before_calling_intercepted_return(self, stage, op, mocker):
-        saved_callback = op.callback
+    def test_restores_callback_before_calling_intercepted_return(self, stage, arbitrary_op, mocker):
+        saved_callback = arbitrary_op.callback
         intercepted_return = mocker.MagicMock()
-        stage._send_op_down_and_intercept_return(op, intercepted_return)
+        stage._send_op_down_and_intercept_return(arbitrary_op, intercepted_return)
         assert intercepted_return.call_args[1]["op"].callback == saved_callback
 
 
 @pytest.mark.describe("PipelineFlow - ._send_event_up()")
 class TestSendEventUp(object):
     @pytest.mark.it("Calls handle_pipeline_event on the previous stage")
-    def test_calls_handle_pipeline_event(self, stage, event, mocker):
+    def test_calls_handle_pipeline_event(self, stage, arbitrary_event, mocker):
         mocker.spy(stage, "handle_pipeline_event")
-        stage.next._send_event_up(event)
+        stage.next._send_event_up(arbitrary_event)
         assert stage.handle_pipeline_event.call_count == 1
-        assert stage.handle_pipeline_event.call_args == mocker.call(event)
+        assert stage.handle_pipeline_event.call_args == mocker.call(arbitrary_event)
 
     @pytest.mark.it(
         "Calls handle_background_exception with a PipelineError if there is no previous stage"
     )
-    def test_no_previous_stage(self, stage, event, mocker):
+    def test_no_previous_stage(self, stage, arbitrary_event, mocker):
         mocker.spy(handle_exceptions, "handle_background_exception")
-        stage._send_event_up(event)
+        stage._send_event_up(arbitrary_event)
         assert handle_exceptions.handle_background_exception.call_count == 1
         assert (
             handle_exceptions.handle_background_exception.call_args[0][0].__class__