<s> import gc <EOL> import os <EOL> import sys <EOL> import socket <EOL> import select <EOL> import unittest <EOL> import pyev <EOL> from whizzer . protocol import Protocol , ProtocolFactory <EOL> from whizzer . server import UnixServer , TcpServer <EOL> from mocks import * <EOL> from common import loop <EOL> fpath = os . path . dirname ( __file__ ) <EOL> class TestServerCreation ( unittest . TestCase ) : <EOL> def test_tcp_server ( self ) : <EOL> factory = ProtocolFactory ( ) <EOL> factory . protocol = Protocol <EOL> server = TcpServer ( loop , factory , "<STR_LIT>" , <NUM_LIT> ) <EOL> server = None <EOL> def test_unix_server ( self ) : <EOL> factory = ProtocolFactory ( ) <EOL> factory . protocol = Protocol <EOL> server = UnixServer ( loop , factory , "<STR_LIT>" ) <EOL> server = None <EOL> gc . collect ( ) <EOL> self . assertTrue ( not os . path . exists ( "<STR_LIT>" ) ) <EOL> class TestUnixServer ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . factory = MockFactory ( ) <EOL> self . factory . protocol = MockProtocol <EOL> self . path = fpath + "<STR_LIT>" <EOL> self . server = UnixServer ( loop , self . factory , self . path ) <EOL> def tearDown ( self ) : <EOL> self . server . shutdown ( ) <EOL> self . server = None <EOL> self . server = None <EOL> self . factory = None <EOL> gc . collect ( ) <EOL> def c_sock ( self ) : <EOL> csock = socket . socket ( socket . AF_UNIX , socket . SOCK_STREAM ) <EOL> csock . setblocking ( False ) <EOL> return csock <EOL> def c_connect ( self , csock ) : <EOL> csock . connect ( self . path ) <EOL> def c_isconnected ( self , csock , testmsg = "<STR_LIT>" ) : <EOL> ( rlist , wlist , xlist ) = select . select ( [ ] , [ csock ] , [ ] , <NUM_LIT:0.0> ) <EOL> if csock in wlist : <EOL> try : <EOL> csock . send ( testmsg ) <EOL> return True <EOL> except IOError as e : <EOL> return False <EOL> else : <EOL> return False <EOL> def test_start ( self ) : <EOL> self . server . start ( ) <EOL> csock = self . c_sock ( ) <EOL> self . c_connect ( csock ) <EOL> loop . start ( pyev . EVRUN_ONCE ) <EOL> self . assertTrue ( self . c_isconnected ( csock ) ) <EOL> def test_stop ( self ) : <EOL> self . server . start ( ) <EOL> csock = self . c_sock ( ) <EOL> self . c_connect ( csock ) <EOL> loop . start ( pyev . EVRUN_ONCE ) <EOL> self . assertTrue ( self . c_isconnected ( csock ) ) <EOL> self . server . stop ( ) <EOL> loop . start ( pyev . EVRUN_ONCE ) <EOL> self . assertTrue ( self . c_isconnected ( csock ) ) <EOL> csock . close ( ) <EOL> csock = None <EOL> loop . start ( pyev . EVRUN_NOWAIT ) <EOL> csock = self . c_sock ( ) <EOL> self . c_connect ( csock ) <EOL> self . assertTrue ( self . factory . builds == <NUM_LIT:1> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
