<s> import datetime <EOL> import json <EOL> from django . db . models import Count <EOL> from django . db . models import FieldDoesNotExist <EOL> from django . db import transaction <EOL> import mox <EOL> from stacktach import dbapi <EOL> from stacktach import models <EOL> from stacktach import utils as stacktach_utils <EOL> from tests . unit import StacktachBaseTestCase <EOL> import utils <EOL> from utils import INSTANCE_ID_1 <EOL> from utils import MESSAGE_ID_1 <EOL> from utils import MESSAGE_ID_2 <EOL> from utils import MESSAGE_ID_3 <EOL> from utils import MESSAGE_ID_4 <EOL> class Length ( mox . Comparator ) : <EOL> def __init__ ( self , l ) : <EOL> self . _len = l <EOL> def equals ( self , rhs ) : <EOL> return self . _len == len ( rhs ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % self . _len <EOL> class DBAPITestCase ( StacktachBaseTestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> dne_exception = models . InstanceExists . DoesNotExist <EOL> mor_exception = models . InstanceExists . MultipleObjectsReturned <EOL> self . mox . StubOutWithMock ( models , '<STR_LIT>' , <EOL> use_mock_anything = True ) <EOL> self . mox . StubOutWithMock ( models , '<STR_LIT>' , <EOL> use_mock_anything = True ) <EOL> self . mox . StubOutWithMock ( models , '<STR_LIT>' , <EOL> use_mock_anything = True ) <EOL> models . RawData . objects = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . _meta = self . mox . CreateMockAnything ( ) <EOL> models . ImageExists . _meta = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects = self . mox . CreateMockAnything ( ) <EOL> models . ImageExists . objects = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . DoesNotExist = dne_exception <EOL> models . ImageExists . DoesNotExist = dne_exception <EOL> models . InstanceExists . MultipleObjectsReturned = mor_exception <EOL> models . ImageExists . MultipleObjectsReturned = mor_exception <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> def make_fake_model ( self ) : <EOL> fake_model = self . mox . CreateMockAnything ( ) <EOL> fake_meta = self . mox . CreateMockAnything ( ) <EOL> fake_model . _meta = fake_meta <EOL> fake_orm = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects = fake_orm <EOL> return fake_model <EOL> def test_get_filter_args ( self ) : <EOL> start_time = datetime . datetime . utcnow ( ) <EOL> start_decimal = utils . decimal_utc ( start_time ) <EOL> end_time = start_time + datetime . timedelta ( days = <NUM_LIT:1> ) <EOL> end_decimal = utils . decimal_utc ( end_time ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_model . _meta . get_field_by_name ( '<STR_LIT>' ) <EOL> fake_model . _meta . get_field_by_name ( '<STR_LIT>' ) <EOL> fake_request . GET = { '<STR_LIT>' : INSTANCE_ID_1 , <EOL> '<STR_LIT>' : str ( start_time ) , <EOL> '<STR_LIT>' : str ( end_time ) } <EOL> self . mox . ReplayAll ( ) <EOL> filter_args = dbapi . _get_filter_args ( fake_model , fake_request ) <EOL> self . mox . VerifyAll ( ) <EOL> self . assertEquals ( filter_args [ '<STR_LIT>' ] , INSTANCE_ID_1 ) <EOL> self . assertEquals ( filter_args . get ( '<STR_LIT>' ) , <EOL> start_decimal ) <EOL> self . assertEquals ( filter_args . get ( '<STR_LIT>' ) , <EOL> end_decimal ) <EOL> def test_get_filter_args_bad_uuid ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( dbapi . BadRequestException , dbapi . _get_filter_args , <EOL> fake_model , fake_request ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_filter_args_bad_min_value ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_model . _meta . get_field_by_name ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( dbapi . BadRequestException , dbapi . _get_filter_args , <EOL> fake_model , fake_request ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_filter_args_bad_max_value ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_model . _meta . get_field_by_name ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( dbapi . BadRequestException , dbapi . _get_filter_args , <EOL> fake_model , fake_request ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_filter_args_bad_range_key ( self ) : <EOL> start_time = datetime . datetime . utcnow ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : str ( start_time ) } <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_model . _meta . get_field_by_name ( '<STR_LIT>' ) . AndRaise ( FieldDoesNotExist ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( dbapi . BadRequestException , dbapi . _get_filter_args , <EOL> fake_model , fake_request ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = None ) . AndReturn ( { } ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects . all ( ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT>' ) . AndReturn ( result ) <EOL> s = slice ( None , dbapi . DEFAULT_LIMIT , None ) <EOL> result . __getitem__ ( s ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects_desc ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = None ) . AndReturn ( { } ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects . all ( ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT>' ) . AndReturn ( result ) <EOL> s = slice ( None , dbapi . DEFAULT_LIMIT , None ) <EOL> result . __getitem__ ( s ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects_asc ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = None ) . AndReturn ( { } ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects . all ( ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT:id>' ) . AndReturn ( result ) <EOL> s = slice ( None , dbapi . DEFAULT_LIMIT , None ) <EOL> result . __getitem__ ( s ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects_limit ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT:1>' } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = None ) . AndReturn ( { } ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects . all ( ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT>' ) . AndReturn ( result ) <EOL> result . __getitem__ ( slice ( None , <NUM_LIT:1> , None ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects_hard_limit ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : str ( dbapi . HARD_LIMIT + <NUM_LIT:1> ) } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = None ) . AndReturn ( { } ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects . all ( ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT>' ) . AndReturn ( result ) <EOL> s = slice ( None , dbapi . HARD_LIMIT , None ) <EOL> result . __getitem__ ( s ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects_offset ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT:1>' } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = None ) . AndReturn ( { } ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects . all ( ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT>' ) . AndReturn ( result ) <EOL> result . __getslice__ ( <NUM_LIT:1> , dbapi . DEFAULT_LIMIT + <NUM_LIT:1> ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects_offset_and_limit ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT:2>' , '<STR_LIT>' : '<STR_LIT:2>' } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = None ) . AndReturn ( { } ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects . all ( ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT>' ) . AndReturn ( result ) <EOL> result . __getslice__ ( <NUM_LIT:2> , <NUM_LIT:4> ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects_with_filter ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> filters = { '<STR_LIT>' : INSTANCE_ID_1 } <EOL> fake_request . GET = filters <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = None ) . AndReturn ( filters ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> fake_model . objects . filter ( ** filters ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT>' ) . AndReturn ( result ) <EOL> s = slice ( None , dbapi . DEFAULT_LIMIT , None ) <EOL> result . __getitem__ ( s ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_db_objects_with_custom_filter ( self ) : <EOL> fake_model = self . make_fake_model ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> filters = { '<STR_LIT>' : INSTANCE_ID_1 } <EOL> custom_filters = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> } } <EOL> fake_request . GET = filters <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_filter_args ( fake_model , fake_request , <EOL> custom_filters = custom_filters ) . AndReturn ( <EOL> filters ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _check_has_field ( fake_model , '<STR_LIT:id>' ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> all_filters = { } <EOL> all_filters . update ( filters ) <EOL> all_filters . update ( custom_filters [ '<STR_LIT>' ] ) <EOL> fake_model . objects . filter ( ** all_filters ) . AndReturn ( result ) <EOL> result . order_by ( '<STR_LIT>' ) . AndReturn ( result ) <EOL> s = slice ( None , dbapi . DEFAULT_LIMIT , None ) <EOL> result . __getitem__ ( s ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> query_result = dbapi . get_db_objects ( fake_model , fake_request , '<STR_LIT:id>' , <EOL> custom_filters = custom_filters ) <EOL> self . assertEquals ( query_result , result ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_exists_no_custom_filters_for_nova ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> objects = self . mox . CreateMockAnything ( ) <EOL> dbapi . get_db_objects ( models . InstanceExists , fake_request , '<STR_LIT:id>' , <EOL> custom_filters = { } ) . AndReturn ( objects ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( objects , dbapi . _exists_extra_values ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_exists ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_exists_no_custom_filters_for_glance ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> objects = self . mox . CreateMockAnything ( ) <EOL> dbapi . get_db_objects ( models . ImageExists , fake_request , '<STR_LIT:id>' , <EOL> custom_filters = { } ) . AndReturn ( objects ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( objects , dbapi . _exists_extra_values ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_exists_glance ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_exists_with_received_min ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> date = str ( datetime . datetime . utcnow ( ) ) <EOL> fake_request . GET = { '<STR_LIT>' : date } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> unix_date = stacktach_utils . str_time_to_unix ( date ) <EOL> custom_filters = { '<STR_LIT>' : { '<STR_LIT>' : unix_date } } <EOL> objects = self . mox . CreateMockAnything ( ) <EOL> dbapi . get_db_objects ( models . InstanceExists , fake_request , '<STR_LIT:id>' , <EOL> custom_filters = custom_filters ) . AndReturn ( objects ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( objects , dbapi . _exists_extra_values ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_exists ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_exists_with_received_max ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> date = str ( datetime . datetime . utcnow ( ) ) <EOL> fake_request . GET = { '<STR_LIT>' : date } <EOL> unix_date = stacktach_utils . str_time_to_unix ( date ) <EOL> custom_filters = { '<STR_LIT>' : { '<STR_LIT>' : unix_date } } <EOL> objects = self . mox . CreateMockAnything ( ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . get_db_objects ( models . InstanceExists , fake_request , '<STR_LIT:id>' , <EOL> custom_filters = custom_filters ) . AndReturn ( objects ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( objects , dbapi . _exists_extra_values ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_exists ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_exists_with_bad_received_min ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_exists ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_exists_with_bad_received_max ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_exists ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_update_tenant_info ( self ) : <EOL> TEST_TENANT = '<STR_LIT:test>' <EOL> models . TenantInfo . objects = self . mox . CreateMockAnything ( ) <EOL> models . TenantType . objects = self . mox . CreateMockAnything ( ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> body_dict = dict ( tenant = TEST_TENANT , <EOL> name = '<STR_LIT>' , <EOL> types = dict ( test_type = '<STR_LIT>' ) ) <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> info = self . mox . CreateMockAnything ( ) <EOL> info_result = self . mox . CreateMockAnything ( ) <EOL> models . TenantInfo . objects . select_for_update ( ) . AndReturn ( info_result ) <EOL> info_result . get ( tenant = TEST_TENANT ) . AndReturn ( info ) <EOL> info . save ( ) <EOL> ttype = self . mox . CreateMockAnything ( ) <EOL> models . TenantType . objects . get ( name = '<STR_LIT>' , value = '<STR_LIT>' ) . AndReturn ( ttype ) <EOL> ttype . __hash__ ( ) . AndReturn ( hash ( '<STR_LIT>' ) ) <EOL> info . save ( ) <EOL> self . mox . ReplayAll ( ) <EOL> dbapi . update_tenant_info ( fake_request , TEST_TENANT ) <EOL> self . assertEqual ( info . name , '<STR_LIT>' ) <EOL> self . assertEqual ( info . types , [ ttype ] ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_batch_update_tenant_info ( self ) : <EOL> TEST_DATE = '<STR_LIT>' <EOL> mock_t1 = self . mox . CreateMock ( models . TenantInfo ) <EOL> mock_t1 . id = <NUM_LIT:1> <EOL> mock_t1 . tenant = '<STR_LIT>' <EOL> mock_t1 . name = '<STR_LIT>' <EOL> mock_t1 . types = self . mox . CreateMockAnything ( ) <EOL> mock_t1 . types . all ( ) . AndReturn ( [ ] ) <EOL> mock_t1 . last_updated = TEST_DATE <EOL> mock_t2 = self . mox . CreateMock ( models . TenantInfo ) <EOL> mock_t2 . id = <NUM_LIT:2> <EOL> mock_t2 . tenant = '<STR_LIT>' <EOL> mock_t2 . name = '<STR_LIT>' <EOL> mock_t2 . last_updated = TEST_DATE <EOL> mock_t2 . types = self . mox . CreateMockAnything ( ) <EOL> mock_t2 . types . all ( ) . AndReturn ( [ ] ) <EOL> TEST_OBJECTS = [ mock_t1 , mock_t2 ] <EOL> mock_tt1 = self . mox . CreateMock ( models . TenantType ) <EOL> mock_tt1 . id = <NUM_LIT:1> <EOL> mock_tt1 . name = '<STR_LIT>' <EOL> mock_tt1 . value = '<STR_LIT>' <EOL> mock_tt2 = self . mox . CreateMock ( models . TenantType ) <EOL> mock_tt2 . id = <NUM_LIT:2> <EOL> mock_tt2 . name = '<STR_LIT>' <EOL> mock_tt2 . value = '<STR_LIT>' <EOL> TEST_TYPES = [ mock_tt1 , mock_tt2 ] <EOL> models . TenantInfo . objects = self . mox . CreateMockAnything ( ) <EOL> models . TenantType . objects = self . mox . CreateMockAnything ( ) <EOL> TypeXref = models . TenantInfo . types . through <EOL> TypeXref . objects = self . mox . CreateMockAnything ( ) <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . datetime . utcnow ( ) . AndReturn ( TEST_DATE ) <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> body_dict = dict ( tenants = [ dict ( tenant = '<STR_LIT>' , <EOL> name = '<STR_LIT>' , <EOL> types = dict ( test_type = '<STR_LIT>' ) ) , <EOL> dict ( tenant = '<STR_LIT>' , <EOL> name = '<STR_LIT>' , <EOL> types = dict ( test_type = '<STR_LIT>' ) ) ] ) <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> info_values = self . mox . CreateMockAnything ( ) <EOL> models . TenantInfo . objects . filter ( tenant__in = mox . SameElementsAs ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) . AndReturn ( info_values ) <EOL> info_values . values ( '<STR_LIT>' ) . AndReturn ( [ dict ( tenant = '<STR_LIT>' ) ] ) <EOL> models . TenantInfo . objects . bulk_create ( mox . And ( <EOL> Length ( <NUM_LIT:1> ) , mox . IsA ( list ) , mox . In ( mox . And ( <EOL> mox . IsA ( models . TenantInfo ) , <EOL> mox . ContainsAttributeValue ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> mox . ContainsAttributeValue ( '<STR_LIT:name>' , '<STR_LIT>' ) , <EOL> mox . ContainsAttributeValue ( '<STR_LIT>' , TEST_DATE ) <EOL> ) ) ) ) <EOL> fake_tenants = self . mox . CreateMockAnything ( ) <EOL> models . TenantInfo . objects . filter ( tenant__in = mox . SameElementsAs ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) . AndReturn ( fake_tenants ) <EOL> fake_tenants . update ( last_updated = TEST_DATE ) <EOL> fake_tenants . __iter__ ( ) . AndReturn ( iter ( TEST_OBJECTS ) ) <EOL> models . TenantType . objects . all ( ) . AndReturn ( TEST_TYPES ) <EOL> mock_query = self . mox . CreateMockAnything ( ) <EOL> TypeXref . objects . filter ( tenantinfo_id__in = [ ] ) . AndReturn ( mock_query ) <EOL> mock_query . delete ( ) <EOL> TypeXref . objects . bulk_create ( mox . And ( <EOL> Length ( <NUM_LIT:2> ) , mox . IsA ( list ) , <EOL> mox . In ( mox . And ( <EOL> mox . IsA ( TypeXref ) , <EOL> mox . ContainsAttributeValue ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> mox . ContainsAttributeValue ( '<STR_LIT>' , <NUM_LIT:1> ) ) ) , <EOL> mox . In ( mox . And ( <EOL> mox . IsA ( TypeXref ) , <EOL> mox . ContainsAttributeValue ( '<STR_LIT>' , <NUM_LIT:2> ) , <EOL> mox . ContainsAttributeValue ( '<STR_LIT>' , <NUM_LIT:2> ) ) ) , <EOL> ) ) <EOL> self . mox . ReplayAll ( ) <EOL> dbapi . batch_update_tenant_info ( fake_request ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> body_dict = { '<STR_LIT>' : <NUM_LIT:200> } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> exists = self . mox . CreateMockAnything ( ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( result ) <EOL> result . get ( message_id = MESSAGE_ID_1 ) . AndReturn ( exists ) <EOL> exists . save ( ) <EOL> self . mox . ReplayAll ( ) <EOL> dbapi . exists_send_status ( fake_request , MESSAGE_ID_1 ) <EOL> self . assertEqual ( exists . send_status , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_accepts_post ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:POST>' <EOL> body_dict = { '<STR_LIT>' : <NUM_LIT:200> } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> exists = self . mox . CreateMockAnything ( ) <EOL> result = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( result ) <EOL> result . get ( message_id = MESSAGE_ID_1 ) . AndReturn ( exists ) <EOL> exists . save ( ) <EOL> self . mox . ReplayAll ( ) <EOL> dbapi . exists_send_status ( fake_request , MESSAGE_ID_1 ) <EOL> self . assertEqual ( exists . send_status , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_not_found ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> body_dict = { '<STR_LIT>' : <NUM_LIT:200> } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> result = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( result ) <EOL> exception = models . InstanceExists . DoesNotExist ( ) <EOL> result . get ( message_id = MESSAGE_ID_1 ) . AndRaise ( exception ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , MESSAGE_ID_1 ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:status>" ) , <NUM_LIT> ) <EOL> msg = "<STR_LIT>" <EOL> msg = msg % MESSAGE_ID_1 <EOL> self . assertEqual ( body . get ( "<STR_LIT:message>" ) , msg ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_multiple_results ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> body_dict = { '<STR_LIT>' : <NUM_LIT:200> } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> result = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( result ) <EOL> exception = models . InstanceExists . MultipleObjectsReturned ( ) <EOL> result . get ( message_id = MESSAGE_ID_1 ) . AndRaise ( exception ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , MESSAGE_ID_1 ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:status>" ) , <NUM_LIT> ) <EOL> msg = "<STR_LIT>" <EOL> msg = msg % MESSAGE_ID_1 <EOL> self . assertEqual ( body . get ( "<STR_LIT:message>" ) , msg ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_wrong_method ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . body = None <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , MESSAGE_ID_1 ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:status>" ) , <NUM_LIT> ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:message>" ) , "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_no_body ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> fake_request . body = None <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , MESSAGE_ID_1 ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:status>" ) , <NUM_LIT> ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:message>" ) , "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_bad_body ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> body_dict = { '<STR_LIT>' : '<STR_LIT:body>' } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , MESSAGE_ID_1 ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:status>" ) , <NUM_LIT> ) <EOL> msg = "<STR_LIT>" <EOL> self . assertEqual ( body . get ( "<STR_LIT:message>" ) , msg ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_accepts_post_when_version_is_not_given ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:POST>' <EOL> messages = { <EOL> MESSAGE_ID_1 : <NUM_LIT> , MESSAGE_ID_2 : <NUM_LIT> <EOL> } <EOL> body_dict = { '<STR_LIT>' : messages } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> self . mox . StubOutWithMock ( transaction , '<STR_LIT>' ) <EOL> trans_obj = self . mox . CreateMockAnything ( ) <EOL> transaction . commit_on_success ( ) . AndReturn ( trans_obj ) <EOL> trans_obj . __enter__ ( ) <EOL> for uuid , code in messages . items ( ) : <EOL> results = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( results ) <EOL> exists = self . mox . CreateMockAnything ( ) <EOL> results . get ( message_id = uuid ) . AndReturn ( exists ) <EOL> exists . save ( ) <EOL> trans_obj . __exit__ ( None , None , None ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> exists . send_status = <NUM_LIT:200> <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_accepts_post_for_nova_and_glance_when_version_is_1 ( <EOL> self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:POST>' <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> messages = { <EOL> '<STR_LIT>' : { MESSAGE_ID_3 : <NUM_LIT> } , <EOL> '<STR_LIT>' : { MESSAGE_ID_1 : <NUM_LIT> , MESSAGE_ID_2 : <NUM_LIT> } <EOL> } <EOL> body_dict = { '<STR_LIT:version>' : <NUM_LIT:1> , '<STR_LIT>' : messages } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> self . mox . StubOutWithMock ( transaction , '<STR_LIT>' ) <EOL> trans_obj = self . mox . CreateMockAnything ( ) <EOL> transaction . commit_on_success ( ) . AndReturn ( trans_obj ) <EOL> trans_obj . __enter__ ( ) <EOL> results1 = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( results1 ) <EOL> exists1 = self . mox . CreateMockAnything ( ) <EOL> results1 . get ( message_id = MESSAGE_ID_3 ) . AndReturn ( exists1 ) <EOL> exists1 . save ( ) <EOL> trans_obj . __exit__ ( None , None , None ) <EOL> trans_obj = self . mox . CreateMockAnything ( ) <EOL> transaction . commit_on_success ( ) . AndReturn ( trans_obj ) <EOL> trans_obj . __enter__ ( ) <EOL> for uuid , code in messages [ '<STR_LIT>' ] . items ( ) : <EOL> query = self . mox . CreateMockAnything ( ) <EOL> models . ImageExists . objects . select_for_update ( ) . AndReturn ( query ) <EOL> existsA = self . mox . CreateMockAnything ( ) <EOL> existsB = self . mox . CreateMockAnything ( ) <EOL> query . filter ( message_id = uuid ) . AndReturn ( [ existsA , existsB ] ) <EOL> existsA . save ( ) <EOL> existsB . save ( ) <EOL> trans_obj . __exit__ ( None , None , None ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_accepts_post_when_version_is_0 ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:POST>' <EOL> messages = { MESSAGE_ID_1 : <NUM_LIT> , MESSAGE_ID_2 : <NUM_LIT> } <EOL> body_dict = { '<STR_LIT:version>' : <NUM_LIT:0> , '<STR_LIT>' : messages } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> self . mox . StubOutWithMock ( transaction , '<STR_LIT>' ) <EOL> trans_obj = self . mox . CreateMockAnything ( ) <EOL> transaction . commit_on_success ( ) . AndReturn ( trans_obj ) <EOL> trans_obj . __enter__ ( ) <EOL> for uuid , code in messages . items ( ) : <EOL> results = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( results ) <EOL> exists = self . mox . CreateMockAnything ( ) <EOL> results . get ( message_id = uuid ) . AndReturn ( exists ) <EOL> exists . save ( ) <EOL> trans_obj . __exit__ ( None , None , None ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_accepts_post_for_nova_and_glance_when_version_is_2 ( <EOL> self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:POST>' <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> messages = { <EOL> '<STR_LIT>' : { MESSAGE_ID_3 : { '<STR_LIT:status>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } , <EOL> MESSAGE_ID_4 : { '<STR_LIT:status>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> } , <EOL> '<STR_LIT>' : { MESSAGE_ID_1 : { '<STR_LIT:status>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } , <EOL> MESSAGE_ID_2 : { '<STR_LIT:status>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } <EOL> } <EOL> } <EOL> body_dict = { '<STR_LIT:version>' : <NUM_LIT:2> , '<STR_LIT>' : messages } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> self . mox . StubOutWithMock ( transaction , '<STR_LIT>' ) <EOL> trans_obj = self . mox . CreateMockAnything ( ) <EOL> transaction . commit_on_success ( ) . AndReturn ( trans_obj ) <EOL> trans_obj . __enter__ ( ) <EOL> for uuid , code in messages [ '<STR_LIT>' ] . items ( ) : <EOL> results = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( results ) <EOL> exists = self . mox . CreateMockAnything ( ) <EOL> results . get ( message_id = uuid ) . AndReturn ( exists ) <EOL> exists . save ( ) <EOL> trans_obj . __exit__ ( None , None , None ) <EOL> trans_obj = self . mox . CreateMockAnything ( ) <EOL> transaction . commit_on_success ( ) . AndReturn ( trans_obj ) <EOL> trans_obj . __enter__ ( ) <EOL> for uuid , code in messages [ '<STR_LIT>' ] . items ( ) : <EOL> results = self . mox . CreateMockAnything ( ) <EOL> models . ImageExists . objects . select_for_update ( ) . AndReturn ( results ) <EOL> existsA = self . mox . CreateMockAnything ( ) <EOL> existsB = self . mox . CreateMockAnything ( ) <EOL> results . filter ( message_id = uuid ) . AndReturn ( [ existsA , existsB ] ) <EOL> existsA . save ( ) <EOL> existsB . save ( ) <EOL> trans_obj . __exit__ ( None , None , None ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_not_found ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> messages = { <EOL> MESSAGE_ID_1 : '<STR_LIT>' , <EOL> } <EOL> body_dict = { '<STR_LIT>' : messages } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> self . mox . StubOutWithMock ( transaction , '<STR_LIT>' ) <EOL> trans_obj = self . mox . CreateMockAnything ( ) <EOL> transaction . commit_on_success ( ) . AndReturn ( trans_obj ) <EOL> trans_obj . __enter__ ( ) <EOL> results = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( results ) <EOL> exception = models . InstanceExists . DoesNotExist ( ) <EOL> results . get ( message_id = MESSAGE_ID_1 ) . AndRaise ( exception ) <EOL> trans_obj . __exit__ ( dbapi . NotFoundException ( ) . __class__ , <EOL> mox . IgnoreArg ( ) , <EOL> mox . IgnoreArg ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:status>" ) , <NUM_LIT> ) <EOL> msg = "<STR_LIT>" <EOL> msg = msg % MESSAGE_ID_1 <EOL> self . assertEqual ( body . get ( "<STR_LIT:message>" ) , msg ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_multiple_results ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> messages = { <EOL> MESSAGE_ID_1 : <NUM_LIT> , <EOL> } <EOL> body_dict = { '<STR_LIT>' : messages } <EOL> body = json . dumps ( body_dict ) <EOL> fake_request . body = body <EOL> self . mox . StubOutWithMock ( transaction , '<STR_LIT>' ) <EOL> trans_obj = self . mox . CreateMockAnything ( ) <EOL> transaction . commit_on_success ( ) . AndReturn ( trans_obj ) <EOL> trans_obj . __enter__ ( ) <EOL> results = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . select_for_update ( ) . AndReturn ( results ) <EOL> exception = models . InstanceExists . MultipleObjectsReturned ( ) <EOL> results . get ( message_id = MESSAGE_ID_1 ) . AndRaise ( exception ) <EOL> trans_obj . __exit__ ( dbapi . APIException ( ) . __class__ , <EOL> mox . IgnoreArg ( ) , <EOL> mox . IgnoreArg ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( "<STR_LIT:status>" ) , <NUM_LIT> ) <EOL> msg = "<STR_LIT>" <EOL> msg = msg % MESSAGE_ID_1 <EOL> self . assertEqual ( body . get ( "<STR_LIT:message>" ) , msg ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_wrong_method ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( '<STR_LIT:status>' ) , <NUM_LIT> ) <EOL> self . assertEqual ( body . get ( '<STR_LIT:message>' ) , "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_no_body ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> fake_request . method = '<STR_LIT>' <EOL> fake_request . body = None <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( '<STR_LIT:status>' ) , <NUM_LIT> ) <EOL> self . assertEqual ( body . get ( '<STR_LIT:message>' ) , "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_empty_body ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> fake_request . body = '<STR_LIT>' <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( '<STR_LIT:status>' ) , <NUM_LIT> ) <EOL> self . assertEqual ( body . get ( '<STR_LIT:message>' ) , "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_send_status_batch_bad_body ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT>' <EOL> fake_request . GET = { '<STR_LIT>' : '<STR_LIT>' } <EOL> body_dict = { '<STR_LIT>' : '<STR_LIT:body>' } <EOL> fake_request . body = json . dumps ( body_dict ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . exists_send_status ( fake_request , '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> body = json . loads ( resp . content ) <EOL> self . assertEqual ( body . get ( '<STR_LIT:status>' ) , <NUM_LIT> ) <EOL> msg = "<STR_LIT>" <EOL> self . assertEqual ( body . get ( '<STR_LIT:message>' ) , msg ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_launches_without_service ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> mock_objects = self . mox . CreateMockAnything ( ) <EOL> launches = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( mock_objects ) . AndReturn ( launches ) <EOL> dbapi . get_db_objects ( models . InstanceUsage , fake_request , <EOL> '<STR_LIT>' ) . AndReturn ( mock_objects ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_launches ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : launches } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_launches_for_glance ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> mock_objects = self . mox . CreateMockAnything ( ) <EOL> launches = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( mock_objects ) . AndReturn ( launches ) <EOL> dbapi . get_db_objects ( models . ImageUsage , fake_request , <EOL> '<STR_LIT>' ) . AndReturn ( mock_objects ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_images ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : launches } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_launches_for_nova ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> mock_objects = self . mox . CreateMockAnything ( ) <EOL> launches = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( mock_objects ) . AndReturn ( launches ) <EOL> dbapi . get_db_objects ( models . InstanceUsage , fake_request , <EOL> '<STR_LIT>' ) . AndReturn ( mock_objects ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_launches ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : launches } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_launch_with_no_service ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> launch = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_model_by_id ( models . InstanceUsage , <NUM_LIT:1> ) . AndReturn ( launch ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . get_usage_launch ( fake_request , <NUM_LIT:1> ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : { '<STR_LIT:a>' : <NUM_LIT:1> } } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_launch_for_nova ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> launch = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_model_by_id ( models . InstanceUsage , <NUM_LIT:1> ) . AndReturn ( launch ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . get_usage_launch ( fake_request , <NUM_LIT:1> ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : { '<STR_LIT:a>' : <NUM_LIT:1> } } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_launch_for_glance ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> launch = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_model_by_id ( models . ImageUsage , <NUM_LIT:1> ) . AndReturn ( launch ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . get_usage_image ( fake_request , <NUM_LIT:1> ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : { '<STR_LIT:a>' : <NUM_LIT:1> } } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_delete_for_nova ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> delete = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_model_by_id ( models . InstanceDeletes , <NUM_LIT:1> ) . AndReturn ( delete ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . get_usage_delete ( fake_request , <NUM_LIT:1> ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : { '<STR_LIT:a>' : <NUM_LIT:1> } } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_delete_for_glance ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> delete = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _get_model_by_id ( models . ImageDeletes , <NUM_LIT:1> ) . AndReturn ( delete ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . get_usage_delete_glance ( fake_request , <NUM_LIT:1> ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : { '<STR_LIT:a>' : <NUM_LIT:1> } } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_deletes_with_no_service ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> mock_objects = self . mox . CreateMockAnything ( ) <EOL> deletes = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( mock_objects ) . AndReturn ( deletes ) <EOL> dbapi . get_db_objects ( models . InstanceDeletes , fake_request , <EOL> '<STR_LIT>' ) . AndReturn ( mock_objects ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_deletes ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : deletes } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_deletes_for_nova ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> mock_objects = self . mox . CreateMockAnything ( ) <EOL> deletes = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( mock_objects ) . AndReturn ( deletes ) <EOL> dbapi . get_db_objects ( models . InstanceDeletes , fake_request , <EOL> '<STR_LIT>' ) . AndReturn ( mock_objects ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_deletes ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : deletes } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_list_usage_deletes_for_glance ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> mock_objects = self . mox . CreateMockAnything ( ) <EOL> deletes = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> self . mox . StubOutWithMock ( dbapi , '<STR_LIT>' ) <EOL> dbapi . _convert_model_list ( mock_objects ) . AndReturn ( deletes ) <EOL> dbapi . get_db_objects ( models . ImageDeletes , fake_request , <EOL> '<STR_LIT>' ) . AndReturn ( mock_objects ) <EOL> self . mox . ReplayAll ( ) <EOL> resp = dbapi . list_usage_deletes_glance ( fake_request ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( json . loads ( resp . content ) , { '<STR_LIT>' : deletes } ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_exist_stats_nova ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> query = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . objects . filter ( ) . AndReturn ( query ) <EOL> query . values ( '<STR_LIT:status>' , '<STR_LIT>' ) . AndReturn ( query ) <EOL> result = [ <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> L , '<STR_LIT>' : <NUM_LIT:1> } <EOL> ] <EOL> query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_usage_exist_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> expected_response = json . dumps ( { '<STR_LIT>' : result } ) <EOL> self . assertEqual ( expected_response , response . content ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_exist_stats_nova_received_min ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> now = datetime . datetime . utcnow ( ) <EOL> fake_request . GET = { '<STR_LIT>' : str ( now ) } <EOL> query = self . mox . CreateMockAnything ( ) <EOL> filters = { '<STR_LIT>' : utils . decimal_utc ( now ) } <EOL> models . InstanceExists . objects . filter ( ** filters ) . AndReturn ( query ) <EOL> query . values ( '<STR_LIT:status>' , '<STR_LIT>' ) . AndReturn ( query ) <EOL> result = [ <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> L , '<STR_LIT>' : <NUM_LIT:1> } <EOL> ] <EOL> query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_usage_exist_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> expected_response = json . dumps ( { '<STR_LIT>' : result } ) <EOL> self . assertEqual ( expected_response , response . content ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_exist_stats_nova_received_max ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> now = datetime . datetime . utcnow ( ) <EOL> fake_request . GET = { '<STR_LIT>' : str ( now ) } <EOL> query = self . mox . CreateMockAnything ( ) <EOL> filters = { '<STR_LIT>' : utils . decimal_utc ( now ) } <EOL> models . InstanceExists . objects . filter ( ** filters ) . AndReturn ( query ) <EOL> query . values ( '<STR_LIT:status>' , '<STR_LIT>' ) . AndReturn ( query ) <EOL> result = [ <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> L , '<STR_LIT>' : <NUM_LIT:1> } <EOL> ] <EOL> query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_usage_exist_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> expected_response = json . dumps ( { '<STR_LIT>' : result } ) <EOL> self . assertEqual ( expected_response , response . content ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_exist_stats_nova_class_field_filter ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> now = datetime . datetime . utcnow ( ) <EOL> fake_request . GET = { '<STR_LIT>' : str ( now ) } <EOL> query = self . mox . CreateMockAnything ( ) <EOL> models . InstanceExists . _meta . get_field_by_name ( '<STR_LIT>' ) <EOL> filters = { '<STR_LIT>' : utils . decimal_utc ( now ) } <EOL> models . InstanceExists . objects . filter ( ** filters ) . AndReturn ( query ) <EOL> query . values ( '<STR_LIT:status>' , '<STR_LIT>' ) . AndReturn ( query ) <EOL> result = [ <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> L , '<STR_LIT>' : <NUM_LIT:1> } <EOL> ] <EOL> query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_usage_exist_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> expected_response = json . dumps ( { '<STR_LIT>' : result } ) <EOL> self . assertEqual ( expected_response , response . content ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_exist_stats_glance ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { } <EOL> query = self . mox . CreateMockAnything ( ) <EOL> models . ImageExists . objects . filter ( ) . AndReturn ( query ) <EOL> query . values ( '<STR_LIT:status>' , '<STR_LIT>' ) . AndReturn ( query ) <EOL> result = [ <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> L , '<STR_LIT>' : <NUM_LIT:1> } <EOL> ] <EOL> query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_usage_exist_stats_glance ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> expected_response = json . dumps ( { '<STR_LIT>' : result } ) <EOL> self . assertEqual ( expected_response , response . content ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_exist_stats_glance_received_min ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> now = datetime . datetime . utcnow ( ) <EOL> fake_request . GET = { '<STR_LIT>' : str ( now ) } <EOL> query = self . mox . CreateMockAnything ( ) <EOL> filters = { '<STR_LIT>' : utils . decimal_utc ( now ) } <EOL> models . ImageExists . objects . filter ( ** filters ) . AndReturn ( query ) <EOL> query . values ( '<STR_LIT:status>' , '<STR_LIT>' ) . AndReturn ( query ) <EOL> result = [ <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> L , '<STR_LIT>' : <NUM_LIT:1> } <EOL> ] <EOL> query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_usage_exist_stats_glance ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> expected_response = json . dumps ( { '<STR_LIT>' : result } ) <EOL> self . assertEqual ( expected_response , response . content ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_exist_stats_glance_received_max ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> now = datetime . datetime . utcnow ( ) <EOL> fake_request . GET = { '<STR_LIT>' : str ( now ) } <EOL> query = self . mox . CreateMockAnything ( ) <EOL> filters = { '<STR_LIT>' : utils . decimal_utc ( now ) } <EOL> models . ImageExists . objects . filter ( ** filters ) . AndReturn ( query ) <EOL> query . values ( '<STR_LIT:status>' , '<STR_LIT>' ) . AndReturn ( query ) <EOL> result = [ <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> L , '<STR_LIT>' : <NUM_LIT:1> } <EOL> ] <EOL> query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_usage_exist_stats_glance ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> expected_response = json . dumps ( { '<STR_LIT>' : result } ) <EOL> self . assertEqual ( expected_response , response . content ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_usage_exist_stats_glance_class_field_filter ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> now = datetime . datetime . utcnow ( ) <EOL> fake_request . GET = { '<STR_LIT>' : str ( now ) } <EOL> query = self . mox . CreateMockAnything ( ) <EOL> models . ImageExists . _meta . get_field_by_name ( '<STR_LIT>' ) <EOL> filters = { '<STR_LIT>' : utils . decimal_utc ( now ) } <EOL> models . ImageExists . objects . filter ( ** filters ) . AndReturn ( query ) <EOL> query . values ( '<STR_LIT:status>' , '<STR_LIT>' ) . AndReturn ( query ) <EOL> result = [ <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> L , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> L , '<STR_LIT>' : <NUM_LIT:1> } <EOL> ] <EOL> query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( result ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_usage_exist_stats_glance ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> expected_response = json . dumps ( { '<STR_LIT>' : result } ) <EOL> self . assertEqual ( expected_response , response . content ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_event_stats ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { '<STR_LIT>' : "<STR_LIT>" } <EOL> mock_query = self . mox . CreateMockAnything ( ) <EOL> models . RawData . objects . values ( '<STR_LIT>' ) . AndReturn ( mock_query ) <EOL> events = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } <EOL> ] <EOL> mock_query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( events ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_event_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( response . content , <EOL> json . dumps ( { '<STR_LIT>' : events } ) ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_event_stats_date_range ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> start = "<STR_LIT>" <EOL> end = "<STR_LIT>" <EOL> fake_request . GET = { '<STR_LIT>' : start , <EOL> '<STR_LIT>' : end , <EOL> '<STR_LIT>' : "<STR_LIT>" } <EOL> mock_query = self . mox . CreateMockAnything ( ) <EOL> filters = { <EOL> '<STR_LIT>' : stacktach_utils . str_time_to_unix ( start ) , <EOL> '<STR_LIT>' : stacktach_utils . str_time_to_unix ( end ) <EOL> } <EOL> models . RawData . objects . filter ( ** filters ) . AndReturn ( mock_query ) <EOL> mock_query . values ( '<STR_LIT>' ) . AndReturn ( mock_query ) <EOL> events = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } <EOL> ] <EOL> mock_query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( events ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_event_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( response . content , <EOL> json . dumps ( { '<STR_LIT>' : events } ) ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_verified_count ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> mock_query = self . mox . CreateMockAnything ( ) <EOL> models . RawData . objects . values ( '<STR_LIT>' ) . AndReturn ( mock_query ) <EOL> events = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } <EOL> ] <EOL> mock_query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( events ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_event_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( response . content , <EOL> json . dumps ( { '<STR_LIT>' : [ events [ <NUM_LIT:0> ] ] } ) ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_verified_count_default ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> mock_query = self . mox . CreateMockAnything ( ) <EOL> models . RawData . objects . values ( '<STR_LIT>' ) . AndReturn ( mock_query ) <EOL> events = [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:100> } <EOL> ] <EOL> mock_query . annotate ( event_count = mox . IsA ( Count ) ) . AndReturn ( events ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_event_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> self . assertEqual ( response . content , <EOL> json . dumps ( { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> } ] } ) ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_verified_count_only_one_range_param_returns_400 ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" } <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_event_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . assertEqual ( json . loads ( response . content ) [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_verified_count_only_large_date_range_returns_400 ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" } <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_event_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . assertEqual ( json . loads ( response . content ) [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_verified_count_wrong_date_format_returns_400 ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" } <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_event_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . assertEqual ( json . loads ( response . content ) [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_get_verified_count_wrong_service_returns_400 ( self ) : <EOL> fake_request = self . mox . CreateMockAnything ( ) <EOL> fake_request . method = '<STR_LIT:GET>' <EOL> fake_request . GET = { '<STR_LIT>' : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" } <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . get_event_stats ( fake_request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> self . assertEqual ( json . loads ( response . content ) [ '<STR_LIT:message>' ] , <EOL> "<STR_LIT>" ) <EOL> self . mox . VerifyAll ( ) <EOL> class StacktachRepairScenarioApi ( StacktachBaseTestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> def test_change_nova_exists_status_for_all_exists ( self ) : <EOL> request = self . mox . CreateMockAnything ( ) <EOL> request . POST = self . mox . CreateMockAnything ( ) <EOL> message_ids = [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> request . POST . _iterlists ( ) . AndReturn ( [ ( '<STR_LIT>' , [ '<STR_LIT>' ] ) , <EOL> ( '<STR_LIT>' , message_ids ) ] ) <EOL> self . mox . StubOutWithMock ( models . InstanceExists , <EOL> '<STR_LIT>' ) <EOL> models . InstanceExists . mark_exists_as_sent_unverified ( message_ids ) . AndReturn ( [ [ ] , [ ] ] ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . repair_stacktach_down ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> response_data = json . loads ( response . content ) <EOL> self . assertEqual ( response_data [ '<STR_LIT>' ] , [ ] ) <EOL> self . assertEqual ( response_data [ '<STR_LIT>' ] , [ ] ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_change_glance_exists_status_for_all_exists ( self ) : <EOL> request = self . mox . CreateMockAnything ( ) <EOL> request . POST = self . mox . CreateMockAnything ( ) <EOL> message_ids = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> request . POST . _iterlists ( ) . AndReturn ( [ ( '<STR_LIT>' , [ '<STR_LIT>' ] ) , <EOL> ( '<STR_LIT>' , message_ids ) ] ) <EOL> self . mox . StubOutWithMock ( models . ImageExists , <EOL> '<STR_LIT>' ) <EOL> models . ImageExists . mark_exists_as_sent_unverified ( message_ids ) . AndReturn ( [ [ ] , [ ] ] ) <EOL> self . mox . ReplayAll ( ) <EOL> response = dbapi . repair_stacktach_down ( request ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> response_data = json . loads ( response . content ) <EOL> self . assertEqual ( response_data [ '<STR_LIT>' ] , [ ] ) <EOL> self . assertEqual ( response_data [ '<STR_LIT>' ] , [ ] ) <EOL> self . mox . VerifyAll ( ) </s>
