# Copyright 2012 New Dream Network, LLC (DreamHost)
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import copy
import os
import re
import sys
import textwrap

from alembic.autogenerate import api as alembic_ag_api
from alembic import config as alembic_config
from alembic.operations import ops as alembic_ops
from alembic import script as alembic_script
import fixtures
import mock
from neutron_lib.utils import helpers
from oslo_utils import fileutils
import pkg_resources
import sqlalchemy as sa
from testtools import matchers

from neutron.conf.db import migration_cli
from neutron.db import migration
from neutron.db.migration import autogen
from neutron.db.migration import cli
from neutron.tests import base
from neutron.tests import tools
from neutron.tests.unit import testlib_api


class FakeConfig(object):
    service = ''


class FakeRevision(object):
    path = 'fakepath'

    def __init__(self, labels=None, down_revision=None, is_branch_point=False):
        if not labels:
            labels = set()
        self.branch_labels = labels
        self.down_revision = down_revision
        self.is_branch_point = is_branch_point
        self.revision = helpers.get_random_string(10)
        self.module = mock.MagicMock()


class MigrationEntrypointsMemento(fixtures.Fixture):
    '''Create a copy of the migration entrypoints map so it can be restored
       during test cleanup.
    '''

    def _setUp(self):
        self.ep_backup = {}
        for proj, ep in migration_cli.migration_entrypoints.items():
            self.ep_backup[proj] = copy.copy(ep)
        self.addCleanup(self.restore)

    def restore(self):
        migration_cli.migration_entrypoints = self.ep_backup


class TestDbMigration(base.BaseTestCase):

    def setUp(self):
        super(TestDbMigration, self).setUp()
        mock.patch('alembic.op.get_bind').start()
        self.mock_alembic_is_offline = mock.patch(
            'alembic.context.is_offline_mode', return_value=False).start()
        self.mock_alembic_is_offline.return_value = False
        self.mock_sa_inspector = mock.patch(
            'sqlalchemy.engine.reflection.Inspector').start()

    def _prepare_mocked_sqlalchemy_inspector(self):
        mock_inspector = mock.MagicMock()
        mock_inspector.get_table_names.return_value = ['foo', 'bar']
        mock_inspector.get_columns.return_value = [{'name': 'foo_column'},
                                                   {'name': 'bar_column'}]
        self.mock_sa_inspector.from_engine.return_value = mock_inspector

    def test_schema_has_table(self):
        self._prepare_mocked_sqlalchemy_inspector()
        self.assertTrue(migration.schema_has_table('foo'))

    def test_schema_has_table_raises_if_offline(self):
        self.mock_alembic_is_offline.return_value = True
        self.assertRaises(RuntimeError, migration.schema_has_table, 'foo')

    def test_schema_has_column_missing_table(self):
        self._prepare_mocked_sqlalchemy_inspector()
        self.assertFalse(migration.schema_has_column('meh', 'meh'))

    def test_schema_has_column(self):
        self._prepare_mocked_sqlalchemy_inspector()
        self.assertTrue(migration.schema_has_column('foo', 'foo_column'))

    def test_schema_has_column_raises_if_offline(self):
        self.mock_alembic_is_offline.return_value = True
        self.assertRaises(RuntimeError, migration.schema_has_column,
                          'foo', 'foo_col')

    def test_schema_has_column_missing_column(self):
        self._prepare_mocked_sqlalchemy_inspector()
        self.assertFalse(migration.schema_has_column(
            'foo', column_name='meh'))


class TestCli(base.BaseTestCase):
    def setUp(self):
        super(TestCli, self).setUp()
        self.do_alembic_cmd_p = mock.patch.object(cli, 'do_alembic_command')
        self.do_alembic_cmd = self.do_alembic_cmd_p.start()
        self.mock_alembic_err = mock.patch('alembic.util.err').start()
        self.mock_alembic_warn = mock.patch('alembic.util.warn').start()
        self.mock_alembic_err.side_effect = SystemExit

        def mocked_root_dir(cfg):
            return os.path.join('/fake/dir', cli._get_project_base(cfg))
        mock_root = mock.patch.object(cli, '_get_package_root_dir').start()
        mock_root.side_effect = mocked_root_dir
        # Avoid creating fake directories
        mock.patch('oslo_utils.fileutils.ensure_tree').start()

        # Set up some configs and entrypoints for tests to chew on
        self.configs = []
        self.projects = ('neutron', 'networking-foo', 'neutron-fwaas')
        ini = os.path.join(os.path.dirname(cli.__file__), 'alembic.ini')
        self.useFixture(MigrationEntrypointsMemento())
        migration_cli.migration_entrypoints = {}
        for project in self.projects:
            config = alembic_config.Config(ini)
            config.set_main_option('neutron_project', project)
            module_name = project.replace('-', '_') + '.db.migration'
            attrs = ('alembic_migrations',)
            script_location = ':'.join([module_name, attrs[0]])
            config.set_main_option('script_location', script_location)
            self.configs.append(config)
            entrypoint = pkg_resources.EntryPoint(project,
                                                  module_name,
                                                  attrs=attrs)
            migration_cli.migration_entrypoints[project] = entrypoint

    def _main_test_helper(self, argv, func_name, exp_kwargs=[{}]):
        with mock.patch.object(sys, 'argv', argv),\
                mock.patch.object(cli, 'run_sanity_checks'),\
                mock.patch.object(cli, 'validate_revisions'):

            cli.main()

            def _append_version_path(args):
                args = copy.copy(args)
                if 'autogenerate' in args and not args['autogenerate']:
                    args['version_path'] = mock.ANY
                return args

            self.do_alembic_cmd.assert_has_calls(
                [mock.call(mock.ANY, func_name, **_append_version_path(kwargs))
                 for kwargs in exp_kwargs]
            )

    def test_stamp(self):
        self._main_test_helper(
            ['prog', 'stamp', 'foo'],
            'stamp',
            [{'revision': 'foo', 'sql': False}]
        )

        self._main_test_helper(
            ['prog', 'stamp', 'foo', '--sql'],
            'stamp',
            [{'revision': 'foo', 'sql': True}]
        )

    def _validate_cmd(self, cmd):
        self._main_test_helper(
            ['prog', cmd],
            cmd,
            [{'verbose': False}])

        self._main_test_helper(
            ['prog', cmd, '--verbose'],
            cmd,
            [{'verbose': True}])

    def test_branches(self):
        self._validate_cmd('branches')

    def test_current(self):
        self._validate_cmd('current')

    def test_history(self):
        self._validate_cmd('history')

    def test_heads(self):
        self._validate_cmd('heads')

    def test_check_migration(self):
        with mock.patch.object(cli, 'validate_head_files') as validate:
            self._main_test_helper(['prog', 'check_migration'], 'branches')
            self.assertEqual(len(self.projects), validate.call_count)

    def _test_database_sync_revision(self, separate_branches=True):
        with mock.patch.object(cli, 'update_head_files') as update:
            if separate_branches:
                mock.patch('os.path.exists').start()
            expected_kwargs = [{
                'message': 'message', 'sql': False, 'autogenerate': True,
            }]
            self._main_test_helper(
                ['prog', 'revision', '--autogenerate', '-m', 'message'],
                'revision',
                expected_kwargs
            )
            self.assertEqual(len(self.projects), update.call_count)
            update.reset_mock()

            expected_kwargs = [{
                'message': 'message',
                'sql': True,
                'autogenerate': False,
                'head': cli._get_branch_head(branch)
            } for branch in cli.MIGRATION_BRANCHES]
            for kwarg in expected_kwargs:
                kwarg['autogenerate'] = False
                kwarg['sql'] = True

            self._main_test_helper(
                ['prog', 'revision', '--sql', '-m', 'message'],
                'revision',
                expected_kwargs
            )
            self.assertEqual(len(self.projects), update.call_count)
            update.reset_mock()

            expected_kwargs = [{
                'message': 'message',
                'sql': False,
                'autogenerate': False,
                'head': 'expand@head'
            }]
            self._main_test_helper(
                ['prog', 'revision', '-m', 'message', '--expand'],
                'revision',
                expected_kwargs
            )
            self.assertEqual(len(self.projects), update.call_count)
            update.reset_mock()

            for kwarg in expected_kwargs:
                kwarg['head'] = 'contract@head'

            self._main_test_helper(
                ['prog', 'revision', '-m', 'message', '--contract'],
                'revision',
                expected_kwargs
            )
            self.assertEqual(len(self.projects), update.call_count)

    def test_database_sync_revision(self):
        self._test_database_sync_revision()

    def test_database_sync_revision_no_branches(self):
        # Test that old branchless approach is still supported
        self._test_database_sync_revision(separate_branches=False)

    def test_upgrade_revision(self):
        self._main_test_helper(
            ['prog', 'upgrade', '--sql', 'head'],
            'upgrade',
            [{'desc': None, 'revision': 'heads', 'sql': True}]
        )

    def test_upgrade_delta(self):
        self._main_test_helper(
            ['prog', 'upgrade', '--delta', '3'],
            'upgrade',
            [{'desc': None, 'revision': '+3', 'sql': False}]
        )

    def test_upgrade_revision_delta(self):
        self._main_test_helper(
            ['prog', 'upgrade', 'kilo', '--delta', '3'],
            'upgrade',
            [{'desc': None, 'revision': 'kilo+3', 'sql': False}]
        )

    def test_upgrade_expand(self):
        self._main_test_helper(
            ['prog', 'upgrade', '--expand'],
            'upgrade',
            [{'desc': cli.EXPAND_BRANCH,
              'revision': 'expand@head',
              'sql': False}]
        )

    def test_upgrade_expand_contract_are_mutually_exclusive(self):
        with testlib_api.ExpectedException(SystemExit):
            self._main_test_helper(
                ['prog', 'upgrade', '--expand --contract'], 'upgrade')

    def _test_upgrade_conflicts_with_revision(self, mode):
        with testlib_api.ExpectedException(SystemExit):
            self._main_test_helper(
                ['prog', 'upgrade', '--%s revision1' % mode], 'upgrade')

    def _test_upgrade_conflicts_with_delta(self, mode):
        with testlib_api.ExpectedException(SystemExit):
            self._main_test_helper(
                ['prog', 'upgrade', '--%s +3' % mode], 'upgrade')

    def _test_revision_autogenerate_conflicts_with_branch(self, branch):
        with testlib_api.ExpectedException(SystemExit):
            self._main_test_helper(
                ['prog', 'revision', '--autogenerate', '--%s' % branch],
                'revision')

    def test_revision_autogenerate_conflicts_with_expand(self):
        self._test_revision_autogenerate_conflicts_with_branch(
            cli.EXPAND_BRANCH)

    def test_revision_autogenerate_conflicts_with_contract(self):
        self._test_revision_autogenerate_conflicts_with_branch(
            cli.CONTRACT_BRANCH)

    def test_upgrade_expand_conflicts_with_revision(self):
        self._test_upgrade_conflicts_with_revision('expand')

    def test_upgrade_contract_conflicts_with_revision(self):
        self._test_upgrade_conflicts_with_revision('contract')

    def test_upgrade_expand_conflicts_with_delta(self):
        self._test_upgrade_conflicts_with_delta('expand')

    def test_upgrade_contract_conflicts_with_delta(self):
        self._test_upgrade_conflicts_with_delta('contract')

    def test_upgrade_contract(self):
        self._main_test_helper(
            ['prog', 'upgrade', '--contract'],
            'upgrade',
            [{'desc': cli.CONTRACT_BRANCH,
              'revision': 'contract@head',
              'sql': False}]
        )

    @mock.patch('alembic.script.ScriptDirectory.walk_revisions')
    def test_upgrade_milestone_expand_before_contract(self, walk_mock):
        c_revs = [FakeRevision(labels={cli.CONTRACT_BRANCH}) for r in range(5)]
        c_revs[1].module.neutron_milestone = [migration.LIBERTY]
        e_revs = [FakeRevision(labels={cli.EXPAND_BRANCH}) for r in range(5)]
        e_revs[3].module.neutron_milestone = [migration.LIBERTY]
        walk_mock.return_value = c_revs + e_revs
        self._main_test_helper(
            ['prog', '--subproject', 'neutron', 'upgrade', 'liberty'],
            'upgrade',
            [{'desc': cli.EXPAND_BRANCH,
              'revision': e_revs[3].revision,
              'sql': False},
             {'desc': cli.CONTRACT_BRANCH,
              'revision': c_revs[1].revision,
              'sql': False}]
        )

    def assert_command_fails(self, command):
        # Avoid cluttering stdout with argparse error messages
        mock.patch('argparse.ArgumentParser._print_message').start()
        with mock.patch.object(sys, 'argv', command), mock.patch.object(
                cli, 'run_sanity_checks'):
            self.assertRaises(SystemExit, cli.main)

    def test_downgrade_fails(self):
        self.assert_command_fails(['prog', 'downgrade', '--sql', 'juno'])

    def test_upgrade_negative_relative_revision_fails(self):
        self.assert_command_fails(['prog', 'upgrade', '-2'])

    def test_upgrade_negative_delta_fails(self):
        self.assert_command_fails(['prog', 'upgrade', '--delta', '-2'])

    def test_upgrade_rejects_delta_with_relative_revision(self):
        self.assert_command_fails(['prog', 'upgrade', '+2', '--delta', '3'])

    def _test_validate_head_files_helper(self, heads, contract_head='',
                                         expand_head=''):
        fake_config = self.configs[0]
        head_files_not_exist = (contract_head == expand_head == '')
        with mock.patch('alembic.script.ScriptDirectory.from_config') as fc,\
                mock.patch('os.path.exists') as os_mock:
            if head_files_not_exist:
                os_mock.return_value = False
            else:
                os_mock.return_value = True

            fc.return_value.get_heads.return_value = heads

            revs = {heads[0]: FakeRevision(labels=cli.CONTRACT_BRANCH),
                    heads[1]: FakeRevision(labels=cli.EXPAND_BRANCH)}
            fc.return_value.get_revision.side_effect = revs.__getitem__
            mock_open_con = self.useFixture(
                tools.OpenFixture(cli._get_contract_head_file_path(
                    fake_config), contract_head + '\n')).mock_open
            mock_open_ex = self.useFixture(
                tools.OpenFixture(cli._get_expand_head_file_path(
                    fake_config), expand_head + '\n')).mock_open

            if contract_head in heads and expand_head in heads:
                cli.validate_head_files(fake_config)
            elif head_files_not_exist:
                cli.validate_head_files(fake_config)
                self.assertTrue(self.mock_alembic_warn.called)
            else:
                self.assertRaises(
                    SystemExit,
                    cli.validate_head_files,
                    fake_config
                )
                self.assertTrue(self.mock_alembic_err.called)

            if contract_head in heads and expand_head in heads:
                mock_open_ex.assert_called_with(
                    cli._get_expand_head_file_path(fake_config))
                mock_open_con.assert_called_with(
                    cli._get_contract_head_file_path(fake_config))

            if not head_files_not_exist:
                fc.assert_called_once_with(fake_config)

    def test_validate_head_files_success(self):
        self._test_validate_head_files_helper(['a', 'b'], contract_head='a',
                                              expand_head='b')

    def test_validate_head_files_missing_file(self):
        self._test_validate_head_files_helper(['a', 'b'])

    def test_validate_head_files_wrong_contents(self):
        self._test_validate_head_files_helper(['a', 'b'], contract_head='c',
                                              expand_head='d')

    @mock.patch.object(fileutils, 'delete_if_exists')
    def test_update_head_files_success(self, *mocks):
        heads = ['a', 'b']
        mock_open_con = self.useFixture(
                    tools.OpenFixture(cli._get_contract_head_file_path(
                        self.configs[0]))).mock_open
        mock_open_ex = self.useFixture(
            tools.OpenFixture(cli._get_expand_head_file_path(
                self.configs[0]))).mock_open
        with mock.patch('alembic.script.ScriptDirectory.from_config') as fc:
            fc.return_value.get_heads.return_value = heads
            revs = {heads[0]: FakeRevision(labels=cli.CONTRACT_BRANCH),
                    heads[1]: FakeRevision(labels=cli.EXPAND_BRANCH)}
            fc.return_value.get_revision.side_effect = revs.__getitem__
            cli.update_head_files(self.configs[0])
            mock_open_con.return_value.write.assert_called_with(
                heads[0] + '\n')
            mock_open_ex.return_value.write.assert_called_with(heads[1] + '\n')

            old_head_file = cli._get_head_file_path(
                self.configs[0])
            old_heads_file = cli._get_heads_file_path(
                self.configs[0])
            delete_if_exists = mocks[0]
            self.assertIn(mock.call(old_head_file),
                          delete_if_exists.call_args_list)
            self.assertIn(mock.call(old_heads_file),
                          delete_if_exists.call_args_list)

    def test_get_project_base(self):
        config = alembic_config.Config()
        config.set_main_option('script_location', 'a.b.c:d')
        proj_base = cli._get_project_base(config)
        self.assertEqual('a', proj_base)

    def test_get_root_versions_dir(self):
        config = alembic_config.Config()
        config.set_main_option('script_location', 'a.b.c:d')
        versions_dir = cli._get_root_versions_dir(config)
        self.assertEqual('/fake/dir/a/a/b/c/d/versions', versions_dir)

    def test_get_subproject_script_location(self):
        foo_ep = cli._get_subproject_script_location('networking-foo')
        expected = 'networking_foo.db.migration:alembic_migrations'
        self.assertEqual(expected, foo_ep)

    def test_get_subproject_script_location_not_installed(self):
        self.assertRaises(
            SystemExit, cli._get_subproject_script_location, 'not-installed')

    def test_get_subproject_base_not_installed(self):
        self.assertRaises(
            SystemExit, cli._get_subproject_base, 'not-installed')

    def test__compare_labels_ok(self):
        labels = {'label1', 'label2'}
        fake_revision = FakeRevision(labels)
        cli._compare_labels(fake_revision, {'label1', 'label2'})

    def test__compare_labels_fail_unexpected_labels(self):
        labels = {'label1', 'label2', 'label3'}
        fake_revision = FakeRevision(labels)
        self.assertRaises(
            SystemExit,
            cli._compare_labels, fake_revision, {'label1', 'label2'})

    @mock.patch.object(cli, '_compare_labels')
    def test__validate_single_revision_labels_branchless_fail_different_labels(
        self, compare_mock):

        fake_down_revision = FakeRevision()
        fake_revision = FakeRevision(down_revision=fake_down_revision)

        script_dir = mock.Mock()
        script_dir.get_revision.return_value = fake_down_revision
        cli._validate_single_revision_labels(script_dir, fake_revision,
                                             label=None)

        expected_labels = set()
        compare_mock.assert_has_calls(
            [mock.call(revision, expected_labels)
             for revision in (fake_revision, fake_down_revision)]
        )

    @mock.patch.object(cli, '_compare_labels')
    def test__validate_single_revision_labels_branches_fail_different_labels(
        self, compare_mock):

        fake_down_revision = FakeRevision()
        fake_revision = FakeRevision(down_revision=fake_down_revision)

        script_dir = mock.Mock()
        script_dir.get_revision.return_value = fake_down_revision
        cli._validate_single_revision_labels(
            script_dir, fake_revision, label='fakebranch')

        expected_labels = {'fakebranch'}
        compare_mock.assert_has_calls(
            [mock.call(revision, expected_labels)
             for revision in (fake_revision, fake_down_revision)]
        )

    @mock.patch.object(cli, '_validate_single_revision_labels')
    def test__validate_revision_validates_branches(self, validate_mock):
        script_dir = mock.Mock()
        fake_revision = FakeRevision()
        branch = cli.MIGRATION_BRANCHES[0]
        fake_revision.path = os.path.join('/fake/path', branch)
        cli._validate_revision(script_dir, fake_revision)
        validate_mock.assert_called_with(
            script_dir, fake_revision, label=branch)

    @mock.patch.object(cli, '_validate_single_revision_labels')
    def test__validate_revision_validates_branchless_migrations(
        self, validate_mock):

        script_dir = mock.Mock()
        fake_revision = FakeRevision()
        cli._validate_revision(script_dir, fake_revision)
        validate_mock.assert_called_with(script_dir, fake_revision)

    @mock.patch.object(cli, '_validate_revision')
    @mock.patch('alembic.script.ScriptDirectory.walk_revisions')
    def test_validate_revisions_walks_thru_all_revisions(
        self, walk_mock, validate_mock):

        revisions = [FakeRevision() for i in range(10)]
        walk_mock.return_value = revisions
        cli.validate_revisions(self.configs[0])
        validate_mock.assert_has_calls(
            [mock.call(mock.ANY, revision) for revision in revisions]
        )

    @mock.patch.object(cli, '_validate_revision')
    @mock.patch('alembic.script.ScriptDirectory.walk_revisions')
    def test_validate_revisions_fails_on_multiple_branch_points(
        self, walk_mock, validate_mock):

        revisions = [FakeRevision(is_branch_point=True) for i in range(2)]
        walk_mock.return_value = revisions
        self.assertRaises(
            SystemExit, cli.validate_revisions, self.configs[0])

    @mock.patch('alembic.script.ScriptDirectory.walk_revisions')
    def test__get_branch_points(self, walk_mock):
        revisions = [FakeRevision(is_branch_point=tools.get_random_boolean)
                     for i in range(50)]
        walk_mock.return_value = revisions
        script_dir = alembic_script.ScriptDirectory.from_config(
            self.configs[0])
        self.assertEqual(set(rev for rev in revisions if rev.is_branch_point),
                         set(cli._get_branch_points(script_dir)))

    @mock.patch.object(cli, '_get_version_branch_path')
    def test_autogen_process_directives(self, get_version_branch_path):

        get_version_branch_path.side_effect = lambda cfg, release, branch: (
            "/foo/expand" if branch == 'expand' else "/foo/contract")

        migration_script = alembic_ops.MigrationScript(
            'eced083f5df',
            # these directives will be split into separate
            # expand/contract scripts
            alembic_ops.UpgradeOps(
                ops=[
                    alembic_ops.CreateTableOp(
                        'organization',
                        [
                            sa.Column('id', sa.Integer(), primary_key=True),
                            sa.Column('name', sa.String(50), nullable=False)
                        ]
                    ),
                    alembic_ops.ModifyTableOps(
                        'user',
                        ops=[
                            alembic_ops.AddColumnOp(
                                'user',
                                sa.Column('organization_id', sa.Integer())
                            ),
                            alembic_ops.CreateForeignKeyOp(
                                'org_fk', 'user', 'organization',
                                ['organization_id'], ['id']
                            ),
                            alembic_ops.DropConstraintOp(
                                'user', 'uq_user_org'
                            ),
                            alembic_ops.DropColumnOp(
                                'user', 'organization_name'
                            )
                        ]
                    )
                ]
            ),
            # these will be discarded
            alembic_ops.DowngradeOps(
                ops=[
                    alembic_ops.AddColumnOp(
                        'user', sa.Column(
                            'organization_name', sa.String(50), nullable=True)
                    ),
                    alembic_ops.CreateUniqueConstraintOp(
                        'uq_user_org', 'user',
                        ['user_name', 'organization_name']
                    ),
                    alembic_ops.ModifyTableOps(
                        'user',
                        ops=[
                            alembic_ops.DropConstraintOp('org_fk', 'user'),
                            alembic_ops.DropColumnOp('user', 'organization_id')
                        ]
                    ),
                    alembic_ops.DropTableOp('organization')
                ]
            ),
            message='create the organization table and '
            'replace user.organization_name'
        )

        directives = [migration_script]
        autogen.process_revision_directives(
            mock.Mock(), mock.Mock(), directives
        )

        expand = directives[0]
        contract = directives[1]
        self.assertEqual("/foo/expand", expand.version_path)
        self.assertEqual("/foo/contract", contract.version_path)
        self.assertTrue(expand.downgrade_ops.is_empty())
        self.assertTrue(contract.downgrade_ops.is_empty())

        def _get_regex(s):
            s = textwrap.dedent(s)
            s = re.escape(s)
            # alembic 0.8.9 added additional leading '# ' before comments
            return s.replace('\\#\\#\\#\\ ', '(# )?### ')

        expected_regex = ("""\
            ### commands auto generated by Alembic - please adjust! ###
                op.create_table('organization',
                sa.Column('id', sa.Integer(), nullable=False),
                sa.Column('name', sa.String(length=50), nullable=False),
                sa.PrimaryKeyConstraint('id')
                )
                op.add_column('user', """
                """sa.Column('organization_id', sa.Integer(), nullable=True))
                op.create_foreign_key('org_fk', 'user', """
                """'organization', ['organization_id'], ['id'])
                ### end Alembic commands ###""")
        self.assertThat(
            alembic_ag_api.render_python_code(expand.upgrade_ops),
            matchers.MatchesRegex(_get_regex(expected_regex)))

        expected_regex = ("""\
            ### commands auto generated by Alembic - please adjust! ###
                op.drop_constraint('user', 'uq_user_org', type_=None)
                op.drop_column('user', 'organization_name')
                ### end Alembic commands ###""")
        self.assertThat(
            alembic_ag_api.render_python_code(contract.upgrade_ops),
            matchers.MatchesRegex(_get_regex(expected_regex)))

    @mock.patch('alembic.script.ScriptDirectory.walk_revisions')
    def test__find_milestone_revisions_one_branch(self, walk_mock):
        c_revs = [FakeRevision(labels={cli.CONTRACT_BRANCH}) for r in range(5)]
        c_revs[1].module.neutron_milestone = [migration.LIBERTY]

        walk_mock.return_value = c_revs
        m = cli._find_milestone_revisions(self.configs[0], 'liberty',
                                          cli.CONTRACT_BRANCH)
        self.assertEqual(1, len(m))
        m = cli._find_milestone_revisions(self.configs[0], 'liberty',
                                          cli.EXPAND_BRANCH)
        self.assertEqual(0, len(m))

    @mock.patch('alembic.script.ScriptDirectory.walk_revisions')
    def test__find_milestone_revisions_two_branches(self, walk_mock):
        c_revs = [FakeRevision(labels={cli.CONTRACT_BRANCH}) for r in range(5)]
        c_revs[1].module.neutron_milestone = [migration.LIBERTY]
        e_revs = [FakeRevision(labels={cli.EXPAND_BRANCH}) for r in range(5)]
        e_revs[3].module.neutron_milestone = [migration.LIBERTY]

        walk_mock.return_value = c_revs + e_revs
        m = cli._find_milestone_revisions(self.configs[0], 'liberty')
        self.assertEqual(2, len(m))

        m = cli._find_milestone_revisions(self.configs[0], 'mitaka')
        self.assertEqual(0, len(m))

    @mock.patch('alembic.script.ScriptDirectory.walk_revisions')
    def test__find_milestone_revisions_branchless(self, walk_mock):
        revisions = [FakeRevision() for r in range(5)]
        revisions[2].module.neutron_milestone = [migration.LIBERTY]

        walk_mock.return_value = revisions
        m = cli._find_milestone_revisions(self.configs[0], 'liberty')
        self.assertEqual(1, len(m))

        m = cli._find_milestone_revisions(self.configs[0], 'mitaka')
        self.assertEqual(0, len(m))


class TestSafetyChecks(base.BaseTestCase):

    def test_validate_revisions(self, *mocks):
        cli.validate_revisions(cli.get_neutron_config())

[((358, 5, 358, 64), 'mock.patch', 'mock.patch', ({(358, 16, 358, 63): '"""alembic.script.ScriptDirectory.walk_revisions"""'}, {}), "('alembic.script.ScriptDirectory.walk_revisions')", False, 'import mock\n'), ((451, 5, 451, 53), 'mock.patch.object', 'mock.patch.object', ({(451, 23, 451, 32): 'fileutils', (451, 34, 451, 52): '"""delete_if_exists"""'}, {}), "(fileutils, 'delete_if_exists')", False, 'import mock\n'), ((517, 5, 517, 46), 'mock.patch.object', 'mock.patch.object', ({(517, 23, 517, 26): 'cli', (517, 28, 517, 45): '"""_compare_labels"""'}, {}), "(cli, '_compare_labels')", False, 'import mock\n'), ((535, 5, 535, 46), 'mock.patch.object', 'mock.patch.object', ({(535, 23, 535, 26): 'cli', (535, 28, 535, 45): '"""_compare_labels"""'}, {}), "(cli, '_compare_labels')", False, 'import mock\n'), ((553, 5, 553, 63), 'mock.patch.object', 'mock.patch.object', ({(553, 23, 553, 26): 'cli', (553, 28, 553, 62): '"""_validate_single_revision_labels"""'}, {}), "(cli, '_validate_single_revision_labels')", False, 'import mock\n'), ((563, 5, 563, 63), 'mock.patch.object', 'mock.patch.object', ({(563, 23, 563, 26): 'cli', (563, 28, 563, 62): '"""_validate_single_revision_labels"""'}, {}), "(cli, '_validate_single_revision_labels')", False, 'import mock\n'), ((572, 5, 572, 49), 'mock.patch.object', 'mock.patch.object', ({(572, 23, 572, 26): 'cli', (572, 28, 572, 48): '"""_validate_revision"""'}, {}), "(cli, '_validate_revision')", False, 'import mock\n'), ((573, 5, 573, 64), 'mock.patch', 'mock.patch', ({(573, 16, 573, 63): '"""alembic.script.ScriptDirectory.walk_revisions"""'}, {}), "('alembic.script.ScriptDirectory.walk_revisions')", False, 'import mock\n'), ((584, 5, 584, 49), 'mock.patch.object', 'mock.patch.object', ({(584, 23, 584, 26): 'cli', (584, 28, 584, 48): '"""_validate_revision"""'}, {}), "(cli, '_validate_revision')", False, 'import mock\n'), ((585, 5, 585, 64), 'mock.patch', 'mock.patch', ({(585, 16, 585, 63): '"""alembic.script.ScriptDirectory.walk_revisions"""'}, {}), "('alembic.script.ScriptDirectory.walk_revisions')", False, 'import mock\n'), ((594, 5, 594, 64), 'mock.patch', 'mock.patch', ({(594, 16, 594, 63): '"""alembic.script.ScriptDirectory.walk_revisions"""'}, {}), "('alembic.script.ScriptDirectory.walk_revisions')", False, 'import mock\n'), ((604, 5, 604, 55), 'mock.patch.object', 'mock.patch.object', ({(604, 23, 604, 26): 'cli', (604, 28, 604, 54): '"""_get_version_branch_path"""'}, {}), "(cli, '_get_version_branch_path')", False, 'import mock\n'), ((712, 5, 712, 64), 'mock.patch', 'mock.patch', ({(712, 16, 712, 63): '"""alembic.script.ScriptDirectory.walk_revisions"""'}, {}), "('alembic.script.ScriptDirectory.walk_revisions')", False, 'import mock\n'), ((725, 5, 725, 64), 'mock.patch', 'mock.patch', ({(725, 16, 725, 63): '"""alembic.script.ScriptDirectory.walk_revisions"""'}, {}), "('alembic.script.ScriptDirectory.walk_revisions')", False, 'import mock\n'), ((739, 5, 739, 64), 'mock.patch', 'mock.patch', ({(739, 16, 739, 63): '"""alembic.script.ScriptDirectory.walk_revisions"""'}, {}), "('alembic.script.ScriptDirectory.walk_revisions')", False, 'import mock\n'), ((56, 24, 56, 53), 'neutron_lib.utils.helpers.get_random_string', 'helpers.get_random_string', ({(56, 50, 56, 52): '10'}, {}), '(10)', False, 'from neutron_lib.utils import helpers\n'), ((57, 22, 57, 38), 'mock.MagicMock', 'mock.MagicMock', ({}, {}), '()', False, 'import mock\n'), ((67, 24, 67, 67), 'neutron.conf.db.migration_cli.migration_entrypoints.items', 'migration_cli.migration_entrypoints.items', ({}, {}), '()', False, 'from neutron.conf.db import migration_cli\n'), ((87, 25, 87, 41), 'mock.MagicMock', 'mock.MagicMock', ({}, {}), '()', False, 'import mock\n'), ((123, 32, 123, 76), 'mock.patch.object', 'mock.patch.object', ({(123, 50, 123, 53): 'cli', (123, 55, 123, 75): '"""do_alembic_command"""'}, {}), "(cli, 'do_alembic_command')", False, 'import mock\n'), ((481, 17, 481, 40), 'alembic.config.Config', 'alembic_config.Config', ({}, {}), '()', True, 'from alembic import config as alembic_config\n'), ((483, 20, 483, 49), 'neutron.db.migration.cli._get_project_base', 'cli._get_project_base', ({(483, 42, 483, 48): 'config'}, {}), '(config)', False, 'from neutron.db.migration import cli\n'), ((487, 17, 487, 40), 'alembic.config.Config', 'alembic_config.Config', ({}, {}), '()', True, 'from alembic import config as alembic_config\n'), ((489, 23, 489, 57), 'neutron.db.migration.cli._get_root_versions_dir', 'cli._get_root_versions_dir', ({(489, 50, 489, 56): 'config'}, {}), '(config)', False, 'from neutron.db.migration import cli\n'), ((493, 17, 493, 70), 'neutron.db.migration.cli._get_subproject_script_location', 'cli._get_subproject_script_location', ({(493, 53, 493, 69): '"""networking-foo"""'}, {}), "('networking-foo')", False, 'from neutron.db.migration import cli\n'), ((508, 8, 508, 64), 'neutron.db.migration.cli._compare_labels', 'cli._compare_labels', ({(508, 28, 508, 41): 'fake_revision', (508, 43, 508, 63): "{'label1', 'label2'}"}, {}), "(fake_revision, {'label1', 'label2'})", False, 'from neutron.db.migration import cli\n'), ((524, 21, 524, 32), 'mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'import mock\n'), ((526, 8, 527, 56), 'neutron.db.migration.cli._validate_single_revision_labels', 'cli._validate_single_revision_labels', (), '', False, 'from neutron.db.migration import cli\n'), ((542, 21, 542, 32), 'mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'import mock\n'), ((544, 8, 545, 58), 'neutron.db.migration.cli._validate_single_revision_labels', 'cli._validate_single_revision_labels', (), '', False, 'from neutron.db.migration import cli\n'), ((555, 21, 555, 32), 'mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'import mock\n'), ((558, 29, 558, 63), 'os.path.join', 'os.path.join', ({(558, 42, 558, 54): '"""/fake/path"""', (558, 56, 558, 62): 'branch'}, {}), "('/fake/path', branch)", False, 'import os\n'), ((559, 8, 559, 57), 'neutron.db.migration.cli._validate_revision', 'cli._validate_revision', ({(559, 31, 559, 41): 'script_dir', (559, 43, 559, 56): 'fake_revision'}, {}), '(script_dir, fake_revision)', False, 'from neutron.db.migration import cli\n'), ((567, 21, 567, 32), 'mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'import mock\n'), ((569, 8, 569, 57), 'neutron.db.migration.cli._validate_revision', 'cli._validate_revision', ({(569, 31, 569, 41): 'script_dir', (569, 43, 569, 56): 'fake_revision'}, {}), '(script_dir, fake_revision)', False, 'from neutron.db.migration import cli\n'), ((579, 8, 579, 47), 'neutron.db.migration.cli.validate_revisions', 'cli.validate_revisions', ({(579, 31, 579, 46): 'self.configs[0]'}, {}), '(self.configs[0])', False, 'from neutron.db.migration import cli\n'), ((599, 21, 600, 28), 'alembic.script.ScriptDirectory.from_config', 'alembic_script.ScriptDirectory.from_config', ({(600, 12, 600, 27): 'self.configs[0]'}, {}), '(self.configs[0])', True, 'from alembic import script as alembic_script\n'), ((718, 12, 719, 62), 'neutron.db.migration.cli._find_milestone_revisions', 'cli._find_milestone_revisions', ({(718, 42, 718, 57): 'self.configs[0]', (718, 59, 718, 68): '"""liberty"""', (719, 42, 719, 61): 'cli.CONTRACT_BRANCH'}, {}), "(self.configs[0], 'liberty', cli.CONTRACT_BRANCH)", False, 'from neutron.db.migration import cli\n'), ((721, 12, 722, 60), 'neutron.db.migration.cli._find_milestone_revisions', 'cli._find_milestone_revisions', ({(721, 42, 721, 57): 'self.configs[0]', (721, 59, 721, 68): '"""liberty"""', (722, 42, 722, 59): 'cli.EXPAND_BRANCH'}, {}), "(self.configs[0], 'liberty', cli.EXPAND_BRANCH)", False, 'from neutron.db.migration import cli\n'), ((733, 12, 733, 69), 'neutron.db.migration.cli._find_milestone_revisions', 'cli._find_milestone_revisions', ({(733, 42, 733, 57): 'self.configs[0]', (733, 59, 733, 68): '"""liberty"""'}, {}), "(self.configs[0], 'liberty')", False, 'from neutron.db.migration import cli\n'), ((736, 12, 736, 68), 'neutron.db.migration.cli._find_milestone_revisions', 'cli._find_milestone_revisions', ({(736, 42, 736, 57): 'self.configs[0]', (736, 59, 736, 67): '"""mitaka"""'}, {}), "(self.configs[0], 'mitaka')", False, 'from neutron.db.migration import cli\n'), ((745, 12, 745, 69), 'neutron.db.migration.cli._find_milestone_revisions', 'cli._find_milestone_revisions', ({(745, 42, 745, 57): 'self.configs[0]', (745, 59, 745, 68): '"""liberty"""'}, {}), "(self.configs[0], 'liberty')", False, 'from neutron.db.migration import cli\n'), ((748, 12, 748, 68), 'neutron.db.migration.cli._find_milestone_revisions', 'cli._find_milestone_revisions', ({(748, 42, 748, 57): 'self.configs[0]', (748, 59, 748, 67): '"""mitaka"""'}, {}), "(self.configs[0], 'mitaka')", False, 'from neutron.db.migration import cli\n'), ((68, 35, 68, 48), 'copy.copy', 'copy.copy', ({(68, 45, 68, 47): 'ep'}, {}), '(ep)', False, 'import copy\n'), ((95, 24, 95, 57), 'neutron.db.migration.schema_has_table', 'migration.schema_has_table', ({(95, 51, 95, 56): '"""foo"""'}, {}), "('foo')", False, 'from neutron.db import migration\n'), ((103, 25, 103, 66), 'neutron.db.migration.schema_has_column', 'migration.schema_has_column', ({(103, 53, 103, 58): '"""meh"""', (103, 60, 103, 65): '"""meh"""'}, {}), "('meh', 'meh')", False, 'from neutron.db import migration\n'), ((107, 24, 107, 72), 'neutron.db.migration.schema_has_column', 'migration.schema_has_column', ({(107, 52, 107, 57): '"""foo"""', (107, 59, 107, 71): '"""foo_column"""'}, {}), "('foo', 'foo_column')", False, 'from neutron.db import migration\n'), ((116, 25, 117, 37), 'neutron.db.migration.schema_has_column', 'migration.schema_has_column', (), '', False, 'from neutron.db import migration\n'), ((139, 27, 139, 56), 'os.path.dirname', 'os.path.dirname', ({(139, 43, 139, 55): 'cli.__file__'}, {}), '(cli.__file__)', False, 'import os\n'), ((143, 21, 143, 47), 'alembic.config.Config', 'alembic_config.Config', ({(143, 43, 143, 46): 'ini'}, {}), '(ini)', True, 'from alembic import config as alembic_config\n'), ((150, 25, 152, 62), 'pkg_resources.EntryPoint', 'pkg_resources.EntryPoint', (), '', False, 'import pkg_resources\n'), ((156, 13, 156, 49), 'mock.patch.object', 'mock.patch.object', ({(156, 31, 156, 34): 'sys', (156, 36, 156, 42): '"""argv"""', (156, 44, 156, 48): 'argv'}, {}), "(sys, 'argv', argv)", False, 'import mock\n'), ((157, 16, 157, 59), 'mock.patch.object', 'mock.patch.object', ({(157, 34, 157, 37): 'cli', (157, 39, 157, 58): '"""run_sanity_checks"""'}, {}), "(cli, 'run_sanity_checks')", False, 'import mock\n'), ((158, 16, 158, 60), 'mock.patch.object', 'mock.patch.object', ({(158, 34, 158, 37): 'cli', (158, 39, 158, 59): '"""validate_revisions"""'}, {}), "(cli, 'validate_revisions')", False, 'import mock\n'), ((160, 12, 160, 22), 'neutron.db.migration.cli.main', 'cli.main', ({}, {}), '()', False, 'from neutron.db.migration import cli\n'), ((210, 13, 210, 58), 'mock.patch.object', 'mock.patch.object', ({(210, 31, 210, 34): 'cli', (210, 36, 210, 57): '"""validate_head_files"""'}, {}), "(cli, 'validate_head_files')", False, 'import mock\n'), ((215, 13, 215, 56), 'mock.patch.object', 'mock.patch.object', ({(215, 31, 215, 34): 'cli', (215, 36, 215, 55): '"""update_head_files"""'}, {}), "(cli, 'update_head_files')", False, 'import mock\n'), ((309, 13, 309, 54), 'neutron.tests.unit.testlib_api.ExpectedException', 'testlib_api.ExpectedException', ({(309, 43, 309, 53): 'SystemExit'}, {}), '(SystemExit)', False, 'from neutron.tests.unit import testlib_api\n'), ((314, 13, 314, 54), 'neutron.tests.unit.testlib_api.ExpectedException', 'testlib_api.ExpectedException', ({(314, 43, 314, 53): 'SystemExit'}, {}), '(SystemExit)', False, 'from neutron.tests.unit import testlib_api\n'), ((319, 13, 319, 54), 'neutron.tests.unit.testlib_api.ExpectedException', 'testlib_api.ExpectedException', ({(319, 43, 319, 53): 'SystemExit'}, {}), '(SystemExit)', False, 'from neutron.tests.unit import testlib_api\n'), ((324, 13, 324, 54), 'neutron.tests.unit.testlib_api.ExpectedException', 'testlib_api.ExpectedException', ({(324, 43, 324, 53): 'SystemExit'}, {}), '(SystemExit)', False, 'from neutron.tests.unit import testlib_api\n'), ((379, 13, 379, 52), 'mock.patch.object', 'mock.patch.object', ({(379, 31, 379, 34): 'sys', (379, 36, 379, 42): '"""argv"""', (379, 44, 379, 51): 'command'}, {}), "(sys, 'argv', command)", False, 'import mock\n'), ((379, 54, 380, 41), 'mock.patch.object', 'mock.patch.object', ({(380, 16, 380, 19): 'cli', (380, 21, 380, 40): '"""run_sanity_checks"""'}, {}), "(cli, 'run_sanity_checks')", False, 'import mock\n'), ((399, 13, 399, 69), 'mock.patch', 'mock.patch', ({(399, 24, 399, 68): '"""alembic.script.ScriptDirectory.from_config"""'}, {}), "('alembic.script.ScriptDirectory.from_config')", False, 'import mock\n'), ((400, 16, 400, 44), 'mock.patch', 'mock.patch', ({(400, 27, 400, 43): '"""os.path.exists"""'}, {}), "('os.path.exists')", False, 'import mock\n'), ((460, 13, 460, 69), 'mock.patch', 'mock.patch', ({(460, 24, 460, 68): '"""alembic.script.ScriptDirectory.from_config"""'}, {}), "('alembic.script.ScriptDirectory.from_config')", False, 'import mock\n'), ((465, 12, 465, 50), 'neutron.db.migration.cli.update_head_files', 'cli.update_head_files', ({(465, 34, 465, 49): 'self.configs[0]'}, {}), '(self.configs[0])', False, 'from neutron.db.migration import cli\n'), ((470, 28, 471, 32), 'neutron.db.migration.cli._get_head_file_path', 'cli._get_head_file_path', ({(471, 16, 471, 31): 'self.configs[0]'}, {}), '(self.configs[0])', False, 'from neutron.db.migration import cli\n'), ((472, 29, 473, 32), 'neutron.db.migration.cli._get_heads_file_path', 'cli._get_heads_file_path', ({(473, 16, 473, 31): 'self.configs[0]'}, {}), '(self.configs[0])', False, 'from neutron.db.migration import cli\n'), ((671, 12, 671, 23), 'mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'import mock\n'), ((671, 25, 671, 36), 'mock.Mock', 'mock.Mock', ({}, {}), '()', False, 'import mock\n'), ((682, 16, 682, 34), 'textwrap.dedent', 'textwrap.dedent', ({(682, 32, 682, 33): 's'}, {}), '(s)', False, 'import textwrap\n'), ((683, 16, 683, 28), 're.escape', 're.escape', ({(683, 26, 683, 27): 's'}, {}), '(s)', False, 'import re\n'), ((700, 12, 700, 65), 'alembic.autogenerate.api.render_python_code', 'alembic_ag_api.render_python_code', ({(700, 46, 700, 64): 'expand.upgrade_ops'}, {}), '(expand.upgrade_ops)', True, 'from alembic.autogenerate import api as alembic_ag_api\n'), ((709, 12, 709, 67), 'alembic.autogenerate.api.render_python_code', 'alembic_ag_api.render_python_code', ({(709, 46, 709, 66): 'contract.upgrade_ops'}, {}), '(contract.upgrade_ops)', True, 'from alembic.autogenerate import api as alembic_ag_api\n'), ((755, 31, 755, 55), 'neutron.db.migration.cli.get_neutron_config', 'cli.get_neutron_config', ({}, {}), '()', False, 'from neutron.db.migration import cli\n'), ((79, 8, 79, 41), 'mock.patch', 'mock.patch', ({(79, 19, 79, 40): '"""alembic.op.get_bind"""'}, {}), "('alembic.op.get_bind')", False, 'import mock\n'), ((80, 39, 81, 66), 'mock.patch', 'mock.patch', (), '', False, 'import mock\n'), ((83, 33, 84, 53), 'mock.patch', 'mock.patch', ({(84, 12, 84, 52): '"""sqlalchemy.engine.reflection.Inspector"""'}, {}), "('sqlalchemy.engine.reflection.Inspector')", False, 'import mock\n'), ((125, 32, 125, 62), 'mock.patch', 'mock.patch', ({(125, 43, 125, 61): '"""alembic.util.err"""'}, {}), "('alembic.util.err')", False, 'import mock\n'), ((126, 33, 126, 64), 'mock.patch', 'mock.patch', ({(126, 44, 126, 63): '"""alembic.util.warn"""'}, {}), "('alembic.util.warn')", False, 'import mock\n'), ((130, 45, 130, 71), 'neutron.db.migration.cli._get_project_base', 'cli._get_project_base', ({(130, 67, 130, 70): 'cfg'}, {}), '(cfg)', False, 'from neutron.db.migration import cli\n'), ((131, 20, 131, 67), 'mock.patch.object', 'mock.patch.object', ({(131, 38, 131, 41): 'cli', (131, 43, 131, 66): '"""_get_package_root_dir"""'}, {}), "(cli, '_get_package_root_dir')", False, 'import mock\n'), ((134, 8, 134, 54), 'mock.patch', 'mock.patch', ({(134, 19, 134, 53): '"""oslo_utils.fileutils.ensure_tree"""'}, {}), "('oslo_utils.fileutils.ensure_tree')", False, 'import mock\n'), ((163, 23, 163, 38), 'copy.copy', 'copy.copy', ({(163, 33, 163, 37): 'args'}, {}), '(args)', False, 'import copy\n'), ((378, 8, 378, 60), 'mock.patch', 'mock.patch', ({(378, 19, 378, 59): '"""argparse.ArgumentParser._print_message"""'}, {}), "('argparse.ArgumentParser._print_message')", False, 'import mock\n'), ((419, 16, 419, 52), 'neutron.db.migration.cli.validate_head_files', 'cli.validate_head_files', ({(419, 40, 419, 51): 'fake_config'}, {}), '(fake_config)', False, 'from neutron.db.migration import cli\n'), ((475, 26, 475, 50), 'mock.call', 'mock.call', ({(475, 36, 475, 49): 'old_head_file'}, {}), '(old_head_file)', False, 'import mock\n'), ((477, 26, 477, 51), 'mock.call', 'mock.call', ({(477, 36, 477, 50): 'old_heads_file'}, {}), '(old_heads_file)', False, 'import mock\n'), ((531, 13, 531, 49), 'mock.call', 'mock.call', ({(531, 23, 531, 31): 'revision', (531, 33, 531, 48): 'expected_labels'}, {}), '(revision, expected_labels)', False, 'import mock\n'), ((549, 13, 549, 49), 'mock.call', 'mock.call', ({(549, 23, 549, 31): 'revision', (549, 33, 549, 48): 'expected_labels'}, {}), '(revision, expected_labels)', False, 'import mock\n'), ((581, 13, 581, 42), 'mock.call', 'mock.call', ({(581, 23, 581, 31): 'mock.ANY', (581, 33, 581, 41): 'revision'}, {}), '(mock.ANY, revision)', False, 'import mock\n'), ((602, 29, 602, 63), 'neutron.db.migration.cli._get_branch_points', 'cli._get_branch_points', ({(602, 52, 602, 62): 'script_dir'}, {}), '(script_dir)', False, 'from neutron.db.migration import cli\n'), ((233, 24, 233, 52), 'neutron.db.migration.cli._get_branch_head', 'cli._get_branch_head', ({(233, 45, 233, 51): 'branch'}, {}), '(branch)', False, 'from neutron.db.migration import cli\n'), ((421, 16, 421, 52), 'neutron.db.migration.cli.validate_head_files', 'cli.validate_head_files', ({(421, 40, 421, 51): 'fake_config'}, {}), '(fake_config)', False, 'from neutron.db.migration import cli\n'), ((433, 20, 433, 63), 'neutron.db.migration.cli._get_expand_head_file_path', 'cli._get_expand_head_file_path', ({(433, 51, 433, 62): 'fake_config'}, {}), '(fake_config)', False, 'from neutron.db.migration import cli\n'), ((435, 20, 435, 65), 'neutron.db.migration.cli._get_contract_head_file_path', 'cli._get_contract_head_file_path', ({(435, 53, 435, 64): 'fake_config'}, {}), '(fake_config)', False, 'from neutron.db.migration import cli\n'), ((455, 38, 456, 40), 'neutron.db.migration.cli._get_contract_head_file_path', 'cli._get_contract_head_file_path', ({(456, 24, 456, 39): 'self.configs[0]'}, {}), '(self.configs[0])', False, 'from neutron.db.migration import cli\n'), ((458, 30, 459, 32), 'neutron.db.migration.cli._get_expand_head_file_path', 'cli._get_expand_head_file_path', ({(459, 16, 459, 31): 'self.configs[0]'}, {}), '(self.configs[0])', False, 'from neutron.db.migration import cli\n'), ((217, 16, 217, 44), 'mock.patch', 'mock.patch', ({(217, 27, 217, 43): '"""os.path.exists"""'}, {}), "('os.path.exists')", False, 'import mock\n'), ((412, 34, 413, 32), 'neutron.db.migration.cli._get_contract_head_file_path', 'cli._get_contract_head_file_path', ({(413, 20, 413, 31): 'fake_config'}, {}), '(fake_config)', False, 'from neutron.db.migration import cli\n'), ((415, 34, 416, 32), 'neutron.db.migration.cli._get_expand_head_file_path', 'cli._get_expand_head_file_path', ({(416, 20, 416, 31): 'fake_config'}, {}), '(fake_config)', False, 'from neutron.db.migration import cli\n'), ((651, 20, 654, 21), 'alembic.operations.ops.CreateUniqueConstraintOp', 'alembic_ops.CreateUniqueConstraintOp', ({(652, 24, 652, 37): '"""uq_user_org"""', (652, 39, 652, 45): '"""user"""', (653, 24, 653, 58): "['user_name', 'organization_name']"}, {}), "('uq_user_org', 'user', ['user_name',\n    'organization_name'])", True, 'from alembic.operations import ops as alembic_ops\n'), ((662, 20, 662, 59), 'alembic.operations.ops.DropTableOp', 'alembic_ops.DropTableOp', ({(662, 44, 662, 58): '"""organization"""'}, {}), "('organization')", True, 'from alembic.operations import ops as alembic_ops\n'), ((649, 49, 649, 62), 'sqlalchemy.String', 'sa.String', ({(649, 59, 649, 61): '50'}, {}), '(50)', True, 'import sqlalchemy as sa\n'), ((619, 44, 619, 56), 'sqlalchemy.Integer', 'sa.Integer', ({}, {}), '()', True, 'import sqlalchemy as sa\n'), ((620, 46, 620, 59), 'sqlalchemy.String', 'sa.String', ({(620, 56, 620, 58): '50'}, {}), '(50)', True, 'import sqlalchemy as sa\n'), ((630, 28, 633, 29), 'alembic.operations.ops.CreateForeignKeyOp', 'alembic_ops.CreateForeignKeyOp', ({(631, 32, 631, 40): '"""org_fk"""', (631, 42, 631, 48): '"""user"""', (631, 50, 631, 64): '"""organization"""', (632, 32, 632, 51): "['organization_id']", (632, 53, 632, 59): "['id']"}, {}), "('org_fk', 'user', 'organization', [\n    'organization_id'], ['id'])", True, 'from alembic.operations import ops as alembic_ops\n'), ((634, 28, 636, 29), 'alembic.operations.ops.DropConstraintOp', 'alembic_ops.DropConstraintOp', ({(635, 32, 635, 38): '"""user"""', (635, 40, 635, 53): '"""uq_user_org"""'}, {}), "('user', 'uq_user_org')", True, 'from alembic.operations import ops as alembic_ops\n'), ((637, 28, 639, 29), 'alembic.operations.ops.DropColumnOp', 'alembic_ops.DropColumnOp', ({(638, 32, 638, 38): '"""user"""', (638, 40, 638, 59): '"""organization_name"""'}, {}), "('user', 'organization_name')", True, 'from alembic.operations import ops as alembic_ops\n'), ((658, 28, 658, 74), 'alembic.operations.ops.DropConstraintOp', 'alembic_ops.DropConstraintOp', ({(658, 57, 658, 65): '"""org_fk"""', (658, 67, 658, 73): '"""user"""'}, {}), "('org_fk', 'user')", True, 'from alembic.operations import ops as alembic_ops\n'), ((659, 28, 659, 79), 'alembic.operations.ops.DropColumnOp', 'alembic_ops.DropColumnOp', ({(659, 53, 659, 59): '"""user"""', (659, 61, 659, 78): '"""organization_id"""'}, {}), "('user', 'organization_id')", True, 'from alembic.operations import ops as alembic_ops\n'), ((628, 61, 628, 73), 'sqlalchemy.Integer', 'sa.Integer', ({}, {}), '()', True, 'import sqlalchemy as sa\n')]