<s> import time <EOL> import unittest <EOL> from unittest . mock import Mock <EOL> import stomp <EOL> from stomp import exception <EOL> from stomp . listener import TestListener <EOL> from stomp . test . testutils import * <EOL> class Test12Connect ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> conn = stomp . Connection12 ( get_default_host ( ) ) <EOL> listener = TestListener ( '<STR_LIT>' ) <EOL> conn . set_listener ( '<STR_LIT>' , listener ) <EOL> conn . start ( ) <EOL> conn . connect ( get_default_user ( ) , get_default_password ( ) , wait = True ) <EOL> self . conn = conn <EOL> self . listener = listener <EOL> self . timestamp = time . strftime ( '<STR_LIT>' ) <EOL> def tearDown ( self ) : <EOL> if self . conn : <EOL> self . conn . disconnect ( receipt = None ) <EOL> def test_send ( self ) : <EOL> queuename = '<STR_LIT>' % self . timestamp <EOL> self . conn . subscribe ( destination = queuename , id = <NUM_LIT:1> , ack = '<STR_LIT>' ) <EOL> self . conn . send ( body = '<STR_LIT>' , destination = queuename , receipt = '<STR_LIT>' ) <EOL> self . listener . wait_on_receipt ( ) <EOL> self . assertTrue ( self . listener . connections == <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertTrue ( self . listener . messages == <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertTrue ( self . listener . errors == <NUM_LIT:0> , '<STR_LIT>' ) <EOL> def test_clientack ( self ) : <EOL> queuename = '<STR_LIT>' % self . timestamp <EOL> self . conn . subscribe ( destination = queuename , id = <NUM_LIT:1> , ack = '<STR_LIT>' ) <EOL> self . conn . send ( body = '<STR_LIT>' , destination = queuename , receipt = '<STR_LIT>' ) <EOL> self . listener . wait_for_message ( ) <EOL> ( headers , _ ) = self . listener . get_latest_message ( ) <EOL> ack_id = headers [ '<STR_LIT>' ] <EOL> self . conn . ack ( ack_id ) <EOL> def test_clientnack ( self ) : <EOL> queuename = '<STR_LIT>' % self . timestamp <EOL> self . conn . subscribe ( destination = queuename , id = <NUM_LIT:1> , ack = '<STR_LIT>' ) <EOL> self . conn . send ( body = '<STR_LIT>' , destination = queuename , receipt = '<STR_LIT>' ) <EOL> self . listener . wait_for_message ( ) <EOL> ( headers , _ ) = self . listener . get_latest_message ( ) <EOL> ack_id = headers [ '<STR_LIT>' ] <EOL> self . conn . nack ( ack_id ) <EOL> def test_timeout ( self ) : <EOL> server = TestStompServer ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) <EOL> try : <EOL> server . start ( ) <EOL> server . add_frame ( '''<STR_LIT>''' ) <EOL> conn = stomp . Connection12 ( [ ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ] ) <EOL> listener = TestListener ( ) <EOL> conn . set_listener ( '<STR_LIT>' , listener ) <EOL> conn . start ( ) <EOL> try : <EOL> conn . connect ( wait = True ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except exception . ConnectFailedException : <EOL> pass <EOL> finally : <EOL> server . stop ( ) <EOL> def test_specialchars ( self ) : <EOL> queuename = '<STR_LIT>' % self . timestamp <EOL> self . conn . subscribe ( destination = queuename , id = <NUM_LIT:1> , ack = '<STR_LIT>' ) <EOL> hdrs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . conn . send ( body = '<STR_LIT>' , headers = hdrs , destination = queuename , receipt = '<STR_LIT>' ) <EOL> self . listener . wait_on_receipt ( ) <EOL> ( headers , _ ) = self . listener . get_latest_message ( ) <EOL> _ = headers [ '<STR_LIT>' ] <EOL> _ = headers [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT>' in headers ) <EOL> self . assertEqual ( '<STR_LIT>' , headers [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( '<STR_LIT>' in headers ) <EOL> self . assertEqual ( '<STR_LIT>' , headers [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( '<STR_LIT>' in headers ) <EOL> self . assertEqual ( '<STR_LIT>' , headers [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( '<STR_LIT>' in headers ) <EOL> self . assertEqual ( '<STR_LIT>' , headers [ '<STR_LIT>' ] ) <EOL> def test_suppress_content_length ( self ) : <EOL> queuename = '<STR_LIT>' % self . timestamp <EOL> self . conn = stomp . Connection12 ( get_default_host ( ) , auto_content_length = False ) <EOL> self . conn . transport = Mock ( ) <EOL> self . conn . send ( body = '<STR_LIT:test>' , destination = queuename , receipt = '<STR_LIT>' ) <EOL> args , kwargs = self . conn . transport . transmit . call_args <EOL> frame = args [ <NUM_LIT:0> ] <EOL> self . assertTrue ( '<STR_LIT>' not in frame . headers ) </s>
