U
vi @ s d dl mZmZmZmZmZmZmZ d dlZd dlm Z m
Z
m Z d dl m Z d dl
mZ d dlmZ ddlmZ dd lmZ dd
lmZ dd lmZ ed ZG d
d deZdS ) )AnyCallableListOptionalTypeVarUnioncastN)
Observableabctyping)
Disposable)VirtualTimeScheduler)Recorded )ColdObservable)
HotObservable MockObserver) ReactiveTest_Tc s e Zd ZdZdZdejeje ee j
d fdd
Z de e
g ee f e e e e e e ee d fdd
Zeee eee f ee d
d d Zeee eee f ee d
d
dZee dddZ ZS )
TestSchedulerzTest time scheduler used for testing applications and libraries
built using Reactive Extensions. All time, both absolute and relative is
specified as integer ticksFN)duetimeactionstatereturnc s( t |tr|n| |}t |||S )a Schedules an action to be executed at the specified virtual
time.
Args:
duetime: Absolute virtual time at which to execute the
action.
action: Action to be executed.
state: State passed to the action to be executed.
Returns:
Disposable object used to cancel the scheduled action
(best effort).
)
isinstancefloatZ
to_secondssuperschedule_absolute)selfr r r __class__ P/opt/alt/python38/lib/python3.8/site-packages/reactivex/testing/testscheduler.pyr s z TestScheduler.schedule_absolute)createcreated
subscribeddisposedr c s |pt j}|pt j}|p t j}| ddd tjttjd fdd
}| || d
tjttjdfdd
}| || d tjttjdfdd
}| || t
S ) a$ Starts the test scheduler and uses the specified virtual
times to invoke the factory function, subscribe to the
resulting sequence, and dispose the subscription.
Args:
create: Factory method to create an observable sequence.
created: Virtual time at which to invoke the factory to
create an observable sequence.
subscribed: Virtual time at which to subscribe to the
created observable sequence.
disposed: Virtual time at which to dispose the
subscription.
Returns:
Observer with timestamped recordings of notification
messages that were received during the virtual time window
when the subscription to the source sequence was active.
N) schedulerr r c s dk r nt t S )z&Called at create time. Defaults to 100N) reactivexneverr r( r )r$ sourcer" r#
action_createR s z*TestScheduler.start..action_createc s rj | dt S )z)Called at subscribe time. Defaults to 200)r( ) subscriber r+ )observerr, subscriptionr" r# action_subscribe\ s z-TestScheduler.start..action_subscribec s r t S )z(Called at dispose time. Defaults to 1000)Zdisposer r+ )r0 r" r# action_disposeg s z+TestScheduler.start..action_dispose)N)N)N) r r% r& r' create_observerr
Z
SchedulerBaser DisposableBaser r start)r r$ r% r& r' r- r1 r2 r )r$ r/ r, r0 r# r5 / s:
zTestScheduler.start)argsr c G s< |r t |d tr |d }ntttt t|}t| |S )a Creates a hot observable using the specified timestamped
notification messages either as a list or by multiple arguments.
Args:
messages: Notifications to surface through the created sequence at
their specified absolute virtual times.
Returns hot observable sequence that can be used to assert the timing
of subscriptions and notifications.
r )r r r r r listr r r6 messagesr" r" r# create_hot_observablet s
z#TestScheduler.create_hot_observablec G s< |r t |d tr |d }ntttt t|}t| |S )a Creates a cold observable using the specified timestamped
notification messages either as an array or arguments.
Args:
args: Notifications to surface through the created sequence
at their specified virtual time offsets from the
sequence subscription time.
Returns:
Cold observable sequence that can be used to assert the
timing of subscriptions and notifications.
r )r r7 r r r r r r8 r" r" r# create_cold_observable s
z$TestScheduler.create_cold_observable)r c C s t | S )zCreates an observer that records received notification messages and
timestamps those. Return an Observer that can be used to assert the
timing of received notifications.
r )r r" r" r# r3 s z TestScheduler.create_observer)N)NNNN) __name__
__module__ __qualname____doc__Z__test__r Z AbsoluteTimeZScheduledActionr r
r4 r r r r r r r r5 r r r r r: r r; r3
__classcell__r" r" r r# r s6 F r )r r r r r r r r r) r r
Zreactivex.disposabler Zreactivex.schedulerr
Zreactivex.testing.recordedr Zcoldobservabler Z
hotobservabler Z mockobserverr Z reactivetestr r r r" r" r" r# s $