witten
/
luminotes
Archived
1
0
Fork 0

Improved schema upgraded for the case when no new database schemas are available. Updated unit tests accordingly.

This commit is contained in:
Dan Helfman 2009-03-06 15:14:17 -08:00
parent d0f9c20e69
commit 27f5f4e1a9
2 changed files with 30 additions and 6 deletions

View File

@ -77,6 +77,10 @@ class Schema_upgrader:
versions.append( ( version, filename ) ) versions.append( ( version, filename ) )
if len( versions ) == 0:
print "no new database schemas available"
return
# sort the schema delta files by version # sort the schema delta files by version
versions.sort( lambda a, b: cmp( a[ 0 ], b[ 0 ] ) ) versions.sort( lambda a, b: cmp( a[ 0 ], b[ 0 ] ) )
@ -91,6 +95,7 @@ class Schema_upgrader:
def schema_version( database, default_version = None ): def schema_version( database, default_version = None ):
try: try:
schema_version = database.select_one( tuple, "select * from schema_version;" ); schema_version = database.select_one( tuple, "select * from schema_version;" );
schema_version = tuple( map( int, schema_version ) )
# if there's no schema version table, then use the default version given. if there's no default # if there's no schema version table, then use the default version given. if there's no default
# version, then assume the from_version is 1.5.4, which was the last version not to include a # version, then assume the from_version is 1.5.4, which was the last version not to include a
# schema_version table # schema_version table

View File

@ -42,29 +42,38 @@ class Test_schema_upgrader( object ):
return contents return contents
def test_upgrade_schema( self ): def test_upgrade_schema( self, to_version = None ):
if not to_version:
to_version = u"5.7.11"
self.fake_files = { self.fake_files = {
u"model/delta/5.6.7.sqlite": u"create table new_table ( foo text ); insert into new_table values ( 'hi' );", u"model/delta/5.6.7.sqlite": u"create table new_table ( foo text ); insert into new_table values ( 'hi' );",
u"model/delta/5.6.8.sqlite": u"insert into new_table values ( 'bye' );", u"model/delta/5.6.8.sqlite": u"insert into new_table values ( 'bye' );",
u"model/delta/5.6.10.sqlite": u"alter table new_table add column bar text;", u"model/delta/5.6.10.sqlite": u"alter table new_table add column bar text;",
u"model/delta/5.7.11.sqlite": u"insert into new_table values ( 'whee', 'stuff' );", u"model/delta/5.7.11.sqlite": u"insert into new_table values ( 'whee', 'stuff' );",
u"model/delta/5.7.18.sqlite": u"insert into new_table values ( 'should not be present', 'nope' );", u"model/delta/5.7.18.sqlite": u"insert into new_table values ( 'more', 'things' );",
} }
self.upgrader.upgrade_schema( u"5.7.11" ) self.upgrader.upgrade_schema( to_version )
result = self.database.select_many( tuple, u"select * from new_table;" ) result = self.database.select_many( tuple, u"select * from new_table;" )
assert result == [ ( u"hi", None ), ( u"bye", None ), ( "whee", "stuff" ) ] if to_version == u"5.7.11":
assert result == [ ( u"hi", None ), ( u"bye", None ), ( "whee", "stuff" ) ]
else:
assert result == [ ( u"hi", None ), ( u"bye", None ), ( "whee", "stuff" ), ( "more", "things" ) ]
result = self.database.select_many( tuple, u"select * from schema_version;" ) result = self.database.select_many( tuple, u"select * from schema_version;" )
assert result == [ ( 5, 7, 11 ) ] if to_version == u"5.7.11":
assert result == [ ( 5, 7, 11 ) ]
else:
assert result == [ ( 5, 7, 18 ) ]
def test_upgrade_schema_with_schema_version_table( self ): def test_upgrade_schema_with_schema_version_table( self ):
self.database.execute( u"create table schema_version ( major numeric, minor numeric, \"release\" numeric );" ) self.database.execute( u"create table schema_version ( major numeric, minor numeric, \"release\" numeric );" )
self.database.execute( u"insert into schema_version values ( 0, 0, 0 );" ) self.database.execute( u"insert into schema_version values ( 0, 0, 0 );" )
self.test_upgrade_schema() self.test_upgrade_schema()
def test_upgrade_schema_with_schema_version_table_and_specific_starting_version( self ): def test_upgrade_schema_with_schema_version_table_and_starting_version( self ):
self.database.execute( u"create table schema_version ( major numeric, minor numeric, \"release\" numeric );" ) self.database.execute( u"create table schema_version ( major numeric, minor numeric, \"release\" numeric );" )
self.database.execute( u"insert into schema_version values ( 5, 6, 6 );" ) self.database.execute( u"insert into schema_version values ( 5, 6, 6 );" )
@ -73,6 +82,16 @@ class Test_schema_upgrader( object ):
self.test_upgrade_schema() self.test_upgrade_schema()
def test_upgrade_schema_with_schema_version_table_and_target_version_without_schema( self ):
self.database.execute( u"create table schema_version ( major numeric, minor numeric, \"release\" numeric );" )
self.database.execute( u"insert into schema_version values ( 0, 0, 0 );" )
self.test_upgrade_schema( to_version = u"5.7.20" )
def test_upgrade_schema_with_schema_version_table_and_starting_version_and_target_version_without_schema( self ):
self.database.execute( u"create table schema_version ( major numeric, minor numeric, \"release\" numeric );" )
self.database.execute( u"insert into schema_version values ( 5, 6, 6 );" )
self.test_upgrade_schema( to_version = u"5.7.20" )
def test_upgrade_schema_with_future_ending_version( self ): def test_upgrade_schema_with_future_ending_version( self ):
self.fake_files = { self.fake_files = {
u"model/delta/5.6.7.sqlite": u"create table new_table ( foo text ); insert into new_table values ( 'hi' );", u"model/delta/5.6.7.sqlite": u"create table new_table ( foo text ); insert into new_table values ( 'hi' );",