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
|
# Copyright 2012, 2013 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
from os import environ
from os.path import join
from os.path import exists
import pymongo
from tempfile import mkdtemp
from unittest import TestCase
from charmworld.models import getconnection
from charmworld.models import getdb
from charmworld.testing.factory import random_string
from charmworld.utils import get_ini
from migrate import DataStore
from migrate import Versions
INI = get_ini()
MONGO_URL = environ.get('TEST_MONGODB')
INI['mongo.url'] = MONGO_URL
SCRIPT_TEMPLATE = """
def upgrade(db, check_value=None):
# Complete this function with the work to be done for this migration.
return {0}
"""
def make_version_files(directory, number, runnable=True):
"""Generate a series of revisions to test against.
:param number: The number of versions to generate from 1 - number
"""
version_files = []
for i in range(1, number + 1):
filename = "{0}_{1}.py".format(str(i).zfill(3), random_string(20))
path = join(directory, filename)
if runnable:
with open(path, 'w') as fh:
# We add in the hard coded number to the template so we can
# tell we hit the right file.
fh.write(SCRIPT_TEMPLATE.format(i))
else:
# We just need the file to exist, forget about the contents.
open(path, 'w').close()
version_files.append(filename)
return version_files
class TestDataStore(TestCase):
"""The data store talks to mongo."""
def setUp(self):
self.connection = getconnection({
'mongo.url': MONGO_URL,
})
self.db = getdb(self.connection, 'juju_test')
def tearDown(self):
# Drop the Mongo database.
connection = pymongo.Connection(MONGO_URL)
connection.drop_database('juju_test')
def test_mongo_version_starts_unset(self):
"""Unset is just None."""
ds = DataStore(self.db)
self.assertIsNone(ds.current_version)
def test_mongo_version_datastore(self):
"""Versioning the db is successfull and starts at 0."""
ds = DataStore(self.db)
ds.version_datastore()
self.assertEqual(0, ds.current_version)
def test_datastore_not_reversionable(self):
"""We can't version a datastore already versioned."""
ds = DataStore(self.db)
ds.version_datastore()
self.assertEqual(0, ds.current_version)
# Now reversion and check for the exception.
self.assertRaises(Exception, ds.version_datastore)
class TestVersions(TestCase):
"""The local versions of the files."""
def setUp(self):
self.tmpdir = mkdtemp()
self.version_files = make_version_files(self.tmpdir, 4)
self.versions = Versions(self.tmpdir)
def tearDown(self):
self.ds = None
self.version = None
def test_latest_wo_migrations(self):
""""""
tmpdir = mkdtemp()
versions = Versions(tmpdir)
self.assertEqual(versions.latest, 0)
def test_latest_version(self):
"""latest returns the number of the newest revision."""
# Note that we're saying to create up to the specified version.
self.assertEqual(self.versions.latest, 4)
def test_version_list(self):
"""versions is the filenames of the version files indexed by order"""
for i, filename in enumerate(self.version_files):
# Note that we're saying to create up to the specified version.
self.assertEqual(filename, self.versions.versions[i + 1])
def test_new_migration_template(self):
"""Create a new migration file and latest is updated."""
self.versions.create_new_version_file('testing basic migration')
self.assertEqual(5, self.versions.latest)
self.assertEqual(
'005_testing_basic_migration.py',
self.versions.versions[5])
# And this new file should exist in our versions directory
self.assertTrue(exists(join(self.tmpdir, self.versions.versions[5])))
class TestVersionUpgrades(TestCase):
"""Functional testing of the actual upgrade process."""
def setUp(self):
self.connection = getconnection({
'mongo.url': MONGO_URL,
})
self.db = getdb(self.connection, 'juju_test')
self.ds = DataStore(self.db)
tmpdir = mkdtemp()
make_version_files(tmpdir, 4, runnable=True)
self.version = Versions(tmpdir)
def tearDown(self):
# Drop the Mongo database.
connection = pymongo.Connection(MONGO_URL)
connection.drop_database('juju_test')
self.ds = None
self.version = None
def test_upgrade_fails_unversioned(self):
"""upgrade will run `upgrade` methods in any migrations > current"""
# We've not yet versioned the db, so it better blow up.
self.assertRaises(Exception, self.version.upgrade, (self.ds, False))
def test_upgrade_wo_migrations(self):
"""Upgrade runs silently if no migrations are created yet."""
self.ds.version_datastore()
tmpdir = mkdtemp()
versions = Versions(tmpdir)
self.assertEqual(versions.upgrade(self.ds, False), None)
def test_upgrade_succeeds(self):
"""upgrade will run the methods and we get a version of 4."""
self.ds.version_datastore()
# We should get 4 versions done. The upgrade returns 4
last_version = self.version.upgrade(self.ds, False)
self.assertEqual(4, last_version)
# and we can query the db for the revision of 4
self.assertEqual(4, self.ds.current_version)
def test_upgrade_succeeds_from_unversioned(self):
"""Forcing init no an unversioned db will upgrade successfully."""
last_version = self.version.upgrade(self.ds, True)
self.assertEqual(4, last_version)
# and we can query the db for the revision of 4
self.assertEqual(4, self.ds.current_version)
|