2
from pysqlite2 import dbapi2 as sqlite3
6
from landscape.lib.persist import Persist
7
from landscape.patch import UpgradeManager, SQLiteUpgradeManager
8
from landscape.tests.helpers import LandscapeTest
11
class PatchTest(LandscapeTest):
14
LandscapeTest.setUp(self)
15
self.persist = Persist()
16
self.manager = UpgradeManager()
18
def test_wb_nopatches(self):
20
Applying no patches should make no change to the database,
21
apart from maybe specifying a default version.
23
self.assertEquals(self.persist._hardmap, {})
24
self.manager.apply(self.persist)
25
self.assertEquals(self.persist._hardmap, {"system-version": 0})
27
def test_one_patch(self):
28
"""Test that patches are called and passed a L{Persist} object."""
30
self.manager.register_upgrader(1, l.append)
31
self.manager.apply(self.persist)
32
self.assertEquals(l, [self.persist])
34
def test_two_patches(self):
35
"""Test that patches are run in order."""
37
self.manager.register_upgrader(2, lambda x: l.append(2))
38
self.manager.register_upgrader(1, lambda x: l.append(1))
40
self.manager.apply(self.persist)
41
self.assertEquals(l, [1, 2])
43
def test_record_version(self):
44
"""When a patch is run it should update the C{system-version}."""
45
self.assertEquals(self.persist.get("system-version"), None)
46
self.manager.register_upgrader(1, lambda x: None)
47
self.manager.apply(self.persist)
48
self.assertEquals(self.persist.get("system-version"), 1)
50
def test_only_apply_unapplied_versions(self):
51
"""Upgraders should only be run if they haven't been run before."""
53
self.manager.register_upgrader(1, lambda x: l.append(1))
54
self.manager.apply(self.persist)
55
self.manager.apply(self.persist)
56
self.assertEquals(l, [1])
58
def test_initialize(self):
59
"""Marking no upgraders as applied should leave the version at 0."""
60
self.manager.initialize(self.persist)
61
self.assertEquals(self.persist.get("system-version"), 0)
63
def test_initialize_with_upgraders(self):
65
After registering some upgraders, initialize should set the
66
version for the new persist to the highest version number
67
available, without running any of the upgraders.
69
self.manager.register_upgrader(1, lambda x: 1/0)
70
self.manager.register_upgrader(5, lambda x: 1/0)
71
self.manager.register_upgrader(3, lambda x: 1/0)
72
self.manager.initialize(self.persist)
74
self.assertEquals(self.persist.get("system-version"), 5)
76
def test_decorated_upgraders_run(self):
78
Upgraders that use the L{upgrader} decorator should
79
automatically register themselves with a given
80
L{UpgradeManager} and be run when the manager applies patches.
82
upgrade_manager = UpgradeManager()
83
@upgrade_manager.upgrader(1)
85
self.persist.set("upgrade-called", True)
87
upgrade_manager.apply(self.persist)
88
self.assertTrue(self.persist.get("upgrade-called"))
91
class SQLitePatchTest(LandscapeTest):
94
LandscapeTest.setUp(self)
95
self.db_filename = self.make_path()
96
self.db = sqlite3.connect(self.db_filename, isolation_level=None)
97
self.cursor = self.db.cursor()
98
self.manager = SQLiteUpgradeManager()
99
self.version_query = "SELECT MAX(version) from patch"
101
def test_no_patches(self):
103
Applying no patches should make no change to the database,
104
apart from maybe specifying a default version.
106
self.manager.initialize(self.cursor)
107
self.manager.apply(self.cursor)
108
self.assertEquals(self.manager.get_database_versions(self.cursor),
111
def test_one_patch(self):
112
"""Test that patches are called and passed a sqlite db object."""
114
self.manager.initialize(self.cursor)
115
self.manager.register_upgrader(1, l.append)
116
self.manager.apply(self.cursor)
117
self.assertEquals(l, [self.cursor])
118
self.cursor.execute(self.version_query)
119
self.assertEquals(self.cursor.fetchone(), (1,))
121
def test_two_patches(self):
122
"""Test that patches are run in order."""
124
self.manager.initialize(self.cursor)
125
self.manager.register_upgrader(2, lambda x: l.append(2))
126
self.manager.register_upgrader(1, lambda x: l.append(1))
128
self.manager.apply(self.cursor)
129
self.assertEquals(l, [1, 2])
130
self.cursor.execute(self.version_query)
131
self.assertEquals(self.cursor.fetchone(), (2,))
133
def test_only_apply_unapplied_versions(self):
134
"""Upgraders should only be run if they haven't been run before."""
138
self.manager.initialize(self.cursor)
139
self.manager.register_upgrader(1, lambda x: patch1.append(1))
140
self.manager.register_upgrader(2, lambda x: patch2.append(1))
141
self.manager.register_upgrader(3, lambda x: patch3.append(1))
142
self.manager.apply_one(2, self.cursor)
143
self.assertEquals((patch1, patch2, patch3), ([], [1], []))
144
self.manager.apply(self.cursor)
145
self.assertEquals((patch1, patch2, patch3), ([1], [1], [1]))
147
def test_initialize_with_upgraders(self):
149
After registering some upgraders, initialize should set the
150
version of the newly created database to the highest version
153
self.manager.register_upgrader(1, lambda x: 1/0)
154
self.manager.register_upgrader(5, lambda x: 1/0)
155
self.manager.register_upgrader(3, lambda x: 1/0)
156
self.manager.initialize(self.cursor)
157
self.assertEquals(self.manager.get_database_versions(self.cursor),
160
def test_decorated_upgraders_run(self):
162
Upgraders that use the L{upgrader} decorator should
163
automatically register themselves with a given
164
L{UpgradeManager} and be run when the manager applies patches.
166
upgrade_manager = SQLiteUpgradeManager()
167
upgrade_manager.initialize(self.cursor)
169
@upgrade_manager.upgrader(1)
173
upgrade_manager.apply(self.cursor)
174
self.assertEquals(l, [self.cursor])