<s> import logging <EOL> import time <EOL> from unittest import TestCase <EOL> from . . import SocketIO , LoggingNamespace , find_callback <EOL> HOST = '<STR_LIT:localhost>' <EOL> PORT = <NUM_LIT> <EOL> DATA = '<STR_LIT>' <EOL> PAYLOAD = { '<STR_LIT>' : '<STR_LIT>' } <EOL> logging . basicConfig ( level = logging . DEBUG ) <EOL> class BaseMixin ( object ) : <EOL> def setUp ( self ) : <EOL> super ( BaseMixin , self ) . setUp ( ) <EOL> self . response_count = <NUM_LIT:0> <EOL> self . wait_time_in_seconds = <NUM_LIT:1> <EOL> def tearDown ( self ) : <EOL> super ( BaseMixin , self ) . tearDown ( ) <EOL> self . socketIO . disconnect ( ) <EOL> def test_disconnect ( self ) : <EOL> '<STR_LIT>' <EOL> namespace = self . socketIO . define ( Namespace ) <EOL> self . assertTrue ( self . socketIO . connected ) <EOL> self . assertFalse ( namespace . called_on_disconnect ) <EOL> self . socketIO . disconnect ( ) <EOL> self . assertTrue ( namespace . called_on_disconnect ) <EOL> self . assertFalse ( self . socketIO . connected ) <EOL> def test_emit ( self ) : <EOL> '<STR_LIT>' <EOL> namespace = self . socketIO . define ( Namespace ) <EOL> self . socketIO . emit ( '<STR_LIT>' ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( namespace . args_by_event , { <EOL> '<STR_LIT>' : ( ) , <EOL> } ) <EOL> def test_emit_with_payload ( self ) : <EOL> '<STR_LIT>' <EOL> namespace = self . socketIO . define ( Namespace ) <EOL> self . socketIO . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( namespace . args_by_event , { <EOL> '<STR_LIT>' : ( PAYLOAD , ) , <EOL> } ) <EOL> def test_emit_with_multiple_payloads ( self ) : <EOL> '<STR_LIT>' <EOL> namespace = self . socketIO . define ( Namespace ) <EOL> self . socketIO . emit ( '<STR_LIT>' , PAYLOAD , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( namespace . args_by_event , { <EOL> '<STR_LIT>' : ( PAYLOAD , PAYLOAD ) , <EOL> } ) <EOL> def test_emit_with_callback ( self ) : <EOL> '<STR_LIT>' <EOL> self . socketIO . emit ( '<STR_LIT>' , self . on_response ) <EOL> self . socketIO . wait_for_callbacks ( seconds = self . wait_time_in_seconds ) <EOL> self . assertEqual ( self . response_count , <NUM_LIT:1> ) <EOL> def test_emit_with_callback_with_payload ( self ) : <EOL> '<STR_LIT>' <EOL> self . socketIO . emit ( <EOL> '<STR_LIT>' , self . on_response ) <EOL> self . socketIO . wait_for_callbacks ( seconds = self . wait_time_in_seconds ) <EOL> self . assertEqual ( self . response_count , <NUM_LIT:1> ) <EOL> def test_emit_with_callback_with_multiple_payloads ( self ) : <EOL> '<STR_LIT>' <EOL> self . socketIO . emit ( <EOL> '<STR_LIT>' , self . on_response ) <EOL> self . socketIO . wait_for_callbacks ( seconds = self . wait_time_in_seconds ) <EOL> self . assertEqual ( self . response_count , <NUM_LIT:1> ) <EOL> def test_emit_with_event ( self ) : <EOL> '<STR_LIT>' <EOL> self . socketIO . on ( '<STR_LIT>' , self . on_response ) <EOL> self . socketIO . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( self . response_count , <NUM_LIT:1> ) <EOL> self . socketIO . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( self . response_count , <NUM_LIT:2> ) <EOL> self . socketIO . off ( '<STR_LIT>' ) <EOL> self . socketIO . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( self . response_count , <NUM_LIT:2> ) <EOL> def test_once ( self ) : <EOL> '<STR_LIT>' <EOL> self . socketIO . once ( '<STR_LIT>' , self . on_response ) <EOL> self . socketIO . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( self . response_count , <NUM_LIT:1> ) <EOL> def test_send ( self ) : <EOL> '<STR_LIT>' <EOL> namespace = self . socketIO . define ( Namespace ) <EOL> self . socketIO . send ( ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( namespace . response , '<STR_LIT>' ) <EOL> def test_send_with_data ( self ) : <EOL> '<STR_LIT>' <EOL> namespace = self . socketIO . define ( Namespace ) <EOL> self . socketIO . send ( DATA ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( namespace . response , DATA ) <EOL> def test_ack ( self ) : <EOL> '<STR_LIT>' <EOL> namespace = self . socketIO . define ( Namespace ) <EOL> self . socketIO . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( namespace . args_by_event , { <EOL> '<STR_LIT>' : ( PAYLOAD , ) , <EOL> '<STR_LIT>' : ( PAYLOAD , ) , <EOL> } ) <EOL> def test_wait_with_disconnect ( self ) : <EOL> '<STR_LIT>' <EOL> self . socketIO . define ( Namespace ) <EOL> self . socketIO . disconnect ( ) <EOL> timeout_in_seconds = <NUM_LIT:5> <EOL> start_time = time . time ( ) <EOL> self . socketIO . wait ( timeout_in_seconds ) <EOL> self . assertTrue ( time . time ( ) - start_time < timeout_in_seconds ) <EOL> def test_namespace_emit ( self ) : <EOL> '<STR_LIT>' <EOL> main_namespace = self . socketIO . define ( Namespace ) <EOL> chat_namespace = self . socketIO . define ( Namespace , '<STR_LIT>' ) <EOL> news_namespace = self . socketIO . define ( Namespace , '<STR_LIT>' ) <EOL> news_namespace . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( main_namespace . args_by_event , { } ) <EOL> self . assertEqual ( chat_namespace . args_by_event , { } ) <EOL> self . assertEqual ( news_namespace . args_by_event , { <EOL> '<STR_LIT>' : ( PAYLOAD , ) , <EOL> } ) <EOL> def test_namespace_ack ( self ) : <EOL> '<STR_LIT>' <EOL> chat_namespace = self . socketIO . define ( Namespace , '<STR_LIT>' ) <EOL> chat_namespace . emit ( '<STR_LIT>' , PAYLOAD ) <EOL> self . socketIO . wait ( self . wait_time_in_seconds ) <EOL> self . assertEqual ( chat_namespace . args_by_event , { <EOL> '<STR_LIT>' : ( PAYLOAD , ) , <EOL> '<STR_LIT>' : ( PAYLOAD , ) , <EOL> } ) <EOL> def on_response ( self , * args ) : <EOL> for arg in args : <EOL> if isinstance ( arg , dict ) : <EOL> self . assertEqual ( arg , PAYLOAD ) <EOL> else : <EOL> self . assertEqual ( arg , DATA ) <EOL> self . response_count += <NUM_LIT:1> <EOL> class Test_XHR_PollingTransport ( BaseMixin , TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( Test_XHR_PollingTransport , self ) . setUp ( ) <EOL> self . socketIO = SocketIO ( HOST , PORT , LoggingNamespace , transports = [ <EOL> '<STR_LIT>' ] , verify = False ) <EOL> self . assertEqual ( self . socketIO . transport_name , '<STR_LIT>' ) <EOL> class Test_WebsocketTransport ( BaseMixin , TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( Test_WebsocketTransport , self ) . setUp ( ) <EOL> self . socketIO = SocketIO ( HOST , PORT , LoggingNamespace , transports = [ <EOL> '<STR_LIT>' , '<STR_LIT>' ] , verify = False ) <EOL> self . assertEqual ( self . socketIO . transport_name , '<STR_LIT>' ) <EOL> class Namespace ( LoggingNamespace ) : <EOL> def initialize ( self ) : <EOL> self . called_on_disconnect = False <EOL> self . args_by_event = { } <EOL> self . response = None <EOL> def on_disconnect ( self ) : <EOL> self . called_on_disconnect = True <EOL> def on_wait_with_disconnect_response ( self ) : <EOL> self . disconnect ( ) <EOL> def on_event ( self , event , * args ) : <EOL> callback , args = find_callback ( args ) <EOL> if callback : <EOL> callback ( * args ) <EOL> self . args_by_event [ event ] = args <EOL> def on_message ( self , data ) : <EOL> self . response = data </s>
