<s> import pytest <EOL> import socket <EOL> import select <EOL> from circuits import Manager <EOL> from circuits . net . events import close , write <EOL> from circuits . core . pollers import Select , Poll , EPoll , KQueue <EOL> from circuits . net . sockets import UDPServer , UDPClient , UDP6Server , UDP6Client <EOL> from . client import Client <EOL> from . server import Server <EOL> def wait_host ( server ) : <EOL> def checker ( obj , attr ) : <EOL> return all ( ( getattr ( obj , a ) for a in attr ) ) <EOL> assert pytest . wait_for ( server , ( "<STR_LIT:host>" , "<STR_LIT:port>" ) , checker ) <EOL> def _pytest_generate_tests ( metafunc , ipv6 ) : <EOL> metafunc . addcall ( funcargs = { "<STR_LIT>" : Select , "<STR_LIT>" : ipv6 } ) <EOL> if hasattr ( select , "<STR_LIT>" ) : <EOL> metafunc . addcall ( funcargs = { "<STR_LIT>" : Poll , "<STR_LIT>" : ipv6 } ) <EOL> if hasattr ( select , "<STR_LIT>" ) : <EOL> metafunc . addcall ( funcargs = { "<STR_LIT>" : EPoll , "<STR_LIT>" : ipv6 } ) <EOL> if hasattr ( select , "<STR_LIT>" ) : <EOL> metafunc . addcall ( funcargs = { "<STR_LIT>" : KQueue , "<STR_LIT>" : ipv6 } ) <EOL> def pytest_generate_tests ( metafunc ) : <EOL> _pytest_generate_tests ( metafunc , ipv6 = False ) <EOL> if socket . has_ipv6 : <EOL> _pytest_generate_tests ( metafunc , ipv6 = True ) <EOL> def test_basic ( Poller , ipv6 ) : <EOL> m = Manager ( ) + Poller ( ) <EOL> if ipv6 : <EOL> udp_server = UDP6Server ( ( "<STR_LIT>" , <NUM_LIT:0> ) ) <EOL> udp_client = UDP6Client ( ( "<STR_LIT>" , <NUM_LIT:0> ) , channel = "<STR_LIT>" ) <EOL> else : <EOL> udp_server = UDPServer ( <NUM_LIT:0> ) <EOL> udp_client = UDPClient ( <NUM_LIT:0> , channel = "<STR_LIT>" ) <EOL> server = Server ( ) + udp_server <EOL> client = Client ( ) + udp_client <EOL> server . register ( m ) <EOL> client . register ( m ) <EOL> m . start ( ) <EOL> try : <EOL> assert pytest . wait_for ( server , "<STR_LIT>" ) <EOL> assert pytest . wait_for ( client , "<STR_LIT>" ) <EOL> wait_host ( server ) <EOL> client . fire ( write ( ( server . host , server . port ) , b"<STR_LIT:foo>" ) ) <EOL> assert pytest . wait_for ( server , "<STR_LIT:data>" , b"<STR_LIT:foo>" ) <EOL> client . fire ( close ( ) ) <EOL> assert pytest . wait_for ( client , "<STR_LIT>" ) <EOL> server . fire ( close ( ) ) <EOL> assert pytest . wait_for ( server , "<STR_LIT>" ) <EOL> finally : <EOL> m . stop ( ) <EOL> def test_close ( Poller , ipv6 ) : <EOL> m = Manager ( ) + Poller ( ) <EOL> server = Server ( ) + UDPServer ( <NUM_LIT:0> ) <EOL> server . register ( m ) <EOL> m . start ( ) <EOL> try : <EOL> assert pytest . wait_for ( server , "<STR_LIT>" ) <EOL> wait_host ( server ) <EOL> host , port = server . host , server . port <EOL> server . fire ( close ( ) ) <EOL> assert pytest . wait_for ( server , "<STR_LIT>" ) <EOL> server . unregister ( ) <EOL> def test ( obj , attr ) : <EOL> return attr not in obj . components <EOL> assert pytest . wait_for ( m , server , value = test ) <EOL> server = Server ( ) + UDPServer ( ( host , port ) ) <EOL> server . register ( m ) <EOL> assert pytest . wait_for ( server , "<STR_LIT>" , timeout = <NUM_LIT> ) <EOL> finally : <EOL> m . stop ( ) </s>
