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

diff --git a/neutron/db/migration/cli.py b/neutron/db/migration/cli.py
index 51b49508a26b6d628f0d769c7e95bce9fb12a90d..5984d657c89f09831c5f81ff7e960c0c18686002 100644 (file)
--- a/neutron/db/migration/cli.py
+++ b/neutron/db/migration/cli.py
@@ -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:

diff --git a/neutron/tests/unit/db/test_migration.py b/neutron/tests/unit/db/test_migration.py
index 9d5588958f1a3596d3b3bcebaa606bc0a349a16c..8d52fa2e48d9a7b2f61795f8a56bcea7b19a9c0a 100644 (file)
--- a/neutron/tests/unit/db/test_migration.py
+++ b/neutron/tests/unit/db/test_migration.py
@@ -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()