<s> import unittest <EOL> from swift . common . middleware import keystoneauth <EOL> from swift . common . swob import Request , Response <EOL> from swift . common . http import HTTP_FORBIDDEN <EOL> from swift . common . utils import split_path <EOL> from swift . proxy . controllers . base import _get_cache_key <EOL> from test . unit import FakeLogger <EOL> UNKNOWN_ID = keystoneauth . UNKNOWN_ID <EOL> def _fake_token_info ( version = '<STR_LIT:2>' ) : <EOL> if version == '<STR_LIT:2>' : <EOL> return { '<STR_LIT>' : '<STR_LIT>' } <EOL> if version == '<STR_LIT:3>' : <EOL> return { '<STR_LIT>' : '<STR_LIT>' } <EOL> def operator_roles ( test_auth ) : <EOL> return list ( test_auth . account_rules [ test_auth . reseller_prefixes [ <NUM_LIT:0> ] ] . get ( <EOL> '<STR_LIT>' ) ) <EOL> def get_account_for_tenant ( test_auth , tenant_id ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( test_auth . reseller_prefixes [ <NUM_LIT:0> ] , tenant_id ) <EOL> def get_identity_headers ( status = '<STR_LIT>' , tenant_id = '<STR_LIT:1>' , <EOL> tenant_name = '<STR_LIT>' , project_domain_name = '<STR_LIT>' , <EOL> project_domain_id = '<STR_LIT>' , <EOL> user_name = '<STR_LIT>' , user_id = '<STR_LIT>' , <EOL> user_domain_name = '<STR_LIT>' , user_domain_id = '<STR_LIT>' , <EOL> role = '<STR_LIT>' , <EOL> service_role = None ) : <EOL> if role is None : <EOL> role = [ ] <EOL> if isinstance ( role , list ) : <EOL> role = '<STR_LIT:U+002C>' . join ( role ) <EOL> res = dict ( X_IDENTITY_STATUS = status , <EOL> X_TENANT_ID = tenant_id , <EOL> X_TENANT_NAME = tenant_name , <EOL> X_PROJECT_ID = tenant_id , <EOL> X_PROJECT_NAME = tenant_name , <EOL> X_PROJECT_DOMAIN_ID = project_domain_id , <EOL> X_PROJECT_DOMAIN_NAME = project_domain_name , <EOL> X_ROLES = role , <EOL> X_USER_NAME = user_name , <EOL> X_USER_ID = user_id , <EOL> X_USER_DOMAIN_NAME = user_domain_name , <EOL> X_USER_DOMAIN_ID = user_domain_id ) <EOL> if service_role : <EOL> res . update ( X_SERVICE_ROLES = service_role ) <EOL> return res <EOL> class FakeApp ( object ) : <EOL> def __init__ ( self , status_headers_body_iter = None ) : <EOL> self . calls = <NUM_LIT:0> <EOL> self . call_contexts = [ ] <EOL> self . status_headers_body_iter = status_headers_body_iter <EOL> if not self . status_headers_body_iter : <EOL> self . status_headers_body_iter = iter ( [ ( '<STR_LIT>' , { } , '<STR_LIT>' ) ] ) <EOL> def __call__ ( self , env , start_response ) : <EOL> self . calls += <NUM_LIT:1> <EOL> self . request = Request . blank ( '<STR_LIT>' , environ = env ) <EOL> if '<STR_LIT>' in env : <EOL> resp = env [ '<STR_LIT>' ] ( self . request ) <EOL> if resp : <EOL> return resp ( env , start_response ) <EOL> context = { '<STR_LIT>' : self . request . method , <EOL> '<STR_LIT>' : self . request . headers } <EOL> self . call_contexts . append ( context ) <EOL> status , headers , body = next ( self . status_headers_body_iter ) <EOL> return Response ( status = status , headers = headers , <EOL> body = body ) ( env , start_response ) <EOL> class SwiftAuth ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . test_auth = keystoneauth . filter_factory ( { } ) ( FakeApp ( ) ) <EOL> self . test_auth . logger = FakeLogger ( ) <EOL> def _make_request ( self , path = None , headers = None , ** kwargs ) : <EOL> if not path : <EOL> path = '<STR_LIT>' % get_account_for_tenant ( self . test_auth , '<STR_LIT:foo>' ) <EOL> return Request . blank ( path , headers = headers , ** kwargs ) <EOL> def _get_successful_middleware ( self ) : <EOL> response_iter = iter ( [ ( '<STR_LIT>' , { } , '<STR_LIT>' ) ] ) <EOL> return keystoneauth . filter_factory ( { } ) ( FakeApp ( response_iter ) ) <EOL> def test_invalid_request_authorized ( self ) : <EOL> role = self . test_auth . reseller_admin_role <EOL> headers = get_identity_headers ( role = role ) <EOL> req = self . _make_request ( '<STR_LIT:/>' , headers = headers ) <EOL> resp = req . get_response ( self . _get_successful_middleware ( ) ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_invalid_request_non_authorized ( self ) : <EOL> req = self . _make_request ( '<STR_LIT:/>' ) <EOL> resp = req . get_response ( self . _get_successful_middleware ( ) ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_confirmed_identity_is_authorized ( self ) : <EOL> role = self . test_auth . reseller_admin_role <EOL> headers = get_identity_headers ( role = role ) <EOL> req = self . _make_request ( '<STR_LIT>' , headers ) <EOL> resp = req . get_response ( self . _get_successful_middleware ( ) ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> def test_detect_reseller_request ( self ) : <EOL> role = self . test_auth . reseller_admin_role <EOL> headers = get_identity_headers ( role = role ) <EOL> req = self . _make_request ( '<STR_LIT>' , headers ) <EOL> req . get_response ( self . _get_successful_middleware ( ) ) <EOL> self . assertTrue ( req . environ . get ( '<STR_LIT>' ) ) <EOL> def test_confirmed_identity_is_not_authorized ( self ) : <EOL> headers = get_identity_headers ( ) <EOL> req = self . _make_request ( '<STR_LIT>' , headers ) <EOL> resp = req . get_response ( self . test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_anonymous_is_authorized_for_permitted_referrer ( self ) : <EOL> req = self . _make_request ( headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> req . acl = '<STR_LIT>' <EOL> resp = req . get_response ( self . _get_successful_middleware ( ) ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> def test_anonymous_with_validtoken_authorized_for_permitted_referrer ( self ) : <EOL> req = self . _make_request ( headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> req . acl = '<STR_LIT>' <EOL> resp = req . get_response ( self . _get_successful_middleware ( ) ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> def test_anonymous_is_not_authorized_for_unknown_reseller_prefix ( self ) : <EOL> req = self . _make_request ( path = '<STR_LIT>' , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> resp = req . get_response ( self . test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_denied_responses ( self ) : <EOL> def get_resp_status ( headers ) : <EOL> req = self . _make_request ( headers = headers ) <EOL> resp = req . get_response ( self . test_auth ) <EOL> return resp . status_int <EOL> self . assertEqual ( get_resp_status ( { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> <NUM_LIT> ) <EOL> self . assertEqual ( get_resp_status ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) , <NUM_LIT> ) <EOL> self . assertEqual ( get_resp_status ( { } ) , <NUM_LIT> ) <EOL> self . assertEqual ( get_resp_status ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } ) , <NUM_LIT> ) <EOL> self . assertEqual ( get_resp_status ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) , <NUM_LIT> ) <EOL> self . assertEqual ( get_resp_status ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) , <NUM_LIT> ) <EOL> self . assertEqual ( get_resp_status ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) , <NUM_LIT> ) <EOL> def test_blank_reseller_prefix ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT>' } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> account = tenant_id = '<STR_LIT:foo>' <EOL> self . assertTrue ( test_auth . _account_matches_tenant ( account , tenant_id ) ) <EOL> def test_reseller_prefix_added_underscore ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT>' } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . assertEqual ( test_auth . reseller_prefixes [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> def test_reseller_prefix_not_added_double_underscores ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT>' } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . assertEqual ( test_auth . reseller_prefixes [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> def test_override_asked_for_but_not_allowed ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT:false>' } <EOL> self . test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> req = self . _make_request ( '<STR_LIT>' , <EOL> environ = { '<STR_LIT>' : True } ) <EOL> resp = req . get_response ( self . test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_override_asked_for_and_allowed ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT:true>' } <EOL> self . test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> req = self . _make_request ( '<STR_LIT>' , <EOL> environ = { '<STR_LIT>' : True } ) <EOL> resp = req . get_response ( self . test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_override_default_allowed ( self ) : <EOL> req = self . _make_request ( '<STR_LIT>' , <EOL> environ = { '<STR_LIT>' : True } ) <EOL> resp = req . get_response ( self . test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_anonymous_options_allowed ( self ) : <EOL> req = self . _make_request ( '<STR_LIT>' , <EOL> environ = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> resp = req . get_response ( self . _get_successful_middleware ( ) ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> def test_identified_options_allowed ( self ) : <EOL> headers = get_identity_headers ( ) <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> req = self . _make_request ( '<STR_LIT>' , <EOL> headers = get_identity_headers ( ) , <EOL> environ = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> resp = req . get_response ( self . _get_successful_middleware ( ) ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> def test_auth_scheme ( self ) : <EOL> req = self . _make_request ( path = '<STR_LIT>' , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> resp = req . get_response ( self . test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> self . assertTrue ( '<STR_LIT>' in resp . headers ) <EOL> def test_project_domain_id_sysmeta_set ( self ) : <EOL> proj_id = '<STR_LIT>' <EOL> proj_domain_id = '<STR_LIT>' <EOL> headers = get_identity_headers ( tenant_id = proj_id , <EOL> project_domain_id = proj_domain_id ) <EOL> account = get_account_for_tenant ( self . test_auth , proj_id ) <EOL> path = '<STR_LIT>' + account <EOL> _ , info_key = _get_cache_key ( account , None ) <EOL> env = { info_key : { '<STR_LIT:status>' : <NUM_LIT:0> , '<STR_LIT>' : { } } , <EOL> '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:3>' ) } <EOL> req = Request . blank ( path , environ = env , headers = headers ) <EOL> req . method = '<STR_LIT:POST>' <EOL> headers_out = { '<STR_LIT>' : proj_domain_id } <EOL> fake_app = FakeApp ( iter ( [ ( '<STR_LIT>' , headers_out , '<STR_LIT>' ) ] ) ) <EOL> test_auth = keystoneauth . filter_factory ( { } ) ( fake_app ) <EOL> resp = req . get_response ( test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> self . assertEqual ( len ( fake_app . call_contexts ) , <NUM_LIT:1> ) <EOL> headers_sent = fake_app . call_contexts [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT>' in headers_sent , <EOL> headers_sent ) <EOL> self . assertEqual ( headers_sent [ '<STR_LIT>' ] , <EOL> proj_domain_id ) <EOL> self . assertTrue ( '<STR_LIT>' in resp . headers ) <EOL> self . assertEqual ( resp . headers [ '<STR_LIT>' ] , <EOL> proj_domain_id ) <EOL> def test_project_domain_id_sysmeta_set_to_unknown ( self ) : <EOL> proj_id = '<STR_LIT>' <EOL> headers = get_identity_headers ( tenant_id = '<STR_LIT>' , <EOL> project_domain_id = '<STR_LIT:default>' , <EOL> role = '<STR_LIT>' ) <EOL> account = get_account_for_tenant ( self . test_auth , proj_id ) <EOL> path = '<STR_LIT>' + account <EOL> _ , info_key = _get_cache_key ( account , None ) <EOL> env = { info_key : { '<STR_LIT:status>' : <NUM_LIT:0> , '<STR_LIT>' : { } } , <EOL> '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:3>' ) } <EOL> req = Request . blank ( path , environ = env , headers = headers ) <EOL> req . method = '<STR_LIT:POST>' <EOL> fake_app = FakeApp ( iter ( [ ( '<STR_LIT>' , { } , '<STR_LIT>' ) ] ) ) <EOL> test_auth = keystoneauth . filter_factory ( { } ) ( fake_app ) <EOL> resp = req . get_response ( test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> self . assertEqual ( len ( fake_app . call_contexts ) , <NUM_LIT:1> ) <EOL> headers_sent = fake_app . call_contexts [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT>' in headers_sent , <EOL> headers_sent ) <EOL> self . assertEqual ( headers_sent [ '<STR_LIT>' ] , <EOL> UNKNOWN_ID ) <EOL> def test_project_domain_id_sysmeta_not_set ( self ) : <EOL> proj_id = '<STR_LIT>' <EOL> headers = get_identity_headers ( tenant_id = proj_id , role = '<STR_LIT>' ) <EOL> account = get_account_for_tenant ( self . test_auth , proj_id ) <EOL> path = '<STR_LIT>' + account <EOL> _ , info_key = _get_cache_key ( account , None ) <EOL> env = { info_key : { '<STR_LIT:status>' : <NUM_LIT:0> , '<STR_LIT>' : { } } , <EOL> '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:2>' ) } <EOL> req = Request . blank ( path , environ = env , headers = headers ) <EOL> req . method = '<STR_LIT:POST>' <EOL> fake_app = FakeApp ( iter ( [ ( '<STR_LIT>' , { } , '<STR_LIT>' ) ] ) ) <EOL> test_auth = keystoneauth . filter_factory ( { } ) ( fake_app ) <EOL> resp = req . get_response ( test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> self . assertEqual ( len ( fake_app . call_contexts ) , <NUM_LIT:1> ) <EOL> headers_sent = fake_app . call_contexts [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> self . assertFalse ( '<STR_LIT>' in headers_sent , <EOL> headers_sent ) <EOL> def test_project_domain_id_sysmeta_set_unknown_with_v2 ( self ) : <EOL> proj_id = '<STR_LIT>' <EOL> headers = get_identity_headers ( tenant_id = '<STR_LIT>' , <EOL> role = '<STR_LIT>' ) <EOL> account = get_account_for_tenant ( self . test_auth , proj_id ) <EOL> path = '<STR_LIT>' + account <EOL> _ , info_key = _get_cache_key ( account , None ) <EOL> env = { info_key : { '<STR_LIT:status>' : <NUM_LIT:0> , '<STR_LIT>' : { } } , <EOL> '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:2>' ) } <EOL> req = Request . blank ( path , environ = env , headers = headers ) <EOL> req . method = '<STR_LIT:POST>' <EOL> fake_app = FakeApp ( iter ( [ ( '<STR_LIT>' , { } , '<STR_LIT>' ) ] ) ) <EOL> test_auth = keystoneauth . filter_factory ( { } ) ( fake_app ) <EOL> resp = req . get_response ( test_auth ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> self . assertEqual ( len ( fake_app . call_contexts ) , <NUM_LIT:1> ) <EOL> headers_sent = fake_app . call_contexts [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> self . assertTrue ( '<STR_LIT>' in headers_sent , <EOL> headers_sent ) <EOL> self . assertEqual ( headers_sent [ '<STR_LIT>' ] , <EOL> UNKNOWN_ID ) <EOL> class SwiftAuthMultiple ( SwiftAuth ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . test_auth = keystoneauth . filter_factory ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } ) ( FakeApp ( ) ) <EOL> self . test_auth . logger = FakeLogger ( ) <EOL> class ServiceTokenFunctionality ( unittest . TestCase ) : <EOL> def _make_authed_request ( self , conf , project_id , path , method = '<STR_LIT:GET>' , <EOL> user_role = '<STR_LIT>' , service_role = None , <EOL> environ = None ) : <EOL> """<STR_LIT>""" <EOL> headers = get_identity_headers ( tenant_id = project_id , <EOL> role = user_role , <EOL> service_role = service_role ) <EOL> ( version , account , _junk , _junk ) = split_path ( path , <NUM_LIT:2> , <NUM_LIT:4> , True ) <EOL> _ , info_key = _get_cache_key ( account , None ) <EOL> env = { info_key : { '<STR_LIT:status>' : <NUM_LIT:0> , '<STR_LIT>' : { } } , <EOL> '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:2>' ) } <EOL> if environ : <EOL> env . update ( environ ) <EOL> req = Request . blank ( path , environ = env , headers = headers ) <EOL> req . method = method <EOL> fake_app = FakeApp ( iter ( [ ( '<STR_LIT>' , { } , '<STR_LIT>' ) ] ) ) <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( fake_app ) <EOL> resp = req . get_response ( test_auth ) <EOL> return resp <EOL> def test_existing_swift_owner_ignored ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' , <EOL> environ = { '<STR_LIT>' : False } , <EOL> user_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' , <EOL> environ = { '<STR_LIT>' : True } , <EOL> user_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' , <EOL> environ = { '<STR_LIT>' : False } ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' , <EOL> environ = { '<STR_LIT>' : True } ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_unknown_prefix ( self ) : <EOL> resp = self . _make_authed_request ( { } , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_authed_for_path_single ( self ) : <EOL> resp = self . _make_authed_request ( { } , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' , <EOL> user_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' , <EOL> user_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> def test_denied_for_path_single ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' , <EOL> user_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' , '<STR_LIT>' , <EOL> method = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_authed_for_primary_path_multiple ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> def test_denied_for_second_path_with_only_operator_role ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> user_role = '<STR_LIT>' , service_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> user_role = '<STR_LIT>' , service_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_authed_for_second_path_with_operator_role_and_service ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' , service_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> def test_denied_for_second_path_with_only_service ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' , user_role = '<STR_LIT>' , <EOL> service_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_denied_for_second_path_for_service_user ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' , user_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' , user_role = '<STR_LIT>' , <EOL> service_role = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_delete_denied_for_second_path ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' , service_role = '<STR_LIT>' , <EOL> method = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT> ) <EOL> def test_delete_of_second_path_by_reseller_admin ( self ) : <EOL> resp = self . _make_authed_request ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' , '<STR_LIT>' , user_role = '<STR_LIT>' , <EOL> method = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_int , <NUM_LIT:200> ) <EOL> class BaseTestAuthorize ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . test_auth = keystoneauth . filter_factory ( { } ) ( FakeApp ( ) ) <EOL> self . test_auth . logger = FakeLogger ( ) <EOL> def _make_request ( self , path , ** kwargs ) : <EOL> return Request . blank ( path , ** kwargs ) <EOL> def _get_account ( self , identity = None ) : <EOL> if not identity : <EOL> identity = self . _get_identity ( ) <EOL> return get_account_for_tenant ( self . test_auth , <EOL> identity [ '<STR_LIT>' ] ) <EOL> def _get_identity ( self , tenant_id = '<STR_LIT>' , tenant_name = '<STR_LIT>' , <EOL> user_id = '<STR_LIT>' , user_name = '<STR_LIT>' , roles = None , <EOL> project_domain_name = '<STR_LIT>' , project_domain_id = '<STR_LIT:foo>' , <EOL> user_domain_name = '<STR_LIT>' , user_domain_id = '<STR_LIT:foo>' ) : <EOL> if roles is None : <EOL> roles = [ ] <EOL> if isinstance ( roles , list ) : <EOL> roles = '<STR_LIT:U+002C>' . join ( roles ) <EOL> return { '<STR_LIT>' : user_id , <EOL> '<STR_LIT>' : user_name , <EOL> '<STR_LIT>' : user_domain_name , <EOL> '<STR_LIT>' : user_domain_id , <EOL> '<STR_LIT>' : tenant_id , <EOL> '<STR_LIT>' : tenant_name , <EOL> '<STR_LIT>' : project_domain_id , <EOL> '<STR_LIT>' : project_domain_name , <EOL> '<STR_LIT>' : roles , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> def _get_env_id ( self , tenant_id = '<STR_LIT>' , tenant_name = '<STR_LIT>' , <EOL> user_id = '<STR_LIT>' , user_name = '<STR_LIT>' , roles = [ ] , <EOL> project_domain_name = '<STR_LIT>' , project_domain_id = '<STR_LIT>' , <EOL> user_domain_name = '<STR_LIT>' , user_domain_id = '<STR_LIT>' , <EOL> auth_version = '<STR_LIT:3>' ) : <EOL> env = self . _get_identity ( tenant_id , tenant_name , user_id , user_name , <EOL> roles , project_domain_name , <EOL> project_domain_id , user_domain_name , <EOL> user_domain_id ) <EOL> token_info = _fake_token_info ( version = auth_version ) <EOL> env . update ( { '<STR_LIT>' : token_info } ) <EOL> return self . test_auth . _keystone_identity ( env ) <EOL> class TestAuthorize ( BaseTestAuthorize ) : <EOL> def _check_authenticate ( self , account = None , identity = None , headers = None , <EOL> exception = None , acl = None , env = None , path = None ) : <EOL> if not identity : <EOL> identity = self . _get_identity ( ) <EOL> if not account : <EOL> account = self . _get_account ( identity ) <EOL> if not path : <EOL> path = '<STR_LIT>' % account <EOL> _ , info_key = _get_cache_key ( account , None ) <EOL> default_env = { '<STR_LIT>' : identity [ '<STR_LIT>' ] , <EOL> info_key : { '<STR_LIT:status>' : <NUM_LIT:200> , '<STR_LIT>' : { } } } <EOL> default_env . update ( identity ) <EOL> if env : <EOL> default_env . update ( env ) <EOL> req = self . _make_request ( path , headers = headers , environ = default_env ) <EOL> req . acl = acl <EOL> env_identity = self . test_auth . _keystone_identity ( req . environ ) <EOL> result = self . test_auth . authorize ( env_identity , req ) <EOL> if exception and not result : <EOL> self . fail ( "<STR_LIT>" % ( str ( exception ) ) ) <EOL> elif exception : <EOL> self . assertEqual ( result . status_int , exception ) <EOL> else : <EOL> self . assertTrue ( result is None ) <EOL> return req <EOL> def test_authorize_fails_for_unauthorized_user ( self ) : <EOL> self . _check_authenticate ( exception = HTTP_FORBIDDEN ) <EOL> def test_authorize_fails_for_invalid_reseller_prefix ( self ) : <EOL> self . _check_authenticate ( account = '<STR_LIT>' , <EOL> exception = HTTP_FORBIDDEN ) <EOL> def test_authorize_succeeds_for_reseller_admin ( self ) : <EOL> roles = [ self . test_auth . reseller_admin_role ] <EOL> identity = self . _get_identity ( roles = roles ) <EOL> req = self . _check_authenticate ( identity = identity ) <EOL> self . assertTrue ( req . environ . get ( '<STR_LIT>' ) ) <EOL> def test_authorize_succeeds_for_insensitive_reseller_admin ( self ) : <EOL> roles = [ self . test_auth . reseller_admin_role . upper ( ) ] <EOL> identity = self . _get_identity ( roles = roles ) <EOL> req = self . _check_authenticate ( identity = identity ) <EOL> self . assertTrue ( req . environ . get ( '<STR_LIT>' ) ) <EOL> def test_authorize_succeeds_as_owner_for_operator_role ( self ) : <EOL> roles = operator_roles ( self . test_auth ) <EOL> identity = self . _get_identity ( roles = roles ) <EOL> req = self . _check_authenticate ( identity = identity ) <EOL> self . assertTrue ( req . environ . get ( '<STR_LIT>' ) ) <EOL> def test_authorize_succeeds_as_owner_for_insensitive_operator_role ( self ) : <EOL> roles = [ r . upper ( ) for r in operator_roles ( self . test_auth ) ] <EOL> identity = self . _get_identity ( roles = roles ) <EOL> req = self . _check_authenticate ( identity = identity ) <EOL> self . assertTrue ( req . environ . get ( '<STR_LIT>' ) ) <EOL> def test_authorize_fails_same_user_and_tenant ( self ) : <EOL> self . test_auth . is_admin = True <EOL> identity = self . _get_identity ( user_name = '<STR_LIT>' , <EOL> tenant_name = '<STR_LIT>' ) <EOL> req = self . _check_authenticate ( identity = identity , <EOL> exception = HTTP_FORBIDDEN ) <EOL> self . assertFalse ( bool ( req . environ . get ( '<STR_LIT>' ) ) ) <EOL> def test_authorize_succeeds_for_container_sync ( self ) : <EOL> env = { '<STR_LIT>' : '<STR_LIT:foo>' , '<STR_LIT>' : '<STR_LIT:127.0.0.1>' } <EOL> headers = { '<STR_LIT>' : '<STR_LIT:foo>' , '<STR_LIT>' : '<STR_LIT:1>' } <EOL> self . _check_authenticate ( env = env , headers = headers ) <EOL> def test_authorize_fails_for_invalid_referrer ( self ) : <EOL> env = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _check_authenticate ( acl = '<STR_LIT>' , env = env , <EOL> exception = HTTP_FORBIDDEN ) <EOL> def test_authorize_fails_for_referrer_without_rlistings ( self ) : <EOL> env = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _check_authenticate ( acl = '<STR_LIT>' , env = env , <EOL> exception = HTTP_FORBIDDEN ) <EOL> def test_authorize_succeeds_for_referrer_with_rlistings ( self ) : <EOL> env = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _check_authenticate ( acl = '<STR_LIT>' , env = env ) <EOL> def test_authorize_succeeds_for_referrer_with_obj ( self ) : <EOL> path = '<STR_LIT>' % self . _get_account ( ) <EOL> env = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _check_authenticate ( acl = '<STR_LIT>' , env = env , path = path ) <EOL> def test_authorize_succeeds_for_user_role_in_roles ( self ) : <EOL> acl = '<STR_LIT>' <EOL> identity = self . _get_identity ( roles = [ acl ] ) <EOL> self . _check_authenticate ( identity = identity , acl = acl ) <EOL> def test_authorize_succeeds_for_tenant_name_user_in_roles ( self ) : <EOL> identity = self . _get_identity ( ) <EOL> user_name = identity [ '<STR_LIT>' ] <EOL> user_id = identity [ '<STR_LIT>' ] <EOL> tenant_name = identity [ '<STR_LIT>' ] <EOL> for user in [ user_id , user_name , '<STR_LIT:*>' ] : <EOL> acl = '<STR_LIT>' % ( tenant_name , user ) <EOL> self . _check_authenticate ( identity = identity , acl = acl ) <EOL> def test_authorize_succeeds_for_tenant_id_user_in_roles ( self ) : <EOL> identity = self . _get_identity ( ) <EOL> user_name = identity [ '<STR_LIT>' ] <EOL> user_id = identity [ '<STR_LIT>' ] <EOL> tenant_id = identity [ '<STR_LIT>' ] <EOL> for user in [ user_id , user_name , '<STR_LIT:*>' ] : <EOL> acl = '<STR_LIT>' % ( tenant_id , user ) <EOL> self . _check_authenticate ( identity = identity , acl = acl ) <EOL> def test_authorize_succeeds_for_wildcard_tenant_user_in_roles ( self ) : <EOL> identity = self . _get_identity ( ) <EOL> user_name = identity [ '<STR_LIT>' ] <EOL> user_id = identity [ '<STR_LIT>' ] <EOL> for user in [ user_id , user_name , '<STR_LIT:*>' ] : <EOL> acl = '<STR_LIT>' % user <EOL> self . _check_authenticate ( identity = identity , acl = acl ) <EOL> def test_cross_tenant_authorization_success ( self ) : <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ) <EOL> def test_cross_tenant_authorization_failure ( self ) : <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) , <EOL> None ) <EOL> def test_cross_tenant_authorization_allow_names ( self ) : <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , allow_names = True ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , allow_names = True ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , allow_names = True ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , allow_names = True ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , allow_names = False ) , <EOL> None ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , allow_names = False ) , <EOL> None ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , allow_names = False ) , <EOL> None ) <EOL> self . assertEqual ( <EOL> self . test_auth . _authorize_cross_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , allow_names = False ) , <EOL> '<STR_LIT>' ) <EOL> def test_delete_own_account_not_allowed ( self ) : <EOL> roles = operator_roles ( self . test_auth ) <EOL> identity = self . _get_identity ( roles = roles ) <EOL> account = self . _get_account ( identity ) <EOL> self . _check_authenticate ( account = account , <EOL> identity = identity , <EOL> exception = HTTP_FORBIDDEN , <EOL> path = '<STR_LIT>' + account , <EOL> env = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_delete_own_account_when_reseller_allowed ( self ) : <EOL> roles = [ self . test_auth . reseller_admin_role ] <EOL> identity = self . _get_identity ( roles = roles ) <EOL> account = self . _get_account ( identity ) <EOL> req = self . _check_authenticate ( account = account , <EOL> identity = identity , <EOL> path = '<STR_LIT>' + account , <EOL> env = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( bool ( req . environ . get ( '<STR_LIT>' ) ) , True ) <EOL> def test_identity_set_up_at_call ( self ) : <EOL> def fake_start_response ( * args , ** kwargs ) : <EOL> pass <EOL> the_env = self . _get_identity ( <EOL> tenant_id = '<STR_LIT:test>' , roles = [ '<STR_LIT>' ] ) <EOL> self . test_auth ( the_env , fake_start_response ) <EOL> subreq = Request . blank ( <EOL> '<STR_LIT>' % get_account_for_tenant ( self . test_auth , '<STR_LIT:test>' ) ) <EOL> subreq . environ . update ( <EOL> self . _get_identity ( tenant_id = '<STR_LIT:test>' , roles = [ '<STR_LIT>' ] ) ) <EOL> authorize_resp = the_env [ '<STR_LIT>' ] ( subreq ) <EOL> self . assertEqual ( authorize_resp , None ) <EOL> def test_names_disallowed_in_acls_outside_default_domain ( self ) : <EOL> id = self . _get_identity ( user_domain_id = '<STR_LIT>' , <EOL> project_domain_id = '<STR_LIT>' ) <EOL> env = { '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:3>' ) } <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env , <EOL> exception = HTTP_FORBIDDEN ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env , <EOL> exception = HTTP_FORBIDDEN ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env , <EOL> exception = HTTP_FORBIDDEN ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> def test_names_allowed_in_acls_inside_default_domain ( self ) : <EOL> id = self . _get_identity ( user_domain_id = '<STR_LIT:default>' , <EOL> project_domain_id = '<STR_LIT:default>' ) <EOL> env = { '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:3>' ) } <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> def test_names_allowed_in_acls_inside_default_domain_with_config ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT:yes>' } <EOL> self . test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . test_auth . logger = FakeLogger ( ) <EOL> id = self . _get_identity ( user_domain_id = '<STR_LIT:default>' , <EOL> project_domain_id = '<STR_LIT:default>' ) <EOL> env = { '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:3>' ) } <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> def test_names_disallowed_in_acls_inside_default_domain ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT:false>' } <EOL> self . test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . test_auth . logger = FakeLogger ( ) <EOL> id = self . _get_identity ( user_domain_id = '<STR_LIT:default>' , <EOL> project_domain_id = '<STR_LIT:default>' ) <EOL> env = { '<STR_LIT>' : _fake_token_info ( version = '<STR_LIT:3>' ) } <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env , <EOL> exception = HTTP_FORBIDDEN ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env , <EOL> exception = HTTP_FORBIDDEN ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env , <EOL> exception = HTTP_FORBIDDEN ) <EOL> acl = '<STR_LIT>' % ( id [ '<STR_LIT>' ] , id [ '<STR_LIT>' ] ) <EOL> self . _check_authenticate ( acl = acl , identity = id , env = env ) <EOL> def test_keystone_identity ( self ) : <EOL> user = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> roles = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> service_roles = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> project = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> user_domain = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> project_domain = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> req = Request . blank ( '<STR_LIT>' ) <EOL> data = self . test_auth . _keystone_identity ( req . environ ) <EOL> self . assertIsNone ( data ) <EOL> req . headers . update ( { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' % roles , <EOL> '<STR_LIT>' : user [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : user [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : project [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : project [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : user_domain [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : user_domain [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : project_domain [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : project_domain [ <NUM_LIT:1> ] } ) <EOL> data = self . test_auth . _keystone_identity ( req . environ ) <EOL> self . assertIsNone ( data ) <EOL> req . headers . update ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> expected = { '<STR_LIT:user>' : user , <EOL> '<STR_LIT>' : project , <EOL> '<STR_LIT>' : list ( roles ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : ( None , None ) , <EOL> '<STR_LIT>' : ( None , None ) , <EOL> '<STR_LIT>' : <NUM_LIT:0> } <EOL> data = self . test_auth . _keystone_identity ( req . environ ) <EOL> self . assertEqual ( expected , data ) <EOL> req . environ [ '<STR_LIT>' ] = _fake_token_info ( version = '<STR_LIT:2>' ) <EOL> expected = { '<STR_LIT:user>' : user , <EOL> '<STR_LIT>' : project , <EOL> '<STR_LIT>' : list ( roles ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : ( None , None ) , <EOL> '<STR_LIT>' : ( None , None ) , <EOL> '<STR_LIT>' : <NUM_LIT:2> } <EOL> data = self . test_auth . _keystone_identity ( req . environ ) <EOL> self . assertEqual ( expected , data ) <EOL> req . environ [ '<STR_LIT>' ] = _fake_token_info ( version = '<STR_LIT:3>' ) <EOL> expected = { '<STR_LIT:user>' : user , <EOL> '<STR_LIT>' : project , <EOL> '<STR_LIT>' : list ( roles ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : user_domain , <EOL> '<STR_LIT>' : project_domain , <EOL> '<STR_LIT>' : <NUM_LIT:3> } <EOL> data = self . test_auth . _keystone_identity ( req . environ ) <EOL> self . assertEqual ( expected , data ) <EOL> req . headers . update ( { '<STR_LIT>' : '<STR_LIT>' % service_roles } ) <EOL> expected = { '<STR_LIT:user>' : user , <EOL> '<STR_LIT>' : project , <EOL> '<STR_LIT>' : list ( roles ) , <EOL> '<STR_LIT>' : list ( service_roles ) , <EOL> '<STR_LIT>' : user_domain , <EOL> '<STR_LIT>' : project_domain , <EOL> '<STR_LIT>' : <NUM_LIT:3> } <EOL> data = self . test_auth . _keystone_identity ( req . environ ) <EOL> self . assertEqual ( expected , data ) <EOL> def test_get_project_domain_id ( self ) : <EOL> sysmeta = { } <EOL> info = { '<STR_LIT>' : sysmeta } <EOL> _ , info_key = _get_cache_key ( '<STR_LIT>' , None ) <EOL> env = { '<STR_LIT>' : '<STR_LIT>' , <EOL> info_key : info } <EOL> info [ '<STR_LIT:status>' ] = <NUM_LIT> <EOL> self . assertEqual ( self . test_auth . _get_project_domain_id ( env ) , <EOL> ( False , None ) ) <EOL> info [ '<STR_LIT:status>' ] = <NUM_LIT:0> <EOL> self . assertEqual ( self . test_auth . _get_project_domain_id ( env ) , <EOL> ( False , None ) ) <EOL> info [ '<STR_LIT:status>' ] = <NUM_LIT:200> <EOL> self . assertEqual ( self . test_auth . _get_project_domain_id ( env ) , <EOL> ( True , None ) ) <EOL> sysmeta [ '<STR_LIT>' ] = '<STR_LIT:default>' <EOL> self . assertEqual ( self . test_auth . _get_project_domain_id ( env ) , <EOL> ( True , '<STR_LIT:default>' ) ) <EOL> class TestIsNameAllowedInACL ( BaseTestAuthorize ) : <EOL> def setUp ( self ) : <EOL> super ( TestIsNameAllowedInACL , self ) . setUp ( ) <EOL> self . default_id = '<STR_LIT:default>' <EOL> def _assert_names_allowed ( self , expected , user_domain_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = None , <EOL> scoped = '<STR_LIT>' ) : <EOL> project_name = '<STR_LIT:foo>' <EOL> account_id = '<STR_LIT>' <EOL> account = get_account_for_tenant ( self . test_auth , account_id ) <EOL> parts = ( '<STR_LIT>' , account , None , None ) <EOL> path = '<STR_LIT>' % parts [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> sysmeta = { } <EOL> if sysmeta_project_domain_id : <EOL> sysmeta = { '<STR_LIT>' : sysmeta_project_domain_id } <EOL> info = { '<STR_LIT:status>' : <NUM_LIT:200> , '<STR_LIT>' : sysmeta } <EOL> _ , info_key = _get_cache_key ( account , None ) <EOL> req = Request . blank ( path , environ = { info_key : info } ) <EOL> if scoped == '<STR_LIT>' : <EOL> project_name = '<STR_LIT>' <EOL> project_id = account_id <EOL> elif scoped == '<STR_LIT>' : <EOL> project_name = '<STR_LIT>' <EOL> project_id = '<STR_LIT>' <EOL> else : <EOL> project_name , project_id , req_project_domain_id = None , None , None <EOL> if user_domain_id : <EOL> id = self . _get_env_id ( tenant_name = project_name , <EOL> tenant_id = project_id , <EOL> user_domain_id = user_domain_id , <EOL> project_domain_id = req_project_domain_id ) <EOL> else : <EOL> id = self . _get_env_id ( tenant_name = project_name , <EOL> tenant_id = project_id , <EOL> auth_version = '<STR_LIT:2>' ) <EOL> actual = self . test_auth . _is_name_allowed_in_acl ( req , parts , id ) <EOL> self . assertEqual ( actual , expected , '<STR_LIT>' <EOL> % ( user_domain_id , req_project_domain_id , <EOL> sysmeta_project_domain_id , scoped ) ) <EOL> def test_is_name_allowed_in_acl_with_token_scoped_to_tenant ( self ) : <EOL> self . _assert_names_allowed ( True , user_domain_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = None ) <EOL> self . _assert_names_allowed ( True , user_domain_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = self . default_id ) <EOL> self . _assert_names_allowed ( True , user_domain_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = UNKNOWN_ID ) <EOL> self . _assert_names_allowed ( True , user_domain_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = '<STR_LIT:foo>' ) <EOL> self . _assert_names_allowed ( True , user_domain_id = self . default_id , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = None ) <EOL> self . _assert_names_allowed ( True , user_domain_id = self . default_id , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = UNKNOWN_ID ) <EOL> self . _assert_names_allowed ( True , user_domain_id = self . default_id , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = '<STR_LIT:bar>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = None ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = self . default_id ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = '<STR_LIT:foo>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = None ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = self . default_id ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = UNKNOWN_ID ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = '<STR_LIT:foo>' ) <EOL> def test_is_name_allowed_in_acl_with_unscoped_token ( self ) : <EOL> self . _assert_names_allowed ( True , user_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = None , <EOL> scoped = False ) <EOL> self . _assert_names_allowed ( True , user_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = self . default_id , <EOL> scoped = False ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = UNKNOWN_ID , <EOL> scoped = False ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = '<STR_LIT:foo>' , <EOL> scoped = False ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = None , <EOL> scoped = False ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = self . default_id , <EOL> scoped = False ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = UNKNOWN_ID , <EOL> scoped = False ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = '<STR_LIT:foo>' , <EOL> scoped = False ) <EOL> def test_is_name_allowed_in_acl_with_token_scoped_to_other_tenant ( self ) : <EOL> self . _assert_names_allowed ( True , user_domain_id = self . default_id , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = None , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( True , user_domain_id = self . default_id , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = self . default_id , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = UNKNOWN_ID , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = '<STR_LIT:foo>' , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = None , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = self . default_id , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = UNKNOWN_ID , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = self . default_id , <EOL> req_project_domain_id = '<STR_LIT:foo>' , <EOL> sysmeta_project_domain_id = '<STR_LIT:foo>' , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = None , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = self . default_id , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = UNKNOWN_ID , <EOL> scoped = '<STR_LIT>' ) <EOL> self . _assert_names_allowed ( False , user_domain_id = '<STR_LIT:foo>' , <EOL> req_project_domain_id = self . default_id , <EOL> sysmeta_project_domain_id = '<STR_LIT:foo>' , <EOL> scoped = '<STR_LIT>' ) <EOL> class TestIsNameAllowedInACLWithConfiguredDomain ( TestIsNameAllowedInACL ) : <EOL> def setUp ( self ) : <EOL> super ( TestIsNameAllowedInACLWithConfiguredDomain , self ) . setUp ( ) <EOL> conf = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . test_auth . logger = FakeLogger ( ) <EOL> self . default_id = '<STR_LIT>' <EOL> class TestSetProjectDomain ( BaseTestAuthorize ) : <EOL> def _assert_set_project_domain ( self , expected , account , req_project_id , <EOL> req_project_domain_id , <EOL> sysmeta_project_domain_id , <EOL> warning = False ) : <EOL> hdr = '<STR_LIT>' <EOL> status = <NUM_LIT:0> if sysmeta_project_domain_id is None else <NUM_LIT:200> <EOL> sysmeta = { } <EOL> if sysmeta_project_domain_id : <EOL> sysmeta [ '<STR_LIT>' ] = sysmeta_project_domain_id <EOL> info = { '<STR_LIT:status>' : status , '<STR_LIT>' : sysmeta } <EOL> _ , info_key = _get_cache_key ( account , None ) <EOL> env = { info_key : info } <EOL> env_id = self . _get_env_id ( tenant_id = req_project_id , <EOL> project_domain_id = req_project_domain_id ) <EOL> self . test_auth . logger = FakeLogger ( ) <EOL> num_warnings = <NUM_LIT:0> <EOL> path = '<STR_LIT>' % account <EOL> for method in [ '<STR_LIT>' , '<STR_LIT:POST>' ] : <EOL> req = Request . blank ( path , environ = env ) <EOL> req . method = method <EOL> path_parts = req . split_path ( <NUM_LIT:1> , <NUM_LIT:4> , True ) <EOL> self . test_auth . _set_project_domain_id ( req , path_parts , env_id ) <EOL> if warning : <EOL> num_warnings += <NUM_LIT:1> <EOL> warnings = self . test_auth . logger . get_lines_for_level ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( warnings ) , num_warnings ) <EOL> self . assertTrue ( warnings [ - <NUM_LIT:1> ] . startswith ( '<STR_LIT>' ) ) <EOL> if expected is not None : <EOL> self . assertTrue ( hdr in req . headers ) <EOL> self . assertEqual ( req . headers [ hdr ] , expected ) <EOL> else : <EOL> self . assertFalse ( hdr in req . headers , req . headers ) <EOL> for method in [ '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> req = Request . blank ( path , environ = env ) <EOL> req . method = method <EOL> self . test_auth . _set_project_domain_id ( req , path_parts , env_id ) <EOL> self . assertFalse ( hdr in req . headers ) <EOL> path = '<STR_LIT>' % account <EOL> for method in [ '<STR_LIT>' ] : <EOL> req = Request . blank ( path , environ = env ) <EOL> req . method = method <EOL> path_parts = req . split_path ( <NUM_LIT:1> , <NUM_LIT:4> , True ) <EOL> self . test_auth . _set_project_domain_id ( req , path_parts , env_id ) <EOL> if warning : <EOL> num_warnings += <NUM_LIT:1> <EOL> warnings = self . test_auth . logger . get_lines_for_level ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( warnings ) , num_warnings ) <EOL> self . assertTrue ( warnings [ - <NUM_LIT:1> ] . startswith ( '<STR_LIT>' ) ) <EOL> if expected is not None : <EOL> self . assertTrue ( hdr in req . headers ) <EOL> self . assertEqual ( req . headers [ hdr ] , expected ) <EOL> else : <EOL> self . assertFalse ( hdr in req . headers ) <EOL> for method in [ '<STR_LIT:POST>' , '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> req = Request . blank ( path , environ = env ) <EOL> req . method = method <EOL> self . test_auth . _set_project_domain_id ( req , path_parts , env_id ) <EOL> self . assertFalse ( hdr in req . headers ) <EOL> path = '<STR_LIT>' % account <EOL> for method in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:POST>' , '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] : <EOL> req = Request . blank ( path , environ = env ) <EOL> req . method = method <EOL> path_parts = req . split_path ( <NUM_LIT:1> , <NUM_LIT:4> , True ) <EOL> self . test_auth . _set_project_domain_id ( req , path_parts , env_id ) <EOL> self . assertFalse ( hdr in req . headers ) <EOL> def test_set_project_domain_id_new_account ( self ) : <EOL> self . _assert_set_project_domain ( '<STR_LIT>' , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT>' , <EOL> sysmeta_project_domain_id = None ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = None ) <EOL> self . _assert_set_project_domain ( UNKNOWN_ID , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = None ) <EOL> self . _assert_set_project_domain ( UNKNOWN_ID , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT:default>' , <EOL> sysmeta_project_domain_id = None ) <EOL> def test_set_project_domain_id_existing_v2_account ( self ) : <EOL> self . _assert_set_project_domain ( '<STR_LIT:default>' , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT:default>' , <EOL> sysmeta_project_domain_id = '<STR_LIT>' ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT>' , <EOL> sysmeta_project_domain_id = '<STR_LIT>' , <EOL> warning = True ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = '<STR_LIT>' ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = '<STR_LIT>' ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = '<STR_LIT>' ) <EOL> def test_set_project_domain_id_existing_account_unknown_domain ( self ) : <EOL> self . _assert_set_project_domain ( '<STR_LIT>' , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT>' , <EOL> sysmeta_project_domain_id = UNKNOWN_ID ) <EOL> self . _assert_set_project_domain ( '<STR_LIT>' , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = UNKNOWN_ID ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = UNKNOWN_ID ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT:default>' , <EOL> sysmeta_project_domain_id = UNKNOWN_ID ) <EOL> def test_set_project_domain_id_existing_known_domain ( self ) : <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT>' , <EOL> sysmeta_project_domain_id = '<STR_LIT>' ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT>' , <EOL> sysmeta_project_domain_id = '<STR_LIT>' , <EOL> warning = True ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = '<STR_LIT>' ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = None , <EOL> req_project_domain_id = None , <EOL> sysmeta_project_domain_id = '<STR_LIT>' ) <EOL> self . _assert_set_project_domain ( None , <EOL> account = '<STR_LIT>' , <EOL> req_project_id = '<STR_LIT>' , <EOL> req_project_domain_id = '<STR_LIT:default>' , <EOL> sysmeta_project_domain_id = '<STR_LIT>' ) <EOL> class ResellerInInfo ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . default_rules = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ ] } <EOL> def test_defaults ( self ) : <EOL> test_auth = keystoneauth . filter_factory ( { } ) ( FakeApp ( ) ) <EOL> self . assertEqual ( test_auth . account_rules [ '<STR_LIT>' ] , self . default_rules ) <EOL> def test_multiple ( self ) : <EOL> conf = { "<STR_LIT>" : "<STR_LIT>" } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . assertEqual ( test_auth . account_rules [ '<STR_LIT>' ] , self . default_rules ) <EOL> self . assertEqual ( test_auth . account_rules [ '<STR_LIT>' ] , self . default_rules ) <EOL> self . assertEqual ( test_auth . account_rules [ '<STR_LIT>' ] , self . default_rules ) <EOL> class PrefixAccount ( unittest . TestCase ) : <EOL> def test_default ( self ) : <EOL> conf = { } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . assertEqual ( get_account_for_tenant ( test_auth , <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , None ) <EOL> self . assertTrue ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_same_as_default ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT>' } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . assertEqual ( get_account_for_tenant ( test_auth , <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , None ) <EOL> self . assertTrue ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_blank_reseller ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT>' } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . assertEqual ( get_account_for_tenant ( test_auth , <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertTrue ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_multiple_resellers ( self ) : <EOL> conf = { '<STR_LIT>' : '<STR_LIT>' } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . assertEqual ( get_account_for_tenant ( test_auth , <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , None ) <EOL> self . assertTrue ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertTrue ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def test_blank_plus_other_reseller ( self ) : <EOL> conf = { '<STR_LIT>' : "<STR_LIT>" } <EOL> test_auth = keystoneauth . filter_factory ( conf ) ( FakeApp ( ) ) <EOL> self . assertEqual ( get_account_for_tenant ( test_auth , <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( <EOL> '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( test_auth . _get_account_prefix ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertTrue ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertTrue ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . assertFalse ( test_auth . _account_matches_tenant ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
