<s> import unittest <EOL> from swift . common . swob import Request , Response <EOL> from swift . common . middleware import gatekeeper <EOL> class FakeApp ( object ) : <EOL> def __init__ ( self , headers = None ) : <EOL> if headers is None : <EOL> headers = { } <EOL> self . headers = headers <EOL> self . req = None <EOL> def __call__ ( self , env , start_response ) : <EOL> self . req = Request ( env ) <EOL> return Response ( request = self . req , body = '<STR_LIT>' , <EOL> headers = self . headers ) ( env , start_response ) <EOL> class FakeMiddleware ( object ) : <EOL> def __init__ ( self , app , conf , header_list = None ) : <EOL> self . app = app <EOL> self . conf = conf <EOL> self . header_list = header_list <EOL> def __call__ ( self , env , start_response ) : <EOL> def fake_resp ( status , response_headers , exc_info = None ) : <EOL> for i in self . header_list : <EOL> response_headers . append ( i ) <EOL> return start_response ( status , response_headers , exc_info ) <EOL> return self . app ( env , fake_resp ) <EOL> class TestGatekeeper ( unittest . TestCase ) : <EOL> methods = [ '<STR_LIT>' , '<STR_LIT:POST>' , '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> allowed_headers = { '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' } <EOL> sysmeta_headers = { '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' , <EOL> '<STR_LIT>' : '<STR_LIT:value>' } <EOL> x_backend_headers = { '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> x_timestamp_headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> forbidden_headers_out = dict ( sysmeta_headers . items ( ) + <EOL> x_backend_headers . items ( ) ) <EOL> forbidden_headers_in = dict ( sysmeta_headers . items ( ) + <EOL> x_backend_headers . items ( ) ) <EOL> shunted_headers_in = dict ( x_timestamp_headers . items ( ) ) <EOL> def _assertHeadersEqual ( self , expected , actual ) : <EOL> for key in expected : <EOL> self . assertTrue ( key . lower ( ) in actual , <EOL> '<STR_LIT>' % ( key , actual ) ) <EOL> def _assertHeadersAbsent ( self , unexpected , actual ) : <EOL> for key in unexpected : <EOL> self . assertTrue ( key . lower ( ) not in actual , <EOL> '<STR_LIT>' % ( key , actual ) ) <EOL> def get_app ( self , app , global_conf , ** local_conf ) : <EOL> factory = gatekeeper . filter_factory ( global_conf , ** local_conf ) <EOL> return factory ( app ) <EOL> def test_ok_header ( self ) : <EOL> req = Request . blank ( '<STR_LIT>' , environ = { '<STR_LIT>' : '<STR_LIT>' } , <EOL> headers = self . allowed_headers ) <EOL> fake_app = FakeApp ( ) <EOL> app = self . get_app ( fake_app , { } ) <EOL> resp = req . get_response ( app ) <EOL> self . assertEqual ( '<STR_LIT>' , resp . status ) <EOL> self . assertEqual ( resp . body , '<STR_LIT>' ) <EOL> self . _assertHeadersEqual ( self . allowed_headers , fake_app . req . headers ) <EOL> def _test_reserved_header_removed_inbound ( self , method ) : <EOL> headers = dict ( self . forbidden_headers_in ) <EOL> headers . update ( self . allowed_headers ) <EOL> headers . update ( self . shunted_headers_in ) <EOL> req = Request . blank ( '<STR_LIT>' , environ = { '<STR_LIT>' : method } , <EOL> headers = headers ) <EOL> fake_app = FakeApp ( ) <EOL> app = self . get_app ( fake_app , { } ) <EOL> resp = req . get_response ( app ) <EOL> self . assertEqual ( '<STR_LIT>' , resp . status ) <EOL> expected_headers = dict ( self . allowed_headers ) <EOL> expected_headers . update ( { '<STR_LIT>' + k : v <EOL> for k , v in self . shunted_headers_in . items ( ) } ) <EOL> self . _assertHeadersEqual ( expected_headers , fake_app . req . headers ) <EOL> unexpected_headers = dict ( self . forbidden_headers_in . items ( ) + <EOL> self . shunted_headers_in . items ( ) ) <EOL> self . _assertHeadersAbsent ( unexpected_headers , fake_app . req . headers ) <EOL> def test_reserved_header_removed_inbound ( self ) : <EOL> for method in self . methods : <EOL> self . _test_reserved_header_removed_inbound ( method ) <EOL> def _test_reserved_header_shunted_inbound ( self , method ) : <EOL> headers = dict ( self . shunted_headers_in ) <EOL> headers . update ( self . allowed_headers ) <EOL> req = Request . blank ( '<STR_LIT>' , environ = { '<STR_LIT>' : method } , <EOL> headers = headers ) <EOL> fake_app = FakeApp ( ) <EOL> app = self . get_app ( fake_app , { } , shunt_inbound_x_timestamp = '<STR_LIT:true>' ) <EOL> resp = req . get_response ( app ) <EOL> self . assertEqual ( '<STR_LIT>' , resp . status ) <EOL> expected_headers = dict ( self . allowed_headers ) <EOL> expected_headers . update ( { '<STR_LIT>' + k : v <EOL> for k , v in self . shunted_headers_in . items ( ) } ) <EOL> self . _assertHeadersEqual ( expected_headers , fake_app . req . headers ) <EOL> self . _assertHeadersAbsent ( self . shunted_headers_in , <EOL> fake_app . req . headers ) <EOL> def test_reserved_header_shunted_inbound ( self ) : <EOL> for method in self . methods : <EOL> self . _test_reserved_header_shunted_inbound ( method ) <EOL> def _test_reserved_header_shunt_bypassed_inbound ( self , method ) : <EOL> headers = dict ( self . shunted_headers_in ) <EOL> headers . update ( self . allowed_headers ) <EOL> req = Request . blank ( '<STR_LIT>' , environ = { '<STR_LIT>' : method } , <EOL> headers = headers ) <EOL> fake_app = FakeApp ( ) <EOL> app = self . get_app ( fake_app , { } , shunt_inbound_x_timestamp = '<STR_LIT:false>' ) <EOL> resp = req . get_response ( app ) <EOL> self . assertEqual ( '<STR_LIT>' , resp . status ) <EOL> expected_headers = dict ( self . allowed_headers . items ( ) + <EOL> self . shunted_headers_in . items ( ) ) <EOL> self . _assertHeadersEqual ( expected_headers , fake_app . req . headers ) <EOL> def test_reserved_header_shunt_bypassed_inbound ( self ) : <EOL> for method in self . methods : <EOL> self . _test_reserved_header_shunt_bypassed_inbound ( method ) <EOL> def _test_reserved_header_removed_outbound ( self , method ) : <EOL> headers = dict ( self . forbidden_headers_out ) <EOL> headers . update ( self . allowed_headers ) <EOL> req = Request . blank ( '<STR_LIT>' , environ = { '<STR_LIT>' : method } ) <EOL> fake_app = FakeApp ( headers = headers ) <EOL> app = self . get_app ( fake_app , { } ) <EOL> resp = req . get_response ( app ) <EOL> self . assertEqual ( '<STR_LIT>' , resp . status ) <EOL> self . _assertHeadersEqual ( self . allowed_headers , resp . headers ) <EOL> self . _assertHeadersAbsent ( self . forbidden_headers_out , resp . headers ) <EOL> def test_reserved_header_removed_outbound ( self ) : <EOL> for method in self . methods : <EOL> self . _test_reserved_header_removed_outbound ( method ) <EOL> def _test_duplicate_headers_not_removed ( self , method , app_hdrs ) : <EOL> def fake_factory ( global_conf , ** local_conf ) : <EOL> conf = global_conf . copy ( ) <EOL> conf . update ( local_conf ) <EOL> headers = [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> def fake_filter ( app ) : <EOL> return FakeMiddleware ( app , conf , headers ) <EOL> return fake_filter <EOL> def fake_start_response ( status , response_headers , exc_info = None ) : <EOL> hdr_list = [ ] <EOL> for k , v in response_headers : <EOL> if k == '<STR_LIT>' : <EOL> hdr_list . append ( v ) <EOL> self . assertTrue ( '<STR_LIT>' in hdr_list ) <EOL> self . assertTrue ( '<STR_LIT>' in hdr_list ) <EOL> self . assertEqual ( len ( hdr_list ) , <NUM_LIT:2> ) <EOL> req = Request . blank ( '<STR_LIT>' , environ = { '<STR_LIT>' : method } ) <EOL> fake_app = FakeApp ( headers = app_hdrs ) <EOL> factory = gatekeeper . filter_factory ( { } ) <EOL> factory_wrap = fake_factory ( { } ) <EOL> app = factory ( factory_wrap ( fake_app ) ) <EOL> app ( req . environ , fake_start_response ) <EOL> def test_duplicate_headers_not_removed ( self ) : <EOL> for method in self . methods : <EOL> for app_hdrs in ( { } , self . forbidden_headers_out ) : <EOL> self . _test_duplicate_headers_not_removed ( method , app_hdrs ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
