<s> from __future__ import absolute_import <EOL> from tornado import web , testing <EOL> from tornado . ioloop import IOLoop <EOL> from pyswagger import SwaggerApp <EOL> from pyswagger . contrib . client . tornado import TornadoClient <EOL> from ... utils import create_pet_db , get_test_data_folder , pet_Mary <EOL> import json <EOL> import six <EOL> sapp = SwaggerApp . _create_ ( get_test_data_folder ( version = '<STR_LIT>' , which = '<STR_LIT>' ) ) <EOL> received_file = None <EOL> received_meta = None <EOL> """<STR_LIT>""" <EOL> class RESTHandler ( web . RequestHandler ) : <EOL> """<STR_LIT>""" <EOL> def initialize ( self , db ) : <EOL> self . db = db <EOL> def prepare ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( RESTHandler , self ) . prepare ( ) <EOL> content_type = self . request . headers . get ( '<STR_LIT:Content-Type>' ) <EOL> if content_type and content_type . startswith ( '<STR_LIT:application/json>' ) : <EOL> if content_type . rfind ( '<STR_LIT>' ) : <EOL> self . json_args = json . loads ( self . request . body . decode ( '<STR_LIT:utf-8>' ) ) <EOL> else : <EOL> raise web . HTTPError ( '<STR_LIT>' + content_type ) <EOL> class PetRequestHandler ( RESTHandler ) : <EOL> """<STR_LIT>""" <EOL> def put ( self ) : <EOL> pet = self . json_args <EOL> if not isinstance ( pet [ '<STR_LIT:id>' ] , int ) : <EOL> self . set_status ( <NUM_LIT> ) <EOL> if not self . db . update_ ( ** pet ) : <EOL> self . set_status ( <NUM_LIT> ) <EOL> else : <EOL> self . set_status ( <NUM_LIT:200> ) <EOL> self . finish ( ) <EOL> def post ( self ) : <EOL> pet = self . json_args <EOL> if self . db . read_ ( pet [ '<STR_LIT:id>' ] ) != None : <EOL> self . set_status ( <NUM_LIT> ) <EOL> else : <EOL> self . db . create_ ( ** pet ) <EOL> self . set_status ( <NUM_LIT:200> ) <EOL> self . finish ( ) <EOL> class PetIdRequestHandler ( RESTHandler ) : <EOL> """<STR_LIT>""" <EOL> def delete ( self , id ) : <EOL> if not self . db . delete_ ( int ( id ) ) : <EOL> self . set_status ( <NUM_LIT> ) <EOL> self . finish ( ) <EOL> def get ( self , id ) : <EOL> pet = self . db . read_ ( int ( id ) ) <EOL> if not pet : <EOL> self . set_status ( <NUM_LIT> ) <EOL> else : <EOL> self . write ( json . dumps ( pet ) ) <EOL> self . finish ( ) <EOL> class ImageRequestHandler ( web . RequestHandler ) : <EOL> """<STR_LIT>""" <EOL> def post ( self ) : <EOL> """<STR_LIT>""" <EOL> global received_file <EOL> global received_meta <EOL> received_file = self . request . files [ '<STR_LIT:file>' ] [ <NUM_LIT:0> ] . body <EOL> received_meta = self . get_argument ( '<STR_LIT>' ) <EOL> """<STR_LIT>""" <EOL> pet_db = create_pet_db ( ) <EOL> app = web . Application ( [ <EOL> ( r'<STR_LIT>' , PetRequestHandler , dict ( db = pet_db ) ) , <EOL> ( r'<STR_LIT>' , PetIdRequestHandler , dict ( db = pet_db ) ) , <EOL> ( r'<STR_LIT>' , ImageRequestHandler ) <EOL> ] , debug = True ) <EOL> class TornadoTestCase ( testing . AsyncHTTPTestCase ) : <EOL> """<STR_LIT:U+0020>""" <EOL> def setUp ( self ) : <EOL> global received_file <EOL> global received_meta <EOL> received_file = received_meta = None <EOL> super ( TornadoTestCase , self ) . setUp ( ) <EOL> self . client = TornadoClient ( ) <EOL> def get_new_ioloop ( self ) : <EOL> return IOLoop . instance ( ) <EOL> def get_app ( self ) : <EOL> global app <EOL> return app <EOL> @ testing . gen_test <EOL> def test_updatePet ( self ) : <EOL> """<STR_LIT>""" <EOL> global pet_db <EOL> resp = yield self . client . request ( <EOL> sapp . op [ '<STR_LIT>' ] ( body = dict ( id = <NUM_LIT:1> , name = '<STR_LIT>' ) ) , <EOL> opt = dict ( <EOL> url_netloc = '<STR_LIT>' + str ( self . get_http_port ( ) ) <EOL> ) ) <EOL> self . assertEqual ( resp . status , <NUM_LIT:200> ) <EOL> self . assertEqual ( pet_db . read_ ( <NUM_LIT:1> ) [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> @ testing . gen_test <EOL> def test_addPet ( self ) : <EOL> """<STR_LIT>""" <EOL> global pet_db <EOL> resp = yield self . client . request ( <EOL> sapp . op [ '<STR_LIT>' ] ( body = dict ( id = <NUM_LIT:5> , name = '<STR_LIT>' ) ) , <EOL> opt = dict ( <EOL> url_netloc = '<STR_LIT>' + str ( self . get_http_port ( ) ) <EOL> ) ) <EOL> self . assertEqual ( resp . status , <NUM_LIT:200> ) <EOL> self . assertEqual ( pet_db . read_ ( <NUM_LIT:5> ) [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> @ testing . gen_test <EOL> def test_deletePet ( self ) : <EOL> """<STR_LIT>""" <EOL> resp = yield self . client . request ( <EOL> sapp . op [ '<STR_LIT>' ] ( petId = <NUM_LIT:5> ) , <EOL> opt = dict ( <EOL> url_netloc = '<STR_LIT>' + str ( self . get_http_port ( ) ) <EOL> ) ) <EOL> self . assertEqual ( resp . status , <NUM_LIT:200> ) <EOL> self . assertEqual ( pet_db . read_ ( <NUM_LIT:5> ) , None ) <EOL> @ testing . gen_test <EOL> def test_getPetById ( self ) : <EOL> """<STR_LIT>""" <EOL> resp = yield self . client . request ( <EOL> sapp . op [ '<STR_LIT>' ] ( petId = <NUM_LIT:2> ) , <EOL> opt = dict ( <EOL> url_netloc = '<STR_LIT>' + str ( self . get_http_port ( ) ) <EOL> ) ) <EOL> self . assertEqual ( resp . status , <NUM_LIT:200> ) <EOL> self . assertEqual ( resp . data , pet_Mary ) <EOL> @ testing . gen_test <EOL> def test_uploadFile ( self ) : <EOL> """<STR_LIT>""" <EOL> global received_file <EOL> global received_meta <EOL> resp = yield self . client . request ( <EOL> sapp . op [ '<STR_LIT>' ] ( <EOL> additionalMetadata = '<STR_LIT>' , file = dict ( data = six . StringIO ( '<STR_LIT>' ) , filename = '<STR_LIT>' ) ) , <EOL> opt = dict ( <EOL> url_netloc = '<STR_LIT>' + str ( self . get_http_port ( ) ) <EOL> ) ) <EOL> self . assertEqual ( resp . status , <NUM_LIT:200> ) <EOL> self . assertEqual ( received_file . decode ( ) , '<STR_LIT>' ) <EOL> self . assertEqual ( received_meta , '<STR_LIT>' ) </s>
