refactored data upgrade tests in test_migrations
The one off testing in test_migrations was starting to get obtuse enough that it was possible for bugs to slip through (which they did already). Create a more general framework for testing data transitions at any db migration. Hook the walk_versions code so as we are about to upgrade to a migration point, attempt to run a _prerun_### function, then follow it up with a _check_### function afterwards. Create some utility functions to make doing things in these _prerun and _check functions pretty concise. This makes these checks db independent, and has it so they'll run however we are running the walk_versions. In doing so, this uncovered a bug in the 147 migration, which is fixed inline. There is no retroactive fix, as 147 was destructive to data, so would have nuked anyone that hit it already. Updated to remove safe_getattr, thanks to jerdfelt for pointer on doing that right. Partially implements blueprint migration-testing-with-data Fixes bug #1108448 Change-Id: I8bd92e3688109bfbdf1da21a175055b7263abf45
This commit is contained in:
Sean Dague
@@ -37,9 +37,9 @@ def upgrade(migrate_engine):
|
||||
if rec['binary'] != 'nova-compute':
|
||||
continue
|
||||
# if zone doesn't exist create
|
||||
result = aggregate_metadata.select().where(aggregate_metadata.c.key ==
|
||||
'availability_zone' and
|
||||
aggregate_metadata.c.key == rec['availability_zone']).execute()
|
||||
result = aggregate_metadata.select().where(
|
||||
aggregate_metadata.c.key == 'availability_zone').where(
|
||||
aggregate_metadata.c.value == rec['availability_zone']).execute()
|
||||
result = [r for r in result]
|
||||
if len(result) > 0:
|
||||
agg_id = result[0].aggregate_id
|
||||
|
||||
@@ -87,6 +87,16 @@ def _have_mysql():
|
||||
return present.lower() in ('', 'true')
|
||||
|
||||
|
||||
def get_table(engine, name):
|
||||
"""Returns an sqlalchemy table dynamically from db.
|
||||
|
||||
Needed because the models don't work for us in migrations
|
||||
as models will be far out of sync with the current data."""
|
||||
metadata = sqlalchemy.schema.MetaData()
|
||||
metadata.bind = engine
|
||||
return sqlalchemy.Table(name, metadata, autoload=True)
|
||||
|
||||
|
||||
class TestMigrations(test.TestCase):
|
||||
"""Test sqlalchemy-migrate migrations."""
|
||||
|
||||
@@ -227,19 +237,11 @@ class TestMigrations(test.TestCase):
|
||||
self.engines["mysqlcitest"] = engine
|
||||
self.test_databases["mysqlcitest"] = connect_string
|
||||
|
||||
# Test that we end in an innodb
|
||||
self._check_mysql_innodb(engine)
|
||||
# Test IP transition
|
||||
self._check_mysql_migration_149(engine)
|
||||
|
||||
def _check_mysql_innodb(self, engine):
|
||||
# build a fully populated mysql database with all the tables
|
||||
self._reset_databases()
|
||||
self._walk_versions(engine, False, False)
|
||||
|
||||
uri = _get_connect_string("mysql", database="information_schema")
|
||||
connection = sqlalchemy.create_engine(uri).connect()
|
||||
|
||||
connection = engine.connect()
|
||||
# sanity check
|
||||
total = connection.execute("SELECT count(*) "
|
||||
"from information_schema.TABLES "
|
||||
@@ -253,92 +255,8 @@ class TestMigrations(test.TestCase):
|
||||
"and TABLE_NAME!='migrate_version'")
|
||||
count = noninnodb.scalar()
|
||||
self.assertEqual(count, 0, "%d non InnoDB tables created" % count)
|
||||
|
||||
def test_migration_149_postgres(self):
|
||||
"""Test updating a table with IPAddress columns."""
|
||||
if not _is_backend_avail('postgres'):
|
||||
self.skipTest("postgres not available")
|
||||
|
||||
connect_string = _get_connect_string("postgres")
|
||||
engine = sqlalchemy.create_engine(connect_string)
|
||||
|
||||
self.engines["postgrescitest"] = engine
|
||||
self.test_databases["postgrescitest"] = connect_string
|
||||
|
||||
self._reset_databases()
|
||||
migration_api.version_control(engine, TestMigrations.REPOSITORY,
|
||||
migration.INIT_VERSION)
|
||||
|
||||
connection = engine.connect()
|
||||
|
||||
self._migrate_up(engine, 148)
|
||||
IPS = ("127.0.0.1", "255.255.255.255", "2001:db8::1:2", "::1")
|
||||
connection.execute("INSERT INTO provider_fw_rules "
|
||||
" (protocol, from_port, to_port, cidr)"
|
||||
"VALUES ('tcp', 1234, 1234, '%s'), "
|
||||
" ('tcp', 1234, 1234, '%s'), "
|
||||
" ('tcp', 1234, 1234, '%s'), "
|
||||
" ('tcp', 1234, 1234, '%s')" % IPS)
|
||||
self.assertEqual('character varying',
|
||||
connection.execute(
|
||||
"SELECT data_type FROM INFORMATION_SCHEMA.COLUMNS "
|
||||
"WHERE table_name='provider_fw_rules' "
|
||||
"AND table_catalog='openstack_citest' "
|
||||
"AND column_name='cidr'").scalar())
|
||||
|
||||
self._migrate_up(engine, 149)
|
||||
self.assertEqual(IPS,
|
||||
tuple(tup[0] for tup in connection.execute(
|
||||
"SELECT cidr from provider_fw_rules").fetchall()))
|
||||
self.assertEqual('inet',
|
||||
connection.execute(
|
||||
"SELECT data_type FROM INFORMATION_SCHEMA.COLUMNS "
|
||||
"WHERE table_name='provider_fw_rules' "
|
||||
"AND table_catalog='openstack_citest' "
|
||||
"AND column_name='cidr'").scalar())
|
||||
connection.close()
|
||||
|
||||
def _check_mysql_migration_149(self, engine):
|
||||
"""Test updating a table with IPAddress columns."""
|
||||
self._reset_databases()
|
||||
migration_api.version_control(engine, TestMigrations.REPOSITORY,
|
||||
migration.INIT_VERSION)
|
||||
|
||||
uri = _get_connect_string("mysql", database="openstack_citest")
|
||||
connection = sqlalchemy.create_engine(uri).connect()
|
||||
|
||||
self._migrate_up(engine, 148)
|
||||
|
||||
IPS = ("127.0.0.1", "255.255.255.255", "2001:db8::1:2", "::1")
|
||||
connection.execute("INSERT INTO provider_fw_rules "
|
||||
" (protocol, from_port, to_port, cidr)"
|
||||
"VALUES ('tcp', 1234, 1234, '%s'), "
|
||||
" ('tcp', 1234, 1234, '%s'), "
|
||||
" ('tcp', 1234, 1234, '%s'), "
|
||||
" ('tcp', 1234, 1234, '%s')" % IPS)
|
||||
self.assertEqual('varchar(255)',
|
||||
connection.execute(
|
||||
"SELECT column_type FROM INFORMATION_SCHEMA.COLUMNS "
|
||||
"WHERE table_name='provider_fw_rules' "
|
||||
"AND table_schema='openstack_citest' "
|
||||
"AND column_name='cidr'").scalar())
|
||||
|
||||
connection.close()
|
||||
|
||||
self._migrate_up(engine, 149)
|
||||
|
||||
connection = sqlalchemy.create_engine(uri).connect()
|
||||
|
||||
self.assertEqual(IPS,
|
||||
tuple(tup[0] for tup in connection.execute(
|
||||
"SELECT cidr from provider_fw_rules").fetchall()))
|
||||
self.assertEqual('varchar(39)',
|
||||
connection.execute(
|
||||
"SELECT column_type FROM INFORMATION_SCHEMA.COLUMNS "
|
||||
"WHERE table_name='provider_fw_rules' "
|
||||
"AND table_schema='openstack_citest' "
|
||||
"AND column_name='cidr'").scalar())
|
||||
|
||||
def _walk_versions(self, engine=None, snake_walk=False, downgrade=True):
|
||||
# Determine latest version script from the repo, then
|
||||
# upgrade from 1 through to the latest, with no data
|
||||
@@ -360,7 +278,7 @@ class TestMigrations(test.TestCase):
|
||||
for version in xrange(migration.INIT_VERSION + 2,
|
||||
TestMigrations.REPOSITORY.latest + 1):
|
||||
# upgrade -> downgrade -> upgrade
|
||||
self._migrate_up(engine, version)
|
||||
self._migrate_up(engine, version, with_data=True)
|
||||
if snake_walk:
|
||||
self._migrate_down(engine, version)
|
||||
self._migrate_up(engine, version)
|
||||
@@ -385,7 +303,19 @@ class TestMigrations(test.TestCase):
|
||||
migration_api.db_version(engine,
|
||||
TestMigrations.REPOSITORY))
|
||||
|
||||
def _migrate_up(self, engine, version):
|
||||
def _migrate_up(self, engine, version, with_data=False):
|
||||
"""migrate up to a new version of the db.
|
||||
|
||||
We allow for data insertion and post checks at every
|
||||
migration version with special _prerun_### and
|
||||
_check_### functions in the main test.
|
||||
"""
|
||||
if with_data:
|
||||
data = None
|
||||
prerun = getattr(self, "_prerun_%d" % version, None)
|
||||
if prerun:
|
||||
data = prerun(engine)
|
||||
|
||||
migration_api.upgrade(engine,
|
||||
TestMigrations.REPOSITORY,
|
||||
version)
|
||||
@@ -393,168 +323,132 @@ class TestMigrations(test.TestCase):
|
||||
migration_api.db_version(engine,
|
||||
TestMigrations.REPOSITORY))
|
||||
|
||||
def test_migration_146(self):
|
||||
name = 'name'
|
||||
az = 'custom_az'
|
||||
if with_data:
|
||||
check = getattr(self, "_check_%d" % version, None)
|
||||
if check:
|
||||
check(engine, data)
|
||||
|
||||
def _145_check():
|
||||
agg = aggregates.select(aggregates.c.id == 1).execute().first()
|
||||
self.assertEqual(name, agg.name)
|
||||
self.assertEqual(az, agg.availability_zone)
|
||||
# migration 146, availability zone transition
|
||||
def _prerun_146(self, engine):
|
||||
data = {
|
||||
'id': 1,
|
||||
'availability_zone': 'custom_az',
|
||||
'aggregate_name': 1,
|
||||
'name': 'name',
|
||||
}
|
||||
|
||||
for key, engine in self.engines.items():
|
||||
migration_api.version_control(engine, TestMigrations.REPOSITORY,
|
||||
migration.INIT_VERSION)
|
||||
migration_api.upgrade(engine, TestMigrations.REPOSITORY, 145)
|
||||
metadata = sqlalchemy.schema.MetaData()
|
||||
metadata.bind = engine
|
||||
aggregates = sqlalchemy.Table('aggregates', metadata,
|
||||
autoload=True)
|
||||
aggregates = get_table(engine, 'aggregates')
|
||||
aggregates.insert().values(data).execute()
|
||||
return data
|
||||
|
||||
aggregates.insert().values(id=1, availability_zone=az,
|
||||
aggregate_name=1, name=name).execute()
|
||||
def _check_146(self, engine, data):
|
||||
aggregate_md = get_table(engine, 'aggregate_metadata')
|
||||
md = aggregate_md.select(
|
||||
aggregate_md.c.aggregate_id == 1).execute().first()
|
||||
self.assertEqual(data['availability_zone'], md['value'])
|
||||
|
||||
_145_check()
|
||||
|
||||
migration_api.upgrade(engine, TestMigrations.REPOSITORY, 146)
|
||||
|
||||
aggregate_metadata = sqlalchemy.Table('aggregate_metadata',
|
||||
metadata, autoload=True)
|
||||
metadata = aggregate_metadata.select(aggregate_metadata.c.
|
||||
aggregate_id == 1).execute().first()
|
||||
self.assertEqual(az, metadata['value'])
|
||||
|
||||
migration_api.downgrade(engine, TestMigrations.REPOSITORY, 145)
|
||||
_145_check()
|
||||
|
||||
def test_migration_147(self):
|
||||
# migration 147, availability zone transition for services
|
||||
def _prerun_147(self, engine):
|
||||
az = 'test_zone'
|
||||
host1 = 'compute-host1'
|
||||
host2 = 'compute-host2'
|
||||
# start at id == 2 because we already inserted one
|
||||
data = [
|
||||
{'id': 1, 'host': host1,
|
||||
'binary': 'nova-compute', 'topic': 'compute',
|
||||
'report_count': 0, 'availability_zone': az},
|
||||
{'id': 2, 'host': 'sched-host',
|
||||
'binary': 'nova-scheduler', 'topic': 'scheduler',
|
||||
'report_count': 0, 'availability_zone': 'ignore_me'},
|
||||
{'id': 3, 'host': host2,
|
||||
'binary': 'nova-compute', 'topic': 'compute',
|
||||
'report_count': 0, 'availability_zone': az},
|
||||
]
|
||||
|
||||
def _146_check():
|
||||
service = services.select(services.c.id == 1).execute().first()
|
||||
self.assertEqual(az, service.availability_zone)
|
||||
self.assertEqual(host1, service.host)
|
||||
service = services.select(services.c.id == 2).execute().first()
|
||||
self.assertNotEqual(az, service.availability_zone)
|
||||
service = services.select(services.c.id == 3).execute().first()
|
||||
self.assertEqual(az, service.availability_zone)
|
||||
self.assertEqual(host2, service.host)
|
||||
services = get_table(engine, 'services')
|
||||
engine.execute(services.insert(), data)
|
||||
return data
|
||||
|
||||
for key, engine in self.engines.items():
|
||||
migration_api.version_control(engine, TestMigrations.REPOSITORY,
|
||||
migration.INIT_VERSION)
|
||||
migration_api.upgrade(engine, TestMigrations.REPOSITORY, 146)
|
||||
metadata = sqlalchemy.schema.MetaData()
|
||||
metadata.bind = engine
|
||||
def _check_147(self, engine, data):
|
||||
aggregate_md = get_table(engine, 'aggregate_metadata')
|
||||
aggregate_hosts = get_table(engine, 'aggregate_hosts')
|
||||
# NOTE(sdague): hard coded to id == 2, because we added to
|
||||
# aggregate_metadata previously
|
||||
for item in data:
|
||||
md = aggregate_md.select(
|
||||
aggregate_md.c.aggregate_id == 2).execute().first()
|
||||
if item['binary'] == "nova-compute":
|
||||
self.assertEqual(item['availability_zone'], md['value'])
|
||||
|
||||
#populate service table
|
||||
services = sqlalchemy.Table('services', metadata,
|
||||
autoload=True)
|
||||
services.insert().values(id=1, host=host1,
|
||||
binary='nova-compute', topic='compute', report_count=0,
|
||||
availability_zone=az).execute()
|
||||
services.insert().values(id=2, host='sched-host',
|
||||
binary='nova-scheduler', topic='scheduler', report_count=0,
|
||||
availability_zone='ignore_me').execute()
|
||||
services.insert().values(id=3, host=host2,
|
||||
binary='nova-compute', topic='compute', report_count=0,
|
||||
availability_zone=az).execute()
|
||||
host = aggregate_hosts.select(
|
||||
aggregate_hosts.c.aggregate_id == 2
|
||||
).execute().first()
|
||||
self.assertEqual(host['host'], data[0]['host'])
|
||||
|
||||
_146_check()
|
||||
# NOTE(sdague): id 3 is just non-existent
|
||||
host = aggregate_hosts.select(
|
||||
aggregate_hosts.c.aggregate_id == 3
|
||||
).execute().first()
|
||||
self.assertEqual(host, None)
|
||||
|
||||
migration_api.upgrade(engine, TestMigrations.REPOSITORY, 147)
|
||||
# migration 149, changes IPAddr storage format
|
||||
def _prerun_149(self, engine):
|
||||
provider_fw_rules = get_table(engine, 'provider_fw_rules')
|
||||
data = [
|
||||
{'protocol': 'tcp', 'from_port': 1234,
|
||||
'to_port': 1234, 'cidr': "127.0.0.1"},
|
||||
{'protocol': 'tcp', 'from_port': 1234,
|
||||
'to_port': 1234, 'cidr': "255.255.255.255"},
|
||||
{'protocol': 'tcp', 'from_port': 1234,
|
||||
'to_port': 1234, 'cidr': "2001:db8::1:2"},
|
||||
{'protocol': 'tcp', 'from_port': 1234,
|
||||
'to_port': 1234, 'cidr': "::1"}
|
||||
]
|
||||
engine.execute(provider_fw_rules.insert(), data)
|
||||
return data
|
||||
|
||||
# check aggregate metadata
|
||||
aggregate_metadata = sqlalchemy.Table('aggregate_metadata',
|
||||
metadata, autoload=True)
|
||||
aggregate_hosts = sqlalchemy.Table('aggregate_hosts',
|
||||
metadata, autoload=True)
|
||||
metadata = aggregate_metadata.select(aggregate_metadata.c.
|
||||
aggregate_id == 1).execute().first()
|
||||
self.assertEqual(az, metadata['value'])
|
||||
self.assertEqual(aggregate_hosts.select(
|
||||
aggregate_hosts.c.aggregate_id == 1).execute().
|
||||
first().host, host1)
|
||||
blank = [h for h in aggregate_hosts.select(
|
||||
aggregate_hosts.c.aggregate_id == 2).execute()]
|
||||
self.assertEqual(blank, [])
|
||||
def _check_149(self, engine, data):
|
||||
provider_fw_rules = get_table(engine, 'provider_fw_rules')
|
||||
result = provider_fw_rules.select().execute()
|
||||
|
||||
migration_api.downgrade(engine, TestMigrations.REPOSITORY, 146)
|
||||
iplist = map(lambda x: x['cidr'], data)
|
||||
|
||||
_146_check()
|
||||
for row in result:
|
||||
self.assertIn(row['cidr'], iplist)
|
||||
|
||||
def test_migration_152(self):
|
||||
# migration 152 - convert deleted from boolean to int
|
||||
def _prerun_152(self, engine):
|
||||
host1 = 'compute-host1'
|
||||
host2 = 'compute-host2'
|
||||
# NOTE(sdague): start at #4 because services data already in table
|
||||
# from 147
|
||||
services_data = [
|
||||
{'id': 4, 'host': host1, 'binary': 'nova-compute',
|
||||
'report_count': 0, 'topic': 'compute', 'deleted': False},
|
||||
{'id': 5, 'host': host1, 'binary': 'nova-compute',
|
||||
'report_count': 0, 'topic': 'compute', 'deleted': True}
|
||||
]
|
||||
volumes_data = [
|
||||
{'id': 'first', 'host': host1, 'deleted': False},
|
||||
{'id': 'second', 'host': host2, 'deleted': True}
|
||||
]
|
||||
|
||||
def _151_check(services, volumes):
|
||||
service = services.select(services.c.id == 1).execute().first()
|
||||
self.assertEqual(False, service.deleted)
|
||||
service = services.select(services.c.id == 2).execute().first()
|
||||
self.assertEqual(True, service.deleted)
|
||||
services = get_table(engine, 'services')
|
||||
engine.execute(services.insert(), services_data)
|
||||
|
||||
volume = volumes.select(volumes.c.id == "first").execute().first()
|
||||
self.assertEqual(False, volume.deleted)
|
||||
volume = volumes.select(volumes.c.id == "second").execute().first()
|
||||
self.assertEqual(True, volume.deleted)
|
||||
volumes = get_table(engine, 'volumes')
|
||||
engine.execute(volumes.insert(), volumes_data)
|
||||
return dict(services=services_data, volumes=volumes_data)
|
||||
|
||||
for key, engine in self.engines.items():
|
||||
migration_api.version_control(engine, TestMigrations.REPOSITORY,
|
||||
migration.INIT_VERSION)
|
||||
migration_api.upgrade(engine, TestMigrations.REPOSITORY, 151)
|
||||
metadata = sqlalchemy.schema.MetaData()
|
||||
metadata.bind = engine
|
||||
def _check_152(self, engine, data):
|
||||
services = get_table(engine, 'services')
|
||||
service = services.select(services.c.id == 4).execute().first()
|
||||
self.assertEqual(0, service.deleted)
|
||||
service = services.select(services.c.id == 5).execute().first()
|
||||
self.assertEqual(service.id, service.deleted)
|
||||
|
||||
# NOTE(boris-42): It is enough to test one table with type of `id`
|
||||
# column Integer and one with type String.
|
||||
services = sqlalchemy.Table('services', metadata, autoload=True)
|
||||
volumes = sqlalchemy.Table('volumes', metadata, autoload=True)
|
||||
|
||||
engine.execute(
|
||||
services.insert(),
|
||||
[
|
||||
{'id': 1, 'host': host1, 'binary': 'nova-compute',
|
||||
'report_count': 0, 'topic': 'compute', 'deleted': False},
|
||||
{'id': 2, 'host': host1, 'binary': 'nova-compute',
|
||||
'report_count': 0, 'topic': 'compute', 'deleted': True}
|
||||
]
|
||||
)
|
||||
|
||||
engine.execute(
|
||||
volumes.insert(),
|
||||
[
|
||||
{'id': 'first', 'host': host1, 'deleted': False},
|
||||
{'id': 'second', 'host': host2, 'deleted': True}
|
||||
]
|
||||
)
|
||||
|
||||
_151_check(services, volumes)
|
||||
|
||||
migration_api.upgrade(engine, TestMigrations.REPOSITORY, 152)
|
||||
# NOTE(boris-42): One more time get from DB info about tables.
|
||||
metadata2 = sqlalchemy.schema.MetaData()
|
||||
metadata2.bind = engine
|
||||
|
||||
services = sqlalchemy.Table('services', metadata2, autoload=True)
|
||||
|
||||
service = services.select(services.c.id == 1).execute().first()
|
||||
self.assertEqual(0, service.deleted)
|
||||
service = services.select(services.c.id == 2).execute().first()
|
||||
self.assertEqual(service.id, service.deleted)
|
||||
|
||||
volumes = sqlalchemy.Table('volumes', metadata2, autoload=True)
|
||||
volume = volumes.select(volumes.c.id == "first").execute().first()
|
||||
self.assertEqual("", volume.deleted)
|
||||
volume = volumes.select(volumes.c.id == "second").execute().first()
|
||||
self.assertEqual(volume.id, volume.deleted)
|
||||
|
||||
migration_api.downgrade(engine, TestMigrations.REPOSITORY, 151)
|
||||
# NOTE(boris-42): One more time get from DB info about tables.
|
||||
metadata = sqlalchemy.schema.MetaData()
|
||||
metadata.bind = engine
|
||||
services = sqlalchemy.Table('services', metadata, autoload=True)
|
||||
volumes = sqlalchemy.Table('volumes', metadata, autoload=True)
|
||||
|
||||
_151_check(services, volumes)
|
||||
volumes = get_table(engine, 'volumes')
|
||||
volume = volumes.select(volumes.c.id == "first").execute().first()
|
||||
self.assertEqual("", volume.deleted)
|
||||
volume = volumes.select(volumes.c.id == "second").execute().first()
|
||||
self.assertEqual(volume.id, volume.deleted)
|
||||
|
||||
Reference in New Issue
Block a user