<s> from contextlib import contextmanager <EOL> import mock <EOL> import os <EOL> import tempfile <EOL> import unittest <EOL> from swift . common . swob import HTTPException , HTTPNoContent , HTTPUnauthorized , HTTPNotFound , Request <EOL> from tests . unit . swift import FakeLogger <EOL> from tests . unit . swift . storlet_middleware import FakeApp <EOL> from storlet_gateway . storlet_docker_gateway import DockerStorletRequest , StorletGatewayDocker <EOL> class TestDockerStorletRequest ( unittest . TestCase ) : <EOL> def test_init ( self ) : <EOL> account = '<STR_LIT:a>' <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> params = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> req = Request . blank ( '<STR_LIT>' , headers = headers ) <EOL> dsreq = DockerStorletRequest ( account , req , params ) <EOL> self . assertEqual ( dsreq . generate_log , False ) <EOL> self . assertEqual ( dsreq . storlet_id , '<STR_LIT>' ) <EOL> self . assertEqual ( dsreq . user_metadata , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( dsreq . account , account ) <EOL> class TestIterLike ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . timeout = <NUM_LIT:10> <EOL> self . content = b'<STR_LIT>' <EOL> self . fd , fname = tempfile . mkstemp ( ) <EOL> os . unlink ( fname ) <EOL> os . write ( self . fd , self . content ) <EOL> self . _reset_fd ( ) <EOL> self . iter_like = StorletGatewayDocker . IterLike ( <EOL> self . fd , self . timeout , None ) <EOL> def tearDown ( self ) : <EOL> pass <EOL> def _fake_select ( self , r , w , x , timeout = <NUM_LIT:0> ) : <EOL> return r , w , x <EOL> def _reset_fd ( self ) : <EOL> os . lseek ( self . fd , <NUM_LIT:0> , os . SEEK_SET ) <EOL> @ contextmanager <EOL> def _mock_select ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , <EOL> self . _fake_select ) : <EOL> yield <EOL> def test_read_with_timeout ( self ) : <EOL> self . assertEqual ( <EOL> self . iter_like . read_with_timeout ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . iter_like . read_with_timeout ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . iter_like . read_with_timeout ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . iter_like . read_with_timeout ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> def test_next ( self ) : <EOL> with self . _mock_select ( ) : <EOL> self . assertEqual ( self . iter_like . next ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( self . iter_like . next ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( self . iter_like . next ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> with self . assertRaises ( StopIteration ) : <EOL> self . iter_like . next ( <NUM_LIT:6> ) <EOL> self . _reset_fd ( ) <EOL> with self . _mock_select ( ) : <EOL> self . assertEqual ( self . iter_like . next ( <NUM_LIT:50> ) , <EOL> b'<STR_LIT>' ) <EOL> with self . assertRaises ( StopIteration ) : <EOL> self . iter_like . next ( <NUM_LIT:50> ) <EOL> def test_read ( self ) : <EOL> with self . _mock_select ( ) : <EOL> self . assertEqual ( self . iter_like . read ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( self . iter_like . read ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( self . iter_like . read ( <NUM_LIT:6> ) , b'<STR_LIT>' ) <EOL> with self . assertRaises ( StopIteration ) : <EOL> self . iter_like . next ( <NUM_LIT:6> ) <EOL> def test_readline ( self ) : <EOL> with self . _mock_select ( ) : <EOL> self . assertEqual ( self . iter_like . readline ( ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( self . iter_like . readline ( <NUM_LIT:2> ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( self . iter_like . readline ( ) , b'<STR_LIT>' ) <EOL> self . assertEqual ( self . iter_like . readline ( <NUM_LIT:100> ) , b'<STR_LIT>' ) <EOL> with self . assertRaises ( StopIteration ) : <EOL> self . iter_like . readline ( ) <EOL> def test_readlines ( self ) : <EOL> with self . _mock_select ( ) : <EOL> self . assertEqual ( <EOL> self . iter_like . readlines ( ) , <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' , b'<STR_LIT>' ] ) <EOL> self . _reset_fd ( ) <EOL> with self . _mock_select ( ) : <EOL> self . assertEqual ( <EOL> self . iter_like . readlines ( <NUM_LIT:7> ) , <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' ] ) <EOL> class TestStorletGatewayDocker ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . sconf = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . logger = FakeLogger ( ) <EOL> self . app = FakeApp ( ) <EOL> self . storlet_container = self . sconf [ '<STR_LIT>' ] <EOL> self . storlet_dependency = self . sconf [ '<STR_LIT>' ] <EOL> self . version = '<STR_LIT>' <EOL> self . account = '<STR_LIT:a>' <EOL> self . container = '<STR_LIT:c>' <EOL> self . obj = '<STR_LIT:o>' <EOL> self . sobj = '<STR_LIT>' <EOL> @ property <EOL> def req_path ( self ) : <EOL> return self . _create_proxy_path ( <EOL> self . version , self . account , self . container , <EOL> self . obj ) <EOL> @ property <EOL> def storlet_path ( self ) : <EOL> return self . _create_proxy_path ( <EOL> self . version , self . account , self . storlet_container , <EOL> self . sobj ) <EOL> def tearDown ( self ) : <EOL> pass <EOL> def _create_gateway ( self ) : <EOL> return StorletGatewayDocker ( <EOL> self . sconf , self . logger , self . app , <EOL> self . version , self . account , self . container , self . obj ) <EOL> def _create_proxy_path ( self , version , account , container , obj ) : <EOL> return '<STR_LIT:/>' . join ( [ '<STR_LIT>' , version , account , container , obj ] ) <EOL> def _create_req ( self , method , headers = None , body = None ) : <EOL> return Request . blank ( <EOL> self . req_path , environ = { '<STR_LIT>' : method } , <EOL> headers = headers , body = body ) <EOL> def _create_storlet_req ( self , method , headers = None , body = None ) : <EOL> if headers is None : <EOL> headers = { } <EOL> headers [ '<STR_LIT>' ] = self . sobj <EOL> return self . _create_req ( method , headers , body ) <EOL> @ contextmanager <EOL> def assertRaisesHttpStatus ( self , status ) : <EOL> with self . assertRaises ( HTTPException ) as e : <EOL> yield <EOL> self . assertEqual ( e . status_int , status ) <EOL> def test_check_mandatory_params ( self ) : <EOL> params = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> StorletGatewayDocker . _check_mandatory_params ( <EOL> params , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> with self . assertRaises ( ValueError ) : <EOL> StorletGatewayDocker . _check_mandatory_params ( <EOL> params , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_validate_storlet_registration ( self ) : <EOL> obj = '<STR_LIT>' <EOL> params = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> StorletGatewayDocker . validate_storlet_registration ( params , obj ) <EOL> params = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> with self . assertRaises ( ValueError ) : <EOL> StorletGatewayDocker . validate_storlet_registration ( params , obj ) <EOL> obj = '<STR_LIT>' <EOL> params = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> with self . assertRaises ( ValueError ) : <EOL> StorletGatewayDocker . validate_storlet_registration ( params , obj ) <EOL> def test_validate_dependency_registration ( self ) : <EOL> obj = '<STR_LIT>' <EOL> params = { '<STR_LIT>' : '<STR_LIT:1.0>' } <EOL> StorletGatewayDocker . validate_dependency_registration ( params , obj ) <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' } <EOL> StorletGatewayDocker . validate_dependency_registration ( params , obj ) <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' } <EOL> with self . assertRaises ( ValueError ) : <EOL> StorletGatewayDocker . validate_dependency_registration ( params , obj ) <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' } <EOL> with self . assertRaises ( ValueError ) : <EOL> StorletGatewayDocker . validate_dependency_registration ( params , obj ) <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' } <EOL> with self . assertRaises ( ValueError ) : <EOL> StorletGatewayDocker . validate_dependency_registration ( params , obj ) <EOL> def test_authorizeStorletExecution ( self ) : <EOL> sheaders = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . app . register ( '<STR_LIT>' , self . storlet_path , HTTPNoContent , sheaders ) <EOL> gw = self . _create_gateway ( ) <EOL> req = self . _create_storlet_req ( '<STR_LIT:GET>' ) <EOL> gw . authorizeStorletExecution ( req ) <EOL> for key in sheaders . keys ( ) : <EOL> self . assertEqual ( sheaders [ key ] , gw . storlet_metadata [ key ] ) <EOL> self . assertEqual ( len ( self . app . get_calls ( ) ) , <NUM_LIT:1> ) <EOL> def test_augmentStorletExcecution ( self ) : <EOL> sheaders = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> oheaders = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> oheaders . update ( sheaders ) <EOL> self . app . register ( '<STR_LIT>' , self . storlet_path , HTTPNoContent , oheaders ) <EOL> gw = self . _create_gateway ( ) <EOL> req = self . _create_storlet_req ( '<STR_LIT:GET>' ) <EOL> gw . authorizeStorletExecution ( req ) <EOL> gw . augmentStorletRequest ( req ) <EOL> for key in sheaders : <EOL> self . assertEqual ( req . headers [ key ] , sheaders [ key ] ) <EOL> self . assertEqual ( req . headers [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( req . headers [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_verify_access ( self ) : <EOL> sheaders = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1.0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . app . register ( '<STR_LIT>' , self . storlet_path , HTTPNoContent , sheaders ) <EOL> gw = self . _create_gateway ( ) <EOL> req = self . _create_storlet_req ( '<STR_LIT:GET>' ) <EOL> headers = gw . _verify_access ( req , self . version , self . account , <EOL> self . storlet_container , self . sobj ) <EOL> for key in sheaders . keys ( ) : <EOL> self . assertEqual ( sheaders [ key ] , headers [ key ] ) <EOL> self . assertEqual ( len ( self . app . get_calls ( ) ) , <NUM_LIT:1> ) <EOL> self . app . reset_all ( ) <EOL> self . app . register ( '<STR_LIT>' , self . storlet_path , HTTPUnauthorized ) <EOL> gw = self . _create_gateway ( ) <EOL> req = self . _create_storlet_req ( '<STR_LIT:GET>' ) <EOL> with self . assertRaisesHttpStatus ( <NUM_LIT> ) : <EOL> gw . _verify_access ( req , self . version , self . account , <EOL> self . storlet_container , self . sobj ) <EOL> self . assertEqual ( len ( self . app . get_calls ( ) ) , <NUM_LIT:1> ) <EOL> self . app . reset_all ( ) <EOL> self . app . register ( '<STR_LIT>' , self . storlet_path , HTTPNotFound ) <EOL> gw = self . _create_gateway ( ) <EOL> req = self . _create_storlet_req ( '<STR_LIT:GET>' ) <EOL> with self . assertRaisesHttpStatus ( <NUM_LIT> ) : <EOL> gw . _verify_access ( req , self . version , self . account , <EOL> self . storlet_container , self . sobj ) <EOL> self . assertEqual ( len ( self . app . get_calls ( ) ) , <NUM_LIT:1> ) <EOL> def test_clean_storlet_stuff_from_request ( self ) : <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> req = self . _create_req ( '<STR_LIT:GET>' , headers = headers ) <EOL> gw = self . _create_gateway ( ) <EOL> gw . _clean_storlet_stuff_from_request ( req . headers ) <EOL> self . assertFalse ( '<STR_LIT>' in req . headers ) <EOL> self . assertEqual ( req . headers [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertFalse ( '<STR_LIT>' in req . headers ) <EOL> self . assertEqual ( req . headers [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_set_metadata_in_headers ( self ) : <EOL> gw = self . _create_gateway ( ) <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> gw . _set_metadata_in_headers ( headers , { } ) <EOL> self . assertEqual ( headers , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> md = { '<STR_LIT>' : '<STR_LIT>' } <EOL> gw . _set_metadata_in_headers ( headers , md ) <EOL> self . assertEqual ( headers , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
