The previous example showed the principles of custom operator testing. However, in this example, the operator is working in a synchronous way. Fortunately, testing operators using several observables emitting items in an asynchronous way is also possible. The key is that the asynchronous part can be simulated, thanks to Subject.

The following example is a test for the buffer operator, where two observables are supposed to emit items in an asynchronous way. By using subjects, it is possible to simulate the order of arrival of each item on each observable:

    def test_nominal(self):
        numbers = Subject()
        windows = Subject()

        expected_numbers = [ [1, 2], [3, 4, 5]]
        expected_error = None
        actual_numbers = []
        actual_error = None

        def on_next(i):
            actual_numbers.append(i)

        def on_error(e):
            nonlocal actual_error
            actual_error = e

        numbers.buffer(windows).subscribe(
            on_next=on_next,
            on_error=on_error
        )

        numbers.on_next(1)
        numbers.on_next(2)
        windows.on_next(True)
        numbers.on_next(3)
        numbers.on_next(4)
        numbers.on_next(5)
        windows.on_next(True)

        self.assertEqual(None, actual_error)
        self.assertEqual(expected_numbers, actual_numbers)

Here, two items are first emitted on the numbers observable. Then the windows observable emits an item, which should complete a first buffer. Then three more items are emitted on the numbers observable, and another on the windows observable and complete a second buffer. Note that the test expects that two items are returned by the buffer operator to pass, which is the case here.

This example, completed with an error test, is available in the GitHub repository (https://github.com/PacktPublishing/Hands-On-Reactive-Programming-with-Python) of this book, in the test_buffer.py script.