1
# Copyright (C) 2006-2010 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for weave repositories.
19
For interface tests see tests/per_repository/*.py.
23
from stat import S_ISDIR
26
from bzrlib.bzrdir import (
29
from bzrlib.errors import (
33
from bzrlib.repository import (
37
from bzrlib.tests import (
38
TestCaseWithTransport,
41
from bzrlib.plugins.weave_fmt.bzrdir import (
44
from bzrlib.plugins.weave_fmt.repository import (
53
class TestFormat6(TestCaseWithTransport):
55
def test_attribute__fetch_order(self):
56
"""Weaves need topological data insertion."""
57
control = BzrDirFormat6().initialize(self.get_url())
58
repo = RepositoryFormat6().initialize(control)
59
self.assertEqual('topological', repo._format._fetch_order)
61
def test_attribute__fetch_uses_deltas(self):
62
"""Weaves do not reuse deltas."""
63
control = BzrDirFormat6().initialize(self.get_url())
64
repo = RepositoryFormat6().initialize(control)
65
self.assertEqual(False, repo._format._fetch_uses_deltas)
67
def test_attribute__fetch_reconcile(self):
68
"""Weave repositories need a reconcile after fetch."""
69
control = BzrDirFormat6().initialize(self.get_url())
70
repo = RepositoryFormat6().initialize(control)
71
self.assertEqual(True, repo._format._fetch_reconcile)
73
def test_no_ancestry_weave(self):
74
control = BzrDirFormat6().initialize(self.get_url())
75
repo = RepositoryFormat6().initialize(control)
76
# We no longer need to create the ancestry.weave file
77
# since it is *never* used.
78
self.assertRaises(NoSuchFile,
79
control.transport.get,
82
def test_supports_external_lookups(self):
83
control = BzrDirFormat6().initialize(self.get_url())
84
repo = RepositoryFormat6().initialize(control)
85
self.assertFalse(repo._format.supports_external_lookups)
90
class TestFormat7(TestCaseWithTransport):
92
def test_attribute__fetch_order(self):
93
"""Weaves need topological data insertion."""
94
control = BzrDirMetaFormat1().initialize(self.get_url())
95
repo = RepositoryFormat7().initialize(control)
96
self.assertEqual('topological', repo._format._fetch_order)
98
def test_attribute__fetch_uses_deltas(self):
99
"""Weaves do not reuse deltas."""
100
control = BzrDirMetaFormat1().initialize(self.get_url())
101
repo = RepositoryFormat7().initialize(control)
102
self.assertEqual(False, repo._format._fetch_uses_deltas)
104
def test_attribute__fetch_reconcile(self):
105
"""Weave repositories need a reconcile after fetch."""
106
control = BzrDirMetaFormat1().initialize(self.get_url())
107
repo = RepositoryFormat7().initialize(control)
108
self.assertEqual(True, repo._format._fetch_reconcile)
110
def test_disk_layout(self):
111
control = BzrDirMetaFormat1().initialize(self.get_url())
112
repo = RepositoryFormat7().initialize(control)
113
# in case of side effects of locking.
117
# format 'Bazaar-NG Repository format 7'
119
# inventory.weave == empty_weave
120
# empty revision-store directory
121
# empty weaves directory
122
t = control.get_repository_transport(None)
123
self.assertEqualDiff('Bazaar-NG Repository format 7',
124
t.get('format').read())
125
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
126
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
127
self.assertEqualDiff('# bzr weave file v5\n'
130
t.get('inventory.weave').read())
131
# Creating a file with id Foo:Bar results in a non-escaped file name on
133
control.create_branch()
134
tree = control.create_workingtree()
135
tree.add(['foo'], ['Foo:Bar'], ['file'])
136
tree.put_file_bytes_non_atomic('Foo:Bar', 'content\n')
138
tree.commit('first post', rev_id='first')
140
if sys.platform != 'win32':
142
self.knownFailure('Foo:Bar cannot be used as a file-id on windows'
145
self.assertEqualDiff(
146
'# bzr weave file v5\n'
148
'1 7fe70820e08a1aac0ef224d9c66ab66831cc4ab1\n'
156
t.get('weaves/74/Foo%3ABar.weave').read())
158
def test_shared_disk_layout(self):
159
control = BzrDirMetaFormat1().initialize(self.get_url())
160
repo = RepositoryFormat7().initialize(control, shared=True)
162
# format 'Bazaar-NG Repository format 7'
163
# inventory.weave == empty_weave
164
# empty revision-store directory
165
# empty weaves directory
166
# a 'shared-storage' marker file.
167
# lock is not present when unlocked
168
t = control.get_repository_transport(None)
169
self.assertEqualDiff('Bazaar-NG Repository format 7',
170
t.get('format').read())
171
self.assertEqualDiff('', t.get('shared-storage').read())
172
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
173
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
174
self.assertEqualDiff('# bzr weave file v5\n'
177
t.get('inventory.weave').read())
178
self.assertFalse(t.has('branch-lock'))
180
def test_creates_lockdir(self):
181
"""Make sure it appears to be controlled by a LockDir existence"""
182
control = BzrDirMetaFormat1().initialize(self.get_url())
183
repo = RepositoryFormat7().initialize(control, shared=True)
184
t = control.get_repository_transport(None)
185
# TODO: Should check there is a 'lock' toplevel directory,
186
# regardless of contents
187
self.assertFalse(t.has('lock/held/info'))
190
self.assertTrue(t.has('lock/held/info'))
192
# unlock so we don't get a warning about failing to do so
195
def test_uses_lockdir(self):
196
"""repo format 7 actually locks on lockdir"""
197
base_url = self.get_url()
198
control = BzrDirMetaFormat1().initialize(base_url)
199
repo = RepositoryFormat7().initialize(control, shared=True)
200
t = control.get_repository_transport(None)
204
# make sure the same lock is created by opening it
205
repo = Repository.open(base_url)
207
self.assertTrue(t.has('lock/held/info'))
209
self.assertFalse(t.has('lock/held/info'))
211
def test_shared_no_tree_disk_layout(self):
212
control = BzrDirMetaFormat1().initialize(self.get_url())
213
repo = RepositoryFormat7().initialize(control, shared=True)
214
repo.set_make_working_trees(False)
216
# format 'Bazaar-NG Repository format 7'
218
# inventory.weave == empty_weave
219
# empty revision-store directory
220
# empty weaves directory
221
# a 'shared-storage' marker file.
222
t = control.get_repository_transport(None)
223
self.assertEqualDiff('Bazaar-NG Repository format 7',
224
t.get('format').read())
225
## self.assertEqualDiff('', t.get('lock').read())
226
self.assertEqualDiff('', t.get('shared-storage').read())
227
self.assertEqualDiff('', t.get('no-working-trees').read())
228
repo.set_make_working_trees(True)
229
self.assertFalse(t.has('no-working-trees'))
230
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
231
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
232
self.assertEqualDiff('# bzr weave file v5\n'
235
t.get('inventory.weave').read())
237
def test_supports_external_lookups(self):
238
control = BzrDirMetaFormat1().initialize(self.get_url())
239
repo = RepositoryFormat7().initialize(control)
240
self.assertFalse(repo._format.supports_external_lookups)
243
class TestInterWeaveRepo(TestCaseWithTransport):
245
def test_is_compatible_and_registered(self):
246
# InterWeaveRepo is compatible when either side
247
# is a format 5/6/7 branch
248
from bzrlib.repofmt import knitrepo
249
formats = [RepositoryFormat5(),
252
incompatible_formats = [RepositoryFormat4(),
253
knitrepo.RepositoryFormatKnit1(),
255
repo_a = self.make_repository('a')
256
repo_b = self.make_repository('b')
257
is_compatible = InterWeaveRepo.is_compatible
258
for source in incompatible_formats:
259
# force incompatible left then right
260
repo_a._format = source
261
repo_b._format = formats[0]
262
self.assertFalse(is_compatible(repo_a, repo_b))
263
self.assertFalse(is_compatible(repo_b, repo_a))
264
for source in formats:
265
repo_a._format = source
266
for target in formats:
267
repo_b._format = target
268
self.assertTrue(is_compatible(repo_a, repo_b))
269
self.assertEqual(InterWeaveRepo,
270
InterRepository.get(repo_a, repo_b).__class__)