← 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_script.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_script.py
 
1
#!/usr/bin/env python
 
2
# -*- coding: utf-8 -*-
 
3
 
 
4
import os
 
5
import sys
 
6
import shutil
 
7
 
 
8
from migrate.versioning import exceptions, version, repository
 
9
from migrate.versioning.script import *
 
10
from migrate.versioning.util import *
 
11
 
 
12
from migrate.tests import fixture
 
13
from migrate.tests.fixture.models import tmp_sql_table
 
14
 
 
15
 
 
16
class TestBaseScript(fixture.Pathed):
 
17
 
 
18
    def test_all(self):
 
19
        """Testing all basic BaseScript operations"""
 
20
        # verify / source / run
 
21
        src = self.tmp()
 
22
        open(src, 'w').close()
 
23
        bscript = BaseScript(src)
 
24
        BaseScript.verify(src)
 
25
        self.assertEqual(bscript.source(), '')
 
26
        self.assertRaises(NotImplementedError, bscript.run, 'foobar')
 
27
 
 
28
 
 
29
class TestPyScript(fixture.Pathed, fixture.DB):
 
30
    cls = PythonScript
 
31
    def test_create(self):
 
32
        """We can create a migration script"""
 
33
        path = self.tmp_py()
 
34
        # Creating a file that doesn't exist should succeed
 
35
        self.cls.create(path)
 
36
        self.assert_(os.path.exists(path))
 
37
        # Created file should be a valid script (If not, raises an error)
 
38
        self.cls.verify(path)
 
39
        # Can't create it again: it already exists
 
40
        self.assertRaises(exceptions.PathFoundError,self.cls.create,path)
 
41
 
 
42
    @fixture.usedb(supported='sqlite')
 
43
    def test_run(self):
 
44
        script_path = self.tmp_py()
 
45
        pyscript = PythonScript.create(script_path)
 
46
        pyscript.run(self.engine, 1)
 
47
        pyscript.run(self.engine, -1)
 
48
 
 
49
        self.assertRaises(exceptions.ScriptError, pyscript.run, self.engine, 0)
 
50
        self.assertRaises(exceptions.ScriptError, pyscript._func, 'foobar')
 
51
 
 
52
        # clean pyc file
 
53
        os.remove(script_path + 'c')
 
54
 
 
55
        # test deprecated upgrade/downgrade with no arguments
 
56
        contents = open(script_path, 'r').read()
 
57
        f = open(script_path, 'w')
 
58
        f.write(contents.replace("upgrade(migrate_engine)", "upgrade()"))
 
59
        f.close()
 
60
 
 
61
        pyscript = PythonScript(script_path)
 
62
        pyscript._module = None
 
63
        try:
 
64
            pyscript.run(self.engine, 1)
 
65
            pyscript.run(self.engine, -1)
 
66
        except exceptions.ScriptError:
 
67
            pass
 
68
        else:
 
69
            self.fail()
 
70
 
 
71
    def test_verify_notfound(self):
 
72
        """Correctly verify a python migration script: nonexistant file"""
 
73
        path = self.tmp_py()
 
74
        self.assertFalse(os.path.exists(path))
 
75
        # Fails on empty path
 
76
        self.assertRaises(exceptions.InvalidScriptError,self.cls.verify,path)
 
77
        self.assertRaises(exceptions.InvalidScriptError,self.cls,path)
 
78
 
 
79
    def test_verify_invalidpy(self):
 
80
        """Correctly verify a python migration script: invalid python file"""
 
81
        path=self.tmp_py()
 
82
        # Create empty file
 
83
        f = open(path,'w')
 
84
        f.write("def fail")
 
85
        f.close()
 
86
        self.assertRaises(Exception,self.cls.verify_module,path)
 
87
        # script isn't verified on creation, but on module reference
 
88
        py = self.cls(path)
 
89
        self.assertRaises(Exception,(lambda x: x.module),py)
 
90
 
 
91
    def test_verify_nofuncs(self):
 
92
        """Correctly verify a python migration script: valid python file; no upgrade func"""
 
93
        path = self.tmp_py()
 
94
        # Create empty file
 
95
        f = open(path, 'w')
 
96
        f.write("def zergling():\n\tprint 'rush'")
 
97
        f.close()
 
98
        self.assertRaises(exceptions.InvalidScriptError, self.cls.verify_module, path)
 
99
        # script isn't verified on creation, but on module reference
 
100
        py = self.cls(path)
 
101
        self.assertRaises(exceptions.InvalidScriptError,(lambda x: x.module),py)
 
102
 
 
103
    @fixture.usedb(supported='sqlite')
 
104
    def test_preview_sql(self):
 
105
        """Preview SQL abstract from ORM layer (sqlite)"""
 
106
        path = self.tmp_py()
 
107
 
 
108
        f = open(path, 'w')
 
109
        content = '''
 
110
from migrate import *
 
111
from sqlalchemy import *
 
112
 
 
113
metadata = MetaData()
 
114
 
 
115
UserGroup = Table('Link', metadata,
 
116
    Column('link1ID', Integer),
 
117
    Column('link2ID', Integer),
 
118
    UniqueConstraint('link1ID', 'link2ID'))
 
119
 
 
120
def upgrade(migrate_engine):
 
121
    metadata.create_all(migrate_engine)
 
122
        '''
 
123
        f.write(content)
 
124
        f.close()
 
125
 
 
126
        pyscript = self.cls(path)
 
127
        SQL = pyscript.preview_sql(self.url, 1)
 
128
        self.assertEqualsIgnoreWhitespace("""
 
129
        CREATE TABLE "Link"
 
130
        ("link1ID" INTEGER,
 
131
        "link2ID" INTEGER,
 
132
        UNIQUE ("link1ID", "link2ID"))
 
133
        """, SQL)
 
134
        # TODO: test: No SQL should be executed!
 
135
 
 
136
    def test_verify_success(self):
 
137
        """Correctly verify a python migration script: success"""
 
138
        path = self.tmp_py()
 
139
        # Succeeds after creating
 
140
        self.cls.create(path)
 
141
        self.cls.verify(path)
 
142
 
 
143
    # test for PythonScript.make_update_script_for_model
 
144
 
 
145
    @fixture.usedb()
 
146
    def test_make_update_script_for_model(self):
 
147
        """Construct script source from differences of two models"""
 
148
 
 
149
        self.setup_model_params()
 
150
        self.write_file(self.first_model_path, self.base_source)
 
151
        self.write_file(self.second_model_path, self.base_source + self.model_source)
 
152
 
 
153
        source_script = self.pyscript.make_update_script_for_model(
 
154
            engine=self.engine,
 
155
            oldmodel=load_model('testmodel_first:meta'),
 
156
            model=load_model('testmodel_second:meta'),
 
157
            repository=self.repo_path,
 
158
        )
 
159
 
 
160
        self.assertTrue('User.create()' in source_script)
 
161
        self.assertTrue('User.drop()' in source_script)
 
162
 
 
163
    #@fixture.usedb()
 
164
    #def test_make_update_script_for_model_equals(self):
 
165
    #    """Try to make update script from two identical models"""
 
166
 
 
167
    #    self.setup_model_params()
 
168
    #    self.write_file(self.first_model_path, self.base_source + self.model_source)
 
169
    #    self.write_file(self.second_model_path, self.base_source + self.model_source)
 
170
 
 
171
    #    source_script = self.pyscript.make_update_script_for_model(
 
172
    #        engine=self.engine,
 
173
    #        oldmodel=load_model('testmodel_first:meta'),
 
174
    #        model=load_model('testmodel_second:meta'),
 
175
    #        repository=self.repo_path,
 
176
    #    )
 
177
 
 
178
    #    self.assertFalse('User.create()' in source_script)
 
179
    #    self.assertFalse('User.drop()' in source_script)
 
180
 
 
181
    def setup_model_params(self):
 
182
        self.script_path = self.tmp_py()
 
183
        self.repo_path = self.tmp()
 
184
        self.first_model_path = os.path.join(self.temp_usable_dir, 'testmodel_first.py')
 
185
        self.second_model_path = os.path.join(self.temp_usable_dir, 'testmodel_second.py')
 
186
 
 
187
        self.base_source = """from sqlalchemy import *\nmeta = MetaData()\n"""
 
188
        self.model_source = """
 
189
User = Table('User', meta,
 
190
    Column('id', Integer, primary_key=True),
 
191
    Column('login', Unicode(40)),
 
192
    Column('passwd', String(40)),
 
193
)"""
 
194
 
 
195
        self.repo = repository.Repository.create(self.repo_path, 'repo')
 
196
        self.pyscript = PythonScript.create(self.script_path)
 
197
 
 
198
    def write_file(self, path, contents):
 
199
        f = open(path, 'w')
 
200
        f.write(contents)
 
201
        f.close()
 
202
        
 
203
 
 
204
class TestSqlScript(fixture.Pathed, fixture.DB):
 
205
 
 
206
    @fixture.usedb()
 
207
    def test_error(self):
 
208
        """Test if exception is raised on wrong script source"""
 
209
        src = self.tmp()
 
210
 
 
211
        f = open(src, 'w')
 
212
        f.write("""foobar""")
 
213
        f.close()
 
214
 
 
215
        sqls = SqlScript(src)
 
216
        self.assertRaises(Exception, sqls.run, self.engine)
 
217
 
 
218
    @fixture.usedb()
 
219
    def test_success(self):
 
220
        """Test sucessful SQL execution"""
 
221
        # cleanup and prepare python script
 
222
        tmp_sql_table.metadata.drop_all(self.engine, checkfirst=True)
 
223
        script_path = self.tmp_py()
 
224
        pyscript = PythonScript.create(script_path)
 
225
 
 
226
        # populate python script
 
227
        contents = open(script_path, 'r').read()
 
228
        contents = contents.replace("pass", "tmp_sql_table.create(migrate_engine)")
 
229
        contents = 'from migrate.tests.fixture.models import tmp_sql_table\n' + contents
 
230
        f = open(script_path, 'w')
 
231
        f.write(contents)
 
232
        f.close()
 
233
 
 
234
        # write SQL script from python script preview
 
235
        pyscript = PythonScript(script_path)
 
236
        src = self.tmp()
 
237
        f = open(src, 'w')
 
238
        f.write(pyscript.preview_sql(self.url, 1))
 
239
        f.close()
 
240
 
 
241
        # run the change
 
242
        sqls = SqlScript(src)
 
243
        sqls.run(self.engine, executemany=False)
 
244
        tmp_sql_table.metadata.drop_all(self.engine, checkfirst=True)