File: //proc/1233/cwd/usr/lib/python3/dist-packages/websocket/tests/__pycache__/test_app.cpython-310.pyc
o
�a � @ s� d Z ddlZddlZddlZddlZddlZej�dd�dkZ ej�dd�Z
e
dkZ dZ G d d
� d
ej
�Zed kr?e�� dS dS ) ab
test_app.py
websocket - WebSocket client library for Python
Copyright 2021 engn33r
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
� N�TEST_WITH_INTERNET�0�1�LOCAL_WS_SERVER_PORTz-1Tc @ s� e Zd ZG dd� d�Zdd� Zdd� Ze�ed�dd � �Z e�e
d
�d d � �Z e�e
d
�d
d� �Z e�e
d
�dd� �Z
e�e
d
�dd� �Ze�e
d
�dd� �Ze�e
d
�dd� �Ze�e
d
�dd� �Ze�e
d
�dd� �ZdS ) �WebSocketAppTestc @ s e Zd ZdZdS )zWebSocketAppTest.NotSetYetzI A marker class for signalling that a value hasn't been set yet.
N)�__name__�
__module__� __qualname__�__doc__� r r �:/usr/lib/python3/dist-packages/websocket/tests/test_app.py� NotSetYet& s r
c C s, t �t� t�� t_t�� t_t�� t_d S �N)�ws� enableTrace� TRACEABLEr r
�keep_running_open�keep_running_close�get_mask_key_id��selfr r r �setUp* s
zWebSocketAppTest.setUpc C s" t �� t _t �� t _t �� t _d S r )r r
r r r r r r r �tearDown1 s
zWebSocketAppTest.tearDownz/Tests using local websocket server are disabledc s> dd� }� fdd�}dd� }t jdt |||d�}|�� d S )
z| A WebSocketApp should keep running as long as its self.keep_running
is not False (in the boolean context).
c _ s | � d� | jt_d| _dS )zn Set the keep_running flag for later inspection and immediately
close the connection.
zhello!FN)�send� keep_runningr r �r �args�kwargsr r r �on_open< s
z1WebSocketAppTest.testKeepRunning.<locals>.on_openc s t |� � �� d S r ��print�close)�wsapp�messager r r �
on_messageD � z4WebSocketAppTest.testKeepRunning.<locals>.on_messagec _ s | j t_dS )z< Set the keep_running flag for the test to use.
N)r r r r r r r �on_closeH s z2WebSocketAppTest.testKeepRunning.<locals>.on_closezws://127.0.0.1:)r r& r$ N)r � WebSocketAppr � run_forever)r r r$ r& �appr r r �testKeepRunning6 s
z WebSocketAppTest.testKeepRunningz%Internet-requiring tests are disabledc C s0 dd� }t jd|d�}| �t|j�t|�� dS )zi A WebSocketApp should forward the received mask_key function down
to the actual socket.
c S s dS )Nz r r r r r �my_mask_key_funcV s z:WebSocketAppTest.testSockMaskKey.<locals>.my_mask_key_funcz wss://stream.meetup.com/2/rsvps)� get_mask_keyN)r r' � assertEqual�idr, )r r+ r) r r r �testSockMaskKeyP s z WebSocketAppTest.testSockMaskKeyc C sB dd� }dd� }t jd||d�}| jt j|jddd tjid
� d S ) zA Test exception handling if ping_interval < ping_timeout
c S � t d� | �� d S �Nz Got a ping!r �r) �msgr r r �on_pingd r% zDWebSocketAppTest.testInvalidPingIntervalPingTimeout.<locals>.on_pingc S r0 �Nz Got a pong! No need to respondr r2 r r r �on_pongh r% zDWebSocketAppTest.testInvalidPingIntervalPingTimeout.<locals>.on_pong� wss://api-pub.bitfinex.com/ws/1�r4 r6 � � � cert_reqs��
ping_interval� ping_timeout�ssloptN�r r' � assertRaises�WebSocketExceptionr( �ssl� CERT_NONE�r r4 r6 r) r r r �"testInvalidPingIntervalPingTimeout_ s "z3WebSocketAppTest.testInvalidPingIntervalPingTimeoutc C s: dd� }dd� }t jd||d�}|jddd tjid
� d S ) z5 Test WebSocketApp proper ping functionality
c S r0 r1 r r2 r r r r4 t r% z2WebSocketAppTest.testPingInterval.<locals>.on_pingc S r0 r5 r r2 r r r r6 x r% z2WebSocketAppTest.testPingInterval.<locals>.on_pongr7 r8 r: r9 r; r< N)r r' r( rC rD rE r r r �testPingIntervalo s z!WebSocketAppTest.testPingIntervalc C � t �d�}|jdddd� dS )z( Test WebSocketApp close opcode
�'wss://tsock.us1.twilio.com/v3/wsconnectr: r9 � Ping payload�r= r> � ping_payloadN�r r' r( �r r) r r r �testOpcodeClose �
z WebSocketAppTest.testOpcodeClosec C rH )z) Test WebSocketApp binary opcode
z'streaming.vn.teslamotors.com/streaming/r: r9 rJ rK NrM rN r r r �testOpcodeBinary� rP z!WebSocketAppTest.testOpcodeBinaryc C �* t �d�}| jt j|jddtjid� dS )z; A WebSocketApp handling of negative ping_interval
r7 ���r; )r= r? Nr@ rN r r r �testBadPingInterval� �
z$WebSocketAppTest.testBadPingIntervalc C rR )z: A WebSocketApp handling of negative ping_timeout
r7 ���r; )r> r? Nr@ rN r r r �testBadPingTimeout� rU z#WebSocketAppTest.testBadPingTimeoutc C s� dd� }t jd|d�}t jt jjdd�}| �ddg|�|�� t jt jjd d�}| �d
d
g|�|�� t �d�}t jt jjd d�}| �d
d
g|�|�� | jt j|jd d � d
S )
zU Test extraction of close frame status code and close reason in WebSocketApp
c S s t d� d S )Nzon_close reached)r )r" �close_status_code� close_msgr r r r& � s z6WebSocketAppTest.testCloseStatusCode.<locals>.on_closerI )r& s �no-init-from-client)�opcode�datai� zno-init-from-client� Nz test if connection is closed)r[ ) r r' �ABNF� OPCODE_CLOSEr- �_get_close_argsrA �"WebSocketConnectionClosedExceptionr )r r& r) �
closeframe�app2r r r �testCloseStatusCode� s
z$WebSocketAppTest.testCloseStatusCodeN)r r r r
r r �unittest�
skipUnless�TEST_WITH_LOCAL_SERVERr* r r/ rF rG rO rQ rT rW rc r r r r r $ s,
r �__main__)r
�os�os.path� websocketr rC rd �environ�getr r rf r �TestCaser r �mainr r r r �<module> s �