← Back to branch summary

~ahasenack/landscape-client/landscape-client-1.5.5-0ubuntu0.9.04.0

« back to all changes in this revision

Viewing changes to landscape/tests/test_patch.py

  • Committer: Bazaar Package Importer
  • Author(s): Rick Clark
  • Date: 2008-09-08 16:35:57 UTC
  • mfrom: (1.1.1 upstream)
  • Revision ID: james.westby@ubuntu.com-20080908163557-l3ixzj5dxz37wnw2
Tags: 1.0.18-0ubuntu1
New upstream release 

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/tests/test_patch.py
 
1
try:
 
2
    from pysqlite2 import dbapi2 as sqlite3
 
3
except:
 
4
    import sqlite3
 
5
 
 
6
from landscape.lib.persist import Persist
 
7
from landscape.patch import UpgradeManager, SQLiteUpgradeManager
 
8
from landscape.tests.helpers import LandscapeTest
 
9
 
 
10
 
 
11
class PatchTest(LandscapeTest):
 
12
 
 
13
    def setUp(self):
 
14
        LandscapeTest.setUp(self)
 
15
        self.persist = Persist()
 
16
        self.manager = UpgradeManager()
 
17
 
 
18
    def test_wb_nopatches(self):
 
19
        """
 
20
        Applying no patches should make no change to the database,
 
21
        apart from maybe specifying a default version.
 
22
        """
 
23
        self.assertEquals(self.persist._hardmap, {})
 
24
        self.manager.apply(self.persist)
 
25
        self.assertEquals(self.persist._hardmap, {"system-version": 0})
 
26
 
 
27
    def test_one_patch(self):
 
28
        """Test that patches are called and passed a L{Persist} object."""
 
29
        l = []
 
30
        self.manager.register_upgrader(1, l.append)
 
31
        self.manager.apply(self.persist)
 
32
        self.assertEquals(l, [self.persist])
 
33
 
 
34
    def test_two_patches(self):
 
35
        """Test that patches are run in order."""
 
36
        l = []
 
37
        self.manager.register_upgrader(2, lambda x: l.append(2))
 
38
        self.manager.register_upgrader(1, lambda x: l.append(1))
 
39
 
 
40
        self.manager.apply(self.persist)
 
41
        self.assertEquals(l, [1, 2])
 
42
 
 
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)
 
49
 
 
50
    def test_only_apply_unapplied_versions(self):
 
51
        """Upgraders should only be run if they haven't been run before."""
 
52
        l = []
 
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])
 
57
 
 
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)
 
62
 
 
63
    def test_initialize_with_upgraders(self):
 
64
        """
 
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.
 
68
        """
 
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)
 
73
 
 
74
        self.assertEquals(self.persist.get("system-version"), 5)
 
75
 
 
76
    def test_decorated_upgraders_run(self):
 
77
        """
 
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.
 
81
        """
 
82
        upgrade_manager = UpgradeManager()
 
83
        @upgrade_manager.upgrader(1)
 
84
        def upgrade(persist):
 
85
            self.persist.set("upgrade-called", True)
 
86
 
 
87
        upgrade_manager.apply(self.persist)
 
88
        self.assertTrue(self.persist.get("upgrade-called"))
 
89
 
 
90
 
 
91
class SQLitePatchTest(LandscapeTest):
 
92
 
 
93
    def setUp(self):
 
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"
 
100
 
 
101
    def test_no_patches(self):
 
102
        """
 
103
        Applying no patches should make no change to the database,
 
104
        apart from maybe specifying a default version.
 
105
        """
 
106
        self.manager.initialize(self.cursor)
 
107
        self.manager.apply(self.cursor)
 
108
        self.assertEquals(self.manager.get_database_versions(self.cursor),
 
109
                          set())
 
110
 
 
111
    def test_one_patch(self):
 
112
        """Test that patches are called and passed a sqlite db object."""
 
113
        l = []
 
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,))
 
120
 
 
121
    def test_two_patches(self):
 
122
        """Test that patches are run in order."""
 
123
        l = []
 
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))
 
127
 
 
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,))
 
132
 
 
133
    def test_only_apply_unapplied_versions(self):
 
134
        """Upgraders should only be run if they haven't been run before."""
 
135
        patch1 = []
 
136
        patch2 = []
 
137
        patch3 = []
 
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]))
 
146
 
 
147
    def test_initialize_with_upgraders(self):
 
148
        """
 
149
        After registering some upgraders, initialize should set the
 
150
        version of the newly created database to the highest version
 
151
        available.
 
152
        """
 
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),
 
158
                          set([1, 3, 5]))
 
159
 
 
160
    def test_decorated_upgraders_run(self):
 
161
        """
 
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.
 
165
        """
 
166
        upgrade_manager = SQLiteUpgradeManager()
 
167
        upgrade_manager.initialize(self.cursor)
 
168
        l = []
 
169
        @upgrade_manager.upgrader(1)
 
170
        def upgrade(db):
 
171
            l.append(db)
 
172
 
 
173
        upgrade_manager.apply(self.cursor)
 
174
        self.assertEquals(l, [self.cursor])