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])
        expected_numbers = [1, 2, 3, 4]
        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

        integer_passthrough(numbers).subscribe(
            on_next=on_next,
            on_error=on_error
        )

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

The callbacks provided to the subscribe call are used to keep track of all events that occurred: the emitted items, as well as the errors. The test simply compares the received values with the ones which are expected. Two asserts are used here:

• One to check that no error has been raised
• The other to check that the received numbers are the expected ones

Note that the assertEqual method compares two lists directly. This avoids comparing the list item per item.

Testing errors is done in a similar way:

    def test_error_on_string(self): 
        numbers = Observable.from_([1, 2, 'c', 4])
        expected_numbers = [1, 2]
        actual_numbers = []
        actual_error = None

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

        def on_error(e):
            nonlocal actual_error
            actual_error = e

        integer_passthrough(numbers).subscribe(
            on_next=on_next,
            on_error=on_error
        )

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

Here, a character is provided in one item of the numbers observable. So it is expected that only the first two items are emitted. However, this version of the test does not check the error, so it fails. This can be seen in the following example:

(venv-rx)$ nosetests test_operator.py 
F.
======================================================================
FAIL: test_error_on_string (test_operator.IntegerPassthroughTestCase)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/xxx/Hands-On-Reactive-Programming-with-Python/ch10/test_operator.py", line 63, in test_error_on_string
    self.assertEqual(None, actual_error)
AssertionError: None != TypeError('Item is not an integer',)
-------------------- >> begin captured logging << --------------------
Rx: DEBUG: CurrentThreadScheduler.schedule(state=None)
Rx: DEBUG: CurrentThreadScheduler.schedule(state=None)
Rx: DEBUG: CurrentThreadScheduler.schedule(state=None)
Rx: DEBUG: CurrentThreadScheduler.schedule(state=None)
Rx: DEBUG: CurrentThreadScheduler.schedule(state=None)
--------------------- >> end captured logging << ---------------------

----------------------------------------------------------------------
Ran 2 tests in 0.005s

FAILED (failures=1)

Fixing the test is done by replacing the first assertEqual statement with following line of code:

self.assertIsInstance(actual_error, TypeError)

Now, with the addition of the preceding line, the test passes:

(venv-rx) $ nosetests test_operator.py 
..
----------------------------------------------------------------------
Ran 2 tests in 0.004s

OK

Another test you can do on this operator is to check that it correctly forwards errors received on the source observable:

    def test_forward_error(self): 
        numbers = Observable.throw(ValueError())
        expected_numbers = []
        actual_numbers = []
        actual_error = None

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

        def on_error(e):
            nonlocal actual_error
            actual_error = e

        integer_passthrough(numbers).subscribe(
            on_next=on_next,
            on_error=on_error
        )

        self.assertIsInstance(actual_error, ValueError)
        self.assertEqual(expected_numbers, actual_numbers)

This test is very similar to the previous one, but this time the numbers observable directly throws an error.

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