<s> """<STR_LIT>""" <EOL> import StringIO <EOL> import unittest <EOL> import urllib <EOL> from google . appengine . api . channel import channel_service_stub <EOL> from google . appengine . tools . devappserver2 import channel <EOL> from google . appengine . tools . devappserver2 import wsgi_test_utils <EOL> class MockChannelServiceStub ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _messages = { } <EOL> self . _connected_channel_tokens = [ ] <EOL> self . _disconnected_channel_client_ids = [ ] <EOL> def set_messages ( self , messages ) : <EOL> self . _messages = messages <EOL> def has_channel_messages ( self , token ) : <EOL> return ( self . _messages . has_key ( token ) and <EOL> self . _messages [ token ] ) <EOL> def get_channel_messages ( self , token ) : <EOL> return self . _messages [ token ] <EOL> def pop_first_message ( self , token ) : <EOL> if self . has_channel_messages ( token ) : <EOL> return self . _messages [ token ] . pop ( <NUM_LIT:0> ) <EOL> else : <EOL> return None <EOL> def clear_channel_messages ( self , token ) : <EOL> del self . _messages [ token ] <EOL> def connect_channel ( self , token ) : <EOL> self . validate_token_and_extract_client_id ( token ) <EOL> self . _connected_channel_tokens . append ( token ) <EOL> def disconnect_channel ( self , client_id ) : <EOL> self . _disconnected_channel_client_ids . append ( client_id ) <EOL> def get_connected_tokens ( self ) : <EOL> return self . _connected_channel_tokens <EOL> def set_connected_tokens ( self , tokens ) : <EOL> self . _connected_tokens = tokens <EOL> def validate_token_and_extract_client_id ( self , token ) : <EOL> if token == '<STR_LIT>' : <EOL> raise channel_service_stub . InvalidTokenError ( ) <EOL> if token == '<STR_LIT>' : <EOL> raise channel_service_stub . TokenTimedOutError ( ) <EOL> return '<STR_LIT>' <EOL> def connect_and_pop_first_message ( self , token ) : <EOL> self . connect_channel ( token ) <EOL> return self . pop_first_message ( token ) <EOL> def _build_environ ( path , query = None ) : <EOL> environ = { <EOL> '<STR_LIT>' : '<STR_LIT:GET>' , <EOL> '<STR_LIT>' : path , <EOL> } <EOL> if query : <EOL> environ [ '<STR_LIT>' ] = urllib . urlencode ( query ) <EOL> return environ <EOL> def _build_poll_environ ( token ) : <EOL> """<STR_LIT>""" <EOL> return _build_environ ( '<STR_LIT>' , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : token } ) <EOL> class DevAppserverChannelTest ( wsgi_test_utils . WSGITestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mock_channel_service_stub = MockChannelServiceStub ( ) <EOL> self . _old_get_channel_stub = channel . _get_channel_stub <EOL> channel . _get_channel_stub = lambda : self . _mock_channel_service_stub <EOL> self . _channel_app = channel . application <EOL> self . _output = StringIO . StringIO ( ) <EOL> def tearDown ( self ) : <EOL> channel . _get_channel_stub = self . _old_get_channel_stub <EOL> def test_channel_request_no_messages ( self ) : <EOL> """<STR_LIT>""" <EOL> environ = _build_poll_environ ( '<STR_LIT:id>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , channel . application , <EOL> environ ) <EOL> def test_channel_request_missing_parameters ( self ) : <EOL> """<STR_LIT>""" <EOL> environ = _build_environ ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , <EOL> channel . application , environ ) <EOL> environ = _build_environ ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT:id>' } ) <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , <EOL> channel . application , environ ) <EOL> def test_channel_request_bad_command ( self ) : <EOL> """<STR_LIT>""" <EOL> environ = _build_environ ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:id>' } ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , <EOL> channel . application , environ ) <EOL> def test_channel_request_bad_token ( self ) : <EOL> """<STR_LIT>""" <EOL> environ = _build_poll_environ ( '<STR_LIT>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , <EOL> channel . application , environ ) <EOL> def test_channel_request_expired_token ( self ) : <EOL> """<STR_LIT>""" <EOL> environ = _build_poll_environ ( '<STR_LIT>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , <EOL> channel . application , environ ) <EOL> def test_channel_request_other_channel_has_messages ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mock_channel_service_stub . set_messages ( { '<STR_LIT:a>' : [ '<STR_LIT:hello>' ] } ) <EOL> environ = _build_poll_environ ( '<STR_LIT:b>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , channel . application , <EOL> environ ) <EOL> def test_channel_request_with_messages ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mock_channel_service_stub . set_messages ( { '<STR_LIT:a>' : [ '<STR_LIT:hello>' ] } ) <EOL> environ = _build_poll_environ ( '<STR_LIT:a>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:5>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT:hello>' , <EOL> channel . application , environ ) <EOL> def test_channel_request_empty_message ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mock_channel_service_stub . set_messages ( { '<STR_LIT:a>' : [ '<STR_LIT>' ] } ) <EOL> environ = _build_poll_environ ( '<STR_LIT:a>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , channel . application , <EOL> environ ) <EOL> def test_channel_request_with_multiple_messages ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mock_channel_service_stub . set_messages ( { '<STR_LIT:a>' : [ '<STR_LIT:hello>' , '<STR_LIT>' ] } ) <EOL> environ = _build_poll_environ ( '<STR_LIT:a>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:5>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT:hello>' , <EOL> channel . application , environ ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , <EOL> channel . application , environ ) <EOL> def test_channel_request_clears_messages ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mock_channel_service_stub . set_messages ( { '<STR_LIT:a>' : [ '<STR_LIT:hello>' ] } ) <EOL> environ = _build_poll_environ ( '<STR_LIT:a>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:5>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT:hello>' , <EOL> channel . application , environ ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , channel . application , <EOL> environ ) <EOL> def test_channel_request_clears_correct_messages ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _mock_channel_service_stub . set_messages ( { '<STR_LIT:a>' : [ '<STR_LIT:hello>' ] , <EOL> '<STR_LIT:b>' : [ '<STR_LIT>' ] } ) <EOL> environ = _build_poll_environ ( '<STR_LIT:a>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:5>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT:hello>' , <EOL> channel . application , environ ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , channel . application , <EOL> environ ) <EOL> environ = _build_poll_environ ( '<STR_LIT:b>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , <EOL> channel . application , environ ) <EOL> def test_channel_request_connects_channel ( self ) : <EOL> """<STR_LIT>""" <EOL> mock_stub = self . _mock_channel_service_stub <EOL> mock_stub . set_connected_tokens ( [ ] ) <EOL> environ = _build_poll_environ ( '<STR_LIT:id>' ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , '<STR_LIT>' , channel . application , <EOL> environ ) <EOL> self . assertListEqual ( [ '<STR_LIT:id>' ] , mock_stub . get_connected_tokens ( ) ) <EOL> def test_channel_request_jsapi ( self ) : <EOL> """<STR_LIT>""" <EOL> environ = _build_environ ( '<STR_LIT>' ) <EOL> js_text = open ( channel . _JSAPI_PATH ) . read ( ) <EOL> expected_headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( len ( js_text ) ) , <EOL> '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> } <EOL> self . assertResponse ( '<STR_LIT>' , expected_headers , js_text , <EOL> channel . application , environ ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
