<s> import thread <EOL> from pypy . rlib . rsocket import * <EOL> from pypy . rlib . rpoll import * <EOL> from pypy . rpython . test . test_llinterp import interpret <EOL> def setup_module ( mod ) : <EOL> rsocket_startup ( ) <EOL> def test_simple ( ) : <EOL> serv = RSocket ( AF_INET , SOCK_STREAM ) <EOL> serv . bind ( INETAddress ( '<STR_LIT:127.0.0.1>' , INADDR_ANY ) ) <EOL> serv . listen ( <NUM_LIT:1> ) <EOL> servaddr = serv . getsockname ( ) <EOL> events = poll ( { serv . fileno ( ) : POLLIN } , timeout = <NUM_LIT:100> ) <EOL> assert len ( events ) == <NUM_LIT:0> <EOL> cli = RSocket ( AF_INET , SOCK_STREAM ) <EOL> cli . setblocking ( False ) <EOL> err = cli . connect_ex ( servaddr ) <EOL> assert err != <NUM_LIT:0> <EOL> events = poll ( { serv . fileno ( ) : POLLIN } , timeout = <NUM_LIT> ) <EOL> assert len ( events ) == <NUM_LIT:1> <EOL> assert events [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] == serv . fileno ( ) <EOL> assert events [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] & POLLIN <EOL> servconn , cliaddr = serv . accept ( ) <EOL> events = poll ( { serv . fileno ( ) : POLLIN , <EOL> cli . fileno ( ) : POLLOUT } , timeout = <NUM_LIT> ) <EOL> assert len ( events ) == <NUM_LIT:1> <EOL> assert events [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] == cli . fileno ( ) <EOL> assert events [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] & POLLOUT <EOL> err = cli . connect_ex ( servaddr ) <EOL> assert err == <NUM_LIT:0> or err == <NUM_LIT> <EOL> events = poll ( { servconn . fileno ( ) : POLLIN , <EOL> cli . fileno ( ) : POLLIN } , timeout = <NUM_LIT:100> ) <EOL> assert len ( events ) == <NUM_LIT:0> <EOL> events = poll ( { servconn . fileno ( ) : POLLOUT , <EOL> cli . fileno ( ) : POLLOUT } , timeout = <NUM_LIT:100> ) <EOL> assert len ( events ) >= <NUM_LIT:1> <EOL> cli . close ( ) <EOL> servconn . close ( ) <EOL> serv . close ( ) <EOL> def test_select ( ) : <EOL> def f ( ) : <EOL> readend , writeend = os . pipe ( ) <EOL> try : <EOL> iwtd , owtd , ewtd = select ( [ readend ] , [ ] , [ ] , <NUM_LIT:0.0> ) <EOL> assert iwtd == owtd == ewtd == [ ] <EOL> os . write ( writeend , '<STR_LIT:X>' ) <EOL> iwtd , owtd , ewtd = select ( [ readend ] , [ ] , [ ] ) <EOL> assert iwtd == [ readend ] <EOL> assert owtd == ewtd == [ ] <EOL> finally : <EOL> os . close ( readend ) <EOL> os . close ( writeend ) <EOL> f ( ) <EOL> interpret ( f , [ ] ) <EOL> def test_select_timeout ( ) : <EOL> from time import time <EOL> def f ( ) : <EOL> a = time ( ) <EOL> iwtd , owtd , ewtd = select ( [ ] , [ ] , [ ] , <NUM_LIT> ) <EOL> diff = time ( ) - a <EOL> assert <NUM_LIT> < diff < <NUM_LIT> <EOL> interpret ( f , [ ] ) <EOL> def test_translate ( ) : <EOL> from pypy . translator . c . test . test_genc import compile <EOL> def func ( ) : <EOL> poll ( { } ) <EOL> compile ( func , [ ] ) </s>
