mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-19 10:51:34 +08:00
Modify behavior of legacy wrapper, add tests, refactor
If the database is combined, running `./manage_db.sh upgrade` should upgrade BOTH models: gxy and tsi. However, if tsi uses a separate database, only the galaxy model should be updated. Also, see this for context: https://github.com/galaxyproject/galaxy/pull/13108#discussion_r789108576 See inline comments for more details. The new tests are examples of new behavior. All refactoring is directly related to the changes in this commit.
This commit is contained in:
Sergey Golitsynskiy
@@ -4,6 +4,7 @@ from typing import (
|
||||
cast,
|
||||
Dict,
|
||||
Iterable,
|
||||
NamedTuple,
|
||||
NewType,
|
||||
NoReturn,
|
||||
Optional,
|
||||
@@ -30,7 +31,6 @@ from sqlalchemy.engine import (
|
||||
from galaxy.model import Base as gxy_base
|
||||
from galaxy.model.database_utils import create_database, database_exists
|
||||
from galaxy.model.mapping import create_additional_database_objects
|
||||
from galaxy.model.migrations.scripts import DatabaseConfig
|
||||
from galaxy.model.tool_shed_install import Base as tsi_base
|
||||
|
||||
ModelId = NewType('ModelId', str)
|
||||
@@ -47,6 +47,12 @@ SQLALCHEMYMIGRATE_LAST_VERSION_TSI = 17
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DatabaseConfig(NamedTuple):
|
||||
url: str
|
||||
template: str
|
||||
encoding: str
|
||||
|
||||
|
||||
class NoVersionTableError(Exception):
|
||||
# The database has no version table (neither SQLAlchemy Migrate, nor Alembic), so it is
|
||||
# impossible to automatically determine the state of the database. Manual update required.
|
||||
|
||||
@@ -1,12 +1,20 @@
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from typing import (
|
||||
List,
|
||||
NamedTuple,
|
||||
Optional,
|
||||
Tuple,
|
||||
)
|
||||
|
||||
import alembic.config
|
||||
|
||||
from galaxy.model.database_utils import is_one_database
|
||||
from galaxy.model.migrations import (
|
||||
DatabaseConfig,
|
||||
GXY,
|
||||
TSI,
|
||||
)
|
||||
from galaxy.util.properties import (
|
||||
find_config_file,
|
||||
get_data_dir,
|
||||
@@ -20,20 +28,6 @@ GXY_CONFIG_PREFIX = 'GALAXY_CONFIG_'
|
||||
TSI_CONFIG_PREFIX = 'GALAXY_INSTALL_CONFIG_'
|
||||
|
||||
|
||||
class DatabaseConfig(NamedTuple):
|
||||
url: str
|
||||
template: str
|
||||
encoding: str
|
||||
|
||||
|
||||
def _pop_config_file(argv: List[str]) -> Optional[str]:
|
||||
if CONFIG_FILE_ARG in argv:
|
||||
pos = argv.index(CONFIG_FILE_ARG)
|
||||
argv.pop(pos) # pop argument name
|
||||
return argv.pop(pos) # pop and return argument value
|
||||
return None
|
||||
|
||||
|
||||
def get_configuration(argv: List[str], cwd: str) -> Tuple[DatabaseConfig, DatabaseConfig, bool]:
|
||||
"""
|
||||
Return a 3-item-tuple with configuration values used for managing databases.
|
||||
@@ -63,10 +57,17 @@ def get_configuration(argv: List[str], cwd: str) -> Tuple[DatabaseConfig, Databa
|
||||
return (gxy_config, tsi_config, is_auto_migrate)
|
||||
|
||||
|
||||
def add_db_urls_to_command_arguments(argv: List[str], cwd: str) -> None:
|
||||
gxy_config, tsi_config, _ = get_configuration(argv, cwd)
|
||||
_insert_x_argument(argv, 'tsi_url', tsi_config.url)
|
||||
_insert_x_argument(argv, 'gxy_url', gxy_config.url)
|
||||
def _pop_config_file(argv: List[str]) -> Optional[str]:
|
||||
if CONFIG_FILE_ARG in argv:
|
||||
pos = argv.index(CONFIG_FILE_ARG)
|
||||
argv.pop(pos) # pop argument name
|
||||
return argv.pop(pos) # pop and return argument value
|
||||
return None
|
||||
|
||||
|
||||
def add_db_urls_to_command_arguments(argv: List[str], gxy_url: str, tsi_url: str) -> None:
|
||||
_insert_x_argument(argv, 'tsi_url', tsi_url)
|
||||
_insert_x_argument(argv, 'gxy_url', gxy_url)
|
||||
|
||||
|
||||
def _insert_x_argument(argv, key: str, value: str) -> None:
|
||||
@@ -75,63 +76,124 @@ def _insert_x_argument(argv, key: str, value: str) -> None:
|
||||
argv.insert(1, '-x')
|
||||
|
||||
|
||||
def invoke_alembic() -> None:
|
||||
"""
|
||||
Invoke the Alembic command line runner.
|
||||
|
||||
Accept 'heads' as the target revision argument to enable upgrading both gxy and tsi in one command.
|
||||
This is consistent with Alembic's CLI, which allows `upgrade heads`. However, this would not work for
|
||||
separate gxy and tsi databases: we can't attach a database url to a revision after Alembic has been
|
||||
invoked with the 'upgrade' command and the 'heads' argument. So, instead we invoke Alembic for each head.
|
||||
"""
|
||||
if 'heads' in sys.argv and 'upgrade' in sys.argv:
|
||||
i = sys.argv.index('heads')
|
||||
sys.argv[i] = f'{GXY}@head'
|
||||
alembic.config.main()
|
||||
sys.argv[i] = f'{TSI}@head'
|
||||
alembic.config.main()
|
||||
else:
|
||||
alembic.config.main()
|
||||
|
||||
|
||||
class LegacyScriptsException(Exception):
|
||||
# Misc. errors caused by incorrect arguments passed to a legacy script.
|
||||
def __init__(self, message: str) -> None:
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
class LegacyScripts:
|
||||
|
||||
LEGACY_CONFIG_FILE_ARG_NAMES = ['-c', '--config', '--config-file']
|
||||
ALEMBIC_CONFIG_FILE_ARG = '--alembic-config' # alembic config file, set in the calling script
|
||||
DEFAULT_DB_ARG = 'default'
|
||||
|
||||
def pop_database_argument(self, argv: List[str]) -> str:
|
||||
def __init__(self, argv: List[str], cwd: Optional[str] = None) -> None:
|
||||
self.argv = argv
|
||||
self.cwd = cwd or os.getcwd()
|
||||
self.database = self.DEFAULT_DB_ARG
|
||||
|
||||
def run(self) -> None:
|
||||
"""
|
||||
If last argument is a valid database name, pop and return it; otherwise return default.
|
||||
Convert legacy arguments to current spec required by Alembic,
|
||||
then add db url arguments required by Alembic
|
||||
"""
|
||||
arg = argv[-1]
|
||||
self.convert_args()
|
||||
add_db_urls_to_command_arguments(self.argv, self.gxy_url, self.tsi_url)
|
||||
|
||||
def convert_args(self) -> None:
|
||||
"""
|
||||
Convert legacy arguments to current spec required by Alembic.
|
||||
|
||||
Note: The following method calls must be done in this sequence.
|
||||
"""
|
||||
self.pop_database_argument()
|
||||
self.rename_config_argument()
|
||||
self.rename_alembic_config_argument()
|
||||
self._get_db_urls()
|
||||
self.convert_version_argument()
|
||||
|
||||
def pop_database_argument(self) -> None:
|
||||
"""
|
||||
If last argument is a valid database name, pop and assign it; otherwise assign default.
|
||||
"""
|
||||
arg = self.argv[-1]
|
||||
if arg in ['galaxy', 'install']:
|
||||
return argv.pop()
|
||||
return 'galaxy'
|
||||
self.database = self.argv.pop()
|
||||
|
||||
def rename_config_argument(self, argv: List[str]) -> None:
|
||||
def rename_config_argument(self) -> None:
|
||||
"""
|
||||
Rename the optional config argument: we can't use '-c' because that option is used by Alembic.
|
||||
"""
|
||||
for arg in self.LEGACY_CONFIG_FILE_ARG_NAMES:
|
||||
if arg in argv:
|
||||
self._rename_arg(argv, arg, CONFIG_FILE_ARG)
|
||||
if arg in self.argv:
|
||||
self._rename_arg(arg, CONFIG_FILE_ARG)
|
||||
return
|
||||
|
||||
def rename_alembic_config_argument(self, argv: List[str]) -> None:
|
||||
def rename_alembic_config_argument(self) -> None:
|
||||
"""
|
||||
Rename argument name: `--alembic-config` to `-c`. There should be no `-c` argument present.
|
||||
"""
|
||||
if '-c' in argv:
|
||||
raise Exception('Cannot rename alembic config argument: `-c` argument present.')
|
||||
self._rename_arg(argv, self.ALEMBIC_CONFIG_FILE_ARG, '-c')
|
||||
if '-c' in self.argv:
|
||||
raise LegacyScriptsException('Cannot rename alembic config argument: `-c` argument present.')
|
||||
self._rename_arg(self.ALEMBIC_CONFIG_FILE_ARG, '-c')
|
||||
|
||||
def convert_version_argument(self, argv: List[str], database: str) -> None:
|
||||
def convert_version_argument(self) -> None:
|
||||
"""
|
||||
Convert legacy version argument to current spec required by Alembic.
|
||||
"""
|
||||
if '--version' in argv:
|
||||
if '--version' in self.argv:
|
||||
# Just remove it: the following argument should be the version/revision identifier.
|
||||
pos = argv.index('--version')
|
||||
argv.pop(pos)
|
||||
pos = self.argv.index('--version')
|
||||
self.argv.pop(pos)
|
||||
else:
|
||||
# If we find --version=foo, extract foo and replace arg with foo (which is the revision identifier)
|
||||
p = re.compile(r'--version=([0-9A-Fa-f]+)')
|
||||
for i, arg in enumerate(argv):
|
||||
for i, arg in enumerate(self.argv):
|
||||
m = p.match(arg)
|
||||
if m:
|
||||
argv[i] = m.group(1)
|
||||
self.argv[i] = m.group(1)
|
||||
return
|
||||
# No version argumen found: construct branch@head argument for an upgrade operation.
|
||||
# No version argument found: construct argument for an upgrade operation.
|
||||
# Raise exception otherwise.
|
||||
if 'upgrade' not in argv:
|
||||
raise Exception('If no `--version` argument supplied, `upgrade` argument is requried')
|
||||
if 'upgrade' not in self.argv:
|
||||
raise LegacyScriptsException('If no `--version` argument supplied, `upgrade` argument is requried')
|
||||
|
||||
if database == 'galaxy':
|
||||
argv.append('gxy@head')
|
||||
elif database == 'install':
|
||||
argv.append('tsi@head')
|
||||
if self._is_one_database(): # upgrade both regardless of database argument
|
||||
self.argv.append('heads')
|
||||
else: # for separate databases, choose one
|
||||
if self.database in ['galaxy', self.DEFAULT_DB_ARG]:
|
||||
self.argv.append('gxy@head')
|
||||
elif self.database == 'install':
|
||||
self.argv.append('tsi@head')
|
||||
|
||||
def _rename_arg(self, argv, old_name, new_name) -> None:
|
||||
pos = argv.index(old_name)
|
||||
argv[pos] = new_name
|
||||
def _rename_arg(self, old_name, new_name) -> None:
|
||||
pos = self.argv.index(old_name)
|
||||
self.argv[pos] = new_name
|
||||
|
||||
def _get_db_urls(self) -> None:
|
||||
gxy_config, tsi_config, _ = get_configuration(self.argv, self.cwd)
|
||||
self.gxy_url = gxy_config.url
|
||||
self.tsi_url = tsi_config.url
|
||||
|
||||
def _is_one_database(self):
|
||||
return is_one_database(self.gxy_url, self.tsi_url)
|
||||
|
||||
@@ -7,7 +7,8 @@ INPUT: | OUTPUT:
|
||||
----------------------------------------------------------
|
||||
upgrade --version=foo | upgrade foo
|
||||
upgrade --version foo | upgrade foo
|
||||
upgrade | upgrade gxy@head
|
||||
upgrade | upgrade heads (if using a combined db for galaxy and install)
|
||||
upgrade | upgrade gxy@head (if using separate dbs for galaxy and install)
|
||||
upgrade install | upgrade tsi@head
|
||||
upgrade --version=bar install | upgrade bar
|
||||
upgrade -c path-to-galaxy.yml | upgrade --galaxy-config path-to-galaxy.yml gxy@head
|
||||
@@ -19,24 +20,18 @@ The optional `-c` argument name is renamed to `--galaxy-config`.
|
||||
import os
|
||||
import sys
|
||||
|
||||
import alembic.config
|
||||
|
||||
sys.path.insert(1, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, 'lib')))
|
||||
|
||||
from galaxy.model.migrations.scripts import (
|
||||
add_db_urls_to_command_arguments,
|
||||
invoke_alembic,
|
||||
LegacyScripts,
|
||||
)
|
||||
|
||||
|
||||
def run():
|
||||
ls = LegacyScripts()
|
||||
target_database = ls.pop_database_argument(sys.argv)
|
||||
ls.rename_config_argument(sys.argv)
|
||||
ls.rename_alembic_config_argument(sys.argv)
|
||||
ls.convert_version_argument(sys.argv, target_database)
|
||||
add_db_urls_to_command_arguments(sys.argv, os.getcwd())
|
||||
alembic.config.main()
|
||||
ls = LegacyScripts(sys.argv, os.getcwd())
|
||||
ls.run()
|
||||
invoke_alembic()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
+10
-20
@@ -8,33 +8,23 @@ import logging
|
||||
import os.path
|
||||
import sys
|
||||
|
||||
import alembic.config
|
||||
|
||||
sys.path.insert(1, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, 'lib')))
|
||||
|
||||
from galaxy.model.migrations import GXY, TSI
|
||||
from galaxy.model.migrations.scripts import add_db_urls_to_command_arguments
|
||||
from galaxy.model.migrations.scripts import (
|
||||
add_db_urls_to_command_arguments,
|
||||
get_configuration,
|
||||
invoke_alembic,
|
||||
)
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def invoke_alembic():
|
||||
add_db_urls_to_command_arguments(sys.argv, os.getcwd())
|
||||
|
||||
# Accept 'heads' as the target revision argument to enable upgrading both gxy and tsi in one command.
|
||||
# This is consistent with Alembic's CLI, which allows `upgrade heads`. However, this would not work for
|
||||
# separate gxy and tsi databases: we can't attach a database url to a revision after Alembic has been
|
||||
# invoked with the 'upgrade' command and the 'heads' argument. So, instead we invoke Alembic for each head.
|
||||
if 'heads' in sys.argv and 'upgrade' in sys.argv:
|
||||
i = sys.argv.index('heads')
|
||||
sys.argv[i] = f'{GXY}@head'
|
||||
alembic.config.main()
|
||||
sys.argv[i] = f'{TSI}@head'
|
||||
alembic.config.main()
|
||||
else:
|
||||
alembic.config.main()
|
||||
def run():
|
||||
gxy_config, tsi_config, _ = get_configuration(sys.argv, os.getcwd())
|
||||
add_db_urls_to_command_arguments(sys.argv, gxy_config.url, tsi_config.url)
|
||||
invoke_alembic()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
invoke_alembic()
|
||||
run()
|
||||
|
||||
@@ -1,81 +1,122 @@
|
||||
import pytest
|
||||
|
||||
from galaxy.model.migrations.scripts import LegacyScripts
|
||||
from galaxy.model.migrations.scripts import (
|
||||
LegacyScripts,
|
||||
LegacyScriptsException,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True) # set combined db for all tests
|
||||
def set_combined(monkeypatch):
|
||||
monkeypatch.setattr(LegacyScripts, '_is_one_database', lambda self: True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def set_separate(monkeypatch):
|
||||
monkeypatch.setattr(LegacyScripts, '_is_one_database', lambda self: False)
|
||||
|
||||
|
||||
class TestLegacyScripts():
|
||||
|
||||
def test_pop_database_name__pop_and_return(self):
|
||||
arg_value = 'install'
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc', arg_value]
|
||||
database = LegacyScripts().pop_database_argument(argv)
|
||||
assert database == arg_value
|
||||
@pytest.mark.parametrize('database_arg', ['galaxy', 'install'])
|
||||
def test_pop_database_name(self, database_arg):
|
||||
# arg_value = 'install'
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc', database_arg]
|
||||
ls = LegacyScripts(argv)
|
||||
|
||||
ls.pop_database_argument()
|
||||
assert ls.database == database_arg
|
||||
assert argv == ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc']
|
||||
|
||||
def test_pop_database_name__pop_and_return_default(self):
|
||||
arg_value = 'galaxy'
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc', arg_value]
|
||||
database = LegacyScripts().pop_database_argument(argv)
|
||||
assert database == arg_value
|
||||
assert argv == ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc']
|
||||
|
||||
def test_pop_database_name__return_default(self):
|
||||
def test_pop_database_name_use_default(self):
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc']
|
||||
database = LegacyScripts().pop_database_argument(argv)
|
||||
assert database == 'galaxy'
|
||||
ls = LegacyScripts(argv)
|
||||
ls.pop_database_argument()
|
||||
assert ls.database == LegacyScripts.DEFAULT_DB_ARG
|
||||
assert argv == ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc']
|
||||
|
||||
@pytest.mark.parametrize('arg_name', LegacyScripts.LEGACY_CONFIG_FILE_ARG_NAMES)
|
||||
def test_rename_config_arg(self, arg_name):
|
||||
# `-c|--config|__config-file` should be renamed to `--galaxy-config`
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', arg_name, 'path-to-galaxy', 'upgrade', '--version=abc']
|
||||
LegacyScripts().rename_config_argument(argv)
|
||||
LegacyScripts(argv).rename_config_argument()
|
||||
assert argv == ['caller', '--alembic-config', 'path-to-alembic', '--galaxy-config', 'path-to-galaxy', 'upgrade', '--version=abc']
|
||||
|
||||
def test_rename_config_arg_reordered_args(self):
|
||||
# `-c|--config|__config-file` should be renamed to `--galaxy-config`
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc', '-c', 'path-to-galaxy']
|
||||
LegacyScripts().rename_config_argument(argv)
|
||||
LegacyScripts(argv).rename_config_argument()
|
||||
assert argv == ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc', '--galaxy-config', 'path-to-galaxy']
|
||||
|
||||
def test_rename_alembic_config_arg(self):
|
||||
# `--alembic-config` should be renamed to `-c`
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc']
|
||||
LegacyScripts().rename_alembic_config_argument(argv)
|
||||
LegacyScripts(argv).rename_alembic_config_argument()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', '--version=abc']
|
||||
|
||||
def test_rename_alembic_config_arg_raises_error_if_c_arg_present(self):
|
||||
# Ensure alembic config arg is renamed AFTER renaming the galaxy config arg. Raise error otherwise.
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', '-c', 'path-to-galaxy', 'upgrade', '--version=abc']
|
||||
with pytest.raises(Exception):
|
||||
LegacyScripts().rename_alembic_config_argument(argv)
|
||||
with pytest.raises(LegacyScriptsException):
|
||||
LegacyScripts(argv).rename_alembic_config_argument()
|
||||
|
||||
def test_convert_version_argument_1(self):
|
||||
# `--version foo` should be converted to `foo`
|
||||
argv = ['caller', '-c', 'path-to-alembic', 'upgrade', '--version', 'abc']
|
||||
LegacyScripts().convert_version_argument(argv, 'galaxy')
|
||||
def test_convert__version_arg_1(self):
|
||||
# `sh manage_db.sh upgrade --version X` >> `... upgrade X`
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version', 'abc']
|
||||
LegacyScripts(argv).convert_args()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', 'abc']
|
||||
|
||||
def test_convert_version_argument_2(self):
|
||||
# `--version=foo` should be converted to `foo`
|
||||
argv = ['caller', '-c', 'path-to-alembic', 'upgrade', '--version=abc']
|
||||
LegacyScripts().convert_version_argument(argv, 'galaxy')
|
||||
def test_convert__version_arg_2(self):
|
||||
# `sh manage_db.sh upgrade --version=X` >> `... upgrade X`
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', '--version=abc']
|
||||
LegacyScripts(argv).convert_args()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', 'abc']
|
||||
|
||||
def test_no_version_argument(self):
|
||||
# No version should be converted to `X@head` where `X` is either gxy or tsi, depending on the target database.
|
||||
database = 'galaxy'
|
||||
argv = ['caller', '-c', 'path-to-alembic', 'upgrade']
|
||||
LegacyScripts().convert_version_argument(argv, database)
|
||||
def test_convert__no_version_no_model_combined_database(self):
|
||||
# `sh manage_db.sh upgrade` >> `... upgrade heads`
|
||||
# No version and no model implies "upgrade the default db (which is galaxy) to its latest version".
|
||||
# If it is combined, we upgrade both models: gxy and tsi.
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade']
|
||||
LegacyScripts(argv).convert_args()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', 'heads']
|
||||
|
||||
def test_convert__no_version_galaxy_model_combined_database_(self):
|
||||
# `sh manage_db.sh upgrade galaxy` >> `... upgrade heads`
|
||||
# same as no model: if combined we upgrade the whole database
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', 'galaxy']
|
||||
LegacyScripts(argv).convert_args()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', 'heads']
|
||||
|
||||
def test_convert__no_version_install_model_combined_database_(self):
|
||||
# `sh manage_db.sh upgrade install` >> `... upgrade heads`
|
||||
# same as no model: if combined we upgrade the whole database
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', 'install']
|
||||
LegacyScripts(argv).convert_args()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', 'heads']
|
||||
|
||||
def test_convert__no_version_no_model_separate_databases(self, set_separate):
|
||||
# `sh manage_db.sh upgrade` >> `... upgrade gxy@head`
|
||||
# No version and no model implies "upgrade the default db (which is galaxy) to its latest version".
|
||||
# Since the tsi model has its own db, we only upgrade the gxy model.
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade']
|
||||
LegacyScripts(argv).convert_args()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', 'gxy@head']
|
||||
|
||||
database = 'install'
|
||||
argv = ['caller', '-c', 'path-to-alembic', 'upgrade']
|
||||
LegacyScripts().convert_version_argument(argv, database)
|
||||
def test_convert__no_version_galaxy_model_separate_databases(self, set_separate):
|
||||
# `sh manage_db.sh upgrade galaxy` >> `... upgrade gxy@head`
|
||||
# No version + a model implies "upgrade the db for the specified model to its latest version".
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', 'galaxy']
|
||||
LegacyScripts(argv).convert_args()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', 'gxy@head']
|
||||
|
||||
def test_convert__no_version_install_model_separate_databases(self, set_separate):
|
||||
# `sh manage_db.sh upgrade install` >> `... upgrade tsi@head`
|
||||
# No version + a model implies "upgrade the db for the specified model to its latest version".
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'upgrade', 'install']
|
||||
LegacyScripts(argv).convert_args()
|
||||
assert argv == ['caller', '-c', 'path-to-alembic', 'upgrade', 'tsi@head']
|
||||
|
||||
def test_downgrade_with_no_version_argument_raises_error(self):
|
||||
database = 'galaxy'
|
||||
argv = ['caller', '-c', 'path-to-alembic', 'downgrade']
|
||||
with pytest.raises(Exception):
|
||||
LegacyScripts().convert_version_argument(argv, database)
|
||||
argv = ['caller', '--alembic-config', 'path-to-alembic', 'downgrade']
|
||||
with pytest.raises(LegacyScriptsException):
|
||||
LegacyScripts(argv).convert_args()
|
||||
|
||||
Reference in New Issue
Block a user