<s> import six <EOL> from nova . api . openstack . compute . legacy_v2 . contrib import used_limits as used_limits_v2 <EOL> from nova . api . openstack . compute import used_limits as used_limits_v21 <EOL> from nova . api . openstack import extensions <EOL> from nova . api . openstack import wsgi <EOL> import nova . context <EOL> from nova import exception <EOL> from nova import quota <EOL> from nova import test <EOL> class FakeRequest ( object ) : <EOL> def __init__ ( self , context , reserved = False ) : <EOL> self . environ = { '<STR_LIT>' : context } <EOL> self . reserved = reserved <EOL> self . GET = { '<STR_LIT>' : <NUM_LIT:1> } if reserved else { } <EOL> class UsedLimitsTestCaseV21 ( test . NoDBTestCase ) : <EOL> used_limit_extension = "<STR_LIT>" <EOL> include_server_group_quotas = True <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( UsedLimitsTestCaseV21 , self ) . setUp ( ) <EOL> self . _set_up_controller ( ) <EOL> self . fake_context = nova . context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def _set_up_controller ( self ) : <EOL> self . ext_mgr = None <EOL> self . controller = used_limits_v21 . UsedLimitsController ( ) <EOL> self . mox . StubOutWithMock ( used_limits_v21 , '<STR_LIT>' ) <EOL> self . authorize = used_limits_v21 . authorize <EOL> def _do_test_used_limits ( self , reserved ) : <EOL> fake_req = FakeRequest ( self . fake_context , reserved = reserved ) <EOL> obj = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> res = wsgi . ResponseObject ( obj ) <EOL> quota_map = { <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> limits = { } <EOL> expected_abs_limits = [ ] <EOL> for display_name , q in six . iteritems ( quota_map ) : <EOL> limits [ q ] = { '<STR_LIT>' : len ( display_name ) , <EOL> '<STR_LIT>' : len ( display_name ) / <NUM_LIT:2> , <EOL> '<STR_LIT>' : len ( display_name ) / <NUM_LIT:3> } <EOL> if ( self . include_server_group_quotas or <EOL> display_name != '<STR_LIT>' ) : <EOL> expected_abs_limits . append ( display_name ) <EOL> def stub_get_project_quotas ( context , project_id , usages = True ) : <EOL> return limits <EOL> self . stubs . Set ( quota . QUOTAS , "<STR_LIT>" , <EOL> stub_get_project_quotas ) <EOL> if self . ext_mgr is not None : <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( False ) <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( <EOL> self . include_server_group_quotas ) <EOL> self . mox . ReplayAll ( ) <EOL> self . controller . index ( fake_req , res ) <EOL> abs_limits = res . obj [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for limit in expected_abs_limits : <EOL> value = abs_limits [ limit ] <EOL> r = limits [ quota_map [ limit ] ] [ '<STR_LIT>' ] if reserved else <NUM_LIT:0> <EOL> self . assertEqual ( limits [ quota_map [ limit ] ] [ '<STR_LIT>' ] + r , value ) <EOL> def test_used_limits_basic ( self ) : <EOL> self . _do_test_used_limits ( False ) <EOL> def test_used_limits_with_reserved ( self ) : <EOL> self . _do_test_used_limits ( True ) <EOL> def test_admin_can_fetch_limits_for_a_given_tenant_id ( self ) : <EOL> project_id = "<STR_LIT>" <EOL> user_id = "<STR_LIT>" <EOL> tenant_id = '<STR_LIT>' <EOL> self . fake_context . project_id = project_id <EOL> self . fake_context . user_id = user_id <EOL> obj = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> target = { <EOL> "<STR_LIT>" : tenant_id , <EOL> "<STR_LIT>" : user_id <EOL> } <EOL> fake_req = FakeRequest ( self . fake_context ) <EOL> fake_req . GET = { '<STR_LIT>' : tenant_id } <EOL> if self . ext_mgr is not None : <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( True ) <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( <EOL> self . include_server_group_quotas ) <EOL> self . authorize ( self . fake_context , target = target ) <EOL> self . mox . StubOutWithMock ( quota . QUOTAS , '<STR_LIT>' ) <EOL> quota . QUOTAS . get_project_quotas ( self . fake_context , '<STR_LIT:%s>' % tenant_id , <EOL> usages = True ) . AndReturn ( { } ) <EOL> self . mox . ReplayAll ( ) <EOL> res = wsgi . ResponseObject ( obj ) <EOL> self . controller . index ( fake_req , res ) <EOL> def test_admin_can_fetch_used_limits_for_own_project ( self ) : <EOL> project_id = "<STR_LIT>" <EOL> user_id = "<STR_LIT>" <EOL> self . fake_context . project_id = project_id <EOL> self . fake_context . user_id = user_id <EOL> obj = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> fake_req = FakeRequest ( self . fake_context ) <EOL> fake_req . GET = { } <EOL> if self . ext_mgr is not None : <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( True ) <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( <EOL> self . include_server_group_quotas ) <EOL> self . mox . StubOutWithMock ( extensions , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( quota . QUOTAS , '<STR_LIT>' ) <EOL> quota . QUOTAS . get_project_quotas ( self . fake_context , '<STR_LIT:%s>' % project_id , <EOL> usages = True ) . AndReturn ( { } ) <EOL> self . mox . ReplayAll ( ) <EOL> res = wsgi . ResponseObject ( obj ) <EOL> self . controller . index ( fake_req , res ) <EOL> def test_non_admin_cannot_fetch_used_limits_for_any_other_project ( self ) : <EOL> project_id = "<STR_LIT>" <EOL> user_id = "<STR_LIT>" <EOL> tenant_id = "<STR_LIT>" <EOL> self . fake_context . project_id = project_id <EOL> self . fake_context . user_id = user_id <EOL> obj = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> target = { <EOL> "<STR_LIT>" : tenant_id , <EOL> "<STR_LIT>" : user_id <EOL> } <EOL> fake_req = FakeRequest ( self . fake_context ) <EOL> fake_req . GET = { '<STR_LIT>' : tenant_id } <EOL> if self . ext_mgr is not None : <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( True ) <EOL> self . authorize ( self . fake_context , target = target ) . AndRaise ( exception . PolicyNotAuthorized ( <EOL> action = self . used_limit_extension ) ) <EOL> self . mox . ReplayAll ( ) <EOL> res = wsgi . ResponseObject ( obj ) <EOL> self . assertRaises ( exception . PolicyNotAuthorized , self . controller . index , <EOL> fake_req , res ) <EOL> def test_used_limits_fetched_for_context_project_id ( self ) : <EOL> project_id = "<STR_LIT>" <EOL> self . fake_context . project_id = project_id <EOL> obj = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> fake_req = FakeRequest ( self . fake_context ) <EOL> if self . ext_mgr is not None : <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( False ) <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( <EOL> self . include_server_group_quotas ) <EOL> self . mox . StubOutWithMock ( quota . QUOTAS , '<STR_LIT>' ) <EOL> quota . QUOTAS . get_project_quotas ( self . fake_context , project_id , <EOL> usages = True ) . AndReturn ( { } ) <EOL> self . mox . ReplayAll ( ) <EOL> res = wsgi . ResponseObject ( obj ) <EOL> self . controller . index ( fake_req , res ) <EOL> def test_used_ram_added ( self ) : <EOL> fake_req = FakeRequest ( self . fake_context ) <EOL> obj = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> } , <EOL> } , <EOL> } <EOL> res = wsgi . ResponseObject ( obj ) <EOL> def stub_get_project_quotas ( context , project_id , usages = True ) : <EOL> return { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } } <EOL> if self . ext_mgr is not None : <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( False ) <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( <EOL> self . include_server_group_quotas ) <EOL> self . stubs . Set ( quota . QUOTAS , "<STR_LIT>" , <EOL> stub_get_project_quotas ) <EOL> self . mox . ReplayAll ( ) <EOL> self . controller . index ( fake_req , res ) <EOL> abs_limits = res . obj [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertIn ( '<STR_LIT>' , abs_limits ) <EOL> self . assertEqual ( <NUM_LIT> , abs_limits [ '<STR_LIT>' ] ) <EOL> def test_no_ram_quota ( self ) : <EOL> fake_req = FakeRequest ( self . fake_context ) <EOL> obj = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : { } , <EOL> } , <EOL> } <EOL> res = wsgi . ResponseObject ( obj ) <EOL> def stub_get_project_quotas ( context , project_id , usages = True ) : <EOL> return { } <EOL> if self . ext_mgr is not None : <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( False ) <EOL> self . ext_mgr . is_loaded ( '<STR_LIT>' ) . AndReturn ( <EOL> self . include_server_group_quotas ) <EOL> self . stubs . Set ( quota . QUOTAS , "<STR_LIT>" , <EOL> stub_get_project_quotas ) <EOL> self . mox . ReplayAll ( ) <EOL> self . controller . index ( fake_req , res ) <EOL> abs_limits = res . obj [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertNotIn ( '<STR_LIT>' , abs_limits ) <EOL> class UsedLimitsTestCaseV2 ( UsedLimitsTestCaseV21 ) : <EOL> used_limit_extension = "<STR_LIT>" <EOL> def _set_up_controller ( self ) : <EOL> self . ext_mgr = self . mox . CreateMock ( extensions . ExtensionManager ) <EOL> self . controller = used_limits_v2 . UsedLimitsController ( self . ext_mgr ) <EOL> self . mox . StubOutWithMock ( used_limits_v2 , '<STR_LIT>' ) <EOL> self . authorize = used_limits_v2 . authorize_for_admin <EOL> class UsedLimitsTestCaseV2WithoutServerGroupQuotas ( UsedLimitsTestCaseV2 ) : <EOL> used_limit_extension = "<STR_LIT>" <EOL> include_server_group_quotas = False </s>
