🔐 Sid Gifari File Manager Pro
v8.0.5 | 2026-08-08 08:11:38 | PHP 8.2.32
📂
/ (Root)
/
opt
/
alt
/
python38
/
lib
/
python3.8
/
site-packages
/
reactivex
/
testing
/
__pycache__
📍 /opt/alt/python38/lib/python3.8/site-packages/reactivex/testing/__pycache__
🔄 Refresh
✏️
Editing: testscheduler.cpython-38.pyc
Read Only
U v�i� � @ 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� de�ZdS )� )�Any�Callable�List�Optional�TypeVar�Union�castN)� Observable�abc�typing)� 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 d�Zee d�dd�Z� ZS )� TestSchedulerz�Test time scheduler used for testing applications and libraries built using Reactive Extensions. All time, both absolute and relative is specified as integer ticksFN)�duetime�action�state�returnc s( t |t�r|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). )� isinstance�floatZ to_seconds�super�schedule_absolute)�selfr r r �� __class__� �P/opt/alt/python38/lib/python3.8/site-packages/reactivex/testing/testscheduler.pyr s z TestScheduler.schedule_absolute)�create�created� subscribed�disposedr c s� |pt j}|pt j}|p t j}| �� �d�d�d tjttjd�� �fdd� }| � ||� d tjttjd����fdd� }| � ||� d tjttjd��fdd� }| � ||� 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)� reactivex�neverr �r( r )r$ �sourcer" r# � action_createR s z*TestScheduler.start.<locals>.action_createc s �r�j � | d��t� S )z)Called at subscribe time. Defaults to 200)r( )� subscriber r+ )�observerr, � subscriptionr" r# �action_subscribe\ s z-TestScheduler.start.<locals>.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.<locals>.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 t�r |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 t�r |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 )z�Creates 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# �<module> s $
💾 Save Changes
❌ Cancel