<s> """<STR_LIT>""" <EOL> import os <EOL> import uuid <EOL> import zmq <EOL> from zmq . tests import BaseZMQTestCase <EOL> from zmq . eventloop . zmqstream import ZMQStream <EOL> from IPython . zmq import session as ss <EOL> class SessionTestCase ( BaseZMQTestCase ) : <EOL> def setUp ( self ) : <EOL> BaseZMQTestCase . setUp ( self ) <EOL> self . session = ss . Session ( ) <EOL> class TestSession ( SessionTestCase ) : <EOL> def test_msg ( self ) : <EOL> """<STR_LIT>""" <EOL> msg = self . session . msg ( '<STR_LIT>' ) <EOL> thekeys = set ( '<STR_LIT>' . split ( ) ) <EOL> s = set ( msg . keys ( ) ) <EOL> self . assertEquals ( s , thekeys ) <EOL> self . assertTrue ( isinstance ( msg [ '<STR_LIT:content>' ] , dict ) ) <EOL> self . assertTrue ( isinstance ( msg [ '<STR_LIT>' ] , dict ) ) <EOL> self . assertTrue ( isinstance ( msg [ '<STR_LIT>' ] , dict ) ) <EOL> self . assertEquals ( msg [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_args ( self ) : <EOL> """<STR_LIT>""" <EOL> s = self . session <EOL> self . assertTrue ( s . pack is ss . default_packer ) <EOL> self . assertTrue ( s . unpack is ss . default_unpacker ) <EOL> self . assertEquals ( s . username , os . environ . get ( '<STR_LIT>' , '<STR_LIT:username>' ) ) <EOL> s = ss . Session ( ) <EOL> self . assertEquals ( s . username , os . environ . get ( '<STR_LIT>' , '<STR_LIT:username>' ) ) <EOL> self . assertRaises ( TypeError , ss . Session , pack = '<STR_LIT>' ) <EOL> self . assertRaises ( TypeError , ss . Session , unpack = '<STR_LIT>' ) <EOL> u = str ( uuid . uuid4 ( ) ) <EOL> s = ss . Session ( username = '<STR_LIT>' , session = u ) <EOL> self . assertEquals ( s . session , u ) <EOL> self . assertEquals ( s . username , '<STR_LIT>' ) <EOL> def test_tracking ( self ) : <EOL> """<STR_LIT>""" <EOL> a , b = self . create_bound_pair ( zmq . PAIR , zmq . PAIR ) <EOL> s = self . session <EOL> stream = ZMQStream ( a ) <EOL> msg = s . send ( a , '<STR_LIT:hello>' , track = False ) <EOL> self . assertTrue ( msg [ '<STR_LIT>' ] is None ) <EOL> msg = s . send ( a , '<STR_LIT:hello>' , track = True ) <EOL> self . assertTrue ( isinstance ( msg [ '<STR_LIT>' ] , zmq . MessageTracker ) ) <EOL> M = zmq . Message ( b'<STR_LIT>' , track = True ) <EOL> msg = s . send ( a , '<STR_LIT:hello>' , buffers = [ M ] , track = True ) <EOL> t = msg [ '<STR_LIT>' ] <EOL> self . assertTrue ( isinstance ( t , zmq . MessageTracker ) ) <EOL> self . assertRaises ( zmq . NotDone , t . wait , <NUM_LIT> ) <EOL> del M <EOL> t . wait ( <NUM_LIT:1> ) <EOL> def test_unique_msg_ids ( self ) : <EOL> """<STR_LIT>""" <EOL> ids = set ( ) <EOL> for i in range ( <NUM_LIT:2> ** <NUM_LIT:12> ) : <EOL> h = self . session . msg_header ( '<STR_LIT:test>' ) <EOL> msg_id = h [ '<STR_LIT>' ] <EOL> self . assertTrue ( msg_id not in ids ) <EOL> ids . add ( msg_id ) <EOL> def test_feed_identities ( self ) : <EOL> """<STR_LIT>""" <EOL> theids = "<STR_LIT>" . split ( ) <EOL> content = dict ( code = '<STR_LIT>' , stuff = object ( ) ) <EOL> themsg = self . session . msg ( '<STR_LIT>' , content = content ) <EOL> pmsg = theids </s>
