<s> import unittest <EOL> import zmqpy <EOL> import time <EOL> from zmqpy . utils . strtypes import asbytes <EOL> from . __init__ import BaseZMQTestCase <EOL> class TestContext ( BaseZMQTestCase ) : <EOL> def test_init ( self ) : <EOL> c1 = zmqpy . Context ( ) <EOL> self . assert_ ( isinstance ( c1 , zmqpy . Context ) ) <EOL> del c1 <EOL> c2 = zmqpy . Context ( ) <EOL> self . assert_ ( isinstance ( c2 , zmqpy . Context ) ) <EOL> del c2 <EOL> c3 = zmqpy . Context ( ) <EOL> self . assert_ ( isinstance ( c3 , zmqpy . Context ) ) <EOL> del c3 <EOL> def test_term ( self ) : <EOL> c = zmqpy . Context ( ) <EOL> c . term ( ) <EOL> self . assert_ ( c . closed ) <EOL> def test_fail_init ( self ) : <EOL> self . assertRaisesErrno ( zmqpy . EINVAL , zmqpy . Context , <NUM_LIT:0> ) <EOL> def test_instance_state ( self ) : <EOL> ctx = zmqpy . Context ( ) <EOL> c2 = zmqpy . Context ( iothreads = <NUM_LIT:2> ) <EOL> self . assertTrue ( ctx . iothreads ) <EOL> c2 . term ( ) <EOL> self . assertTrue ( ctx . closed ) <EOL> c3 = zmqpy . Context ( ) <EOL> c4 = zmqpy . Context ( ) <EOL> self . assertFalse ( c3 . closed ) <EOL> self . assertFalse ( c4 . closed ) <EOL> c4 . term ( ) <EOL> self . assertTrue ( ctx . closed ) <EOL> def test_term_hang ( self ) : <EOL> rep , req = self . create_bound_pair ( zmqpy . XREP , zmqpy . XREQ ) <EOL> req . setsockopt ( zmqpy . LINGER , <NUM_LIT:0> ) <EOL> req . send ( asbytes ( '<STR_LIT:hello>' ) , copy = False ) <EOL> req . close ( ) <EOL> rep . close ( ) <EOL> self . context . term ( ) <EOL> def test_many_sockets ( self ) : <EOL> """<STR_LIT>""" <EOL> ctx = self . context <EOL> for i in range ( <NUM_LIT:16> ) : <EOL> sockets = [ ctx . socket ( zmqpy . REP ) for i in range ( <NUM_LIT> ) ] <EOL> [ s . close ( ) for s in sockets ] <EOL> time . sleep ( <NUM_LIT> ) <EOL> ctx . term ( ) <EOL> for s in sockets : <EOL> self . assertTrue ( s . closed ) <EOL> def test_term_close ( self ) : <EOL> """<STR_LIT>""" <EOL> sockets = [ self . context . socket ( zmqpy . REP ) for i in range ( <NUM_LIT> ) ] <EOL> [ s . close ( ) for s in sockets [ : : <NUM_LIT:2> ] ] <EOL> self . context . term ( ) <EOL> for s in sockets : <EOL> self . assertTrue ( s . closed ) <EOL> def test_set_linger ( self ) : <EOL> ctx = zmqpy . Context ( ) <EOL> self . assertEquals ( ctx . linger , <NUM_LIT:1> ) <EOL> ctx . set_linger ( <NUM_LIT:10> ) <EOL> self . assertEquals ( ctx . linger , <NUM_LIT:10> ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
