<s> """<STR_LIT>""" <EOL> import sys <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> import unittest2 as unittest <EOL> else : <EOL> import unittest <EOL> import os <EOL> import tempfile <EOL> import shutil <EOL> from ioflo . aid . sixing import * <EOL> from ioflo . aid . consoling import getConsole <EOL> from ioflo . aio import wiring <EOL> from ioflo . aio . uxd import uxding <EOL> console = getConsole ( ) <EOL> def setUpModule ( ) : <EOL> console . reinit ( verbosity = console . Wordage . concise ) <EOL> def tearDownModule ( ) : <EOL> console . reinit ( verbosity = console . Wordage . concise ) <EOL> class BasicTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> pass <EOL> def tearDown ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> pass <EOL> def testSocketUxdNb ( self ) : <EOL> """<STR_LIT>""" <EOL> console . terse ( "<STR_LIT>" . format ( self . testSocketUxdNb . __doc__ ) ) <EOL> console . reinit ( verbosity = console . Wordage . profuse ) <EOL> userDirpath = os . path . join ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:test>' ) <EOL> userDirpath = os . path . abspath ( os . path . expanduser ( userDirpath ) ) <EOL> if not os . path . exists ( userDirpath ) : <EOL> os . makedirs ( userDirpath ) <EOL> tempDirpath = tempfile . mkdtemp ( prefix = "<STR_LIT:test>" , suffix = "<STR_LIT>" , dir = userDirpath ) <EOL> sockDirpath = os . path . join ( tempDirpath , '<STR_LIT>' ) <EOL> if not os . path . exists ( sockDirpath ) : <EOL> os . makedirs ( sockDirpath ) <EOL> logDirpath = os . path . join ( tempDirpath , '<STR_LIT>' ) <EOL> if not os . path . exists ( logDirpath ) : <EOL> os . makedirs ( logDirpath ) <EOL> wireLog = wiring . WireLog ( path = logDirpath ) <EOL> result = wireLog . reopen ( prefix = '<STR_LIT>' , midfix = '<STR_LIT>' ) <EOL> ha = os . path . join ( sockDirpath , '<STR_LIT>' ) <EOL> alpha = uxding . SocketUxdNb ( ha = ha , umask = <NUM_LIT> , wlog = wireLog ) <EOL> self . assertIs ( alpha . opened , False ) <EOL> result = alpha . reopen ( ) <EOL> self . assertIs ( result , True ) <EOL> self . assertIs ( alpha . opened , True ) <EOL> self . assertEqual ( alpha . ha , ha ) <EOL> ha = os . path . join ( sockDirpath , '<STR_LIT>' ) <EOL> beta = uxding . SocketUxdNb ( ha = ha , umask = <NUM_LIT> ) <EOL> result = beta . reopen ( ) <EOL> self . assertIs ( result , True ) <EOL> self . assertEqual ( beta . ha , ha ) <EOL> ha = os . path . join ( sockDirpath , '<STR_LIT>' ) <EOL> gamma = uxding . SocketUxdNb ( ha = ha , umask = <NUM_LIT> ) <EOL> result = gamma . reopen ( ) <EOL> self . assertIs ( result , True ) <EOL> self . assertEqual ( gamma . ha , ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> alpha . send ( txMsg , beta . ha ) <EOL> rxMsg , sa = beta . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , alpha . ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> alpha . send ( txMsg , gamma . ha ) <EOL> rxMsg , sa = gamma . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , alpha . ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> alpha . send ( txMsg , alpha . ha ) <EOL> rxMsg , sa = alpha . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , alpha . ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> beta . send ( txMsg , alpha . ha ) <EOL> rxMsg , sa = alpha . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , beta . ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> beta . send ( txMsg , gamma . ha ) <EOL> rxMsg , sa = gamma . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , beta . ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> beta . send ( txMsg , beta . ha ) <EOL> rxMsg , sa = beta . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , beta . ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> gamma . send ( txMsg , alpha . ha ) <EOL> rxMsg , sa = alpha . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , gamma . ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> gamma . send ( txMsg , beta . ha ) <EOL> rxMsg , sa = beta . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , gamma . ha ) <EOL> txMsg = b"<STR_LIT>" <EOL> gamma . send ( txMsg , gamma . ha ) <EOL> rxMsg , sa = gamma . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , gamma . ha ) <EOL> pairs = [ ( alpha , beta ) , ( alpha , gamma ) , ( alpha , alpha ) , <EOL> ( beta , alpha ) , ( beta , gamma ) , ( beta , beta ) , <EOL> ( gamma , alpha ) , ( gamma , beta ) , ( gamma , gamma ) , ] <EOL> names = [ ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ] <EOL> for i , pair in enumerate ( pairs ) : <EOL> txer , rxer = pair <EOL> txName , rxName = names [ i ] <EOL> txMsg = ns2b ( "<STR_LIT>" . format ( txName . capitalize ( ) , rxName . capitalize ( ) ) ) <EOL> txer . send ( txMsg , rxer . ha ) <EOL> rxMsg , sa = rxer . receive ( ) <EOL> self . assertEqual ( txMsg , rxMsg ) <EOL> self . assertEqual ( sa , txer . ha ) <EOL> rxMsg , sa = alpha . receive ( ) <EOL> self . assertIs ( b'<STR_LIT>' , rxMsg ) <EOL> self . assertIs ( None , sa ) <EOL> rxMsg , sa = beta . receive ( ) <EOL> self . assertIs ( b'<STR_LIT>' , rxMsg ) <EOL> self . assertIs ( None , sa ) <EOL> rxMsg , sa = gamma . receive ( ) <EOL> self . assertIs ( b'<STR_LIT>' , rxMsg ) <EOL> self . assertIs ( None , sa ) <EOL> alpha . close ( ) <EOL> beta . close ( ) <EOL> gamma . close ( ) <EOL> self . assertIs ( alpha . opened , False ) <EOL> wireLog . close ( ) <EOL> shutil . rmtree ( tempDirpath ) <EOL> console . reinit ( verbosity = console . Wordage . concise ) <EOL> def runOne ( test ) : <EOL> '''<STR_LIT>''' <EOL> test = BasicTestCase ( test ) <EOL> suite = unittest . TestSuite ( [ test ] ) <EOL> unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) . run ( suite ) <EOL> def runSome ( ) : <EOL> """<STR_LIT>""" <EOL> tests = [ ] <EOL> names = [ <EOL> '<STR_LIT>' , <EOL> ] <EOL> tests . extend ( map ( BasicTestCase , names ) ) <EOL> suite = unittest . TestSuite ( tests ) <EOL> unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) . run ( suite ) <EOL> def runAll ( ) : <EOL> """<STR_LIT>""" <EOL> suite = unittest . TestSuite ( ) <EOL> suite . addTest ( unittest . TestLoader ( ) . loadTestsFromTestCase ( BasicTestCase ) ) <EOL> unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) . run ( suite ) <EOL> if __name__ == '<STR_LIT:__main__>' and __package__ is None : <EOL> runSome ( ) </s>
