File: //opt/alt/python37/lib/python3.7/site-packages/tests/unit/__pycache__/test_chunking.cpython-37.pyc
B
�P�e� � @ s� d Z ddlZddlZddlZddlZddlZddlmZmZm Z m
Z
m Z m Z m
Z
ej�de eeg�dd� �Zdd� Zej�de eeg�d d
� �Zej�de eeg�d d � �Zd
d� Zdd� Zdd� Zdd� ZdS )z-pytests for various GELF UDP message chunkers� N)�GELFTruncatingChunker�GELFWarningChunker�BaseGELFChunker�BaseGELFHandler�
SYSLOG_LEVELS�GELFChunkOverflowWarning� GELFTruncationFailureWarning� gelf_chunkerc
C s� d}d}t | dd��|��}t�dd�t�dd�dft�dd �t�dd�d
ft�dd�t�dd�d fg}t|�t|�ks|t�xrt|�D ]f\}}|| \}}} ||d d� ks�t�||d
d� ks�t�||dd� ks�t�| |dd � ks�t�q�W d S )zTest various GELF chunkerss 12345s � )�
chunk_size�br � s 12� s 34� 5N�
� � )�list�
chunk_message�struct�pack�len�AssertionError� enumerate)
r �message�header�chunks�expected�index�chunkZexpected_indexZexpected_chunks_countZexpected_chunk� r �I/opt/alt/python37/lib/python3.7/site-packages/tests/unit/test_chunking.py�test_gelf_chunking s r" c C sl d}t | d �}xV| D ]N}t |�|k rB|||t |� d � 7 }q||dt�d� t |� d � 7 }qW |S )N� r r
ZQBB)r r �calcsize)r Z
gelf_bytesZbsizer r r r! �"rebuild_gelf_bytes_from_udp_chunks0 s
&r% c C sH t � �t�dtjd d dd d ��}t| dd��|��}t|�dksDt�d S )N�test_gelf_chunkersZ
1111111111r
)r � ) r �
makePickle�logging� LogRecord�INFOr r r r )r r r r r r! r& ; s
r& c C sH t � �t�dtjd d dd d ��}t| dd��|��}t|�dksDt�d S )N�test_gelf_chunkers_overflow�� 1111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111r )r r' ) r r( r) r* r+ r r r r )r r r r r r! r, H s r, c C s� t dd��t�dtjd d dd d ��} t�t�� tt ddd��
| ��}W d Q R X t |�dks`t �t
|��d�}t�|�}|d d
ks�t �|d dks�t �|d t�tjtj�ks�t �d S )
NF)�compress�)test_chunk_overflow_truncate_uncompressedr- r
)r r. r' zUTF-8�_chunk_overflowT�
short_message�level)r r( r) r* r+ �pytest�warnsr r r r r r r% �decode�json�loadsr �get�ERROR)r r �payload� glef_jsonr r r! r/ [ s$
r/ c C s� t dd��t�dtjd d dd d d ��} t�t�� tt ddd��
| ��}W d Q R X t |�dksdt �t
�t|���d �}t�|�}|d
dks�t �|d dd ks�t �|d t�tjtj�ks�t �d S )
NT)r. �'test_chunk_overflow_truncate_compressedZ 123412345i� r
)r r. r' zUTF-8r0 r1 r2 )r r( r) r* r+ r3 r4 r r r r r r �zlib�
decompressr% r5 r6 r7 r r8 r9 )r r r: r; r r r! r< s s&
r< c C sL t � �t�dtjd d dd d ��} t�t�� tt d��
| �� W d Q R X d S )N�!test_chunk_overflow_truncate_failr- r ) r r( r) r* r+ r3 r4 r r r r )r r r r! r? � s r? c C sT t dd d��t�dtjd d dd d ��} t�t�� tt d��
| �� W d Q R X d S )Nz this is a really long facilityi� )�facilityr? zreasonable messager
) r r( r) r* r+ r3 r4 r r r r )r r r r! �7test_chunk_overflow_truncate_fail_large_inherited_field� s rA )�__doc__r6 r) r r= r3 Zgraypy.handlerr r r r r r r �markZ parametrizer" r% r& r, r/ r< r? rA r r r r! �<module> s" $