Hands-On Reactive Programming with Python

Testing a custom operator

Now let's see how to test a custom operator. Consider the following code:

def integer_passthrough(numbers):    def on_subscribe(observer):        def on_next(i):            if type(i) is int:                observer.on_next(i)            else:                observer.on_error(TypeError("Item is not an integer"))        disposable = numbers.subscribe(            on_next=on_next,            on_error= lambda e: observer.on_error(e),            on_completed= lambda: observer.on_completed()        )        return disposable    return Observable.create(on_subscribe)

The integer_passthrough function takes an observable of numbers as input and returns it directly. If a received item is not a number, then an error is raised. The nominal case can be tested in the following way:

    def test_next(self): numbers = Observable.from_([1, 2, 3, 4]) ...