<s> from tests import EventSourceTestCase <EOL> from tornado_eventsource . event_source_client import eventsource_connect <EOL> from tornado . iostream import StreamClosedError <EOL> from mock import patch <EOL> class PostMessageTest ( EventSourceTestCase ) : <EOL> def test_returns_405_if_post ( self ) : <EOL> response = self . fetch ( '<STR_LIT:/>' , method = '<STR_LIT:POST>' , body = '<STR_LIT:a>' ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> def test_returns_405_if_put ( self ) : <EOL> response = self . fetch ( '<STR_LIT:/>' , method = '<STR_LIT>' , body = '<STR_LIT:a>' ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_stream_closed_error ( self , write_mock ) : <EOL> write_mock . side_effect = StreamClosedError <EOL> event_source = eventsource_connect ( url = self . get_url ( '<STR_LIT:/>' ) , callback = self . stop ) <EOL> self . wait ( ) <EOL> self . assertRaises ( Exception , event_source . result ) <EOL> def test_check_connection_error ( self ) : <EOL> response = self . fetch ( '<STR_LIT>' , method = '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . code , <NUM_LIT> ) <EOL> def test_should_set_custom_headers ( self ) : <EOL> event_source = eventsource_connect ( url = self . get_url ( '<STR_LIT:/>' ) , callback = self . stop ) <EOL> self . wait ( ) <EOL> headers = event_source . result ( ) . headers <EOL> self . assertEqual ( headers [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ '<STR_LIT>' ] , '<STR_LIT:*>' ) <EOL> self . assertEqual ( headers [ '<STR_LIT:Content-Type>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( headers [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_get_message_on_open ( self ) : <EOL> event_source = eventsource_connect ( url = self . get_url ( '<STR_LIT:/>' ) , callback = self . stop ) <EOL> self . wait ( ) <EOL> event = event_source . result ( ) . events [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . name , '<STR_LIT>' ) <EOL> self . assertEqual ( event . data , '<STR_LIT>' ) <EOL> def test_get_message_with_retry_on_open ( self ) : <EOL> event_source = eventsource_connect ( url = self . get_url ( '<STR_LIT>' ) , callback = self . stop ) <EOL> self . wait ( ) <EOL> event = event_source . result ( ) . events [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . name , '<STR_LIT>' ) <EOL> self . assertEqual ( event . data , '<STR_LIT>' ) <EOL> self . assertEqual ( event . retry , <NUM_LIT> ) <EOL> def test_get_message_with_url_params ( self ) : <EOL> event_source = eventsource_connect ( url = self . get_url ( '<STR_LIT>' ) , callback = self . stop ) <EOL> self . wait ( ) <EOL> event = event_source . result ( ) . events [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . name , '<STR_LIT>' ) <EOL> self . assertEqual ( event . data , '<STR_LIT>' ) <EOL> def test_get_message_without_name ( self ) : <EOL> event_source = eventsource_connect ( url = self . get_url ( '<STR_LIT>' ) , callback = self . stop ) <EOL> self . wait ( ) <EOL> event = event_source . result ( ) . events [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . name , None ) <EOL> self . assertEqual ( event . data , '<STR_LIT>' ) <EOL> def test_get_message_with_multiline ( self ) : <EOL> event_source = eventsource_connect ( url = self . get_url ( '<STR_LIT>' ) , callback = self . stop ) <EOL> self . wait ( ) <EOL> event = event_source . result ( ) . events [ <NUM_LIT:0> ] <EOL> self . assertEqual ( event . name , None ) <EOL> self . assertEqual ( event . data , '<STR_LIT>' ) </s>
