U
vi @ s d dl mZ d dlmZmZmZmZmZmZm Z m
Z
d dl m Z d dl
Z
d dl
mZmZ d dlmZmZmZ d dlmZ d dlmZ d d lmZmZ d
d lmZ d
d l m Z d
d
l m Z e Z G dd deZ!edeee!ddf dddZ"ee ejee f ee e dddZ#dS ) )contextmanager)AnyDict GeneratorList
NamedTupleOptionalTupleUnion)warnN)
Observabletyping) NotificationOnErrorOnNext)parse)NewThreadScheduler)Callable RelativeTime ) ReactiveTest)Recorded)
TestSchedulerc @ s e Zd ZU eeee eg ee f f geee f e d< ee
e e ee
e
f ef e e gee f e d< ee
e e ee
e
f ef e e gee f e d< ee
e e ee
e
f ef e e geee f e d< dS )MarblesContextstartZcoldhotexpN)__name__
__module__ __qualname__r r
r r r r __annotations__strr r float Exception r$ r$ J/opt/alt/python38/lib/python3.8/site-packages/reactivex/testing/marbles.pyr s$
(
r ?)timespanreturnc # s< t dddddddfdd ttt tg tt f f ttt d fd
d }dttt ttt
f tf tt ttt d fd
d
}dttt ttt
f tf tt tt d fdd
}dttt ttt
f tf tt tt d fdd
}zt ||||V W 5 dX dS )as
Initialize a :class:`rx.testing.TestScheduler` and return a namedtuple
containing the following functions that wrap its methods.
:func:`cold()`:
Parse a marbles string and return a cold observable
:func:`hot()`:
Parse a marbles string and return a hot observable
:func:`start()`:
Start the test scheduler, invoke the create function,
subscribe to the resulting sequence, dispose the subscription and
return the resulting records
:func:`exp()`:
Parse a marbles string and return a list of records
Examples:
>>> with marbles_testing() as (start, cold, hot, exp):
... obs = hot("-a-----b---c-|")
... ex = exp( "-a-----b---c-|")
... results = start(obs)
... assert results == ex
The underlying test scheduler is initialized with the following
parameters:
- created time = 100.0s
- subscribed = 200.0s
- disposed = 1000.0s
**IMPORTANT**: regarding :func:`hot()`, a marble declared as the
first character will be skipped by the test scheduler.
E.g. hot("a--b--") will only emit b.
g Y@g @@g i@FNr( c s( rt dtdd r$t dtdd d S )NzAcontext functions should not be called outside of with statement. )
stacklevelz?start() should only be called one time inside a with statement.)r UserWarningr$ )outside_of_context start_calledr$ r% checkT s z marbles_testing..check)creater( c sP t | tr0| tt d fdd }|}n| }j|d}d|jS )Nr) c s S )Nr$ r$ Zcreate_r$ r% default_createl s z;marbles_testing..test_start..default_create)r0 created
subscribeddisposedT)
isinstancer r r messages)r0 r2 Zcreate_functionZ
mock_observer)r/ r3 r5 schedulerr. r4 r1 r%
test_startc s
z#marbles_testing..test_start)stringlookuperrorr( c s t | ||d}t|S )N)r' Z
time_shiftr; r< )r messages_to_records)r: r; r< r7 )r4 r' r$ r%
test_expected| s z&marbles_testing..test_expectedc s t j| ||dS )N)r' r; r< ) reactivexZ from_marbles)r: r; r< )r/ r' r$ r% test_cold s z"marbles_testing..test_coldc s t j| ||d}|S )N)r' Zduetimer; r< r8 )r? r )r: r; r< Zhot_obs)r/ r8 r4 r' r$ r% test_hot s z!marbles_testing..test_hotT)NN)NN)NN)
r r
r r r r r r! r r r" r# r )r' r9 r> r@ rA r$ )r/ r3 r5 r- r8 r. r4 r' r% marbles_testing% sJ (
rB )r7 r( c C s g }| D ]v}|\}}t |ttfr,t|}n
|jd }t |trPt||j}n$t |trjt ||j
}n
t |}| | q|S )z\
Helper function to convert messages returned by parse() to a list of
Recorded.
i )
r6 intr" microsecondsr r Zon_nextvaluer Zon_error exceptionZ on_completedappend)r7 recordsmessagetimeZ notificationZtime_recordr$ r$ r% r= s
r= )r& )$
contextlibr r
r r r r r r r r
warningsr r? r Zreactivex.notificationr r r Z reactivex.observable.marblesr Zreactivex.schedulerr Zreactivex.typingr r Z reactivetestr Zrecordedr Z
testschedulerr Znew_thread_schedulerr rB r= r$ r$ r$ r% s. (