← Back to branch summary

~ubuntu-branches/ubuntu/saucy/migrate/saucy-proposed

« back to all changes in this revision

Viewing changes to migrate/tests/versioning/test_schema.py

  • Committer: Bazaar Package Importer
  • Author(s): Jan Dittberner
  • Date: 2010-07-12 00:24:57 UTC
  • mfrom: (1.1.5 upstream) (2.1.8 sid)
  • Revision ID: james.westby@ubuntu.com-20100712002457-4j2fdmco4u9kqzm5
Upload to unstable.

Show diffs side-by-side

added added

removed removed

Lines of Context:
migrate/tests/versioning/test_schema.py
 
1
#!/usr/bin/env python
 
2
# -*- coding: utf-8 -*-
 
3
 
 
4
import os
 
5
import shutil
 
6
 
 
7
from migrate.versioning.schema import *
 
8
from migrate.versioning import script, exceptions, schemadiff
 
9
 
 
10
from sqlalchemy import *
 
11
 
 
12
from migrate.tests import fixture
 
13
 
 
14
 
 
15
class TestControlledSchema(fixture.Pathed, fixture.DB):
 
16
    # Transactions break postgres in this test; we'll clean up after ourselves
 
17
    level = fixture.DB.CONNECT
 
18
 
 
19
    def setUp(self):
 
20
        super(TestControlledSchema, self).setUp()
 
21
        self.path_repos = self.temp_usable_dir + '/repo/'
 
22
        self.repos = Repository.create(self.path_repos, 'repo_name')
 
23
 
 
24
    def _setup(self, url):
 
25
        self.setUp()
 
26
        super(TestControlledSchema, self)._setup(url)
 
27
        self.cleanup()
 
28
 
 
29
    def _teardown(self):
 
30
        super(TestControlledSchema, self)._teardown()
 
31
        self.cleanup()
 
32
        self.tearDown()
 
33
 
 
34
    def cleanup(self):
 
35
        # drop existing version table if necessary
 
36
        try:
 
37
            ControlledSchema(self.engine, self.repos).drop()
 
38
        except:
 
39
            # No table to drop; that's fine, be silent
 
40
            pass
 
41
 
 
42
    def tearDown(self):
 
43
        self.cleanup()
 
44
        super(TestControlledSchema, self).tearDown()
 
45
 
 
46
    @fixture.usedb()
 
47
    def test_version_control(self):
 
48
        """Establish version control on a particular database"""
 
49
        # Establish version control on this database
 
50
        dbcontrol = ControlledSchema.create(self.engine, self.repos)
 
51
 
 
52
        # Trying to create another DB this way fails: table exists
 
53
        self.assertRaises(exceptions.DatabaseAlreadyControlledError,
 
54
            ControlledSchema.create, self.engine, self.repos)
 
55
        
 
56
        # We can load a controlled DB this way, too
 
57
        dbcontrol0 = ControlledSchema(self.engine, self.repos)
 
58
        self.assertEquals(dbcontrol, dbcontrol0)
 
59
 
 
60
        # We can also use a repository path, instead of a repository
 
61
        dbcontrol0 = ControlledSchema(self.engine, self.repos.path)
 
62
        self.assertEquals(dbcontrol, dbcontrol0)
 
63
 
 
64
        # We don't have to use the same connection
 
65
        engine = create_engine(self.url)
 
66
        dbcontrol0 = ControlledSchema(engine, self.repos.path)
 
67
        self.assertEquals(dbcontrol, dbcontrol0)
 
68
 
 
69
        # Clean up: 
 
70
        dbcontrol.drop()
 
71
 
 
72
        # Attempting to drop vc from a db without it should fail
 
73
        self.assertRaises(exceptions.DatabaseNotControlledError, dbcontrol.drop)
 
74
 
 
75
        # No table defined should raise error
 
76
        self.assertRaises(exceptions.DatabaseNotControlledError,
 
77
            ControlledSchema, self.engine, self.repos)
 
78
 
 
79
    @fixture.usedb()
 
80
    def test_version_control_specified(self):
 
81
        """Establish version control with a specified version"""
 
82
        # Establish version control on this database
 
83
        version = 0
 
84
        dbcontrol = ControlledSchema.create(self.engine, self.repos, version)
 
85
        self.assertEquals(dbcontrol.version, version)
 
86
        
 
87
        # Correct when we load it, too
 
88
        dbcontrol = ControlledSchema(self.engine, self.repos)
 
89
        self.assertEquals(dbcontrol.version, version)
 
90
 
 
91
        dbcontrol.drop()
 
92
 
 
93
        # Now try it with a nonzero value
 
94
        version = 10
 
95
        for i in range(version):
 
96
            self.repos.create_script('')
 
97
        self.assertEquals(self.repos.latest, version)
 
98
 
 
99
        # Test with some mid-range value
 
100
        dbcontrol = ControlledSchema.create(self.engine,self.repos, 5)
 
101
        self.assertEquals(dbcontrol.version, 5)
 
102
        dbcontrol.drop()
 
103
 
 
104
        # Test with max value
 
105
        dbcontrol = ControlledSchema.create(self.engine, self.repos, version)
 
106
        self.assertEquals(dbcontrol.version, version)
 
107
        dbcontrol.drop()
 
108
 
 
109
    @fixture.usedb()
 
110
    def test_version_control_invalid(self):
 
111
        """Try to establish version control with an invalid version"""
 
112
        versions = ('Thirteen', '-1', -1, '' , 13)
 
113
        # A fresh repository doesn't go up to version 13 yet
 
114
        for version in versions:
 
115
            #self.assertRaises(ControlledSchema.InvalidVersionError,
 
116
            # Can't have custom errors with assertRaises...
 
117
            try:
 
118
                ControlledSchema.create(self.engine, self.repos, version)
 
119
                self.assert_(False, repr(version))
 
120
            except exceptions.InvalidVersionError:
 
121
                pass
 
122
 
 
123
    @fixture.usedb()
 
124
    def test_changeset(self):
 
125
        """Create changeset from controlled schema"""
 
126
        dbschema = ControlledSchema.create(self.engine, self.repos)
 
127
        
 
128
        # empty schema doesn't have changesets
 
129
        cs = dbschema.changeset()
 
130
        self.assertEqual(cs, {})
 
131
 
 
132
        for i in range(5):
 
133
            self.repos.create_script('')
 
134
        self.assertEquals(self.repos.latest, 5)
 
135
 
 
136
        cs = dbschema.changeset(5)
 
137
        self.assertEqual(len(cs), 5)
 
138
 
 
139
        # cleanup
 
140
        dbschema.drop()
 
141
 
 
142
    @fixture.usedb()
 
143
    def test_upgrade_runchange(self):
 
144
        dbschema = ControlledSchema.create(self.engine, self.repos)
 
145
        
 
146
        for i in range(10):
 
147
            self.repos.create_script('')
 
148
 
 
149
        self.assertEquals(self.repos.latest, 10)
 
150
 
 
151
        dbschema.upgrade(10)
 
152
 
 
153
        self.assertRaises(ValueError, dbschema.upgrade, 'a')
 
154
        self.assertRaises(exceptions.InvalidVersionError, dbschema.runchange, 20, '', 1)
 
155
 
 
156
        # TODO: test for table version in db
 
157
 
 
158
        # cleanup
 
159
        dbschema.drop()
 
160
 
 
161
    @fixture.usedb()
 
162
    def test_create_model(self):
 
163
        """Test workflow to generate create_model"""
 
164
        model = ControlledSchema.create_model(self.engine, self.repos, declarative=False)
 
165
        self.assertTrue(isinstance(model, basestring))
 
166
 
 
167
        model = ControlledSchema.create_model(self.engine, self.repos.path, declarative=True)
 
168
        self.assertTrue(isinstance(model, basestring))
 
169
 
 
170
    @fixture.usedb()
 
171
    def test_compare_model_to_db(self):
 
172
        meta = self.construct_model()
 
173
 
 
174
        diff = ControlledSchema.compare_model_to_db(self.engine, meta, self.repos)
 
175
        self.assertTrue(isinstance(diff, schemadiff.SchemaDiff))
 
176
 
 
177
        diff = ControlledSchema.compare_model_to_db(self.engine, meta, self.repos.path)
 
178
        self.assertTrue(isinstance(diff, schemadiff.SchemaDiff))
 
179
        meta.drop_all(self.engine)
 
180
 
 
181
    @fixture.usedb()
 
182
    def test_update_db_from_model(self):
 
183
        dbschema = ControlledSchema.create(self.engine, self.repos)
 
184
 
 
185
        meta = self.construct_model()
 
186
    
 
187
        dbschema.update_db_from_model(meta)
 
188
 
 
189
        # TODO: test for table version in db
 
190
 
 
191
        # cleanup
 
192
        dbschema.drop()
 
193
        meta.drop_all(self.engine)
 
194
 
 
195
    def construct_model(self):
 
196
        meta = MetaData()
 
197
 
 
198
        user = Table('temp_model_schema', meta, Column('id', Integer), Column('user', String(245)))
 
199
 
 
200
        return meta
 
201
 
 
202
    # TODO: test how are tables populated in db