<s> import time <EOL> import unittest <EOL> from nose . tools import raises <EOL> from ryu . lib import hub <EOL> hub . patch ( ) <EOL> class MyException ( BaseException ) : <EOL> pass <EOL> class Test_hub ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> pass <EOL> def tearDown ( self ) : <EOL> pass <EOL> @ raises ( hub . Timeout ) <EOL> def test_0_timeout1 ( self ) : <EOL> with hub . Timeout ( <NUM_LIT:0.1> ) : <EOL> hub . sleep ( <NUM_LIT:1> ) <EOL> @ raises ( MyException ) <EOL> def test_0_timeout2 ( self ) : <EOL> with hub . Timeout ( <NUM_LIT:0.1> , MyException ) : <EOL> hub . sleep ( <NUM_LIT:1> ) <EOL> def test_0_timeout3 ( self ) : <EOL> with hub . Timeout ( <NUM_LIT:1> ) : <EOL> hub . sleep ( <NUM_LIT:0.1> ) <EOL> hub . sleep ( <NUM_LIT:2> ) <EOL> def test_spawn_event1 ( self ) : <EOL> def _child ( ev , result ) : <EOL> hub . sleep ( <NUM_LIT:1> ) <EOL> result . append ( <NUM_LIT:1> ) <EOL> ev . set ( ) <EOL> ev = hub . Event ( ) <EOL> result = [ ] <EOL> with hub . Timeout ( <NUM_LIT:2> ) : <EOL> hub . spawn ( _child , ev , result ) <EOL> ev . wait ( ) <EOL> assert len ( result ) == <NUM_LIT:1> <EOL> def test_spawn_event2 ( self ) : <EOL> def _child ( ev , result ) : <EOL> hub . sleep ( <NUM_LIT:1> ) <EOL> result . append ( <NUM_LIT:1> ) <EOL> ev . set ( ) <EOL> ev = hub . Event ( ) <EOL> result = [ ] <EOL> with hub . Timeout ( <NUM_LIT:2> ) : <EOL> t = hub . spawn ( _child , ev , result ) <EOL> ev . wait ( timeout = <NUM_LIT:0.5> ) <EOL> assert len ( result ) == <NUM_LIT:0> <EOL> ev . wait ( ) <EOL> assert len ( result ) == <NUM_LIT:1> <EOL> def test_spawn_event3 ( self ) : <EOL> def _child ( ev , ev2 , result ) : <EOL> ev2 . wait ( ) <EOL> hub . sleep ( <NUM_LIT:0.5> ) <EOL> result . append ( <NUM_LIT:1> ) <EOL> ev . set ( ) <EOL> ev = hub . Event ( ) <EOL> ev2 = hub . Event ( ) <EOL> result = [ ] <EOL> with hub . Timeout ( <NUM_LIT:2> ) : <EOL> hub . spawn ( _child , ev , ev2 , result ) <EOL> hub . spawn ( _child , ev , ev2 , result ) <EOL> hub . sleep ( <NUM_LIT:0.5> ) <EOL> ev2 . set ( ) <EOL> ev . wait ( timeout = <NUM_LIT:1> ) <EOL> assert len ( result ) == <NUM_LIT:2> <EOL> def test_spawn_select1 ( self ) : <EOL> import select <EOL> import socket <EOL> def _child ( s1 ) : <EOL> hub . sleep ( <NUM_LIT:0.5> ) <EOL> s1 . send ( b"<STR_LIT>" ) <EOL> s1 , s2 = socket . socketpair ( ) <EOL> with hub . Timeout ( <NUM_LIT:1> ) : <EOL> hub . spawn ( _child , s1 ) <EOL> select . select ( [ s2 . fileno ( ) ] , [ ] , [ ] ) <EOL> select . select ( [ s2 . fileno ( ) ] , [ ] , [ ] ) <EOL> @ raises ( MyException ) <EOL> def test_select1 ( self ) : <EOL> import select <EOL> import socket <EOL> s1 , s2 = socket . socketpair ( ) <EOL> with hub . Timeout ( <NUM_LIT:1> , MyException ) : <EOL> select . select ( [ s2 . fileno ( ) ] , [ ] , [ ] ) <EOL> def test_select2 ( self ) : <EOL> import select <EOL> with hub . Timeout ( <NUM_LIT:1> , MyException ) : <EOL> select . select ( [ ] , [ ] , [ ] , <NUM_LIT:0> ) <EOL> def test_select3 ( self ) : <EOL> import select <EOL> import socket <EOL> s1 , s2 = socket . socketpair ( ) <EOL> with hub . Timeout ( <NUM_LIT:1> , MyException ) : <EOL> list = [ s1 . fileno ( ) , s2 . fileno ( ) ] <EOL> rlist , wlist , xlist = select . select ( list , list , list ) <EOL> assert not s1 . fileno ( ) in rlist <EOL> assert not s2 . fileno ( ) in rlist <EOL> assert ( s1 . fileno ( ) in wlist ) or ( s2 . fileno ( ) in wlist ) <EOL> assert not s1 . fileno ( ) in xlist <EOL> assert not s2 . fileno ( ) in xlist <EOL> def test_spawn_joinall ( self ) : <EOL> def _child ( ev2 , result ) : <EOL> ev2 . wait ( ) <EOL> hub . sleep ( <NUM_LIT:0.5> ) <EOL> result . append ( <NUM_LIT:1> ) <EOL> raise BaseException ( "<STR_LIT>" ) <EOL> ev2 = hub . Event ( ) <EOL> threads = [ ] <EOL> result = [ ] <EOL> with hub . Timeout ( <NUM_LIT:2> ) : <EOL> threads . append ( hub . spawn ( _child , ev2 , result ) ) <EOL> threads . append ( hub . spawn ( _child , ev2 , result ) ) <EOL> hub . sleep ( <NUM_LIT:0.5> ) <EOL> ev2 . set ( ) <EOL> hub . joinall ( threads ) <EOL> assert len ( result ) == <NUM_LIT:2> <EOL> def test_spawn_kill_joinall ( self ) : <EOL> def _child ( ev2 , result ) : <EOL> ev2 . wait ( ) <EOL> result . append ( <NUM_LIT:1> ) <EOL> ev2 = hub . Event ( ) <EOL> threads = [ ] <EOL> result = [ ] <EOL> with hub . Timeout ( <NUM_LIT:2> ) : <EOL> threads . append ( hub . spawn ( _child , ev2 , result ) ) <EOL> threads . append ( hub . spawn ( _child , ev2 , result ) ) <EOL> hub . sleep ( <NUM_LIT:0.5> ) <EOL> for t in threads : <EOL> hub . kill ( t ) <EOL> hub . joinall ( threads ) <EOL> assert len ( result ) == <NUM_LIT:0> <EOL> def test_spawn_kill_nowait_joinall ( self ) : <EOL> def _child ( result ) : <EOL> result . append ( <NUM_LIT:1> ) <EOL> threads = [ ] <EOL> result = [ ] <EOL> with hub . Timeout ( <NUM_LIT:2> ) : <EOL> threads . append ( hub . spawn ( _child , result ) ) <EOL> for t in threads : <EOL> hub . kill ( t ) <EOL> hub . joinall ( threads ) <EOL> assert len ( result ) == <NUM_LIT:0> <EOL> def test_spawn_kill_die_joinall ( self ) : <EOL> def _child ( result ) : <EOL> result . append ( <NUM_LIT:1> ) <EOL> threads = [ ] <EOL> result = [ ] <EOL> with hub . Timeout ( <NUM_LIT:2> ) : <EOL> threads . append ( hub . spawn ( _child , result ) ) <EOL> threads . append ( hub . spawn ( _child , result ) ) <EOL> hub . sleep ( <NUM_LIT:0.5> ) <EOL> for t in threads : <EOL> hub . kill ( t ) <EOL> hub . joinall ( threads ) <EOL> assert len ( result ) == <NUM_LIT:2> <EOL> def test_spawn_exception_joinall ( self ) : <EOL> def _child ( ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> threads = [ ] <EOL> with hub . Timeout ( <NUM_LIT:2> ) : <EOL> threads . append ( hub . spawn ( _child ) ) <EOL> threads . append ( hub . spawn ( _child ) ) <EOL> hub . sleep ( <NUM_LIT:0.5> ) <EOL> hub . joinall ( threads ) <EOL> def test_event1 ( self ) : <EOL> ev = hub . Event ( ) <EOL> ev . set ( ) <EOL> with hub . Timeout ( <NUM_LIT:1> ) : <EOL> ev . wait ( ) <EOL> def test_event2 ( self ) : <EOL> ev = hub . Event ( ) <EOL> ev . set ( ) <EOL> ev . set ( ) </s>
