Hands-On Reactive Programming with Python

Injecting asynchronous obervables

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 ...