← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_repository_vf/test_repository.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-03-14 12:08:53 UTC
  • mfrom: (5718.3.1 more-per-vf)
  • Revision ID: pqm@pqm.ubuntu.com-20110314120853-ot5tnefih540reoa
(jelmer) Skip more tests for repositories that don't implement the full
 versionedfiles API. (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_repository_vf/test_repository.py
17
17
"""Tests for repository implementations - tests a repository format."""
18
18
 
19
19
from bzrlib import (
 
20
    errors,
 
21
    tests,
20
22
    versionedfile,
21
23
    )
22
24
 
89
91
        repo = tree.branch.repository
90
92
        self.assertIsInstance(repo.signatures,
91
93
            versionedfile.VersionedFiles)
 
94
 
 
95
    def test_exposed_versioned_files_are_marked_dirty(self):
 
96
        repo = self.make_repository('.')
 
97
        repo.lock_write()
 
98
        signatures = repo.signatures
 
99
        revisions = repo.revisions
 
100
        inventories = repo.inventories
 
101
        repo.unlock()
 
102
        self.assertRaises(errors.ObjectNotLocked,
 
103
            signatures.keys)
 
104
        self.assertRaises(errors.ObjectNotLocked,
 
105
            revisions.keys)
 
106
        self.assertRaises(errors.ObjectNotLocked,
 
107
            inventories.keys)
 
108
        self.assertRaises(errors.ObjectNotLocked,
 
109
            signatures.add_lines, ('foo',), [], [])
 
110
        self.assertRaises(errors.ObjectNotLocked,
 
111
            revisions.add_lines, ('foo',), [], [])
 
112
        self.assertRaises(errors.ObjectNotLocked,
 
113
            inventories.add_lines, ('foo',), [], [])
 
114
 
 
115
 
 
116
class TestCaseWithComplexRepository(TestCaseWithRepository):
 
117
 
 
118
    scenarios = all_repository_vf_format_scenarios()
 
119
 
 
120
    def setUp(self):
 
121
        super(TestCaseWithComplexRepository, self).setUp()
 
122
        tree_a = self.make_branch_and_tree('a')
 
123
        self.bzrdir = tree_a.branch.bzrdir
 
124
        # add a corrupt inventory 'orphan'
 
125
        # this may need some generalising for knits.
 
126
        tree_a.lock_write()
 
127
        try:
 
128
            tree_a.branch.repository.start_write_group()
 
129
            try:
 
130
                inv_file = tree_a.branch.repository.inventories
 
131
                inv_file.add_lines(('orphan',), [], [])
 
132
            except:
 
133
                tree_a.branch.repository.commit_write_group()
 
134
                raise
 
135
            else:
 
136
                tree_a.branch.repository.abort_write_group()
 
137
        finally:
 
138
            tree_a.unlock()
 
139
        # add a real revision 'rev1'
 
140
        tree_a.commit('rev1', rev_id='rev1', allow_pointless=True)
 
141
        # add a real revision 'rev2' based on rev1
 
142
        tree_a.commit('rev2', rev_id='rev2', allow_pointless=True)
 
143
        # add a reference to a ghost
 
144
        tree_a.add_parent_tree_id('ghost1')
 
145
        try:
 
146
            tree_a.commit('rev3', rev_id='rev3', allow_pointless=True)
 
147
        except errors.RevisionNotPresent:
 
148
            raise tests.TestNotApplicable(
 
149
                "Cannot test with ghosts for this format.")
 
150
        # add another reference to a ghost, and a second ghost.
 
151
        tree_a.add_parent_tree_id('ghost1')
 
152
        tree_a.add_parent_tree_id('ghost2')
 
153
        tree_a.commit('rev4', rev_id='rev4', allow_pointless=True)
 
154
 
 
155
    def test_revision_trees(self):
 
156
        revision_ids = ['rev1', 'rev2', 'rev3', 'rev4']
 
157
        repository = self.bzrdir.open_repository()
 
158
        repository.lock_read()
 
159
        self.addCleanup(repository.unlock)
 
160
        trees1 = list(repository.revision_trees(revision_ids))
 
161
        trees2 = [repository.revision_tree(t) for t in revision_ids]
 
162
        self.assertEqual(len(trees1), len(trees2))
 
163
        for tree1, tree2 in zip(trees1, trees2):
 
164
            self.assertFalse(tree2.changes_from(tree1).has_changed())
 
165
 
 
166
    def test_get_deltas_for_revisions(self):
 
167
        repository = self.bzrdir.open_repository()
 
168
        repository.lock_read()
 
169
        self.addCleanup(repository.unlock)
 
170
        revisions = [repository.get_revision(r) for r in
 
171
                     ['rev1', 'rev2', 'rev3', 'rev4']]
 
172
        deltas1 = list(repository.get_deltas_for_revisions(revisions))
 
173
        deltas2 = [repository.get_revision_delta(r.revision_id) for r in
 
174
                   revisions]
 
175
        self.assertEqual(deltas1, deltas2)
 
176
 
 
177
    def test_all_revision_ids(self):
 
178
        # all_revision_ids -> all revisions
 
179
        self.assertEqual(set(['rev1', 'rev2', 'rev3', 'rev4']),
 
180
            set(self.bzrdir.open_repository().all_revision_ids()))
 
181
 
 
182
    def test_get_ancestry_missing_revision(self):
 
183
        # get_ancestry(revision that is in some data but not fully installed
 
184
        # -> NoSuchRevision
 
185
        self.assertRaises(errors.NoSuchRevision,
 
186
                          self.bzrdir.open_repository().get_ancestry, 'orphan')
 
187
 
 
188
    def test_get_unordered_ancestry(self):
 
189
        repo = self.bzrdir.open_repository()
 
190
        self.assertEqual(set(repo.get_ancestry('rev3')),
 
191
                         set(repo.get_ancestry('rev3', topo_sorted=False)))
 
192
 
 
193
    def test_reserved_id(self):
 
194
        repo = self.make_repository('repository')
 
195
        repo.lock_write()
 
196
        repo.start_write_group()
 
197
        try:
 
198
            self.assertRaises(errors.ReservedId, repo.add_inventory,
 
199
                'reserved:', None, None)
 
200
            self.assertRaises(errors.ReservedId, repo.add_inventory_by_delta,
 
201
                "foo", [], 'reserved:', None)
 
202
            self.assertRaises(errors.ReservedId, repo.add_revision,
 
203
                'reserved:', None)
 
204
        finally:
 
205
            repo.abort_write_group()
 
206
            repo.unlock()