a
    ñ»jP'  ã                   @   s~   d dl Z d dlZd dlmZmZmZ d dlmZ d dlm	Z	 d dl
mZmZ d dlmZ G dd„ de jƒZed	krze  ¡  dS )
é    N)ÚMockÚpatchÚ	MagicMock)ÚABNF)Ú	WebSocket)ÚWebSocketProtocolExceptionÚWebSocketPayloadException)ÚSSLErrorc                   @   sL   e Zd Zdd„ Zdd„ Zdd„ Zdd„ Zd	d
„ Zdd„ Zdd„ Z	dd„ Z
dS )ÚLargePayloadTestc              	   C   s–   g d¢}|D ]„\}}| j ||d��^ d| }t |tj¡}| ¡ }|  |t¡ |  t|ƒ|k¡ |  	t|j
ƒ|¡ W d  ƒ q1 s†0    Y  qdS )z:Test WebSocket frame length encoding at various boundaries)	)é}   zSingle byte length)é~   zTwo byte length start)é   zTwo byte length)éÿÿ  zTwo byte length max)é   zEight byte length start)é @  z16KB boundary)é@  zJust over 16KB)é €  Z32KB)i   Z128KB)ÚlengthÚdescriptionó   AN)ÚsubTestr   Úcreate_frameÚOPCODE_BINARYÚformatÚassertIsInstanceÚbytesÚ
assertTrueÚlenÚassertEqualÚdata)ÚselfZ
test_casesr   r   ÚpayloadÚframeÚ	formatted© r$   úIE:\soccer\ml_env\Lib\site-packages\websocket/tests/test_large_payloads.pyÚ%test_frame_length_encoding_boundaries   s    z6LargePayloadTest.test_frame_length_encoding_boundariesc                    s�   dd }g ‰d}t dt|ƒ|ƒD ]}ˆ |||| … ¡ q d‰ ‡ ‡fdd„}ddlm} ||dd	�}| t|ƒ¡}|  ||¡ |  ˆ d
¡ dS )zHTest receiving large payloads in chunks (simulating the 16KB recv issue)ó   Br   r   r   c                    s$   ˆ t ˆƒkrdS ˆˆ  }ˆ d7 ‰ |S ©Nó    é   ©r   ©ÚbufsizeÚresult©Ú
call_countÚchunksr$   r%   Ú	mock_recvO   s
    zCLargePayloadTest.test_recv_large_payload_chunked.<locals>.mock_recv©Úframe_bufferT©Zskip_utf8_validationr*   N)Úranger   ÚappendÚwebsocket._abnfr4   Úrecv_strictr   ÚassertGreater)r    Úlarge_payloadÚ
chunk_sizeÚir2   r4   Úfbr.   r$   r/   r%   Útest_recv_large_payload_chunkedA   s    	z0LargePayloadTest.test_recv_large_payload_chunkedc                    sr   d}g ‰ ‡ fdd„}ddl m} ||dd�}| |¡}|  t|ƒ|¡ |  tdd	„ ˆ D ƒƒ¡ |  tˆ ƒd
¡ dS )z&Simulate SSL BAD_LENGTH error scenarior   c                    s(   ˆ   | ¡ | dkrtdƒ‚dt| dƒ S )Nr   z[SSL: BAD_LENGTH] unknown erroró   C)r7   r	   Úmin)r-   ©Z
recv_callsr$   r%   Úmock_recv_with_ssl_limitk   s    
zTLargePayloadTest.test_ssl_large_payload_simulation.<locals>.mock_recv_with_ssl_limitr   r3   Tr5   c                 s   s   | ]}|d kV  qdS )r   Nr$   )Ú.0Úcallr$   r$   r%   Ú	<genexpr>|   r)   zELargePayloadTest.test_ssl_large_payload_simulation.<locals>.<genexpr>r*   N)r8   r4   r9   r   r   r   Úallr:   )r    Zpayload_sizerC   r4   r>   r.   r$   rB   r%   Ú!test_ssl_large_payload_simulationc   s    
z2LargePayloadTest.test_ssl_large_payload_simulationc              	   C   sÆ   g d¢}|D ]´}| j |d��” d| }t |tj¡}| ¡ }|  |t¡ |  t|j	ƒ|¡ d}|tj
k rrd| }n|tjk r†d| }nd| }|  t|ƒ|| ¡ W d  ƒ q1 s¶0    Y  qdS )z6Test frame formatting with various large payload sizes)iÿ?  r   r   r   r   r   )Úsizeó   Dé   é   é
   N)r   r   r   r   r   r   r   r   r   r   ZLENGTH_7Z	LENGTH_16)r    Z
test_sizesrI   r!   r"   r#   Z	mask_sizeZexpected_header_sizer$   r$   r%   Ú test_frame_format_large_payloads€   s$    
ÿ
ÿÿz1LargePayloadTest.test_frame_format_large_payloadsc                    sŠ   t ƒ }g ‰ ‡ fdd„}||_d|j_tƒ }||_d|_dd }tdƒ�0}dd	„ |_| 	|¡}|  
|d
¡ W d  ƒ n1 s|0    Y  dS )z?Test that large payloads are sent in chunks to avoid SSL issuesc                    s   ˆ   t| ƒ¡ t| ƒS ©N)r7   r   )r   ©Zsent_chunksr$   r%   Ú	mock_send¯   s    zDLargePayloadTest.test_send_large_payload_chunking.<locals>.mock_sendg      >@Tó   Er   zwebsocket._core.sendc                 S   s   t |ƒS rO   r+   )Úsockr   r$   r$   r%   Ú<lambda>À   r)   zCLargePayloadTest.test_send_large_payload_chunking.<locals>.<lambda>r   N)r   ÚsendÚ
gettimeoutZreturn_valuer   rS   Ú	connectedr   Zside_effectZsend_binaryr:   )r    Z	mock_sockrQ   Úwsr;   Zmock_send_funcr.   r$   rP   r%   Ú test_send_large_payload_chunking§   s    


z1LargePayloadTest.test_send_large_payload_chunkingc                 C   sŠ   dd }t  |t j¡}| ¡ }|  |t¡ t dd¡d }t ddddt jd|ƒ}|  	t
¡� |jdd	� W d
  ƒ n1 s|0    Y  d
S )z.Test UTF-8 validation with large text payloadsu   Hello ä¸–ç•Œ! iÐ  z!Hiè  s   ÿþ invalid utf8r*   r   Fr5   N)r   r   ZOPCODE_TEXTr   r   r   ÚstructÚpackZOPCODE_CLOSEZassertRaisesr   Úvalidate)r    Z
large_textr"   r#   Zinvalid_utf8_close_datar$   r$   r%   Útest_utf8_validation_large_textÈ   s    z0LargePayloadTest.test_utf8_validation_large_textc                    sv   dd ‰‡fdd„t dtˆƒdƒD ƒ‰d‰ ‡ ‡fdd„}dd	lm} ||d
d�}| d¡}|  |ˆ¡ |  ˆ d¡ dS )z9Test frame buffer with edge cases that could trigger bugsó   Fr   c                    s   g | ]}ˆ ||d  … ‘qS )é   r$   )rD   r=   )Úpayload_16kr$   r%   Ú
<listcomp>æ   r)   zALargePayloadTest.test_frame_buffer_edge_cases.<locals>.<listcomp>r   r_   c                    s$   ˆ t ˆƒkrdS ˆˆ  }ˆ d7 ‰ |S r(   r+   r,   r/   r$   r%   r2   ê   s
    z@LargePayloadTest.test_frame_buffer_edge_cases.<locals>.mock_recvr3   Tr5   rK   N)r6   r   r8   r4   r9   r   )r    r2   r4   r>   r.   r$   )r0   r1   r`   r%   Útest_frame_buffer_edge_casesß   s    
z-LargePayloadTest.test_frame_buffer_edge_casesc                 C   sD   d}d| }t  |t j¡}| ¡ }|  |t¡ |  t|jƒ|¡ dS )z4Test behavior at WebSocket maximum frame size limitsi   ó   GN)	r   r   r   r   r   r   r   r   r   )r    Z
large_sizer!   r"   r#   r$   r$   r%   Útest_max_frame_size_limitsû   s    z+LargePayloadTest.test_max_frame_size_limitsN)Ú__name__Ú
__module__Ú__qualname__r&   r?   rH   rN   rY   r]   rb   rd   r$   r$   r$   r%   r
      s   ""'!r
   Ú__main__)ZunittestrZ   Zunittest.mockr   r   r   r8   r   Zwebsocket._corer   Zwebsocket._exceptionsr   r   Zwebsocket._ssl_compatr	   ZTestCaser
   re   Úmainr$   r$   r$   r%   Ú<module>   s    s