<s> """<STR_LIT>""" <EOL> import contextlib <EOL> from alembic import script <EOL> import mock <EOL> from oslo_db import exception as db_exc <EOL> from oslo_db . sqlalchemy import enginefacade <EOL> from oslo_db . sqlalchemy import test_base <EOL> from oslo_db . sqlalchemy import test_migrations <EOL> from oslo_db . sqlalchemy import utils as db_utils <EOL> from oslo_log import log as logging <EOL> from oslo_utils import uuidutils <EOL> import sqlalchemy <EOL> import sqlalchemy . exc <EOL> from ironic . common . i18n import _LE <EOL> from ironic . db . sqlalchemy import migration <EOL> from ironic . db . sqlalchemy import models <EOL> from ironic . tests import base <EOL> LOG = logging . getLogger ( __name__ ) <EOL> def _get_connect_string ( backend , user , passwd , database ) : <EOL> """<STR_LIT>""" <EOL> if backend == "<STR_LIT>" : <EOL> backend = "<STR_LIT>" <EOL> elif backend == "<STR_LIT>" : <EOL> backend = "<STR_LIT>" <EOL> else : <EOL> raise Exception ( "<STR_LIT>" % backend ) <EOL> return ( "<STR_LIT>" <EOL> % { '<STR_LIT>' : backend , '<STR_LIT:user>' : user , '<STR_LIT>' : passwd , <EOL> '<STR_LIT>' : database } ) <EOL> def _is_backend_avail ( backend , user , passwd , database ) : <EOL> try : <EOL> connect_uri = _get_connect_string ( backend , user , passwd , database ) <EOL> engine = sqlalchemy . create_engine ( connect_uri ) <EOL> connection = engine . connect ( ) <EOL> except Exception : <EOL> return False <EOL> else : <EOL> connection . close ( ) <EOL> engine . dispose ( ) <EOL> return True <EOL> @ contextlib . contextmanager <EOL> def patch_with_engine ( engine ) : <EOL> with mock . patch . object ( enginefacade . get_legacy_facade ( ) , <EOL> '<STR_LIT>' ) as patch_engine : <EOL> patch_engine . return_value = engine <EOL> yield <EOL> class WalkVersionsMixin ( object ) : <EOL> def _walk_versions ( self , engine = None , alembic_cfg = None ) : <EOL> with patch_with_engine ( engine ) : <EOL> script_directory = script . ScriptDirectory . from_config ( alembic_cfg ) <EOL> self . assertIsNone ( self . migration_api . version ( alembic_cfg ) ) <EOL> versions = [ ver for ver in script_directory . walk_revisions ( ) ] <EOL> for version in reversed ( versions ) : <EOL> self . _migrate_up ( engine , alembic_cfg , <EOL> version . revision , with_data = True ) <EOL> def _migrate_up ( self , engine , config , version , with_data = False ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if with_data : <EOL> data = None <EOL> pre_upgrade = getattr ( <EOL> self , "<STR_LIT>" % version , None ) <EOL> if pre_upgrade : <EOL> data = pre_upgrade ( engine ) <EOL> self . migration_api . upgrade ( version , config = config ) <EOL> self . assertEqual ( version , self . migration_api . version ( config ) ) <EOL> if with_data : <EOL> check = getattr ( self , "<STR_LIT>" % version , None ) <EOL> if check : <EOL> check ( engine , data ) <EOL> except Exception : <EOL> LOG . error ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> { '<STR_LIT:version>' : version , '<STR_LIT>' : engine } ) <EOL> raise <EOL> class TestWalkVersions ( base . TestCase , WalkVersionsMixin ) : <EOL> def setUp ( self ) : <EOL> super ( TestWalkVersions , self ) . setUp ( ) <EOL> self . migration_api = mock . MagicMock ( ) <EOL> self . engine = mock . MagicMock ( ) <EOL> self . config = mock . MagicMock ( ) <EOL> self . versions = [ mock . Mock ( revision = '<STR_LIT>' ) , mock . Mock ( revision = '<STR_LIT>' ) ] <EOL> def test_migrate_up ( self ) : <EOL> self . migration_api . version . return_value = '<STR_LIT>' <EOL> self . _migrate_up ( self . engine , self . config , '<STR_LIT>' ) <EOL> self . migration_api . upgrade . assert_called_with ( '<STR_LIT>' , <EOL> config = self . config ) <EOL> self . migration_api . version . assert_called_with ( self . config ) <EOL> def test_migrate_up_with_data ( self ) : <EOL> test_value = { "<STR_LIT:a>" : <NUM_LIT:1> , "<STR_LIT:b>" : <NUM_LIT:2> } <EOL> self . migration_api . version . return_value = '<STR_LIT>' <EOL> self . _pre_upgrade_141 = mock . MagicMock ( ) <EOL> self . _pre_upgrade_141 . return_value = test_value <EOL> self . _check_141 = mock . MagicMock ( ) <EOL> self . _migrate_up ( self . engine , self . config , '<STR_LIT>' , True ) <EOL> self . _pre_upgrade_141 . assert_called_with ( self . engine ) <EOL> self . _check_141 . assert_called_with ( self . engine , test_value ) <EOL> @ mock . patch . object ( script , '<STR_LIT>' ) <EOL> @ mock . patch . object ( WalkVersionsMixin , '<STR_LIT>' ) <EOL> def test_walk_versions_all_default ( self , _migrate_up , script_directory ) : <EOL> fc = script_directory . from_config ( ) <EOL> fc . walk_revisions . return_value = self . versions <EOL> self . migration_api . version . return_value = None <EOL> self . _walk_versions ( self . engine , self . config ) <EOL> self . migration_api . version . assert_called_with ( self . config ) <EOL> upgraded = [ mock . call ( self . engine , self . config , v . revision , <EOL> with_data = True ) for v in reversed ( self . versions ) ] <EOL> self . assertEqual ( self . _migrate_up . call_args_list , upgraded ) <EOL> @ mock . patch . object ( script , '<STR_LIT>' ) <EOL> @ mock . patch . object ( WalkVersionsMixin , '<STR_LIT>' ) <EOL> def test_walk_versions_all_false ( self , _migrate_up , script_directory ) : <EOL> fc = script_directory . from_config ( ) <EOL> fc . walk_revisions . return_value = self . versions <EOL> self . migration_api . version . return_value = None <EOL> self . _walk_versions ( self . engine , self . config ) <EOL> upgraded = [ mock . call ( self . engine , self . config , v . revision , <EOL> with_data = True ) for v in reversed ( self . versions ) ] <EOL> self . assertEqual ( upgraded , self . _migrate_up . call_args_list ) <EOL> class MigrationCheckersMixin ( object ) : <EOL> def setUp ( self ) : <EOL> super ( MigrationCheckersMixin , self ) . setUp ( ) <EOL> self . config = migration . _alembic_config ( ) <EOL> self . migration_api = migration <EOL> def test_walk_versions ( self ) : <EOL> self . _walk_versions ( self . engine , self . config ) <EOL> def test_connect_fail ( self ) : <EOL> """<STR_LIT>""" <EOL> if _is_backend_avail ( self . FIXTURE . DRIVER , "<STR_LIT>" , <EOL> self . FIXTURE . USERNAME , self . FIXTURE . DBNAME ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def _check_21b331f883ef ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in nodes . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIsInstance ( nodes . c . provision_updated_at . type , <EOL> sqlalchemy . types . DateTime ) <EOL> def _check_3cb628139ea4 ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in nodes . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertTrue ( isinstance ( nodes . c . console_enabled . type , <EOL> sqlalchemy . types . Boolean ) or <EOL> isinstance ( nodes . c . console_enabled . type , <EOL> sqlalchemy . types . Integer ) ) <EOL> def _check_31baaf680d2b ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in nodes . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIsInstance ( nodes . c . instance_info . type , <EOL> sqlalchemy . types . TEXT ) <EOL> def _check_3bea56f25597 ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> instance_uuid = '<STR_LIT>' <EOL> data = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' : instance_uuid } <EOL> nodes . insert ( ) . values ( data ) . execute ( ) <EOL> data [ '<STR_LIT>' ] = uuidutils . generate_uuid ( ) <EOL> self . assertRaises ( db_exc . DBDuplicateEntry , <EOL> nodes . insert ( ) . execute , data ) <EOL> def _check_242cc6a923b3 ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in nodes . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIsInstance ( nodes . c . maintenance_reason . type , <EOL> sqlalchemy . types . String ) <EOL> def _pre_upgrade_5674c57409b9 ( self , engine ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> data = [ { '<STR_LIT>' : uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' : None } ] <EOL> nodes . insert ( ) . values ( data ) . execute ( ) <EOL> return data <EOL> def _check_5674c57409b9 ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> result = engine . execute ( nodes . select ( ) ) <EOL> def _get_state ( uuid ) : <EOL> for row in data : <EOL> if row [ '<STR_LIT>' ] == uuid : <EOL> return row [ '<STR_LIT>' ] <EOL> for row in result : <EOL> old = _get_state ( row [ '<STR_LIT>' ] ) <EOL> new = row [ '<STR_LIT>' ] <EOL> if old is None : <EOL> self . assertEqual ( '<STR_LIT>' , new ) <EOL> else : <EOL> self . assertEqual ( old , new ) <EOL> def _check_bb59b63f55a ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in nodes . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIsInstance ( nodes . c . driver_internal_info . type , <EOL> sqlalchemy . types . TEXT ) <EOL> def _check_4f399b21ae71 ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in nodes . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIsInstance ( nodes . c . clean_step . type , <EOL> sqlalchemy . types . String ) <EOL> def _check_789acc877671 ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in nodes . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIsInstance ( nodes . c . raid_config . type , <EOL> sqlalchemy . types . String ) <EOL> self . assertIsInstance ( nodes . c . target_raid_config . type , <EOL> sqlalchemy . types . String ) <EOL> def _check_2fb93ffd2af1 ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> bigstring = '<STR_LIT:a>' * <NUM_LIT:255> <EOL> uuid = uuidutils . generate_uuid ( ) <EOL> data = { '<STR_LIT>' : uuid , '<STR_LIT:name>' : bigstring } <EOL> nodes . insert ( ) . execute ( data ) <EOL> node = nodes . select ( nodes . c . uuid == uuid ) . execute ( ) . first ( ) <EOL> self . assertEqual ( bigstring , node [ '<STR_LIT:name>' ] ) <EOL> def _check_516faf1bb9b1 ( self , engine , data ) : <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> bigstring = '<STR_LIT:a>' * <NUM_LIT:255> <EOL> uuid = uuidutils . generate_uuid ( ) <EOL> data = { '<STR_LIT>' : uuid , '<STR_LIT>' : bigstring } <EOL> nodes . insert ( ) . execute ( data ) <EOL> node = nodes . select ( nodes . c . uuid == uuid ) . execute ( ) . first ( ) <EOL> self . assertEqual ( bigstring , node [ '<STR_LIT>' ] ) <EOL> def _check_48d6c242bb9b ( self , engine , data ) : <EOL> node_tags = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in node_tags . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIsInstance ( node_tags . c . tag . type , <EOL> sqlalchemy . types . String ) <EOL> nodes = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> data = { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT:name>' : '<STR_LIT>' } <EOL> nodes . insert ( ) . execute ( data ) <EOL> data = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> node_tags . insert ( ) . execute ( data ) <EOL> tag = node_tags . select ( node_tags . c . node_id == '<STR_LIT>' ) . execute ( ) . first ( ) <EOL> self . assertEqual ( '<STR_LIT>' , tag [ '<STR_LIT>' ] ) <EOL> def _check_5ea1b0d310e ( self , engine , data ) : <EOL> portgroup = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in portgroup . c ] <EOL> expected_names = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:id>' , '<STR_LIT>' , '<STR_LIT:name>' , <EOL> '<STR_LIT>' , '<STR_LIT:address>' , '<STR_LIT>' ] <EOL> self . assertEqual ( sorted ( expected_names ) , sorted ( col_names ) ) <EOL> self . assertIsInstance ( portgroup . c . created_at . type , <EOL> sqlalchemy . types . DateTime ) <EOL> self . assertIsInstance ( portgroup . c . updated_at . type , <EOL> sqlalchemy . types . DateTime ) <EOL> self . assertIsInstance ( portgroup . c . id . type , <EOL> sqlalchemy . types . Integer ) <EOL> self . assertIsInstance ( portgroup . c . uuid . type , <EOL> sqlalchemy . types . String ) <EOL> self . assertIsInstance ( portgroup . c . name . type , <EOL> sqlalchemy . types . String ) <EOL> self . assertIsInstance ( portgroup . c . node_id . type , <EOL> sqlalchemy . types . Integer ) <EOL> self . assertIsInstance ( portgroup . c . address . type , <EOL> sqlalchemy . types . String ) <EOL> self . assertIsInstance ( portgroup . c . extra . type , <EOL> sqlalchemy . types . TEXT ) <EOL> ports = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> col_names = [ column . name for column in ports . c ] <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIn ( '<STR_LIT>' , col_names ) <EOL> self . assertIsInstance ( ports . c . portgroup_id . type , <EOL> sqlalchemy . types . Integer ) <EOL> self . assertTrue ( isinstance ( ports . c . pxe_enabled . type , <EOL> sqlalchemy . types . Boolean ) or <EOL> isinstance ( ports . c . pxe_enabled . type , <EOL> sqlalchemy . types . Integer ) ) <EOL> def _pre_upgrade_f6fdb920c182 ( self , engine ) : <EOL> ports = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> data = [ { '<STR_LIT>' : uuidutils . generate_uuid ( ) , '<STR_LIT>' : None } , <EOL> { '<STR_LIT>' : uuidutils . generate_uuid ( ) , '<STR_LIT>' : None } ] <EOL> ports . insert ( ) . values ( data ) . execute ( ) <EOL> return data <EOL> def _check_f6fdb920c182 ( self , engine , data ) : <EOL> ports = db_utils . get_table ( engine , '<STR_LIT>' ) <EOL> result = engine . execute ( ports . select ( ) ) <EOL> def _was_inserted ( uuid ) : <EOL> for row in data : <EOL> if row [ '<STR_LIT>' ] == uuid : <EOL> return True <EOL> for row in result : <EOL> if _was_inserted ( row [ '<STR_LIT>' ] ) : <EOL> self . assertTrue ( row [ '<STR_LIT>' ] ) <EOL> def test_upgrade_and_version ( self ) : <EOL> with patch_with_engine ( self . engine ) : <EOL> self . migration_api . upgrade ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( self . migration_api . version ( ) ) <EOL> def test_create_schema_and_version ( self ) : <EOL> with patch_with_engine ( self . engine ) : <EOL> self . migration_api . create_schema ( ) <EOL> self . assertIsNotNone ( self . migration_api . version ( ) ) <EOL> def test_upgrade_and_create_schema ( self ) : <EOL> with patch_with_engine ( self . engine ) : <EOL> self . migration_api . upgrade ( '<STR_LIT>' ) <EOL> self . assertRaises ( db_exc . DbMigrationError , <EOL> self . migration_api . create_schema ) <EOL> def test_upgrade_twice ( self ) : <EOL> with patch_with_engine ( self . engine ) : <EOL> self . migration_api . upgrade ( '<STR_LIT>' ) <EOL> v1 = self . migration_api . version ( ) <EOL> self . migration_api . upgrade ( '<STR_LIT>' ) <EOL> v2 = self . migration_api . version ( ) <EOL> self . assertNotEqual ( v1 , v2 ) <EOL> class TestMigrationsMySQL ( MigrationCheckersMixin , <EOL> WalkVersionsMixin , <EOL> test_base . MySQLOpportunisticTestCase ) : <EOL> pass <EOL> class TestMigrationsPostgreSQL ( MigrationCheckersMixin , <EOL> WalkVersionsMixin , <EOL> test_base . PostgreSQLOpportunisticTestCase ) : <EOL> pass <EOL> class ModelsMigrationSyncMixin ( object ) : <EOL> def get_metadata ( self ) : <EOL> return models . Base . metadata <EOL> def get_engine ( self ) : <EOL> return self . engine <EOL> def db_sync ( self , engine ) : <EOL> with patch_with_engine ( engine ) : <EOL> migration . upgrade ( '<STR_LIT>' ) <EOL> class ModelsMigrationsSyncMysql ( ModelsMigrationSyncMixin , <EOL> test_migrations . ModelsMigrationsSync , <EOL> test_base . MySQLOpportunisticTestCase ) : <EOL> pass <EOL> class ModelsMigrationsSyncPostgres ( ModelsMigrationSyncMixin , <EOL> test_migrations . ModelsMigrationsSync , <EOL> test_base . PostgreSQLOpportunisticTestCase ) : <EOL> pass </s>
