o
    èó·ij6  ã                   @   s  d dl Z d dlZe j dd¡r‚d dlmZ d dlmZ d dlm	Z	 d dl
mZ d dl
mZ d d	l
mZ d d
l
mZ d dlmZ d dlmZ d dlmZmZ d dlmZ d dlmZ edƒdd„ ƒZdZdZG dd„ dejƒZG dd„ dejƒZG dd„ dejƒZ dS dS )é    NZUSE_TWISTEDF)Úunittest)ÚIPv4Address)ÚClock)ÚWebSocketServerProtocol)ÚWebSocketServerFactory)ÚWebSocketClientProtocol)ÚWebSocketClientFactory)ÚPerMessageDeflate)ÚFakeTransport)Ú	MagicMockÚpatch)Úreplace_loop)Ú	b64decodezbase64.b64encodec                    s€   d| _ tdgd�}t|_| ¡  | tdddƒ¡}tƒ |_| 	¡  t
|_| ¡  g ‰ ‡ fdd„}||_|jdi |¤Ž d	 ˆ ¡S )a[  
        Kind-of hack-y; maybe better to re-factor the Protocol to have a
        frame-encoder method-call? Anyway, makes a throwaway protocol
        encode a frame for us, collects the .sendData call and returns
        the data that would have gone out. Accepts all the kwargs that
        WebSocketClientProtocol.sendFrame() accepts.
        s   QIatSt9QkZPyS4QQfdufO8TgkL0=úwamp.2.json©Ú	protocolsÚTCPz	127.0.0.9éþÿ  c                    s   ˆ   | ¡ d S ©N)Úappend)ÚdÚargs©Údata© úrC:\Users\psant\Downloads\PAF backend\school\venv\Lib\site-packages\autobahn/websocket/test/test_websocket_frame.pyÚcollectI   s   z$create_client_frame.<locals>.collectó    Nr   )Úreturn_valuer   r   ÚprotocolÚdoStartÚbuildProtocolr   r
   Ú	transportÚconnectionMadeÚmock_handshake_serverr   ÚprocessHandshakeZsendDataZ	sendFrameÚjoin)Zb64patchÚkwargsÚfactoryÚprotor   r   r   r   Úcreate_client_frame0   s   
r*   s
  GET / HTTP/1.1
User-Agent: AutobahnPython/0.10.2
Host: localhost:80
Upgrade: WebSocket
Connection: Upgrade
Pragma: no-cache
Cache-Control: no-cache
Sec-WebSocket-Key: 6Jid6RgXpH0RVegaNSs/4g==
Sec-WebSocket-Protocol: wamp.2.json
Sec-WebSocket-Version: 13

s  HTTP/1.1 101 Switching Protocols
Server: AutobahnPython/0.10.2
X-Powered-By: AutobahnPython/0.10.2
Upgrade: WebSocket
Connection: Upgrade
Sec-WebSocket-Protocol: wamp.2.json
Sec-WebSocket-Accept: QIatSt9QkZPyS4QQfdufO8TgkL0=

�~[1,"crossbar",{"roles":{"subscriber":{"features":{"publisher_identification":true,"pattern_based_subscription":true,"subscription_revocation":true}},"publisher":{"features":{"publisher_identification":true,"publisher_exclusion":true,"subscriber_blackwhite_listing":true}},"caller":{"features":{"caller_identification":true,"progressive_call_results":true}},"callee":{"features":{"progressive_call_results":true,"pattern_based_registration":true,"registration_revocation":true,"shared_registration":true,"caller_identification":true}}}}]c                   @   s   e Zd Zdd„ Zdd„ ZdS )ÚTestDeflatec              	   C   s<   t dddddddd�}d}| ¡  | |¡}|  |d¡ d S )NFé   é   é
   )Ú	is_serverÚserver_no_context_takeoverÚclient_no_context_takeoverÚserver_max_window_bitsÚclient_max_window_bitsÚ	mem_levelZmax_message_sizeó   «¨£`Œ‚Q0
FÁP Ï@©®s
   xxxxxxxxxx©r	   Zstart_decompress_messageZdecompress_message_dataÚassertEqual©ÚselfÚdecoderZcompressed_datar   r   r   r   Útest_max_sizeW   s   ù
zTestDeflate.test_max_sizec                 C   s:   t dddddd d�}d}| ¡  | |¡}|  |d¡ d S )NFr,   )r/   r0   r1   r2   r3   r4   r5   sÐ  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxr6   r8   r   r   r   Útest_no_max_sizel   s   ú

zTestDeflate.test_no_max_sizeN)Ú__name__Ú
__module__Ú__qualname__r;   r<   r   r   r   r   r+   U   s    r+   c                   @   s,   e Zd Zdd„ Zdd„ Zdd„ Zdd„ Zd	S )
Ú
TestClientc                 C   óV   t dgd�| _t| j_| j ¡  | j tdddƒ¡| _tƒ | _	| j	| j_	| j 
¡  d S ©Nr   r   r   z	127.0.0.1r   )r   r(   r   r   r    r!   r   r)   r
   r"   r#   ©r9   r   r   r   ÚsetUp   ó   

zTestClient.setUpc                 C   ó*   | j jr
| j j ¡  | j ¡  | `| ` d S r   ©r)   ZopenHandshakeTimeoutCallÚcancelr(   ZdoStoprC   r   r   r   ÚtearDown‰   ó
   
zTestClient.tearDownc                 C   s8   d| j _| j j| j _d| j _tƒ | j _| j  dd ¡ d S )NTé   éè  )r)   ZechoCloseCodeReasonÚ
STATE_OPENÚstateZwebsocket_versionr   ÚsendCloseFrameÚonCloseFramerC   r   r   r   Útest_missing_reason_raw‘   s
   
z"TestClient.test_missing_reason_rawc                 C   s  	 tdƒ| j_t| j_| j ¡  |  	| jj
tj¡ |  | jjdk¡ ttƒ ƒ�R}| j ¡  | j dd¡ |  	t|jƒd¡ |  	|jd j| jj¡ |  	|jd  ¡ | jj¡ | | jjd ¡ |  | jj ¡ ¡ |  	| jj
tj¡ W d  ƒ dS 1 s|w   Y  dS )zR
            make a delayed call to drop the connection (client-side)
            Fz6Jid6RgXpH0RVegaNSs/4g==r   rL   s
   raw reasonrK   N)Úprintr)   r(   Ú_logr   Zwebsocket_keyr$   r   r%   r7   rN   r   rM   Ú
assertTrueZserverConnectionDropTimeoutr   r   rO   rP   ÚlenÚcallsÚfuncZonServerConnectionDropTimeoutZgetTimeÚadvanceÚcloseHandshakeTimeoutr"   Zabort_calledÚSTATE_CLOSED©r9   Úreactorr   r   r   Útest_unclean_timeout_client›   s    

"ïz&TestClient.test_unclean_timeout_clientN)r=   r>   r?   rD   rI   rQ   r]   r   r   r   r   r@   ~   s
    

r@   c                   @   s4   e Zd Zdd„ Zdd„ Zdd„ Zdd„ Zd	d
„ ZdS )ÚTestPingc                 C   rA   rB   )r   r(   r   r   r    r!   r   r)   r   r"   r#   rC   r   r   r   rD   ¾   rE   zTestPing.setUpc                 C   rF   r   rG   rC   r   r   r   rI   È   rJ   zTestPing.tearDownc                 C   sÂ   t | j_| j ¡  |  | jjtjk¡ tt	ƒ ƒ�?}| j 
¡  |  t|jƒd¡ | | jjd ¡ |  dt| jjjd ƒ¡ |  | jjjj¡ |  | jjtj¡ W d  ƒ dS 1 sZw   Y  dS )zD
            make a delayed call to drop the connection
            rK   zcall.abortConnection()éÿÿÿÿN)Úmock_handshake_clientr)   r   r%   rT   rN   r   rM   r   r   rO   r7   rU   rV   rX   rY   Ústrr"   Zmethod_callsZabortConnectionÚcalledrZ   r[   r   r   r   Útest_unclean_timeoutÐ   s   

"ñzTestPing.test_unclean_timeoutc                 C   s  d| j _d| j _ttƒ ƒ�u}t| j _| j  ¡  |  | j j	t
jk¡ |  dt|jƒ¡ | d¡ |  | jjj¡ | jjjd d }t|d gƒ}t d|¡\}|d@ }|  d|¡ |  dt|jƒ¡ |  | j j	| j j¡ | d¡ |  | j j	| j j¡ W d	  ƒ d	S 1 s…w   Y  d	S )
z>
            autoping and autoping-timeout timing
            é   é   rK   r   ÚBiÿÿÿé	   é   N)r)   ÚautoPingIntervalÚautoPingTimeoutr   r   r`   r   r%   rT   rN   r   rM   r7   rU   rV   rX   r"   Úwriterb   Z	call_argsÚbytesÚstructÚunpackZassertNotEqualrZ   )r9   r\   r   Ú_dataÚopcoder   r   r   Útest_auto_pingpong_timeoutð   s&   


"ßz#TestPing.test_auto_pingpong_timeoutc                 C   sÖ   d| j _d| j _ttƒ ƒ�T}t| j _| j  ¡  |  | j j	t
jk¡ |  dt|jƒ¡ | d¡ |  dt|jƒ¡ |jd }td| j jd�}|| j _| j  ¡  | j  ¡  |  |j¡ W d  ƒ dS 1 sdw   Y  dS )zF
            auto-ping with correct reply cancels timeout
            rd   re   rK   r   r.   )rp   ÚpayloadN)r)   ri   rj   r   r   r`   r   r%   rT   rN   r   rM   r7   rU   rV   rX   r*   ZautoPingPendingZprocessDataÚ	cancelled)r9   r\   Ztimeout_callÚframer   r   r   Útest_auto_ping_got_pong  s    




"äz TestPing.test_auto_ping_got_pongN)r=   r>   r?   rD   rI   rc   rq   ru   r   r   r   r   r^   ½   s    
 +r^   )!Úosrm   ÚenvironÚgetZtwisted.trialr   Ztwisted.internet.addressr   Ztwisted.internet.taskr   Zautobahn.twisted.websocketr   r   r   r   Z#autobahn.websocket.compress_deflater	   Zautobahn.testutilr
   Zunittest.mockr   r   Ztxaio.testutilr   Úbase64r   r*   r`   r$   ZTestCaser+   r@   r^   r   r   r   r   Ú<module>   s0   
 )? �á