openstack
/
neutron
Code Issues Proposed changes

neutron-db-manage: fix check_migration for branch-less migration directories 

I3823900bc5aaf7757c37edb804027cf4d9c757ab introduced support for
multi-branch migration directories in neutron-db-manage. That broke
check_migration for those projects without multiple branches.

The tool should properly handle both types of directories for forseable
future.

Related-Bug: #1475804
Change-Id: Ie4d20f5119018ffdebe2929b961c5ddb314ed734
This commit is contained in:
Ihar Hrachyshka 2015-07-18 00:48:30 +02:00
parent dbe7cac34b
commit 04f71b22e7
2 changed files with 54 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
neutron/db/migration/cli.py
View File

@ -27,6 +27,7 @@ from neutron.common import repos
# TODO(ihrachyshka): maintain separate HEAD files per branch
HEAD_FILENAME = 'HEAD'
HEADS_FILENAME = 'HEADS'
CURRENT_RELEASE = "liberty"
MIGRATION_BRANCHES = ('expand', 'contract')
@ -164,7 +165,7 @@ def validate_heads_file(config):
'''Check that HEADS file contains the latest heads for each branch.'''
script = alembic_script.ScriptDirectory.from_config(config)
expected_heads = _get_sorted_heads(script)
heads_path = _get_heads_file_path(CONF)
heads_path = _get_active_head_file_path(CONF)
try:
with open(heads_path) as file_:
observed_heads = file_.read().split()
@ -181,7 +182,7 @@ def update_heads_file(config):
'''Update HEADS file with the latest branch heads.'''
script = alembic_script.ScriptDirectory.from_config(config)
heads = _get_sorted_heads(script)
heads_path = _get_heads_file_path(CONF)
heads_path = _get_active_head_file_path(CONF)
with open(heads_path, 'w+') as f:
f.write('\n'.join(heads))
@ -247,6 +248,13 @@ def _get_root_versions_dir(neutron_config):
'db/migration/alembic_migrations/versions')
def _get_head_file_path(neutron_config):
'''Return the path of the file that contains single head.'''
return os.path.join(
_get_root_versions_dir(neutron_config),
HEAD_FILENAME)
def _get_heads_file_path(neutron_config):
'''Return the path of the file that contains all latest heads, sorted.'''
return os.path.join(
@ -254,6 +262,15 @@ def _get_heads_file_path(neutron_config):
HEADS_FILENAME)
def _get_active_head_file_path(neutron_config):
'''Return the path of the file that contains latest head(s), depending on
whether multiple branches are used.
'''
if _separate_migration_branches_supported(neutron_config):
return _get_heads_file_path(neutron_config)
return _get_head_file_path(neutron_config)
def _get_version_branch_path(neutron_config, branch=None):
version_path = _get_root_versions_dir(neutron_config)
if branch:

50
neutron/tests/unit/db/test_migration.py
View File

@ -22,6 +22,10 @@ from neutron.db.migration import cli
from neutron.tests import base
class FakeConfig(object):
service = ''
class TestDbMigration(base.BaseTestCase):
def setUp(self):
@ -112,9 +116,6 @@ class TestCli(base.BaseTestCase):
def _test_database_sync_revision(self, separate_branches=True):
with mock.patch.object(cli, 'update_heads_file') as update:
class FakeConfig(object):
service = ''
fake_config = FakeConfig()
if separate_branches:
expected_kwargs = [
@ -196,28 +197,38 @@ class TestCli(base.BaseTestCase):
def test_upgrade_rejects_delta_with_relative_revision(self):
self.assert_command_fails(['prog', 'upgrade', '+2', '--delta', '3'])
def _test_validate_heads_file_helper(self, heads, file_content=None):
def _test_validate_heads_file_helper(self, heads, file_heads=None,
branchless=False):
if file_heads is None:
file_heads = []
fake_config = FakeConfig()
with mock.patch('alembic.script.ScriptDirectory.from_config') as fc:
fc.return_value.get_heads.return_value = heads
fc.return_value.get_current_head.return_value = heads[0]
with mock.patch('six.moves.builtins.open') as mock_open:
mock_open.return_value.__enter__ = lambda s: s
mock_open.return_value.__exit__ = mock.Mock()
mock_open.return_value.read.return_value = file_content
mock_open.return_value.read.return_value = (
'\n'.join(file_heads))
with mock.patch('os.path.isfile') as is_file:
is_file.return_value = file_content is not None
is_file.return_value = bool(file_heads)
if file_content in heads:
cli.validate_heads_file(mock.sentinel.config)
if all(head in file_heads for head in heads):
cli.validate_heads_file(fake_config)
else:
self.assertRaises(
SystemExit,
cli.validate_heads_file,
mock.sentinel.config
fake_config
)
self.mock_alembic_err.assert_called_once_with(mock.ANY)
fc.assert_called_once_with(mock.sentinel.config)
if branchless:
mock_open.assert_called_with(
cli._get_head_file_path(fake_config))
else:
mock_open.assert_called_with(
cli._get_heads_file_path(fake_config))
fc.assert_called_once_with(fake_config)
def test_validate_heads_file_multiple_heads(self):
self._test_validate_heads_file_helper(['a', 'b'])
@ -226,10 +237,20 @@ class TestCli(base.BaseTestCase):
self._test_validate_heads_file_helper(['a'])
def test_validate_heads_file_wrong_contents(self):
self._test_validate_heads_file_helper(['a'], 'b')
self._test_validate_heads_file_helper(['a'], ['b'])
def test_validate_head_success(self):
self._test_validate_heads_file_helper(['a'], 'a')
def test_validate_heads_success(self):
self._test_validate_heads_file_helper(['a'], ['a'])
@mock.patch.object(cli, '_separate_migration_branches_supported',
return_value=False)
def test_validate_heads_file_branchless_failure(self, *args):
self._test_validate_heads_file_helper(['a'], ['b'], branchless=True)
@mock.patch.object(cli, '_separate_migration_branches_supported',
return_value=False)
def test_validate_heads_file_branchless_success(self, *args):
self._test_validate_heads_file_helper(['a'], ['a'], branchless=True)
def test_update_heads_file_two_heads(self):
with mock.patch('alembic.script.ScriptDirectory.from_config') as fc:
@ -258,7 +279,6 @@ class TestCli(base.BaseTestCase):
with mock.patch('alembic.script.ScriptDirectory.from_config') as fc:
heads = ('a', 'b')
fc.return_value.get_heads.return_value = heads
fc.return_value.get_current_head.return_value = heads
with mock.patch('six.moves.builtins.open') as mock_open:
mock_open.return_value.__enter__ = lambda s: s
mock_open.return_value.__exit__ = mock.Mock()