<s> from __future__ import absolute_import <EOL> import os <EOL> import multiprocessing <EOL> import socket <EOL> import time <EOL> import pytest <EOL> import thriftpy <EOL> thriftpy . install_import_hook ( ) <EOL> from thriftpy . rpc import make_server , client_context <EOL> addressbook = thriftpy . load ( os . path . join ( os . path . dirname ( __file__ ) , <EOL> "<STR_LIT>" ) ) <EOL> unix_sock = "<STR_LIT>" <EOL> class Dispatcher ( object ) : <EOL> def __init__ ( self ) : <EOL> self . ab = addressbook . AddressBook ( ) <EOL> self . ab . people = { } <EOL> def ping ( self ) : <EOL> return True <EOL> def hello ( self , name ) : <EOL> return "<STR_LIT>" + name <EOL> def add ( self , person ) : <EOL> self . ab . people [ person . name ] = person <EOL> return True <EOL> def remove ( self , name ) : <EOL> try : <EOL> self . ab . people . pop ( name ) <EOL> return True <EOL> except KeyError : <EOL> raise addressbook . PersonNotExistsError ( <EOL> "<STR_LIT>" . format ( name ) ) <EOL> def get ( self , name ) : <EOL> try : <EOL> return self . ab . people [ name ] <EOL> except KeyError : <EOL> raise addressbook . PersonNotExistsError ( <EOL> "<STR_LIT>" . format ( name ) ) <EOL> def book ( self ) : <EOL> return self . ab <EOL> def get_phonenumbers ( self , name , count ) : <EOL> p = [ self . ab . people [ name ] . phones [ <NUM_LIT:0> ] ] if name in self . ab . people else [ ] <EOL> return p * count <EOL> def get_phones ( self , name ) : <EOL> phone_numbers = self . ab . people [ name ] . phones <EOL> return dict ( ( p . type , p . number ) for p in phone_numbers ) <EOL> def sleep ( self , ms ) : <EOL> time . sleep ( ms / <NUM_LIT> ) <EOL> return True <EOL> @ pytest . fixture ( scope = "<STR_LIT>" ) <EOL> def server ( request ) : <EOL> server = make_server ( addressbook . AddressBookService , Dispatcher ( ) , <EOL> unix_socket = unix_sock ) <EOL> ps = multiprocessing . Process ( target = server . serve ) <EOL> ps . start ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> def fin ( ) : <EOL> if ps . is_alive ( ) : <EOL> ps . terminate ( ) <EOL> try : <EOL> os . remove ( unix_sock ) <EOL> except IOError : <EOL> pass <EOL> request . addfinalizer ( fin ) <EOL> @ pytest . fixture ( scope = "<STR_LIT>" ) <EOL> def person ( ) : <EOL> phone1 = addressbook . PhoneNumber ( ) <EOL> phone1 . type = addressbook . PhoneType . MOBILE <EOL> phone1 . number = '<STR_LIT>' <EOL> phone2 = addressbook . PhoneNumber ( ) <EOL> phone2 . type = addressbook . PhoneType . HOME <EOL> phone2 . number = '<STR_LIT>' <EOL> phone3 = addressbook . PhoneNumber ( ) <EOL> alice = addressbook . Person ( ) <EOL> alice . name = "<STR_LIT>" <EOL> alice . phones = [ phone1 , phone2 , phone3 ] <EOL> alice . created_at = int ( time . time ( ) ) <EOL> return alice <EOL> def client ( timeout = <NUM_LIT> ) : <EOL> return client_context ( addressbook . AddressBookService , <EOL> unix_socket = unix_sock , timeout = timeout ) <EOL> def test_void_api ( server ) : <EOL> with client ( ) as c : <EOL> assert c . ping ( ) is None <EOL> def test_string_api ( server ) : <EOL> with client ( ) as c : <EOL> assert c . hello ( "<STR_LIT>" ) == "<STR_LIT>" <EOL> def test_huge_res ( server ) : <EOL> with client ( ) as c : <EOL> big_str = "<STR_LIT>" * <NUM_LIT> <EOL> assert c . hello ( big_str ) == "<STR_LIT>" + big_str <EOL> def test_tstruct_req ( person ) : <EOL> with client ( ) as c : <EOL> assert c . add ( person ) is True <EOL> def test_tstruct_res ( person ) : <EOL> with client ( ) as c : <EOL> assert person == c . get ( "<STR_LIT>" ) <EOL> def test_complex_tstruct ( ) : <EOL> with client ( ) as c : <EOL> assert len ( c . get_phonenumbers ( "<STR_LIT>" , <NUM_LIT:0> ) ) == <NUM_LIT:0> <EOL> assert len ( c . get_phonenumbers ( "<STR_LIT>" , <NUM_LIT:1000> ) ) == <NUM_LIT:1000> <EOL> def test_exception ( ) : <EOL> with pytest . raises ( addressbook . PersonNotExistsError ) : <EOL> with client ( ) as c : <EOL> c . remove ( "<STR_LIT>" ) <EOL> def test_client_timeout ( ) : <EOL> with pytest . raises ( socket . timeout ) : <EOL> with client ( timeout = <NUM_LIT> ) as c : <EOL> c . sleep ( <NUM_LIT:1000> ) </s>
