<s> import copy <EOL> import mock <EOL> import unittest <EOL> from contextlib import contextmanager <EOL> from swift . common . swob import Request , HTTPOk , HTTPCreated , HTTPAccepted , HTTPBadRequest , HTTPNotFound <EOL> from storlet_middleware import storlet_handler <EOL> from storlet_middleware . storlet_handler import StorletProxyHandler , StorletObjectHandler , BaseStorletHandler <EOL> from tests . unit . swift import FakeLogger <EOL> from tests . unit . swift . storlet_middleware import FakeApp <EOL> DEFAULT_CONFIG = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> class TestStorletMiddleware ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . got_statuses = [ ] <EOL> self . got_headers = [ ] <EOL> self . conf = copy . copy ( DEFAULT_CONFIG ) <EOL> self . app = FakeApp ( ) <EOL> def tearDown ( self ) : <EOL> pass <EOL> def get_app ( self , app , global_conf , ** local_conf ) : <EOL> with mock . patch ( '<STR_LIT>' ) as get_fake_logger : <EOL> get_fake_logger . return_value = FakeLogger ( ) <EOL> factory = storlet_handler . filter_factory ( global_conf , ** local_conf ) <EOL> return factory ( app ) <EOL> def start_response ( self , status , headers ) : <EOL> self . got_statuses . append ( status ) <EOL> self . got_headers . append ( headers ) <EOL> def test_load_app ( self ) : <EOL> try : <EOL> self . get_app ( self . app , self . conf ) <EOL> except Exception : <EOL> self . fail ( '<STR_LIT>' ) <EOL> @ contextmanager <EOL> def fake_acc_info ( acc_info ) : <EOL> with mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) as ai : <EOL> ai . return_value = acc_info <EOL> yield <EOL> @ contextmanager <EOL> def storlet_enabled ( ) : <EOL> acc_info = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT:true>' } } <EOL> with fake_acc_info ( acc_info ) : <EOL> yield <EOL> @ contextmanager <EOL> def authorize_storlet_execution ( ) : <EOL> with mock . patch ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) as ase : <EOL> yield ase <EOL> class TestStorletMiddlewareProxy ( TestStorletMiddleware ) : <EOL> def setUp ( self ) : <EOL> super ( TestStorletMiddlewareProxy , self ) . setUp ( ) <EOL> self . conf [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> def test_GET_without_storlets ( self ) : <EOL> def basic_get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } ) <EOL> self . app . register ( '<STR_LIT:GET>' , path , HTTPOk , body = '<STR_LIT>' ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> self . app . reset_all ( ) <EOL> for target in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> path = '<STR_LIT:/>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , target ] ) <EOL> basic_get ( path ) <EOL> def test_GET_with_storlets ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( target , <EOL> self . app . get_calls ( ) [ - <NUM_LIT:1> ] [ <NUM_LIT:1> ] ) <EOL> self . assertIn ( '<STR_LIT>' , <EOL> self . app . get_calls ( ) [ - <NUM_LIT:1> ] [ <NUM_LIT:2> ] ) <EOL> acc_info = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT:true>' } } <EOL> with fake_acc_info ( acc_info ) : <EOL> get ( target ) <EOL> def test_GET_with_storlets_disabled_account ( self ) : <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> acc_info = { '<STR_LIT>' : { } } <EOL> with fake_acc_info ( acc_info ) : <EOL> get ( '<STR_LIT>' ) <EOL> def test_GET_with_storlets_object_404 ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPNotFound ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> with storlet_enabled ( ) : <EOL> get ( target ) <EOL> def test_GET_with_storlets_and_http_range ( self ) : <EOL> target = '<STR_LIT>' <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> with storlet_enabled ( ) : <EOL> get ( target ) <EOL> def test_GET_with_storlets_and_storlet_range ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req_range = '<STR_LIT>' <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : req_range } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp . read ( ) , '<STR_LIT>' ) <EOL> resp_headers = dict ( self . got_headers [ - <NUM_LIT:1> ] ) <EOL> self . assertFalse ( '<STR_LIT>' in resp_headers ) <EOL> self . assertEqual ( resp_headers [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> raw_req = self . app . get_calls ( '<STR_LIT:GET>' , path ) [ <NUM_LIT:0> ] <EOL> for key in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> self . assertEqual ( raw_req [ <NUM_LIT:2> ] [ key ] , req_range ) <EOL> with storlet_enabled ( ) : <EOL> get ( target ) <EOL> def test_GET_slo_without_storlets ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:True>' } , <EOL> body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> get ( target ) <EOL> def test_GET_slo_with_storlets ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:True>' } , <EOL> body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( resp . read ( ) , '<STR_LIT>' ) <EOL> with storlet_enabled ( ) : <EOL> get ( target ) <EOL> def test_PUT_without_storlets ( self ) : <EOL> def basic_put ( path ) : <EOL> self . app . register ( '<STR_LIT>' , path , HTTPCreated ) <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . app . reset_all ( ) <EOL> with authorize_storlet_execution ( ) as ase : <EOL> for target in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> path = '<STR_LIT:/>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , target ] ) <EOL> basic_put ( path ) <EOL> self . assertEqual ( <NUM_LIT:0> , ase . call_count ) <EOL> def test_PUT_with_storlets ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT>' , target , HTTPCreated ) <EOL> def put ( path ) : <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> body = '<STR_LIT>' ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> put_calls = self . app . get_calls ( '<STR_LIT>' , target ) <EOL> self . assertEqual ( len ( put_calls ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( put_calls [ - <NUM_LIT:1> ] [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> with storlet_enabled ( ) : <EOL> put ( target ) <EOL> def test_PUT_copy_without_storlets ( self ) : <EOL> target = '<STR_LIT>' <EOL> copy_from = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT>' , target , HTTPCreated ) <EOL> def copy ( ) : <EOL> req = Request . blank ( target , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = { '<STR_LIT>' : copy_from , <EOL> '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> with authorize_storlet_execution ( ) as ase : <EOL> copy ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , ase . call_count ) <EOL> def test_PUT_copy_with_storlets ( self ) : <EOL> source = '<STR_LIT>' <EOL> target = '<STR_LIT>' <EOL> copy_from = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , source , HTTPOk , body = '<STR_LIT>' ) <EOL> self . app . register ( '<STR_LIT>' , target , HTTPCreated ) <EOL> def copy ( target , source , copy_from ) : <EOL> req = Request . blank ( target , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = { '<STR_LIT>' : copy_from , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> get_calls = self . app . get_calls ( '<STR_LIT:GET>' , source ) <EOL> self . assertEqual ( len ( get_calls ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( get_calls [ - <NUM_LIT:1> ] [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( get_calls [ - <NUM_LIT:1> ] [ <NUM_LIT:1> ] , source ) <EOL> put_calls = self . app . get_calls ( '<STR_LIT>' , target ) <EOL> self . assertEqual ( len ( put_calls ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( put_calls [ - <NUM_LIT:1> ] [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> with storlet_enabled ( ) : <EOL> copy ( target , source , copy_from ) <EOL> def test_COPY_verb_without_storlets ( self ) : <EOL> source = '<STR_LIT>' <EOL> destination = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT>' , source , HTTPCreated ) <EOL> def copy ( ) : <EOL> req = Request . blank ( source , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = { '<STR_LIT>' : destination , <EOL> '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> with mock . patch ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) as m : <EOL> copy ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , m . call_count ) <EOL> def test_COPY_verb_with_storlets ( self ) : <EOL> source = '<STR_LIT>' <EOL> target = '<STR_LIT>' <EOL> destination = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , source , HTTPOk , body = '<STR_LIT>' ) <EOL> self . app . register ( '<STR_LIT>' , target , HTTPCreated ) <EOL> def copy ( target , source , destination ) : <EOL> req = Request . blank ( source , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = { '<STR_LIT>' : destination , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> get_calls = self . app . get_calls ( '<STR_LIT:GET>' , source ) <EOL> self . assertEqual ( len ( get_calls ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( get_calls [ - <NUM_LIT:1> ] [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( get_calls [ - <NUM_LIT:1> ] [ <NUM_LIT:1> ] , source ) <EOL> put_calls = self . app . get_calls ( '<STR_LIT>' , target ) <EOL> self . assertEqual ( len ( put_calls ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( put_calls [ - <NUM_LIT:1> ] [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> with storlet_enabled ( ) : <EOL> copy ( target , source , destination ) <EOL> def test_copy_with_unsupported_headers ( self ) : <EOL> target = '<STR_LIT>' <EOL> def copy ( method , copy_header ) : <EOL> base_headers = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> } <EOL> base_headers . update ( copy_header ) <EOL> req = Request . blank ( target , environ = { '<STR_LIT>' : method } , <EOL> headers = base_headers ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> with storlet_enabled ( ) : <EOL> self . assertRaises ( HTTPBadRequest , <EOL> copy ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT:a>' } ) ) <EOL> self . assertRaises ( HTTPBadRequest , <EOL> copy ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) ) <EOL> self . assertRaises ( HTTPBadRequest , <EOL> copy ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT:a>' } ) ) <EOL> self . assertRaises ( HTTPBadRequest , <EOL> copy ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) ) <EOL> def test_PUT_storlet ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT>' , target , HTTPCreated ) <EOL> def put ( path ) : <EOL> sheaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = sheaders ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> with storlet_enabled ( ) : <EOL> put ( target ) <EOL> def test_PUT_storlet_mandatory_parameter_fails ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT>' , target , HTTPCreated ) <EOL> drop_headers = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> def put ( path , header , assertion ) : <EOL> sheaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> del sheaders [ header ] <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = sheaders ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> for header , assertion in drop_headers : <EOL> with storlet_enabled ( ) : <EOL> put ( target , header , assertion ) <EOL> def test_POST_storlet ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:POST>' , target , HTTPAccepted ) <EOL> def post ( path ) : <EOL> sheaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT:POST>' } , <EOL> headers = sheaders ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> with storlet_enabled ( ) : <EOL> post ( target ) <EOL> def test_GET_storlet ( self ) : <EOL> target = '<STR_LIT>' <EOL> sheaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , headers = sheaders , <EOL> body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> resp_headers = dict ( self . got_headers [ - <NUM_LIT:1> ] ) <EOL> for key in sheaders : <EOL> self . assertEqual ( resp_headers [ key ] , sheaders [ key ] ) <EOL> with storlet_enabled ( ) : <EOL> get ( target ) <EOL> def test_PUT_dependency ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT>' , target , HTTPCreated ) <EOL> def put ( path ) : <EOL> sheaders = { '<STR_LIT>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = sheaders ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> with storlet_enabled ( ) : <EOL> put ( target ) <EOL> def test_POST_dependency ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:POST>' , target , HTTPAccepted ) <EOL> def post ( path ) : <EOL> sheaders = { '<STR_LIT>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT:POST>' } , <EOL> headers = sheaders ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> with storlet_enabled ( ) : <EOL> post ( target ) <EOL> def test_GET_dependency ( self ) : <EOL> target = '<STR_LIT>' <EOL> sheaders = { '<STR_LIT>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , headers = sheaders , <EOL> body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> resp_headers = dict ( self . got_headers [ - <NUM_LIT:1> ] ) <EOL> for key in sheaders : <EOL> self . assertEqual ( resp_headers [ key ] , sheaders [ key ] ) <EOL> with storlet_enabled ( ) : <EOL> get ( target ) <EOL> class TestStorletMiddlewareObject ( TestStorletMiddleware ) : <EOL> def setUp ( self ) : <EOL> super ( TestStorletMiddlewareObject , self ) . setUp ( ) <EOL> self . conf [ '<STR_LIT>' ] = '<STR_LIT:object>' <EOL> def test_call_unsupported_method ( self ) : <EOL> def call ( method ) : <EOL> path = '<STR_LIT>' <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : method } , <EOL> headers = headers ) <EOL> app = self . get_app ( lambda env , start : None , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> for method in ( '<STR_LIT:POST>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> call ( method ) <EOL> def test_PUT_objet_into_storlets_container ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT>' , target , HTTPCreated ) <EOL> def put ( path ) : <EOL> sheaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> req = Request . blank ( path , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = sheaders ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> put ( target ) <EOL> def test_GET_without_storlets ( self ) : <EOL> def basic_get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' } ) <EOL> self . app . register ( '<STR_LIT:GET>' , path , HTTPOk , body = '<STR_LIT>' ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> self . app . reset_all ( ) <EOL> for target in ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> basic_get ( target ) <EOL> def test_GET_with_storlets ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp . read ( ) , '<STR_LIT>' ) <EOL> get ( target ) <EOL> def test_GET_with_storlets_and_http_range ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> get ( target ) <EOL> def test_GET_with_storlets_and_storlet_range ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> get ( target ) <EOL> def test_GET_slo_manifest_with_storlets ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:True>' } , <EOL> body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> get ( target ) <EOL> def test_GET_slo_segment_with_storlets ( self ) : <EOL> target = '<STR_LIT>' <EOL> self . app . register ( '<STR_LIT:GET>' , target , HTTPOk , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:True>' } , <EOL> body = '<STR_LIT>' ) <EOL> def get ( path ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> app = self . get_app ( self . app , self . conf ) <EOL> resp = app ( req . environ , self . start_response ) <EOL> self . assertEqual ( '<STR_LIT>' , self . got_statuses [ - <NUM_LIT:1> ] ) <EOL> self . assertEqual ( resp , [ '<STR_LIT>' ] ) <EOL> get ( target ) <EOL> class TestStorletBaseHandler ( unittest . TestCase ) : <EOL> def test_init_failed_via_base_handler ( self ) : <EOL> def assert_not_implemented ( method , path , headers ) : <EOL> req = Request . blank ( <EOL> path , environ = { '<STR_LIT>' : method } , <EOL> headers = headers ) <EOL> try : <EOL> BaseStorletHandler ( <EOL> req , mock . MagicMock ( ) , mock . MagicMock ( ) , mock . MagicMock ( ) ) <EOL> except NotImplementedError : <EOL> pass <EOL> except Exception as e : <EOL> self . fail ( '<STR_LIT>' % <EOL> ( repr ( e ) , path , method , headers ) ) <EOL> for method in ( '<STR_LIT>' , '<STR_LIT:GET>' , '<STR_LIT:POST>' ) : <EOL> for path in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> for headers in ( { } , { '<STR_LIT>' : '<STR_LIT>' } ) : <EOL> assert_not_implemented ( method , path , headers ) <EOL> class TestStorletProxyHandler ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . handler_class = StorletProxyHandler <EOL> def test_init_handler ( self ) : <EOL> req = Request . blank ( <EOL> '<STR_LIT>' , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> with storlet_enabled ( ) : <EOL> handler = self . handler_class ( <EOL> req , mock . MagicMock ( ) , mock . MagicMock ( ) , mock . MagicMock ( ) ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . request . path ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . api_version ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . account ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . container ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . obj ) <EOL> req = Request . blank ( <EOL> '<STR_LIT>' , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> handler . request = req <EOL> self . assertEqual ( '<STR_LIT>' , handler . request . path ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . api_version ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . account ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . container ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . obj ) <EOL> with self . assertRaises ( AttributeError ) : <EOL> handler . api_version = '<STR_LIT:1>' <EOL> with self . assertRaises ( AttributeError ) : <EOL> handler . account = '<STR_LIT>' <EOL> with self . assertRaises ( AttributeError ) : <EOL> handler . container = '<STR_LIT>' <EOL> with self . assertRaises ( AttributeError ) : <EOL> handler . obj = '<STR_LIT>' <EOL> class TestStorletObjectHandler ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . handler_class = StorletObjectHandler <EOL> def test_init_handler ( self ) : <EOL> req = Request . blank ( <EOL> '<STR_LIT>' , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> handler = self . handler_class ( <EOL> req , mock . MagicMock ( ) , mock . MagicMock ( ) , mock . MagicMock ( ) ) <EOL> self . assertEqual ( '<STR_LIT:0>' , handler . api_version ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . account ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . container ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . obj ) <EOL> req = Request . blank ( <EOL> '<STR_LIT>' , environ = { '<STR_LIT>' : '<STR_LIT:GET>' } , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> handler . request = req <EOL> self . assertEqual ( '<STR_LIT>' , handler . request . path ) <EOL> self . assertEqual ( '<STR_LIT:0>' , handler . api_version ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . account ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . container ) <EOL> self . assertEqual ( '<STR_LIT>' , handler . obj ) <EOL> with self . assertRaises ( AttributeError ) : <EOL> handler . api_version = '<STR_LIT:1>' <EOL> with self . assertRaises ( AttributeError ) : <EOL> handler . account = '<STR_LIT>' <EOL> with self . assertRaises ( AttributeError ) : <EOL> handler . container = '<STR_LIT>' <EOL> with self . assertRaises ( AttributeError ) : <EOL> handler . obj = '<STR_LIT>' <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
