forked from openlp/openlp
This prevents updates from running (which should only run on outdated databases that existed before)
bzr-revno: 2408
This commit is contained in:
commit
16399e2ea5
@ -96,9 +96,10 @@ def upgrade_db(url, upgrade):
|
|||||||
mapper(Metadata, metadata_table)
|
mapper(Metadata, metadata_table)
|
||||||
version_meta = session.query(Metadata).get('version')
|
version_meta = session.query(Metadata).get('version')
|
||||||
if version_meta is None:
|
if version_meta is None:
|
||||||
version_meta = Metadata.populate(key='version', value='0')
|
# Tables have just been created - fill the version field with the most recent version
|
||||||
|
version = upgrade.__version__
|
||||||
|
version_meta = Metadata.populate(key='version', value=version)
|
||||||
session.add(version_meta)
|
session.add(version_meta)
|
||||||
version = 0
|
|
||||||
else:
|
else:
|
||||||
version = int(version_meta.value)
|
version = int(version_meta.value)
|
||||||
if version > upgrade.__version__:
|
if version > upgrade.__version__:
|
||||||
|
@ -33,7 +33,6 @@ backend for the Songs plugin
|
|||||||
import logging
|
import logging
|
||||||
|
|
||||||
from sqlalchemy import Column, ForeignKey, types
|
from sqlalchemy import Column, ForeignKey, types
|
||||||
from sqlalchemy.exc import OperationalError
|
|
||||||
from sqlalchemy.sql.expression import func, false, null, text
|
from sqlalchemy.sql.expression import func, false, null, text
|
||||||
|
|
||||||
from openlp.core.lib.db import get_upgrade_op
|
from openlp.core.lib.db import get_upgrade_op
|
||||||
@ -57,16 +56,13 @@ def upgrade_1(session, metadata):
|
|||||||
:param session:
|
:param session:
|
||||||
:param metadata:
|
:param metadata:
|
||||||
"""
|
"""
|
||||||
try:
|
op = get_upgrade_op(session)
|
||||||
op = get_upgrade_op(session)
|
op.drop_table('media_files_songs')
|
||||||
op.drop_table('media_files_songs')
|
op.add_column('media_files', Column('song_id', types.Integer(), server_default=null()))
|
||||||
op.add_column('media_files', Column('song_id', types.Integer(), server_default=null()))
|
op.add_column('media_files', Column('weight', types.Integer(), server_default=text('0')))
|
||||||
op.add_column('media_files', Column('weight', types.Integer(), server_default=text('0')))
|
if metadata.bind.url.get_dialect().name != 'sqlite':
|
||||||
if metadata.bind.url.get_dialect().name != 'sqlite':
|
# SQLite doesn't support ALTER TABLE ADD CONSTRAINT
|
||||||
# SQLite doesn't support ALTER TABLE ADD CONSTRAINT
|
op.create_foreign_key('fk_media_files_song_id', 'media_files', 'songs', ['song_id', 'id'])
|
||||||
op.create_foreign_key('fk_media_files_song_id', 'media_files', 'songs', ['song_id', 'id'])
|
|
||||||
except OperationalError:
|
|
||||||
log.info('Upgrade 1 has already been run')
|
|
||||||
|
|
||||||
|
|
||||||
def upgrade_2(session, metadata):
|
def upgrade_2(session, metadata):
|
||||||
@ -75,12 +71,9 @@ def upgrade_2(session, metadata):
|
|||||||
|
|
||||||
This upgrade adds a create_date and last_modified date to the songs table
|
This upgrade adds a create_date and last_modified date to the songs table
|
||||||
"""
|
"""
|
||||||
try:
|
op = get_upgrade_op(session)
|
||||||
op = get_upgrade_op(session)
|
op.add_column('songs', Column('create_date', types.DateTime(), default=func.now()))
|
||||||
op.add_column('songs', Column('create_date', types.DateTime(), default=func.now()))
|
op.add_column('songs', Column('last_modified', types.DateTime(), default=func.now()))
|
||||||
op.add_column('songs', Column('last_modified', types.DateTime(), default=func.now()))
|
|
||||||
except OperationalError:
|
|
||||||
log.info('Upgrade 2 has already been run')
|
|
||||||
|
|
||||||
|
|
||||||
def upgrade_3(session, metadata):
|
def upgrade_3(session, metadata):
|
||||||
@ -89,14 +82,11 @@ def upgrade_3(session, metadata):
|
|||||||
|
|
||||||
This upgrade adds a temporary song flag to the songs table
|
This upgrade adds a temporary song flag to the songs table
|
||||||
"""
|
"""
|
||||||
try:
|
op = get_upgrade_op(session)
|
||||||
op = get_upgrade_op(session)
|
if metadata.bind.url.get_dialect().name == 'sqlite':
|
||||||
if metadata.bind.url.get_dialect().name == 'sqlite':
|
op.add_column('songs', Column('temporary', types.Boolean(create_constraint=False), server_default=false()))
|
||||||
op.add_column('songs', Column('temporary', types.Boolean(create_constraint=False), server_default=false()))
|
else:
|
||||||
else:
|
op.add_column('songs', Column('temporary', types.Boolean(), server_default=false()))
|
||||||
op.add_column('songs', Column('temporary', types.Boolean(), server_default=false()))
|
|
||||||
except OperationalError:
|
|
||||||
log.info('Upgrade 3 has already been run')
|
|
||||||
|
|
||||||
|
|
||||||
def upgrade_4(session, metadata):
|
def upgrade_4(session, metadata):
|
||||||
@ -105,17 +95,14 @@ def upgrade_4(session, metadata):
|
|||||||
|
|
||||||
This upgrade adds a column for author type to the authors_songs table
|
This upgrade adds a column for author type to the authors_songs table
|
||||||
"""
|
"""
|
||||||
try:
|
# Since SQLite doesn't support changing the primary key of a table, we need to recreate the table
|
||||||
# Since SQLite doesn't support changing the primary key of a table, we need to recreate the table
|
# and copy the old values
|
||||||
# and copy the old values
|
op = get_upgrade_op(session)
|
||||||
op = get_upgrade_op(session)
|
op.create_table('authors_songs_tmp',
|
||||||
op.create_table('authors_songs_tmp',
|
Column('author_id', types.Integer(), ForeignKey('authors.id'), primary_key=True),
|
||||||
Column('author_id', types.Integer(), ForeignKey('authors.id'), primary_key=True),
|
Column('song_id', types.Integer(), ForeignKey('songs.id'), primary_key=True),
|
||||||
Column('song_id', types.Integer(), ForeignKey('songs.id'), primary_key=True),
|
Column('author_type', types.String(), primary_key=True,
|
||||||
Column('author_type', types.String(), primary_key=True,
|
nullable=False, server_default=text('""')))
|
||||||
nullable=False, server_default=text('""')))
|
op.execute('INSERT INTO authors_songs_tmp SELECT author_id, song_id, "" FROM authors_songs')
|
||||||
op.execute('INSERT INTO authors_songs_tmp SELECT author_id, song_id, "" FROM authors_songs')
|
op.drop_table('authors_songs')
|
||||||
op.drop_table('authors_songs')
|
op.rename_table('authors_songs_tmp', 'authors_songs')
|
||||||
op.rename_table('authors_songs_tmp', 'authors_songs')
|
|
||||||
except OperationalError:
|
|
||||||
log.info('Upgrade 4 has already been run')
|
|
||||||
|
@ -32,7 +32,6 @@ backend for the SongsUsage plugin
|
|||||||
"""
|
"""
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
from sqlalchemy.exc import OperationalError
|
|
||||||
from sqlalchemy import Column, types
|
from sqlalchemy import Column, types
|
||||||
|
|
||||||
from openlp.core.lib.db import get_upgrade_op
|
from openlp.core.lib.db import get_upgrade_op
|
||||||
@ -50,9 +49,6 @@ def upgrade_1(session, metadata):
|
|||||||
:param session: SQLAlchemy Session object
|
:param session: SQLAlchemy Session object
|
||||||
:param metadata: SQLAlchemy MetaData object
|
:param metadata: SQLAlchemy MetaData object
|
||||||
"""
|
"""
|
||||||
try:
|
op = get_upgrade_op(session)
|
||||||
op = get_upgrade_op(session)
|
op.add_column('songusage_data', Column('plugin_name', types.Unicode(20), server_default=''))
|
||||||
op.add_column('songusage_data', Column('plugin_name', types.Unicode(20), server_default=''))
|
op.add_column('songusage_data', Column('source', types.Unicode(10), server_default=''))
|
||||||
op.add_column('songusage_data', Column('source', types.Unicode(10), server_default=''))
|
|
||||||
except OperationalError:
|
|
||||||
log.info('Upgrade 1 has already taken place')
|
|
||||||
|
@ -125,3 +125,31 @@ class TestDB(TestCase):
|
|||||||
|
|
||||||
# THEN: The type should be correct
|
# THEN: The type should be correct
|
||||||
self.assertEqual(author_type, AuthorType.Words)
|
self.assertEqual(author_type, AuthorType.Words)
|
||||||
|
|
||||||
|
def test_author_get_display_name(self):
|
||||||
|
"""
|
||||||
|
Test that the display name of an author is correct
|
||||||
|
"""
|
||||||
|
# GIVEN: An author
|
||||||
|
author = Author()
|
||||||
|
author.display_name = "John Doe"
|
||||||
|
|
||||||
|
# WHEN: We call the get_display_name() function
|
||||||
|
display_name = author.get_display_name()
|
||||||
|
|
||||||
|
# THEN: It should return only the name
|
||||||
|
self.assertEqual("John Doe", display_name)
|
||||||
|
|
||||||
|
def test_author_get_display_name_with_type(self):
|
||||||
|
"""
|
||||||
|
Test that the display name of an author with a type is correct
|
||||||
|
"""
|
||||||
|
# GIVEN: An author
|
||||||
|
author = Author()
|
||||||
|
author.display_name = "John Doe"
|
||||||
|
|
||||||
|
# WHEN: We call the get_display_name() function
|
||||||
|
display_name = author.get_display_name(AuthorType.Words)
|
||||||
|
|
||||||
|
# THEN: It should return the name with the type in brackets
|
||||||
|
self.assertEqual("John Doe (Words)", display_name)
|
||||||
|
Loading…
Reference in New Issue
Block a user