File: //usr/lib/python3/dist-packages/twisted/test/__pycache__/test_htb.cpython-310.pyc
o
�bR � @ s� d dd� Z ddlmZ ddlmZ ddlmZ G dd � d �ZG d
d � d ej�Z G d d
� d
ej
�Z G dd� de �Z G dd� de �Z
G dd� de �ZdS )z$Revision: 1.3 $� ���� )�htb)�unittest� )�
DummyConsumerc @ s e Zd ZdZdd� Zdd� ZdS )�
DummyClockr c C s
|| _ d S �N��time)�self�when� r �7/usr/lib/python3/dist-packages/twisted/test/test_htb.py�set
s
zDummyClock.setc C s | j S r r
�r r r r �__call__ s zDummyClock.__call__N)�__name__�
__module__� __qualname__r r r r r r r r
s r c @ s e Zd ZdZdZdS )�
SomeBucket�d � N)r r r �maxburst�rater r r r r s r c @ s e Zd Zdd� Zdd� ZdS )�TestBucketBasec C s t j| _t� | _| jt _d S r )r r �
_realTimeFuncr �clockr r r r �setUp s zTestBucketBase.setUpc C s | j t_d S r )r r r r r r r �tearDown s zTestBucketBase.tearDownN)r r r r r r r r r r s r c @ �$ e Zd Zdd� Zdd� Zdd� ZdS )� BucketTestsc C s t � }|�d�}| �d|� dS )z1
Testing the size of the bucket.
�� r N)r �add� assertEqual�r �b�fitr r r �testBucketSize$ s
zBucketTests.testBucketSizec C s6 t � }|�d�}| j�d� |�d�}| �d|� dS )z2
Testing the bucket's drain rate.
r" �
� N)r r# r r r$ r% r r r �testBucketDrain, s
zBucketTests.testBucketDrainc C sP t � }|�d� | j�d� |�� }| �|� | j�d� |�� }| �|� dS )z\
L{htb.Bucket.drip} returns C{True} if the bucket is empty after that drip.
r* � r) N)r r# r r �drip� assertFalse�
assertTrue)r r& �emptyr r r �test_bucketEmpty6 s
z BucketTests.test_bucketEmptyN)r r r r( r+ r1 r r r r r! # s
r! c @ r )�BucketNestingTestsc C s. t �| � t� | _t| j�| _t| j�| _d S r )r r r �parent�child1�child2r r r r r E s
zBucketNestingTests.setUpc C s( | j �d� | j�d�}| �d|� d S )N�Z r) )r4 r# r5 r$ �r r' r r r �testBucketParentSizeK s z'BucketNestingTests.testBucketParentSizec C s< d| j _| j�d� | j�d� | j�d�}| �d|� d S )Nr r r) )r3 r r4 r# r r r$ r7 r r r �testBucketParentRateQ s
z'BucketNestingTests.testBucketParentRateN)r r r r r8 r9 r r r r r2 D s r2 c @ r )�ConsumerShaperTestsc C s0 t �| � t� | _t� | _t�| j| j�| _d S r ) r r r �
underlyingr �bucketr �ShapedConsumer�shapedr r r r r c s
zConsumerShaperTests.setUpc C sT d}| j �d� | j�d� | j�|� | j�� | �t| j �
� �|| j j � d S )Nr) r �dxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx) r< r# r> �writer r �resumeProducingr$ �lenr; �getvaluer )r �delta_tr r r �testRatei s
"z ConsumerShaperTests.testRatec C s. | � | jjd� | j�� | � | jjd� d S )Nr r )r$ r< � _refcountr> �
stopProducingr r r r �testBucketRefss s
z"ConsumerShaperTests.testBucketRefsN)r r r r rE rH r r r r r: b s
r: N)� __version__�twisted.protocolsr �
twisted.trialr �test_pcpr r �Bucketr �TestCaser r! r2 r: r r r r �<module> s
!