Using the tools provided in reactivex.testing, it is possible to create tests for your own observables, custom operators and subscriptions.
Additionally, tests can be used to help understand the behaviors of existing operators.
# This assumes that you are using pytest but unittest or others would work just as well
# Import the testing tools
from reactivex.testing import ReactiveTest, TestScheduler
from reactivex import operators
def test_double():
# Create a scheduler
scheduler = TestScheduler()
# Define one or more source
source = scheduler.create_hot_observable(
ReactiveTest.on_next(250, 3),
ReactiveTest.on_next(350, 5),
)
# Define how the observable/operator is used on the source
def create():
return source.pipe(operators.map(lambda x: 2 * x))
# trigger subscription and record emissions
results = scheduler.start(create)
# check the messages and potentially subscriptions
assert results.messages == [
ReactiveTest.on_next(250, 6),
ReactiveTest.on_next(350, 10),
]
Whether your custom operator is created using a composition of operators or with full control, you can easily test various situations and combinations
Surprised about the timestamps (@500, @600, …) for the result messages? Then read below about the timeline.
When scheduler.start
is called, the test scheduler starts moving its virtual clock forward.
Some important timestamps are however hidden as defaults, as listed below.
These values can be modified using kwargs in the scheduler.start(...)
call:
created
[100]: When is the observable created.
That is when the create
function seen in the basic example is called.
subscribed
[200]: When does the subscription occur.
This explains the above emission timestamps:
consider the first emission @500; given that we are using a cold observable,
and subscribe to it at 200, the source’s timeline starts at 200 and only 300 ticks later, it emits.
disposed
[1000]: When the subscription is disposed
Gotchas when modifying these values:
Do not use 0 as values for created/subscribed since the code would ignore it.
If you change subscribed
to be lower than 100, you need to change created
as well,
otherwise nothing will happen.
As we saw in the previous section, we can use reactivex.from_marbles to create observables for our tests.
An example of using to_marbles for the assertion is shown in test_hot
There is a simplified flow available in reactivex.testing.marbles and here’s an example:
def test_start_with():
from reactivex.testing.marbles import marbles_testing
with marbles_testing() as (start, cold, hot, exp):
source = cold('------1-2-3-|')
outcome = exp('a-----1-2-3-|', {"a": None}) # can use lookups if needed
obs = source.pipe(
operators.start_with(None)
)
# Note that start accepts the observable directly,
# without the need for a "create" function
results = start(obs)
assert results == outcome
This method makes for very quick to write, and easy to read, tests. At this moment however, it does not allow for testing subscriptions.
An observable created directly from Observable
can be just as easily tested.
In this example, we will additionally test a case where a
Disposable
is used.
def test_my_observable_factory():
from reactivex.disposable import Disposable, CompositeDisposable
a = 42
def factory(observer: Observer, scheduler=None):
def increment():
nonlocal a
a += 1
sub = Disposable(action=increment)
return CompositeDisposable(
sub,
reactivex.timer(20, scheduler=scheduler).subscribe(observer)
)
scheduler = TestScheduler()
result = scheduler.start(lambda: Observable(factory))
assert result.messages == [
on_next(220, 0),
on_completed(220)
]
assert a == 43 # shows that our Disposable's action was as expected
Going back to the in_sequence_or_throw operator, we did not test the error case; Let’s remedy that below.
def test_in_sequence_or_throw_error():
scheduler = TestScheduler()
source = reactivex.from_marbles('--1-4-3-', timespan=50, scheduler=scheduler)
result = scheduler.start(lambda: source.pipe(
in_sequence_or_throw(),
), created=1, subscribed=30)
assert result.messages == [
on_next(30+100, 1),
on_error(230, ValueError('Sequence error'))
]
# At times it's better not to test the exact exception,
# maybe its message changes with time or other reasons
# We can test a specific notification's details as follows:
first_notification, error_notification = result.messages
assert first_notification.time == 130
assert error_notification.time == 230
assert first_notification.value.kind == 'N' # Notification
assert error_notification.value.kind == 'E' # E for errors
assert first_notification.value.value == 1
assert type(error_notification.value.exception) == ValueError # look at .exception for errors
scheduler.start
only allows for a single subscription.
Some cases like e.g. operators.partition
require more.
The examples below showcase some less commonly needed testing tools.
def test_multiple():
scheduler = TestScheduler()
source = reactivex.from_marbles('-1-4-3-|', timespan=50, scheduler=scheduler)
odd, even = source.pipe(
operators.partition(lambda x: x % 2),
)
steven = scheduler.create_observer()
todd = scheduler.create_observer()
even.subscribe(steven)
odd.subscribe(todd)
# Note! Since the subscription is not created within
# `scheduler.start` below, the usual `subscribed` delay of t=200
# is not in effect. The subscriptions therefore occur at t=0
scheduler.start()
assert steven.messages == [
on_next(150, 4),
on_completed(350)
]
assert todd.messages == [
on_next(50, 1),
on_next(250, 3),
on_completed(350)
]
from reactivex.testing.subscription import Subscription
def test_subscriptions():
scheduler = TestScheduler()
source = scheduler.create_cold_observable() # "infinite"
subs = []
shared = source.pipe(
operators.share()
)
# Creating our story:
# first sub is set to occur at t=200; this creates a sub on source
scheduler.schedule_relative(200, lambda *_: subs.append(shared.subscribe(scheduler=scheduler)))
# second sub does not create a new sub on source, due to the `share` operator
scheduler.schedule_relative(300, lambda *_: subs.append(shared.subscribe(scheduler=scheduler)))
# second sub ends
scheduler.schedule_relative(500, lambda *_: subs[1].dispose())
# first sub ends… and since there is no sub remaining, the only sub on source should be disposed too
scheduler.schedule_relative(600, lambda *_: subs[0].dispose())
# no existing sub on source, therefore this will create a new one
# we never dispose of it; we will test that infinite sub in the assertions
scheduler.schedule_relative(900, lambda *_: subs.append(shared.subscribe(scheduler=scheduler)))
scheduler.start()
# Check that the submissions on the source are as expected
assert source.subscriptions == [
Subscription(200, 600), # only one sub from 200 to 600
Subscription(900), # represents an infinite subscription
]
def test_hot():
scheduler = TestScheduler()
# hot starts at 0 but sub starts at 200 so we'll miss 190
source = scheduler.create_hot_observable(
on_next(190, 5),
on_next(300, 42),
on_completed(500)
)
result = scheduler.start(lambda: source.pipe(
operators.to_marbles(timespan=20, scheduler=scheduler)
))
message = result.messages[0]
# the subscription starts at 200;
# since `source` is a hot observable, the notification @190 will not be caught
# the next notification is at 300 ticks,
# which, on our subscription, will show at 100 ticks (300-200 from subscription delay)
# or 5 "-" each representing 20 ticks (timespan=20 in `to_marbles`).
# Then the "42" notification is received
# and then nothing for another 200 ticks, which is equal to 10 "-", before complete
assert message.value.value == '-----(42)----------|'