← Back to branch summary

~ubuntu-branches/ubuntu/saucy/nova/saucy-proposed

« back to all changes in this revision

Viewing changes to nova/tests/test_migration_utils.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Chuck Short, Adam Gandelman
  • Date: 2013-02-22 09:27:29 UTC
  • mfrom: (1.1.68)
  • Revision ID: package-import@ubuntu.com-20130222092729-nn3gt8rf97uvts77
Tags: 2013.1.g3-0ubuntu1
https://launchpad.net/bugs/1110567
[ Chuck Short ]
* New usptream release. 
* debian/patches/debian/patches/fix-ubuntu-tests.patch: Refreshed.
* debian/nova-baremetal.logrotate: Fix logfile path.
* debian/control, debian/nova-spiceproxy.{install, logrotate, upstart}:
  Add spice html5 proxy support.
* debian/nova-novncproxy.upstart: Start on runlevel [2345]
* debian/rules: Call testr directly since run_tests.sh -N gives weird return
  value when tests pass.
* debian/pyddist-overrides: Add websockify.
* debian/nova-common.postinst: Removed config file conversion, since
  the option is no longer available. (LP: #1110567)
* debian/control: Add python-pyasn1 as a dependency.
* debian/control: Add python-oslo-config as a dependency.
* debian/control: Suggest sysfsutils, sg3-utils, multipath-tools for fibre
  channel support.

[ Adam Gandelman ]
* debian/control: Fix typo (websocikfy -> websockify).

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/test_migration_utils.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright (c) 2013 Boris Pavlovic (boris@pavlovic.me).
 
4
# All Rights Reserved.
 
5
#
 
6
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
7
#    not use this file except in compliance with the License. You may obtain
 
8
#    a copy of the License at
 
9
#
 
10
#         http://www.apache.org/licenses/LICENSE-2.0
 
11
#
 
12
#    Unless required by applicable law or agreed to in writing, software
 
13
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
14
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
15
#    License for the specific language governing permissions and limitations
 
16
#    under the License.
 
17
 
 
18
from migrate.changeset import UniqueConstraint
 
19
from sqlalchemy import Integer, BigInteger, DateTime, String
 
20
from sqlalchemy import MetaData, Table, Column
 
21
from sqlalchemy.sql import select
 
22
 
 
23
from nova.db.sqlalchemy import utils
 
24
from nova import exception
 
25
from nova.tests import test_migrations
 
26
 
 
27
 
 
28
class TestMigrationUtils(test_migrations.BaseMigrationTestCase):
 
29
    """Class for testing utils that are used in db migrations."""
 
30
 
 
31
    def test_utils_drop_unique_constraint(self):
 
32
        table_name = "__test_tmp_table__"
 
33
        uc_name = 'uniq_foo'
 
34
        values = [
 
35
            {'id': 1, 'a': 3, 'foo': 10},
 
36
            {'id': 2, 'a': 2, 'foo': 20},
 
37
            {'id': 3, 'a': 1, 'foo': 30}
 
38
        ]
 
39
        for key, engine in self.engines.items():
 
40
            meta = MetaData()
 
41
            meta.bind = engine
 
42
            test_table = Table(table_name, meta,
 
43
                               Column('id', Integer, primary_key=True,
 
44
                                      nullable=False),
 
45
                               Column('a', Integer),
 
46
                               Column('foo', Integer),
 
47
                               UniqueConstraint('a', name='uniq_a'),
 
48
                               UniqueConstraint('foo', name=uc_name))
 
49
            test_table.create()
 
50
 
 
51
            engine.execute(test_table.insert(), values)
 
52
            # NOTE(boris-42): This method is generic UC dropper.
 
53
            utils.drop_unique_constraint(engine, table_name, uc_name, 'foo')
 
54
 
 
55
            s = test_table.select().order_by(test_table.c.id)
 
56
            rows = engine.execute(s).fetchall()
 
57
 
 
58
            for i in xrange(0, len(values)):
 
59
                v = values[i]
 
60
                self.assertEqual((v['id'], v['a'], v['foo']), rows[i])
 
61
 
 
62
            # NOTE(boris-42): Update data about Table from DB.
 
63
            meta = MetaData()
 
64
            meta.bind = engine
 
65
            test_table = Table(table_name, meta, autoload=True)
 
66
            constraints = filter(lambda c: c.name == uc_name,
 
67
                                 test_table.constraints)
 
68
            self.assertEqual(len(constraints), 0)
 
69
            self.assertEqual(len(test_table.constraints), 1)
 
70
 
 
71
            test_table.drop()
 
72
 
 
73
    def test_util_drop_unique_constraint_with_not_supported_sqlite_type(self):
 
74
        table_name = "__test_tmp_table__"
 
75
        uc_name = 'uniq_foo'
 
76
        values = [
 
77
            {'id': 1, 'a': 3, 'foo': 10},
 
78
            {'id': 2, 'a': 2, 'foo': 20},
 
79
            {'id': 3, 'a': 1, 'foo': 30}
 
80
        ]
 
81
 
 
82
        for key, engine in self.engines.items():
 
83
            meta = MetaData()
 
84
            meta.bind = engine
 
85
            test_table = Table(table_name, meta,
 
86
                               Column('id', Integer, primary_key=True,
 
87
                                      nullable=False),
 
88
                               Column('a', Integer),
 
89
                               Column('foo', BigInteger, default=0),
 
90
                               UniqueConstraint('a', name='uniq_a'),
 
91
                               UniqueConstraint('foo', name=uc_name))
 
92
            test_table.create()
 
93
 
 
94
            engine.execute(test_table.insert(), values)
 
95
            if key == "sqlite":
 
96
                # NOTE(boris-42): Missing info about column `foo` that has
 
97
                #                 unsupported type BigInteger.
 
98
                self.assertRaises(exception.NovaException,
 
99
                                  utils.drop_unique_constraint,
 
100
                                  engine, table_name, uc_name, 'foo')
 
101
 
 
102
                # NOTE(boris-42): Wrong type of foo instance. it should be
 
103
                #                 instance of sqlalchemy.Column.
 
104
                self.assertRaises(exception.NovaException,
 
105
                                  utils.drop_unique_constraint,
 
106
                                  engine, table_name, uc_name, 'foo',
 
107
                                  foo=Integer())
 
108
 
 
109
            foo = Column('foo', BigInteger, default=0)
 
110
            utils.drop_unique_constraint(engine, table_name, uc_name, 'foo',
 
111
                                         foo=foo)
 
112
 
 
113
            s = test_table.select().order_by(test_table.c.id)
 
114
            rows = engine.execute(s).fetchall()
 
115
 
 
116
            for i in xrange(0, len(values)):
 
117
                v = values[i]
 
118
                self.assertEqual((v['id'], v['a'], v['foo']), rows[i])
 
119
 
 
120
            # NOTE(boris-42): Update data about Table from DB.
 
121
            meta = MetaData()
 
122
            meta.bind = engine
 
123
            test_table = Table(table_name, meta, autoload=True)
 
124
            constraints = filter(lambda c: c.name == uc_name,
 
125
                                 test_table.constraints)
 
126
            self.assertEqual(len(constraints), 0)
 
127
            self.assertEqual(len(test_table.constraints), 1)
 
128
            test_table.drop()
 
129
 
 
130
    def _populate_db_for_drop_duplicate_entries(self, engine, meta,
 
131
                                                table_name):
 
132
        values = [
 
133
            {'id': 11, 'a': 3, 'b': 10, 'c': 'abcdef'},
 
134
            {'id': 12, 'a': 5, 'b': 10, 'c': 'abcdef'},
 
135
            {'id': 13, 'a': 6, 'b': 10, 'c': 'abcdef'},
 
136
            {'id': 14, 'a': 7, 'b': 10, 'c': 'abcdef'},
 
137
            {'id': 21, 'a': 1, 'b': 20, 'c': 'aa'},
 
138
            {'id': 31, 'a': 1, 'b': 20, 'c': 'bb'},
 
139
            {'id': 41, 'a': 1, 'b': 30, 'c': 'aef'},
 
140
            {'id': 42, 'a': 2, 'b': 30, 'c': 'aef'},
 
141
            {'id': 43, 'a': 3, 'b': 30, 'c': 'aef'}
 
142
        ]
 
143
 
 
144
        test_table = Table(table_name, meta,
 
145
                           Column('id', Integer, primary_key=True,
 
146
                                  nullable=False),
 
147
                           Column('a', Integer),
 
148
                           Column('b', Integer),
 
149
                           Column('c', String),
 
150
                           Column('deleted', Integer, default=0),
 
151
                           Column('deleted_at', DateTime),
 
152
                           Column('updated_at', DateTime))
 
153
 
 
154
        test_table.create()
 
155
        engine.execute(test_table.insert(), values)
 
156
        return test_table, values
 
157
 
 
158
    def test_drop_old_duplicate_entries_from_table(self):
 
159
        table_name = "__test_tmp_table__"
 
160
 
 
161
        for key, engine in self.engines.items():
 
162
            meta = MetaData()
 
163
            meta.bind = engine
 
164
            test_table, values = self.\
 
165
                    _populate_db_for_drop_duplicate_entries(engine, meta,
 
166
                                                            table_name)
 
167
 
 
168
            utils.drop_old_duplicate_entries_from_table(engine, table_name,
 
169
                                                        False, 'b', 'c')
 
170
 
 
171
            uniq_values = set()
 
172
            expected_ids = []
 
173
            for value in sorted(values, key=lambda x: x['id'], reverse=True):
 
174
                uniq_value = (('b', value['b']), ('c', value['c']))
 
175
                if uniq_value in uniq_values:
 
176
                    continue
 
177
                uniq_values.add(uniq_value)
 
178
                expected_ids.append(value['id'])
 
179
 
 
180
            real_ids = [row[0] for row in
 
181
                        engine.execute(select([test_table.c.id])).fetchall()]
 
182
 
 
183
            self.assertEqual(len(real_ids), len(expected_ids))
 
184
            for id_ in expected_ids:
 
185
                self.assertTrue(id_ in real_ids)
 
186
 
 
187
    def test_drop_old_duplicate_entries_from_table_soft_delete(self):
 
188
        table_name = "__test_tmp_table__"
 
189
 
 
190
        for key, engine in self.engines.items():
 
191
            meta = MetaData()
 
192
            meta.bind = engine
 
193
            table, values = self.\
 
194
                    _populate_db_for_drop_duplicate_entries(engine, meta,
 
195
                                                            table_name)
 
196
            utils.drop_old_duplicate_entries_from_table(engine, table_name,
 
197
                                                        True, 'b', 'c')
 
198
            uniq_values = set()
 
199
            expected_values = []
 
200
            soft_deleted_values = []
 
201
 
 
202
            for value in sorted(values, key=lambda x: x['id'], reverse=True):
 
203
                uniq_value = (('b', value['b']), ('c', value['c']))
 
204
                if uniq_value in uniq_values:
 
205
                    soft_deleted_values.append(value)
 
206
                    continue
 
207
                uniq_values.add(uniq_value)
 
208
                expected_values.append(value)
 
209
 
 
210
            base_select = table.select()
 
211
 
 
212
            rows_select = base_select.\
 
213
                                where(table.c.deleted != table.c.id)
 
214
            row_ids = [row['id'] for row in
 
215
                            engine.execute(rows_select).fetchall()]
 
216
            self.assertEqual(len(row_ids), len(expected_values))
 
217
            for value in expected_values:
 
218
                self.assertTrue(value['id'] in row_ids)
 
219
 
 
220
            deleted_rows_select = base_select.\
 
221
                                    where(table.c.deleted == table.c.id)
 
222
            deleted_rows_ids = [row['id'] for row in
 
223
                                engine.execute(deleted_rows_select).fetchall()]
 
224
            self.assertEqual(len(deleted_rows_ids),
 
225
                             len(values) - len(row_ids))
 
226
            for value in soft_deleted_values:
 
227
                self.assertTrue(value['id'] in deleted_rows_ids)