<s> import os <EOL> import uuid <EOL> import mock <EOL> from oslo_config import fixture <EOL> from oslotest import base <EOL> from oslotest import mockpatch <EOL> from rally . common import db <EOL> from rally import plugins <EOL> from tests . unit import fakes <EOL> class DatabaseFixture ( fixture . Config ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( DatabaseFixture , self ) . setUp ( ) <EOL> db_url = os . environ . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> db . engine_reset ( ) <EOL> self . conf . set_default ( "<STR_LIT>" , db_url , group = "<STR_LIT>" ) <EOL> db . schema_cleanup ( ) <EOL> db . schema_create ( ) <EOL> class TestCase ( base . BaseTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( TestCase , self ) . setUp ( ) <EOL> self . addCleanup ( mock . patch . stopall ) <EOL> plugins . load ( ) <EOL> def _test_atomic_action_timer ( self , atomic_actions , name ) : <EOL> action_duration = atomic_actions . get ( name ) <EOL> self . assertIsNotNone ( action_duration ) <EOL> self . assertIsInstance ( action_duration , float ) <EOL> def assertSequenceEqual ( self , iterable_1 , iterable_2 , msg = None ) : <EOL> self . assertEqual ( tuple ( iterable_1 ) , tuple ( iterable_2 ) , msg ) <EOL> class DBTestCase ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( DBTestCase , self ) . setUp ( ) <EOL> self . useFixture ( DatabaseFixture ( ) ) <EOL> class ScenarioTestCase ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> benchmark_utils = "<STR_LIT>" <EOL> patch_benchmark_utils = True <EOL> def client_factory ( self , client_type , version = None , admin = False ) : <EOL> """<STR_LIT>""" <EOL> return mock . MagicMock ( client_type = client_type , <EOL> version = version , <EOL> admin = admin ) <EOL> def clients ( self , client_type , version = None , admin = False ) : <EOL> """<STR_LIT>""" <EOL> key = ( client_type , version , admin ) <EOL> if key not in self . _clients : <EOL> self . _clients [ key ] = self . client_factory ( client_type , <EOL> version = version , <EOL> admin = admin ) <EOL> return self . _clients [ key ] <EOL> def admin_clients ( self , client_type , version = None ) : <EOL> """<STR_LIT>""" <EOL> return self . clients ( client_type , version = version , admin = True ) <EOL> def client_created ( self , client_type , version = None , admin = False ) : <EOL> """<STR_LIT>""" <EOL> key = ( client_type , version , admin ) <EOL> return key in self . _clients <EOL> def get_client_mocks ( self ) : <EOL> base_path = "<STR_LIT>" <EOL> return [ <EOL> mock . patch ( <EOL> "<STR_LIT>" % base_path , <EOL> mock . Mock ( side_effect = self . clients ) ) , <EOL> mock . patch ( <EOL> "<STR_LIT>" % base_path , <EOL> mock . Mock ( side_effect = self . admin_clients ) ) <EOL> ] <EOL> def setUp ( self ) : <EOL> super ( ScenarioTestCase , self ) . setUp ( ) <EOL> if self . patch_benchmark_utils : <EOL> self . mock_resource_is = mockpatch . Patch ( <EOL> self . benchmark_utils + "<STR_LIT>" ) <EOL> self . mock_get_from_manager = mockpatch . Patch ( <EOL> self . benchmark_utils + "<STR_LIT>" ) <EOL> self . mock_wait_for = mockpatch . Patch ( <EOL> self . benchmark_utils + "<STR_LIT>" ) <EOL> self . mock_wait_for_delete = mockpatch . Patch ( <EOL> self . benchmark_utils + "<STR_LIT>" ) <EOL> self . mock_wait_for_status = mockpatch . Patch ( <EOL> self . benchmark_utils + "<STR_LIT>" ) <EOL> self . useFixture ( self . mock_resource_is ) <EOL> self . useFixture ( self . mock_get_from_manager ) <EOL> self . useFixture ( self . mock_wait_for ) <EOL> self . useFixture ( self . mock_wait_for_delete ) <EOL> self . useFixture ( self . mock_wait_for_status ) <EOL> self . mock_sleep = mockpatch . Patch ( "<STR_LIT>" ) <EOL> self . useFixture ( self . mock_sleep ) <EOL> self . _clients = { } <EOL> self . _client_mocks = self . get_client_mocks ( ) <EOL> for patcher in self . _client_mocks : <EOL> patcher . start ( ) <EOL> self . context = get_test_context ( ) <EOL> def tearDown ( self ) : <EOL> for patcher in self . _client_mocks : <EOL> patcher . stop ( ) <EOL> super ( ScenarioTestCase , self ) . tearDown ( ) <EOL> class ContextClientAdapter ( object ) : <EOL> def __init__ ( self , endpoint , test_case ) : <EOL> self . endpoint = endpoint <EOL> self . test_case = test_case <EOL> def mock_client ( self , name , version = None ) : <EOL> admin = self . endpoint . startswith ( "<STR_LIT>" ) <EOL> client = self . test_case . clients ( name , version = version , admin = admin ) <EOL> if not isinstance ( client . return_value , mock . Mock ) : <EOL> return client . return_value <EOL> if client . side_effect is not None : <EOL> return client ( ) <EOL> return client <EOL> def __getattr__ ( self , name ) : <EOL> return lambda version = None : self . mock_client ( name , version ) <EOL> class ContextTestCase ( ScenarioTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ContextTestCase , self ) . setUp ( ) <EOL> self . _adapters = { } <EOL> def context_client ( self , endpoint , api_info = None ) : <EOL> if endpoint not in self . _adapters : <EOL> self . _adapters [ endpoint ] = ContextClientAdapter ( endpoint , self ) <EOL> return self . _adapters [ endpoint ] <EOL> def get_client_mocks ( self ) : <EOL> return [ <EOL> mock . patch ( <EOL> "<STR_LIT>" , <EOL> mock . Mock ( side_effect = self . context_client ) ) <EOL> ] <EOL> class FakeClientsScenarioTestCase ( ScenarioTestCase ) : <EOL> """<STR_LIT>""" <EOL> def client_factory ( self , client_type , version = None , admin = False ) : <EOL> return getattr ( self . _fake_clients , client_type ) ( ) <EOL> def setUp ( self ) : <EOL> super ( FakeClientsScenarioTestCase , self ) . setUp ( ) <EOL> self . _fake_clients = fakes . FakeClients ( ) <EOL> def get_test_context ( ) : <EOL> return { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : str ( uuid . uuid4 ( ) ) <EOL> } <EOL> } </s>
