<s> import mock <EOL> import zmq <EOL> import zmq . utils . jsonapi as json <EOL> from circus . tests . support import TestCase , EasyTestSuite <EOL> from circus . stats . publisher import StatsPublisher <EOL> class TestStatsPublisher ( TestCase ) : <EOL> def test_publish ( self ) : <EOL> publisher = StatsPublisher ( ) <EOL> publisher . socket . close ( ) <EOL> publisher . socket = mock . MagicMock ( ) <EOL> stat = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:foo>' : '<STR_LIT:bar>' } <EOL> publisher . publish ( '<STR_LIT>' , stat ) <EOL> publisher . socket . send_multipart . assert_called_with ( <EOL> [ b'<STR_LIT>' , json . dumps ( stat ) ] ) <EOL> def test_publish_reraise_zmq_errors ( self ) : <EOL> publisher = StatsPublisher ( ) <EOL> publisher . socket = mock . MagicMock ( ) <EOL> publisher . socket . closed = False <EOL> publisher . socket . send_multipart . side_effect = zmq . ZMQError ( ) <EOL> stat = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:foo>' : '<STR_LIT:bar>' } <EOL> self . assertRaises ( zmq . ZMQError , publisher . publish , '<STR_LIT>' , stat ) <EOL> def test_publish_silent_zmq_errors_when_socket_closed ( self ) : <EOL> publisher = StatsPublisher ( ) <EOL> publisher . socket = mock . MagicMock ( ) <EOL> publisher . socket . closed = True <EOL> publisher . socket . send_multipart . side_effect = zmq . ZMQError ( ) <EOL> stat = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:foo>' : '<STR_LIT:bar>' } <EOL> publisher . publish ( '<STR_LIT>' , stat ) <EOL> test_suite = EasyTestSuite ( __name__ ) </s>
