1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import shutil
from migrate import exceptions
from migrate.versioning.repository import *
from migrate.versioning.script import *
from nose.tools import raises
from migrate.tests import fixture
from datetime import datetime
class TestRepository(fixture.Pathed):
def test_create(self):
"""Repositories are created successfully"""
path = self.tmp_repos()
name = 'repository_name'
# Creating a repository that doesn't exist should succeed
repo = Repository.create(path, name)
config_path = repo.config.path
manage_path = os.path.join(repo.path, 'manage.py')
self.assert_(repo)
# Files should actually be created
self.assert_(os.path.exists(path))
self.assert_(os.path.exists(config_path))
self.assert_(os.path.exists(manage_path))
# Can't create it again: it already exists
self.assertRaises(exceptions.PathFoundError, Repository.create, path, name)
return path
def test_load(self):
"""We should be able to load information about an existing repository"""
# Create a repository to load
path = self.test_create()
repos = Repository(path)
self.assert_(repos)
self.assert_(repos.config)
self.assert_(repos.config.get('db_settings', 'version_table'))
# version_table's default isn't none
self.assertNotEquals(repos.config.get('db_settings', 'version_table'), 'None')
def test_load_notfound(self):
"""Nonexistant repositories shouldn't be loaded"""
path = self.tmp_repos()
self.assert_(not os.path.exists(path))
self.assertRaises(exceptions.InvalidRepositoryError, Repository, path)
def test_load_invalid(self):
"""Invalid repos shouldn't be loaded"""
# Here, invalid=empty directory. There may be other conditions too,
# but we shouldn't need to test all of them
path = self.tmp_repos()
os.mkdir(path)
self.assertRaises(exceptions.InvalidRepositoryError, Repository, path)
class TestVersionedRepository(fixture.Pathed):
"""Tests on an existing repository with a single python script"""
def setUp(self):
super(TestVersionedRepository, self).setUp()
Repository.clear()
self.path_repos = self.tmp_repos()
Repository.create(self.path_repos, 'repository_name')
def test_version(self):
"""We should correctly detect the version of a repository"""
repos = Repository(self.path_repos)
# Get latest version, or detect if a specified version exists
self.assertEquals(repos.latest, 0)
# repos.latest isn't an integer, but a VerNum
# (so we can't just assume the following tests are correct)
self.assert_(repos.latest >= 0)
self.assert_(repos.latest < 1)
# Create a script and test again
repos.create_script('')
self.assertEquals(repos.latest, 1)
self.assert_(repos.latest >= 0)
self.assert_(repos.latest >= 1)
self.assert_(repos.latest < 2)
# Create a new script and test again
repos.create_script('')
self.assertEquals(repos.latest, 2)
self.assert_(repos.latest >= 0)
self.assert_(repos.latest >= 1)
self.assert_(repos.latest >= 2)
self.assert_(repos.latest < 3)
def test_timestmap_numbering_version(self):
repos = Repository(self.path_repos)
repos.config.set('db_settings', 'use_timestamp_numbering', 'True')
# Get latest version, or detect if a specified version exists
self.assertEquals(repos.latest, 0)
# repos.latest isn't an integer, but a VerNum
# (so we can't just assume the following tests are correct)
self.assert_(repos.latest >= 0)
self.assert_(repos.latest < 1)
# Create a script and test again
now = int(datetime.utcnow().strftime('%Y%m%d%H%M%S'))
repos.create_script('')
print repos.latest
self.assertEquals(repos.latest, now)
def test_source(self):
"""Get a script object by version number and view its source"""
# Load repository and commit script
repo = Repository(self.path_repos)
repo.create_script('')
repo.create_script_sql('postgres', 'foo bar')
# Source is valid: script must have an upgrade function
# (not a very thorough test, but should be plenty)
source = repo.version(1).script().source()
self.assertTrue(source.find('def upgrade') >= 0)
import pprint; pprint.pprint(repo.version(2).sql)
source = repo.version(2).script('postgres', 'upgrade').source()
self.assertEqual(source.strip(), '')
def test_latestversion(self):
"""Repository.version() (no params) returns the latest version"""
repos = Repository(self.path_repos)
repos.create_script('')
self.assert_(repos.version(repos.latest) is repos.version())
self.assert_(repos.version() is not None)
def test_changeset(self):
"""Repositories can create changesets properly"""
# Create a nonzero-version repository of empty scripts
repos = Repository(self.path_repos)
for i in range(10):
repos.create_script('')
def check_changeset(params, length):
"""Creates and verifies a changeset"""
changeset = repos.changeset('postgres', *params)
self.assertEquals(len(changeset), length)
self.assertTrue(isinstance(changeset, Changeset))
uniq = list()
# Changesets are iterable
for version, change in changeset:
self.assert_(isinstance(change, BaseScript))
# Changes aren't identical
self.assert_(id(change) not in uniq)
uniq.append(id(change))
return changeset
# Upgrade to a specified version...
cs = check_changeset((0, 10), 10)
self.assertEquals(cs.keys().pop(0),0 ) # 0 -> 1: index is starting version
self.assertEquals(cs.keys().pop(), 9) # 9 -> 10: index is starting version
self.assertEquals(cs.start, 0) # starting version
self.assertEquals(cs.end, 10) # ending version
check_changeset((0, 1), 1)
check_changeset((0, 5), 5)
check_changeset((0, 0), 0)
check_changeset((5, 5), 0)
check_changeset((10, 10), 0)
check_changeset((5, 10), 5)
# Can't request a changeset of higher version than this repository
self.assertRaises(Exception, repos.changeset, 'postgres', 5, 11)
self.assertRaises(Exception, repos.changeset, 'postgres', -1, 5)
# Upgrade to the latest version...
cs = check_changeset((0,), 10)
self.assertEquals(cs.keys().pop(0), 0)
self.assertEquals(cs.keys().pop(), 9)
self.assertEquals(cs.start, 0)
self.assertEquals(cs.end, 10)
check_changeset((1,), 9)
check_changeset((5,), 5)
check_changeset((9,), 1)
check_changeset((10,), 0)
# run changes
cs.run('postgres', 'upgrade')
# Can't request a changeset of higher/lower version than this repository
self.assertRaises(Exception, repos.changeset, 'postgres', 11)
self.assertRaises(Exception, repos.changeset, 'postgres', -1)
# Downgrade
cs = check_changeset((10, 0),10)
self.assertEquals(cs.keys().pop(0), 10) # 10 -> 9
self.assertEquals(cs.keys().pop(), 1) # 1 -> 0
self.assertEquals(cs.start, 10)
self.assertEquals(cs.end, 0)
check_changeset((10, 5), 5)
check_changeset((5, 0), 5)
def test_many_versions(self):
"""Test what happens when lots of versions are created"""
repos = Repository(self.path_repos)
for i in range(1001):
repos.create_script('')
# since we normally create 3 digit ones, let's see if we blow up
self.assert_(os.path.exists('%s/versions/1000.py' % self.path_repos))
self.assert_(os.path.exists('%s/versions/1001.py' % self.path_repos))
# TODO: test manage file
# TODO: test changeset
|
