|
Server : LiteSpeed System : Linux host 5.15.0-161-generic #171-Ubuntu SMP Sat Oct 11 08:17:01 UTC 2025 x86_64 User : idnco5810 ( 1093) PHP Version : 8.2.29 Disable Function : NONE Directory : /lib/python3/dist-packages/twisted/internet/test/__pycache__/ |
Upload File : |
o
�b� � @ s� d Z ddlmZmZ ddlmZ ddlmZ ddlm Z ddl
m Z G dd� de�Z G d d
� d
e�Z
e� �e �� � e� �e
�� � d S ) z/
Tests for implementations of L{IReactorTime}.
� )�IReactorThreads� IReactorTime)�ReactorBuilder)�msg)�platform)�SkipTestc @ s& e Zd ZdZefZdd� Zdd� ZdS )�TimeTestsBuilderzA
Builder for defining tests relating to L{IReactorTime}.
c C s" | � � }|�d|j� |�� dS )z?
The reactor can be stopped by a delayed call.
r N)� buildReactor� callLater�stop�run)�self�reactor� r �A/usr/lib/python3/dist-packages/twisted/internet/test/test_time.py� test_delayedCallStopsReactor s z-TimeTestsBuilder.test_delayedCallStopsReactorc s� | � � �t���rdd� � ntd����dd d dd� �}�fd d
��� �� ��fd d�� ��� | �|�� � | �|�� � � d S )
z
Scheduling a delayed call at a point in the extreme future does not
prevent normal reactor operation.
c S s t d|d� | �|� d S )Nz.Thread-based event-source scheduling %(event)r)�format�event)r �callFromThread)r r r r r � eventSource& s �z=TimeTestsBuilder.test_distantDelayedCall.<locals>.eventSourcez=Do not know how to synthesize non-time event to stop the test� � � c S s d S �Nr r r r r �<lambda>2 s z:TimeTestsBuilder.test_distantDelayedCall.<locals>.<lambda>c s t d� � �� d S )NzStopping the reactor)r r r )r r r r 4 s z6TimeTestsBuilder.test_distantDelayedCall.<locals>.stopc s
� ���S r r r �r r r r r r < s
N)
r r �
providedByr r
r �
assertTrue�active�assertIn�getDelayedCalls)r
� delayedCallr r r �test_distantDelayedCall s
� z(TimeTestsBuilder.test_distantDelayedCallN)�__name__�
__module__� __qualname__�__doc__r �requiredInterfacesr r" r r r r r s
r c @ s6 e Zd ZdZefZe�� rdgZnddgZdd� Z dS )�GlibTimeTestsBuilderza
Builder for defining tests relating to L{IReactorTime} for reactors based
off glib.
z/twisted.internet.gtk2reactor.PortableGtkReactorz*twisted.internet.glib2reactor.Glib2Reactorz(twisted.internet.gtk2reactor.Gtk2Reactorc s^ ddl }| �� �g �� �fdd�}��fdd�� ��|jd|� | ��d� | ��d g� dS )
z�
A
L{reactor.callLater<twisted.internet.interfaces.IReactorTime.callLater>}
call scheduled from a C{gobject.timeout_add}
call is run on time.
r Nc s �� d� � dS )Nr )r
r )�callbackr r r � gscheduleh s z8GlibTimeTestsBuilder.test_timeout_add.<locals>.gschedulec s �� d� � �� d S )NT)�appendr r )r �resultr r r) l s
z7GlibTimeTestsBuilder.test_timeout_add.<locals>.callback�
� T)�gobjectr �callWhenRunning� timeout_add�
runReactor� assertEqual)r
r/ r* r )r) r r, r �test_timeout_add[ s z%GlibTimeTestsBuilder.test_timeout_addN)
r# r$ r% r& r r' r � isWindows� _reactorsr4 r r r r r( K s � r( N)r& �twisted.internet.interfacesr r �#twisted.internet.test.reactormixinsr �twisted.python.logr �twisted.python.runtimer �twisted.trial.unittestr r r( �globals�update�makeTestCaseClassesr r r r �<module> s <*