<s> import uuid <EOL> import alembic <EOL> from datetime import datetime <EOL> from oslo_serialization import jsonutils <EOL> import six <EOL> import sqlalchemy as sa <EOL> from sqlalchemy . exc import DataError <EOL> from sqlalchemy . exc import IntegrityError <EOL> from nailgun import consts <EOL> from nailgun . db import db <EOL> from nailgun . db import dropdb <EOL> from nailgun . db . migration import ALEMBIC_CONFIG <EOL> from nailgun . test import base <EOL> _prepare_revision = '<STR_LIT>' <EOL> _test_revision = '<STR_LIT>' <EOL> master_node_settings_before_migration = None <EOL> def setup_module ( ) : <EOL> dropdb ( ) <EOL> alembic . command . upgrade ( ALEMBIC_CONFIG , _prepare_revision ) <EOL> prepare ( ) <EOL> global master_node_settings_before_migration <EOL> master_node_settings_before_migration = jsonutils . loads ( <EOL> get_master_node_settings ( ) ) <EOL> alembic . command . upgrade ( ALEMBIC_CONFIG , _test_revision ) <EOL> def get_master_node_settings ( ) : <EOL> meta = base . reflect_db_metadata ( ) <EOL> master_node_settings_table = meta . tables [ '<STR_LIT>' ] <EOL> settings = db . execute ( sa . select ( <EOL> [ master_node_settings_table . c . settings ] ) ) . scalar ( ) <EOL> db ( ) . commit ( ) <EOL> return settings <EOL> def prepare ( ) : <EOL> meta = base . reflect_db_metadata ( ) <EOL> releaseid = insert_table_row ( <EOL> meta . tables [ '<STR_LIT>' ] , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:version>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:state>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : jsonutils . dumps ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : { } <EOL> } <EOL> } ) <EOL> } <EOL> ) <EOL> clusterid = insert_table_row ( <EOL> meta . tables [ '<STR_LIT>' ] , <EOL> { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : releaseid , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> ) <EOL> db . execute ( <EOL> meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ <EOL> { '<STR_LIT>' : clusterid , '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : clusterid , '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : clusterid , '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : clusterid , '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : clusterid , '<STR_LIT:name>' : consts . NODE_GROUPS . default } , <EOL> ] ) <EOL> netconfigid = insert_table_row ( <EOL> meta . tables [ '<STR_LIT>' ] , <EOL> { <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : None , <EOL> } ) <EOL> db . execute ( <EOL> meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ { <EOL> '<STR_LIT:id>' : netconfigid , <EOL> '<STR_LIT>' : [ ] , <EOL> '<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> result = db . execute ( <EOL> meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> '<STR_LIT:version>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ <EOL> { '<STR_LIT>' : '<STR_LIT>' } <EOL> ] ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> } ] <EOL> ) <EOL> pluginid_a = result . inserted_primary_key [ <NUM_LIT:0> ] <EOL> result = db . execute ( <EOL> meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> '<STR_LIT:version>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ <EOL> { '<STR_LIT>' : '<STR_LIT>' } <EOL> ] ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> } ] <EOL> ) <EOL> pluginid_b = result . inserted_primary_key [ <NUM_LIT:0> ] <EOL> db . execute ( <EOL> meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ <EOL> { <EOL> '<STR_LIT>' : clusterid , <EOL> '<STR_LIT>' : pluginid_a <EOL> } , <EOL> { <EOL> '<STR_LIT>' : clusterid , <EOL> '<STR_LIT>' : pluginid_b <EOL> } <EOL> ] <EOL> ) <EOL> db . execute ( <EOL> meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ { <EOL> '<STR_LIT>' : clusterid , <EOL> '<STR_LIT>' : jsonutils . dumps ( { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : pluginid_a , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:value>' : '<STR_LIT:value>' , <EOL> '<STR_LIT:type>' : '<STR_LIT:text>' , <EOL> '<STR_LIT:description>' : '<STR_LIT:description>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT:label>' : '<STR_LIT:label>' <EOL> } <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : pluginid_b , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> } <EOL> } ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( { } ) , <EOL> } ] ) <EOL> db . commit ( ) <EOL> def insert_table_row ( table , row_data ) : <EOL> result = db . execute ( <EOL> table . insert ( ) , <EOL> [ row_data ] <EOL> ) <EOL> return result . inserted_primary_key [ <NUM_LIT:0> ] <EOL> class TestNodeGroupsMigration ( base . BaseAlembicMigrationTest ) : <EOL> def test_name_cluster_unique_constraint_migration ( self ) : <EOL> names = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . name ] ) ) . fetchall ( ) <EOL> self . assertIn ( ( '<STR_LIT>' , ) , names ) <EOL> self . assertIn ( ( '<STR_LIT>' , ) , names ) <EOL> self . assertIn ( ( '<STR_LIT>' , ) , names ) <EOL> self . assertIn ( ( '<STR_LIT>' , ) , names ) <EOL> def test_unique_name_fields_insert_duplicated ( self ) : <EOL> nodegroup = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . cluster_id , <EOL> self . meta . tables [ '<STR_LIT>' ] . c . name ] ) ) . fetchone ( ) <EOL> with self . assertRaisesRegexp ( IntegrityError , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) : <EOL> insert_table_row ( self . meta . tables [ '<STR_LIT>' ] , <EOL> { '<STR_LIT>' : nodegroup [ '<STR_LIT>' ] , <EOL> '<STR_LIT:name>' : nodegroup [ '<STR_LIT:name>' ] } ) <EOL> def test_unique_name_fields_insert_unique ( self ) : <EOL> nodegroup = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . cluster_id , <EOL> self . meta . tables [ '<STR_LIT>' ] . c . name ] ) ) . fetchone ( ) <EOL> db . execute ( self . meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ { '<STR_LIT>' : nodegroup [ '<STR_LIT>' ] , <EOL> '<STR_LIT:name>' : six . text_type ( uuid . uuid4 ( ) ) } ] ) <EOL> def test_node_group_has_default_field ( self ) : <EOL> nodegroup = db . execute ( <EOL> sa . select ( <EOL> [ self . meta . tables [ '<STR_LIT>' ] . c . is_default ] ) ) . fetchone ( ) <EOL> self . assertEqual ( nodegroup [ '<STR_LIT>' ] , False ) <EOL> def test_existing_default_groups_are_updated_by_default_flag ( self ) : <EOL> nodegroups = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . name , <EOL> self . meta . tables [ '<STR_LIT>' ] . c . is_default ] ) <EOL> ) <EOL> for ng in nodegroups : <EOL> if ng [ <NUM_LIT:0> ] == consts . NODE_GROUPS . default : <EOL> self . assertTrue ( ng [ <NUM_LIT:1> ] ) <EOL> else : <EOL> self . assertFalse ( ng [ <NUM_LIT:1> ] ) <EOL> class TestReleaseMigrations ( base . BaseAlembicMigrationTest ) : <EOL> def test_release_is_deployable_deleted ( self ) : <EOL> self . assertNotIn ( '<STR_LIT>' , <EOL> [ c . name for c in self . meta . tables [ '<STR_LIT>' ] . c ] ) <EOL> def test_releases_are_manageonly ( self ) : <EOL> states = [ r [ <NUM_LIT:0> ] for r in db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . state ] ) ) . fetchall ( ) ] <EOL> for state in states : <EOL> self . assertEqual ( state , '<STR_LIT>' ) <EOL> class TestTaskStatus ( base . BaseAlembicMigrationTest ) : <EOL> def test_pending_status_saving ( self ) : <EOL> db . execute ( <EOL> self . meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ <EOL> { <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : consts . TASK_NAMES . node_deletion , <EOL> '<STR_LIT:message>' : None , <EOL> '<STR_LIT:status>' : consts . TASK_STATUSES . pending , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT:result>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT:1> <EOL> } <EOL> ] ) <EOL> result = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . status ] ) ) <EOL> for row in result . fetchall ( ) : <EOL> status = row [ <NUM_LIT:0> ] <EOL> self . assertEqual ( status , consts . TASK_STATUSES . pending ) <EOL> class TestTaskNameMigration ( base . BaseAlembicMigrationTest ) : <EOL> def test_task_name_enum ( self ) : <EOL> added_task_names = ( '<STR_LIT>' , ) <EOL> tasks_table = self . meta . tables [ '<STR_LIT>' ] <EOL> for name in added_task_names : <EOL> insert_table_row ( tasks_table , <EOL> { '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : str ( uuid . uuid4 ( ) ) , <EOL> '<STR_LIT:status>' : '<STR_LIT>' } ) <EOL> with self . assertRaisesRegexp ( DataError , '<STR_LIT>' <EOL> '<STR_LIT>' ) : <EOL> insert_table_row ( tasks_table , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( uuid . uuid4 ( ) ) , <EOL> '<STR_LIT:status>' : '<STR_LIT>' } ) <EOL> class TestNodeErrorTypeMigration ( base . BaseAlembicMigrationTest ) : <EOL> def test_node_error_type_enum ( self ) : <EOL> added_error_types = ( '<STR_LIT>' , ) <EOL> nodes_table = self . meta . tables [ '<STR_LIT>' ] <EOL> for error_type in added_error_types : <EOL> insert_table_row ( nodes_table , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT:error>' , <EOL> '<STR_LIT>' : error_type , <EOL> '<STR_LIT>' : str ( uuid . uuid4 ( ) ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : datetime . now ( ) } ) <EOL> inserted_count = db . execute ( <EOL> sa . select ( [ sa . func . count ( nodes_table . c . error_type ) ] ) . <EOL> where ( sa . and_ ( nodes_table . c . error_type == error_type ) ) <EOL> ) . fetchone ( ) [ <NUM_LIT:0> ] <EOL> self . assertEqual ( inserted_count , <NUM_LIT:1> ) <EOL> with self . assertRaisesRegexp ( DataError , '<STR_LIT>' <EOL> '<STR_LIT>' ) : <EOL> insert_table_row ( nodes_table , <EOL> { '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT:error>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : str ( uuid . uuid4 ( ) ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : datetime . now ( ) } ) <EOL> class TestNeutronConfigInternalFloatingNames ( base . BaseAlembicMigrationTest ) : <EOL> def test_internal_name_and_floating_names_are_added ( self ) : <EOL> columns = [ t . name for t in self . meta . tables [ '<STR_LIT>' ] . columns ] <EOL> self . assertIn ( '<STR_LIT>' , columns ) <EOL> self . assertIn ( '<STR_LIT>' , columns ) <EOL> def test_internal_name_and_floating_names_defaults ( self ) : <EOL> result = db . execute ( <EOL> sa . select ( [ <EOL> self . meta . tables [ '<STR_LIT>' ] . c . internal_name , <EOL> self . meta . tables [ '<STR_LIT>' ] . c . floating_name ] ) ) <EOL> for internal_name , floating_name in result : <EOL> self . assertEqual ( '<STR_LIT>' , internal_name ) <EOL> self . assertEqual ( '<STR_LIT>' , floating_name ) <EOL> def test_neutron_config_is_updated_in_releases ( self ) : <EOL> result = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . networks_metadata ] ) ) <EOL> for networks_metadata , in result : <EOL> networks_metadata = jsonutils . loads ( networks_metadata ) <EOL> neutron_config = networks_metadata [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertEqual ( '<STR_LIT>' , neutron_config [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , neutron_config [ '<STR_LIT>' ] ) <EOL> class TestClusterPluginsMigration ( base . BaseAlembicMigrationTest ) : <EOL> def _get_enabled ( self , plugin_name ) : <EOL> plugins = self . meta . tables [ '<STR_LIT>' ] <EOL> cluster_plugins = self . meta . tables [ '<STR_LIT>' ] <EOL> query = sa . select ( [ cluster_plugins . c . enabled ] ) . select_from ( <EOL> sa . join ( <EOL> cluster_plugins , plugins , <EOL> cluster_plugins . c . plugin_id == plugins . c . id ) ) . where ( plugins . c . name == plugin_name ) <EOL> return db . execute ( query ) . fetchone ( ) [ <NUM_LIT:0> ] <EOL> def test_plugin_a_is_enabled ( self ) : <EOL> enabled = self . _get_enabled ( '<STR_LIT>' ) <EOL> self . assertTrue ( enabled ) <EOL> def test_plugin_b_is_disabled ( self ) : <EOL> enabled = self . _get_enabled ( '<STR_LIT>' ) <EOL> self . assertFalse ( enabled ) <EOL> def test_moving_plugin_attributes ( self ) : <EOL> clusters = self . meta . tables [ '<STR_LIT>' ] <EOL> attributes = self . meta . tables [ '<STR_LIT>' ] <EOL> plugins = self . meta . tables [ '<STR_LIT>' ] <EOL> cluster_plugins = self . meta . tables [ '<STR_LIT>' ] <EOL> query = sa . select ( [ attributes . c . editable ] ) . select_from ( <EOL> sa . join ( <EOL> attributes , clusters , <EOL> attributes . c . cluster_id == clusters . c . id ) ) <EOL> result = jsonutils . loads ( db . execute ( query ) . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertItemsEqual ( result , { } ) <EOL> query = sa . select ( [ cluster_plugins . c . attributes ] ) . select_from ( <EOL> sa . join ( <EOL> cluster_plugins , plugins , <EOL> cluster_plugins . c . plugin_id == plugins . c . id ) ) . where ( plugins . c . name == '<STR_LIT>' ) <EOL> result = jsonutils . loads ( db . execute ( query ) . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertNotIn ( '<STR_LIT>' , result ) <EOL> self . assertItemsEqual ( result [ '<STR_LIT>' ] , { <EOL> '<STR_LIT:value>' : '<STR_LIT:value>' , <EOL> '<STR_LIT:type>' : '<STR_LIT:text>' , <EOL> '<STR_LIT:description>' : '<STR_LIT:description>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT:label>' : '<STR_LIT:label>' <EOL> } ) <EOL> class TestBaremetalFields ( base . BaseAlembicMigrationTest ) : <EOL> def test_baremetal_fields_saving ( self ) : <EOL> baremetal_gateway = '<STR_LIT>' <EOL> baremetal_range = jsonutils . dumps ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> result = db . execute ( <EOL> self . meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ { <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : None , <EOL> } ] ) <EOL> db . execute ( <EOL> self . meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ { <EOL> '<STR_LIT:id>' : result . inserted_primary_key [ <NUM_LIT:0> ] , <EOL> '<STR_LIT>' : [ ] , <EOL> '<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> '<STR_LIT>' : baremetal_gateway , <EOL> '<STR_LIT>' : baremetal_range , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ] ) <EOL> result = db . execute ( <EOL> sa . select ( <EOL> [ self . meta . tables [ '<STR_LIT>' ] . c . baremetal_gateway , <EOL> self . meta . tables [ '<STR_LIT>' ] . c . baremetal_range ] ) ) . fetchall ( ) <EOL> self . assertIn ( ( baremetal_gateway , baremetal_range ) , result ) <EOL> class TestComponentsMigration ( base . BaseAlembicMigrationTest ) : <EOL> def test_new_component_metadata_field_exists_and_empty ( self ) : <EOL> column_values = [ <EOL> ( self . meta . tables [ '<STR_LIT>' ] . c . components_metadata , [ ] ) , <EOL> ( self . meta . tables [ '<STR_LIT>' ] . c . components_metadata , [ ] ) , <EOL> ( self . meta . tables [ '<STR_LIT>' ] . c . components , [ ] ) <EOL> ] <EOL> result = db . execute ( sa . select ( <EOL> [ item [ <NUM_LIT:0> ] for item in column_values ] ) ) <EOL> db_values = result . fetchone ( ) <EOL> for idx , db_value in enumerate ( db_values ) : <EOL> self . assertEqual ( jsonutils . loads ( db_value ) , column_values [ idx ] [ <NUM_LIT:1> ] ) <EOL> def test_hotplug_field_exists ( self ) : <EOL> result = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . is_hotpluggable ] ) <EOL> ) <EOL> self . assertTrue ( all ( x [ <NUM_LIT:0> ] is None for x in result ) ) <EOL> class TestMasterNodeSettingsMigration ( base . BaseAlembicMigrationTest ) : <EOL> def test_bootstrap_field_exists_and_filled ( self ) : <EOL> settings = get_master_node_settings ( ) <EOL> bootstrap_settings = { <EOL> "<STR_LIT:error>" : { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:10> <EOL> } <EOL> } <EOL> self . assertEqual ( <EOL> bootstrap_settings , <EOL> settings [ '<STR_LIT>' ] <EOL> ) <EOL> def test_ui_settings_field_exists_and_has_default_value ( self ) : <EOL> settings = get_master_node_settings ( ) <EOL> ui_settings = settings [ '<STR_LIT>' ] <EOL> self . assertItemsEqual ( ui_settings [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertItemsEqual ( ui_settings [ '<STR_LIT>' ] , { } ) <EOL> self . assertItemsEqual ( ui_settings [ '<STR_LIT>' ] , [ { '<STR_LIT:status>' : '<STR_LIT>' } ] ) <EOL> self . assertItemsEqual ( ui_settings [ '<STR_LIT>' ] , { } ) <EOL> self . assertItemsEqual ( ui_settings [ '<STR_LIT>' ] , [ ] ) <EOL> self . assertItemsEqual ( ui_settings [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_master_node_settings_old_data_not_modified ( self ) : <EOL> settings = get_master_node_settings ( ) <EOL> settings . pop ( '<STR_LIT>' ) <EOL> settings . pop ( '<STR_LIT>' ) <EOL> self . assertDictEqual ( <EOL> master_node_settings_before_migration , <EOL> settings <EOL> ) <EOL> class TestClusterPluginLinks ( base . BaseAlembicMigrationTest ) : <EOL> def test_cluster_plugin_links_creation ( self ) : <EOL> clusters = self . meta . tables [ '<STR_LIT>' ] <EOL> cluster_plugin_links = self . meta . tables [ '<STR_LIT>' ] <EOL> cluster_id = db . execute ( sa . select ( [ clusters ] ) ) . scalar ( ) <EOL> db . execute ( <EOL> cluster_plugin_links . insert ( ) , <EOL> [ { <EOL> '<STR_LIT>' : cluster_id , <EOL> '<STR_LIT:title>' : '<STR_LIT:title>' , <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT:description>' , <EOL> '<STR_LIT>' : False <EOL> } ] ) <EOL> class TestOpenstackConfigMigration ( base . BaseAlembicMigrationTest ) : <EOL> def test_openstack_configs_table_saving ( self ) : <EOL> result = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . id ] ) ) <EOL> cluster_id = result . fetchone ( ) [ <NUM_LIT:0> ] <EOL> db . execute ( <EOL> self . meta . tables [ '<STR_LIT>' ] . insert ( ) , <EOL> [ { <EOL> '<STR_LIT>' : cluster_id , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : datetime . now ( ) , <EOL> '<STR_LIT>' : jsonutils . dumps ( { <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : { } , <EOL> } ) , <EOL> } ] <EOL> ) <EOL> result = db . execute ( <EOL> sa . select ( [ self . meta . tables [ '<STR_LIT>' ] . c . cluster_id ] ) ) <EOL> config = result . fetchone ( ) <EOL> self . assertEqual ( config [ <NUM_LIT:0> ] , cluster_id ) <EOL> class TestPluginLinks ( base . BaseAlembicMigrationTest ) : <EOL> def test_plugin_links_creation ( self ) : <EOL> plugins = self . meta . tables [ '<STR_LIT>' ] <EOL> plugin_links = self . meta . tables [ '<STR_LIT>' ] <EOL> plugin_id = db . execute ( sa . select ( [ plugins ] ) ) . scalar ( ) <EOL> plugin_link_data = { <EOL> '<STR_LIT>' : plugin_id , <EOL> '<STR_LIT:title>' : '<STR_LIT:title>' , <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT:description>' , <EOL> '<STR_LIT>' : False <EOL> } <EOL> link_id = db . execute ( <EOL> plugin_links . insert ( ) , <EOL> [ plugin_link_data ] <EOL> ) . inserted_primary_key [ <NUM_LIT:0> ] <EOL> fetched_data = db . execute ( sa . select ( [ plugin_links ] ) ) . fetchone ( ) <EOL> self . assertEqual ( link_id , fetched_data [ <NUM_LIT:0> ] ) <EOL> class TestOswlStats ( base . BaseAlembicMigrationTest ) : <EOL> def test_oswl_stats_creation ( self ) : <EOL> oswl = self . meta . tables [ '<STR_LIT>' ] <EOL> expected_version_info = { "<STR_LIT>" : "<STR_LIT>" } <EOL> oswl_data = { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : datetime . utcnow ( ) . date ( ) , <EOL> '<STR_LIT>' : datetime . utcnow ( ) . time ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:x>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : expected_version_info <EOL> } <EOL> expected_id = db . execute ( <EOL> oswl . insert ( ) , [ oswl_data ] ) . inserted_primary_key [ <NUM_LIT:0> ] <EOL> actual_id , actual_version_info = db . execute ( <EOL> sa . select ( [ oswl . c . id , oswl . c . version_info ] ) <EOL> ) . first ( ) <EOL> self . assertEqual ( expected_id , actual_id ) <EOL> self . assertEqual ( expected_version_info , actual_version_info ) </s>
