1
# Copyright (C) 2011, 2016 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 __future__ import absolute_import
25
from cStringIO import StringIO
26
from stat import S_ISDIR
29
from bzrlib.bzrdir import (
32
from bzrlib.errors import (
36
from bzrlib.repository import (
40
from bzrlib.serializer import (
41
format_registry as serializer_format_registry,
43
from bzrlib.tests import (
45
TestCaseWithTransport,
48
from bzrlib.plugins.weave_fmt import xml4
49
from bzrlib.plugins.weave_fmt.bzrdir import (
52
from bzrlib.plugins.weave_fmt.repository import (
61
class TestFormat6(TestCaseWithTransport):
63
def test_attribute__fetch_order(self):
64
"""Weaves need topological data insertion."""
65
control = BzrDirFormat6().initialize(self.get_url())
66
repo = RepositoryFormat6().initialize(control)
67
self.assertEqual('topological', repo._format._fetch_order)
69
def test_attribute__fetch_uses_deltas(self):
70
"""Weaves do not reuse deltas."""
71
control = BzrDirFormat6().initialize(self.get_url())
72
repo = RepositoryFormat6().initialize(control)
73
self.assertEqual(False, repo._format._fetch_uses_deltas)
75
def test_attribute__fetch_reconcile(self):
76
"""Weave repositories need a reconcile after fetch."""
77
control = BzrDirFormat6().initialize(self.get_url())
78
repo = RepositoryFormat6().initialize(control)
79
self.assertEqual(True, repo._format._fetch_reconcile)
81
def test_no_ancestry_weave(self):
82
control = BzrDirFormat6().initialize(self.get_url())
83
repo = RepositoryFormat6().initialize(control)
84
# We no longer need to create the ancestry.weave file
85
# since it is *never* used.
86
self.assertRaises(NoSuchFile,
87
control.transport.get,
90
def test_supports_external_lookups(self):
91
control = BzrDirFormat6().initialize(self.get_url())
92
repo = RepositoryFormat6().initialize(control)
93
self.assertFalse(repo._format.supports_external_lookups)
98
class TestFormat7(TestCaseWithTransport):
100
def test_attribute__fetch_order(self):
101
"""Weaves need topological data insertion."""
102
control = BzrDirMetaFormat1().initialize(self.get_url())
103
repo = RepositoryFormat7().initialize(control)
104
self.assertEqual('topological', repo._format._fetch_order)
106
def test_attribute__fetch_uses_deltas(self):
107
"""Weaves do not reuse deltas."""
108
control = BzrDirMetaFormat1().initialize(self.get_url())
109
repo = RepositoryFormat7().initialize(control)
110
self.assertEqual(False, repo._format._fetch_uses_deltas)
112
def test_attribute__fetch_reconcile(self):
113
"""Weave repositories need a reconcile after fetch."""
114
control = BzrDirMetaFormat1().initialize(self.get_url())
115
repo = RepositoryFormat7().initialize(control)
116
self.assertEqual(True, repo._format._fetch_reconcile)
118
def test_disk_layout(self):
119
control = BzrDirMetaFormat1().initialize(self.get_url())
120
repo = RepositoryFormat7().initialize(control)
121
# in case of side effects of locking.
125
# format 'Bazaar-NG Repository format 7'
127
# inventory.weave == empty_weave
128
# empty revision-store directory
129
# empty weaves directory
130
t = control.get_repository_transport(None)
131
self.assertEqualDiff('Bazaar-NG Repository format 7',
132
t.get('format').read())
133
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
134
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
135
self.assertEqualDiff('# bzr weave file v5\n'
138
t.get('inventory.weave').read())
139
# Creating a file with id Foo:Bar results in a non-escaped file name on
141
control.create_branch()
142
tree = control.create_workingtree()
143
tree.add(['foo'], ['Foo:Bar'], ['file'])
144
tree.put_file_bytes_non_atomic('Foo:Bar', 'content\n')
146
tree.commit('first post', rev_id='first')
148
if sys.platform != 'win32':
150
self.knownFailure('Foo:Bar cannot be used as a file-id on windows'
153
self.assertEqualDiff(
154
'# bzr weave file v5\n'
156
'1 7fe70820e08a1aac0ef224d9c66ab66831cc4ab1\n'
164
t.get('weaves/74/Foo%3ABar.weave').read())
166
def test_shared_disk_layout(self):
167
control = BzrDirMetaFormat1().initialize(self.get_url())
168
repo = RepositoryFormat7().initialize(control, shared=True)
170
# format 'Bazaar-NG Repository format 7'
171
# inventory.weave == empty_weave
172
# empty revision-store directory
173
# empty weaves directory
174
# a 'shared-storage' marker file.
175
# lock is not present when unlocked
176
t = control.get_repository_transport(None)
177
self.assertEqualDiff('Bazaar-NG Repository format 7',
178
t.get('format').read())
179
self.assertEqualDiff('', t.get('shared-storage').read())
180
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
181
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
182
self.assertEqualDiff('# bzr weave file v5\n'
185
t.get('inventory.weave').read())
186
self.assertFalse(t.has('branch-lock'))
188
def test_creates_lockdir(self):
189
"""Make sure it appears to be controlled by a LockDir existence"""
190
control = BzrDirMetaFormat1().initialize(self.get_url())
191
repo = RepositoryFormat7().initialize(control, shared=True)
192
t = control.get_repository_transport(None)
193
# TODO: Should check there is a 'lock' toplevel directory,
194
# regardless of contents
195
self.assertFalse(t.has('lock/held/info'))
198
self.assertTrue(t.has('lock/held/info'))
200
# unlock so we don't get a warning about failing to do so
203
def test_uses_lockdir(self):
204
"""repo format 7 actually locks on lockdir"""
205
base_url = self.get_url()
206
control = BzrDirMetaFormat1().initialize(base_url)
207
repo = RepositoryFormat7().initialize(control, shared=True)
208
t = control.get_repository_transport(None)
212
# make sure the same lock is created by opening it
213
repo = Repository.open(base_url)
215
self.assertTrue(t.has('lock/held/info'))
217
self.assertFalse(t.has('lock/held/info'))
219
def test_shared_no_tree_disk_layout(self):
220
control = BzrDirMetaFormat1().initialize(self.get_url())
221
repo = RepositoryFormat7().initialize(control, shared=True)
222
repo.set_make_working_trees(False)
224
# format 'Bazaar-NG Repository format 7'
226
# inventory.weave == empty_weave
227
# empty revision-store directory
228
# empty weaves directory
229
# a 'shared-storage' marker file.
230
t = control.get_repository_transport(None)
231
self.assertEqualDiff('Bazaar-NG Repository format 7',
232
t.get('format').read())
233
## self.assertEqualDiff('', t.get('lock').read())
234
self.assertEqualDiff('', t.get('shared-storage').read())
235
self.assertEqualDiff('', t.get('no-working-trees').read())
236
repo.set_make_working_trees(True)
237
self.assertFalse(t.has('no-working-trees'))
238
self.assertTrue(S_ISDIR(t.stat('revision-store').st_mode))
239
self.assertTrue(S_ISDIR(t.stat('weaves').st_mode))
240
self.assertEqualDiff('# bzr weave file v5\n'
243
t.get('inventory.weave').read())
245
def test_supports_external_lookups(self):
246
control = BzrDirMetaFormat1().initialize(self.get_url())
247
repo = RepositoryFormat7().initialize(control)
248
self.assertFalse(repo._format.supports_external_lookups)
251
class TestInterWeaveRepo(TestCaseWithTransport):
253
def test_is_compatible_and_registered(self):
254
# InterWeaveRepo is compatible when either side
255
# is a format 5/6/7 branch
256
from bzrlib.repofmt import knitrepo
257
formats = [RepositoryFormat5(),
260
incompatible_formats = [RepositoryFormat4(),
261
knitrepo.RepositoryFormatKnit1(),
263
repo_a = self.make_repository('a')
264
repo_b = self.make_repository('b')
265
is_compatible = InterWeaveRepo.is_compatible
266
for source in incompatible_formats:
267
# force incompatible left then right
268
repo_a._format = source
269
repo_b._format = formats[0]
270
self.assertFalse(is_compatible(repo_a, repo_b))
271
self.assertFalse(is_compatible(repo_b, repo_a))
272
for source in formats:
273
repo_a._format = source
274
for target in formats:
275
repo_b._format = target
276
self.assertTrue(is_compatible(repo_a, repo_b))
277
self.assertEqual(InterWeaveRepo,
278
InterRepository.get(repo_a, repo_b).__class__)
281
_working_inventory_v4 = """<inventory file_id="TREE_ROOT">
282
<entry file_id="bar-20050901064931-73b4b1138abc9cd2" kind="file" name="bar" parent_id="TREE_ROOT" />
283
<entry file_id="foo-20050801201819-4139aa4a272f4250" kind="directory" name="foo" parent_id="TREE_ROOT" />
284
<entry file_id="bar-20050824000535-6bc48cfad47ed134" kind="file" name="bar" parent_id="foo-20050801201819-4139aa4a272f4250" />
288
_revision_v4 = """<revision committer="Martin Pool <mbp@sourcefrog.net>"
289
inventory_id="mbp@sourcefrog.net-20050905080035-e0439293f8b6b9f9"
290
inventory_sha1="e79c31c1deb64c163cf660fdedd476dd579ffd41"
291
revision_id="mbp@sourcefrog.net-20050905080035-e0439293f8b6b9f9"
292
timestamp="1125907235.212"
294
<message>- start splitting code for xml (de)serialization away from objects
295
preparatory to supporting multiple formats by a single library
298
<revision_ref revision_id="mbp@sourcefrog.net-20050905063503-43948f59fa127d92" revision_sha1="7bdf4cc8c5bdac739f8cf9b10b78cf4b68f915ff" />
304
class TestSerializer(TestCase):
305
"""Test serializer"""
307
def test_registry(self):
308
self.assertIs(xml4.serializer_v4,
309
serializer_format_registry.get('4'))
311
def test_canned_inventory(self):
312
"""Test unpacked a canned inventory v4 file."""
313
inp = StringIO(_working_inventory_v4)
314
inv = xml4.serializer_v4.read_inventory(inp)
315
self.assertEqual(len(inv), 4)
316
self.assertTrue(inv.has_id('bar-20050901064931-73b4b1138abc9cd2'))
318
def test_unpack_revision(self):
319
"""Test unpacking a canned revision v4"""
320
inp = StringIO(_revision_v4)
321
rev = xml4.serializer_v4.read_revision(inp)
322
eq = self.assertEqual
324
"Martin Pool <mbp@sourcefrog.net>")
326
"mbp@sourcefrog.net-20050905080035-e0439293f8b6b9f9")
327
eq(len(rev.parent_ids), 1)
328
eq(rev.parent_ids[0],
329
"mbp@sourcefrog.net-20050905063503-43948f59fa127d92")
added
removed
bzrlib/plugins/weave_fmt/test_repository.py
