14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Tests for branch implementations - tests a branch format."""
22
from bzrlib import branch, bzrdir, errors, gpg, transactions, repository
23
from bzrlib.branch import Branch, needs_read_lock, needs_write_lock
24
from bzrlib.delta import TreeDelta
25
from bzrlib.errors import (FileExists,
28
UninitializableFormat,
31
from bzrlib.osutils import getcwd
32
from bzrlib.tests import TestCase, TestCaseWithTransport, TestSkipped
33
from bzrlib.tests.bzrdir_implementations.test_bzrdir import TestCaseWithBzrDir
34
from bzrlib.trace import mutter
35
from bzrlib.transport import get_transport
36
from bzrlib.transport.http import HttpServer
37
from bzrlib.transport.memory import MemoryServer
38
from bzrlib.upgrade import upgrade
39
from bzrlib.workingtree import WorkingTree
42
# TODO: Make a branch using basis branch, and check that it
43
# doesn't request any files that could have been avoided, by
44
# hooking into the Transport.
47
class TestCaseWithBranch(TestCaseWithBzrDir):
50
super(TestCaseWithBranch, self).setUp()
54
if self.branch is None:
55
self.branch = self.make_branch('')
58
def make_branch(self, relpath, format=None):
59
repo = self.make_repository(relpath, format=format)
60
# fixme RBC 20060210 this isnt necessarily a fixable thing,
61
# Skipped is the wrong exception to raise.
63
return self.branch_format.initialize(repo.bzrdir)
64
except errors.UninitializableFormat:
65
raise TestSkipped('Uninitializable branch format')
67
def make_repository(self, relpath, shared=False, format=None):
68
made_control = self.make_bzrdir(relpath, format=format)
69
return made_control.create_repository(shared=shared)
72
class TestBranch(TestCaseWithBranch):
74
def test_append_revisions(self):
75
"""Test appending more than one revision"""
76
br = self.get_branch()
17
from bzrlib.selftest import InTempDir
21
class TestAppendRevisions(InTempDir):
22
"""Test appending more than one revision"""
24
from bzrlib.branch import Branch
25
br = Branch(".", init=True)
77
26
br.append_revision("rev1")
78
27
self.assertEquals(br.revision_history(), ["rev1",])
79
28
br.append_revision("rev2", "rev3")
80
29
self.assertEquals(br.revision_history(), ["rev1", "rev2", "rev3"])
82
def test_fetch_revisions(self):
83
"""Test fetch-revision operation."""
84
get_transport(self.get_url()).mkdir('b1')
85
get_transport(self.get_url()).mkdir('b2')
86
wt = self.make_branch_and_tree('b1')
88
b2 = self.make_branch('b2')
89
file('b1/foo', 'w').write('hello')
90
wt.add(['foo'], ['foo-id'])
91
wt.commit('lala!', rev_id='revision-1', allow_pointless=False)
94
self.assertEqual((1, []), b2.fetch(b1))
96
rev = b2.repository.get_revision('revision-1')
97
tree = b2.repository.revision_tree('revision-1')
98
self.assertEqual(tree.get_file_text('foo-id'), 'hello')
100
def test_get_revision_delta(self):
101
tree_a = self.make_branch_and_tree('a')
102
self.build_tree(['a/foo'])
103
tree_a.add('foo', 'file1')
104
tree_a.commit('rev1', rev_id='rev1')
105
self.build_tree(['a/vla'])
106
tree_a.add('vla', 'file2')
107
tree_a.commit('rev2', rev_id='rev2')
109
delta = tree_a.branch.get_revision_delta(1)
110
self.assertIsInstance(delta, TreeDelta)
111
self.assertEqual([('foo', 'file1', 'file')], delta.added)
112
delta = tree_a.branch.get_revision_delta(2)
113
self.assertIsInstance(delta, TreeDelta)
114
self.assertEqual([('vla', 'file2', 'file')], delta.added)
116
def get_unbalanced_tree_pair(self):
117
"""Return two branches, a and b, with one file in a."""
118
get_transport(self.get_url()).mkdir('a')
119
tree_a = self.make_branch_and_tree('a')
120
file('a/b', 'wb').write('b')
122
tree_a.commit("silly commit", rev_id='A')
124
get_transport(self.get_url()).mkdir('b')
125
tree_b = self.make_branch_and_tree('b')
126
return tree_a, tree_b
128
def get_balanced_branch_pair(self):
129
"""Returns br_a, br_b as with one commit in a, and b has a's stores."""
130
tree_a, tree_b = self.get_unbalanced_tree_pair()
131
tree_b.branch.repository.fetch(tree_a.branch.repository)
132
return tree_a, tree_b
134
def test_clone_branch(self):
135
"""Copy the stores from one branch to another"""
136
tree_a, tree_b = self.get_balanced_branch_pair()
137
tree_b.commit("silly commit")
139
# this fails to test that the history from a was not used.
140
dir_c = tree_a.bzrdir.clone('c', basis=tree_b.bzrdir)
141
self.assertEqual(tree_a.branch.revision_history(),
142
dir_c.open_branch().revision_history())
144
def test_clone_partial(self):
145
"""Copy only part of the history of a branch."""
146
# TODO: RBC 20060208 test with a revision not on revision-history.
147
# what should that behaviour be ? Emailed the list.
148
wt_a = self.make_branch_and_tree('a')
149
self.build_tree(['a/one'])
151
wt_a.commit('commit one', rev_id='1')
152
self.build_tree(['a/two'])
154
wt_a.commit('commit two', rev_id='2')
155
repo_b = self.make_repository('b')
156
wt_a.bzrdir.open_repository().copy_content_into(repo_b)
157
br_b = wt_a.bzrdir.open_branch().clone(repo_b.bzrdir, revision_id='1')
158
self.assertEqual(br_b.last_revision(), '1')
160
def test_sprout_partial(self):
161
# test sprouting with a prefix of the revision-history.
162
# also needs not-on-revision-history behaviour defined.
163
wt_a = self.make_branch_and_tree('a')
164
self.build_tree(['a/one'])
166
wt_a.commit('commit one', rev_id='1')
167
self.build_tree(['a/two'])
169
wt_a.commit('commit two', rev_id='2')
170
repo_b = self.make_repository('b')
171
wt_a.bzrdir.open_repository().copy_content_into(repo_b)
172
br_b = wt_a.bzrdir.open_branch().sprout(repo_b.bzrdir, revision_id='1')
173
self.assertEqual(br_b.last_revision(), '1')
175
def test_clone_branch_nickname(self):
176
# test the nick name is preserved always
177
raise TestSkipped('XXX branch cloning is not yet tested..')
179
def test_clone_branch_parent(self):
180
# test the parent is preserved always
181
raise TestSkipped('XXX branch cloning is not yet tested..')
183
def test_sprout_branch_nickname(self):
184
# test the nick name is reset always
185
raise TestSkipped('XXX branch sprouting is not yet tested..')
187
def test_sprout_branch_parent(self):
188
source = self.make_branch('source')
189
target = source.bzrdir.sprout(self.get_url('target')).open_branch()
190
self.assertEqual(source.bzrdir.root_transport.base, target.get_parent())
192
def test_record_initial_ghost_merge(self):
193
"""A pending merge with no revision present is still a merge."""
194
wt = self.make_branch_and_tree('.')
196
wt.add_pending_merge('non:existent@rev--ision--0--2')
197
wt.commit('pretend to merge nonexistent-revision', rev_id='first')
198
rev = branch.repository.get_revision(branch.last_revision())
199
self.assertEqual(len(rev.parent_ids), 1)
200
# parent_sha1s is not populated now, WTF. rbc 20051003
201
self.assertEqual(len(rev.parent_sha1s), 0)
202
self.assertEqual(rev.parent_ids[0], 'non:existent@rev--ision--0--2')
204
def test_bad_revision(self):
205
self.assertRaises(errors.InvalidRevisionId,
206
self.get_branch().repository.get_revision,
210
# compare the gpg-to-sign info for a commit with a ghost and
211
# an identical tree without a ghost
212
# fetch missing should rewrite the TOC of weaves to list newly available parents.
214
def test_pending_merges(self):
215
"""Tracking pending-merged revisions."""
216
wt = self.make_branch_and_tree('.')
218
self.assertEquals(wt.pending_merges(), [])
219
wt.add_pending_merge('foo@azkhazan-123123-abcabc')
220
self.assertEquals(wt.pending_merges(), ['foo@azkhazan-123123-abcabc'])
221
wt.add_pending_merge('foo@azkhazan-123123-abcabc')
222
self.assertEquals(wt.pending_merges(), ['foo@azkhazan-123123-abcabc'])
223
wt.add_pending_merge('wibble@fofof--20050401--1928390812')
224
self.assertEquals(wt.pending_merges(),
225
['foo@azkhazan-123123-abcabc',
226
'wibble@fofof--20050401--1928390812'])
227
wt.commit("commit from base with two merges")
228
rev = b.repository.get_revision(b.revision_history()[0])
229
self.assertEquals(len(rev.parent_ids), 2)
230
self.assertEquals(rev.parent_ids[0],
231
'foo@azkhazan-123123-abcabc')
232
self.assertEquals(rev.parent_ids[1],
233
'wibble@fofof--20050401--1928390812')
234
# list should be cleared when we do a commit
235
self.assertEquals(wt.pending_merges(), [])
237
def test_sign_existing_revision(self):
238
wt = self.make_branch_and_tree('.')
240
wt.commit("base", allow_pointless=True, rev_id='A')
241
from bzrlib.testament import Testament
242
strategy = gpg.LoopbackGPGStrategy(None)
243
branch.repository.sign_revision('A', strategy)
244
self.assertEqual(Testament.from_revision(branch.repository,
245
'A').as_short_text(),
246
branch.repository.get_signature_text('A'))
248
def test_store_signature(self):
249
wt = self.make_branch_and_tree('.')
251
branch.repository.store_revision_signature(
252
gpg.LoopbackGPGStrategy(None), 'FOO', 'A')
253
self.assertRaises(errors.NoSuchRevision,
254
branch.repository.has_signature_for_revision_id,
256
wt.commit("base", allow_pointless=True, rev_id='A')
257
self.assertEqual('FOO',
258
branch.repository.get_signature_text('A'))
260
def test_branch_keeps_signatures(self):
261
wt = self.make_branch_and_tree('source')
262
wt.commit('A', allow_pointless=True, rev_id='A')
263
wt.branch.repository.sign_revision('A',
264
gpg.LoopbackGPGStrategy(None))
265
#FIXME: clone should work to urls,
266
# wt.clone should work to disks.
267
self.build_tree(['target/'])
268
d2 = wt.bzrdir.clone('target')
269
self.assertEqual(wt.branch.repository.get_signature_text('A'),
270
d2.open_repository().get_signature_text('A'))
272
def test_nicks(self):
273
"""Branch nicknames"""
274
t = get_transport(self.get_url())
276
branch = self.make_branch('bzr.dev')
277
self.assertEqual(branch.nick, 'bzr.dev')
278
t.move('bzr.dev', 'bzr.ab')
279
branch = Branch.open(self.get_url('bzr.ab'))
280
self.assertEqual(branch.nick, 'bzr.ab')
281
branch.nick = "Aaron's branch"
282
branch.nick = "Aaron's branch"
286
branch.control_files.controlfilename("branch.conf")
290
self.assertEqual(branch.nick, "Aaron's branch")
291
t.move('bzr.ab', 'integration')
292
branch = Branch.open(self.get_url('integration'))
293
self.assertEqual(branch.nick, "Aaron's branch")
294
branch.nick = u"\u1234"
295
self.assertEqual(branch.nick, u"\u1234")
297
def test_commit_nicks(self):
298
"""Nicknames are committed to the revision"""
299
get_transport(self.get_url()).mkdir('bzr.dev')
300
wt = self.make_branch_and_tree('bzr.dev')
302
branch.nick = "My happy branch"
303
wt.commit('My commit respect da nick.')
304
committed = branch.repository.get_revision(branch.last_revision())
305
self.assertEqual(committed.properties["branch-nick"],
308
def test_create_open_branch_uses_repository(self):
310
repo = self.make_repository('.', shared=True)
311
except errors.IncompatibleFormat:
313
repo.bzrdir.root_transport.mkdir('child')
314
child_dir = self.bzrdir_format.initialize('child')
316
child_branch = self.branch_format.initialize(child_dir)
317
except errors.UninitializableFormat:
318
# branch references are not default init'able.
320
self.assertEqual(repo.bzrdir.root_transport.base,
321
child_branch.repository.bzrdir.root_transport.base)
322
child_branch = branch.Branch.open(self.get_url('child'))
323
self.assertEqual(repo.bzrdir.root_transport.base,
324
child_branch.repository.bzrdir.root_transport.base)
326
def test_format_description(self):
327
tree = self.make_branch_and_tree('tree')
328
text = tree.branch._format.get_format_description()
329
self.failUnless(len(text))
331
def test_check_branch_report_results(self):
332
"""Checking a branch produces results which can be printed"""
333
branch = self.make_branch('.')
334
result = branch.check()
335
# reports results through logging
336
result.report_results(verbose=True)
337
result.report_results(verbose=False)
339
def test_get_commit_builder(self):
340
self.assertIsInstance(self.make_branch(".").get_commit_builder([]),
341
repository.CommitBuilder)
343
def test_generate_revision_history(self):
344
"""Create a fake revision history easily."""
345
tree = self.make_branch_and_tree('.')
346
rev1 = tree.commit('foo')
347
orig_history = tree.branch.revision_history()
348
rev2 = tree.commit('bar', allow_pointless=True)
349
tree.branch.generate_revision_history(rev1)
350
self.assertEqual(orig_history, tree.branch.revision_history())
353
class ChrootedTests(TestCaseWithBranch):
354
"""A support class that provides readonly urls outside the local namespace.
356
This is done by checking if self.transport_server is a MemoryServer. if it
357
is then we are chrooted already, if it is not then an HttpServer is used
362
super(ChrootedTests, self).setUp()
363
if not self.transport_server == MemoryServer:
364
self.transport_readonly_server = HttpServer
366
def test_open_containing(self):
367
self.assertRaises(NotBranchError, Branch.open_containing,
368
self.get_readonly_url(''))
369
self.assertRaises(NotBranchError, Branch.open_containing,
370
self.get_readonly_url('g/p/q'))
371
branch = self.make_branch('.')
372
branch, relpath = Branch.open_containing(self.get_readonly_url(''))
373
self.assertEqual('', relpath)
374
branch, relpath = Branch.open_containing(self.get_readonly_url('g/p/q'))
375
self.assertEqual('g/p/q', relpath)
378
class InstrumentedTransaction(object):
381
self.calls.append('finish')
387
class TestDecorator(object):
393
self._calls.append('lr')
395
def lock_write(self):
396
self._calls.append('lw')
399
self._calls.append('ul')
402
def do_with_read(self):
406
def except_with_read(self):
410
def do_with_write(self):
414
def except_with_write(self):
418
class TestDecorators(TestCase):
420
def test_needs_read_lock(self):
421
branch = TestDecorator()
422
self.assertEqual(1, branch.do_with_read())
423
self.assertEqual(['lr', 'ul'], branch._calls)
425
def test_excepts_in_read_lock(self):
426
branch = TestDecorator()
427
self.assertRaises(RuntimeError, branch.except_with_read)
428
self.assertEqual(['lr', 'ul'], branch._calls)
430
def test_needs_write_lock(self):
431
branch = TestDecorator()
432
self.assertEqual(2, branch.do_with_write())
433
self.assertEqual(['lw', 'ul'], branch._calls)
435
def test_excepts_in_write_lock(self):
436
branch = TestDecorator()
437
self.assertRaises(RuntimeError, branch.except_with_write)
438
self.assertEqual(['lw', 'ul'], branch._calls)
441
class TestBranchTransaction(TestCaseWithBranch):
444
super(TestBranchTransaction, self).setUp()
447
def test_default_get_transaction(self):
448
"""branch.get_transaction on a new branch should give a PassThrough."""
449
self.failUnless(isinstance(self.get_branch().get_transaction(),
450
transactions.PassThroughTransaction))
452
def test__set_new_transaction(self):
453
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
455
def test__set_over_existing_transaction_raises(self):
456
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
457
self.assertRaises(errors.LockError,
458
self.get_branch()._set_transaction,
459
transactions.ReadOnlyTransaction())
461
def test_finish_no_transaction_raises(self):
462
self.assertRaises(errors.LockError, self.get_branch()._finish_transaction)
464
def test_finish_readonly_transaction_works(self):
465
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
466
self.get_branch()._finish_transaction()
467
self.assertEqual(None, self.get_branch().control_files._transaction)
469
def test_unlock_calls_finish(self):
470
self.get_branch().lock_read()
471
transaction = InstrumentedTransaction()
472
self.get_branch().control_files._transaction = transaction
473
self.get_branch().unlock()
474
self.assertEqual(['finish'], transaction.calls)
476
def test_lock_read_acquires_ro_transaction(self):
477
self.get_branch().lock_read()
478
self.failUnless(isinstance(self.get_branch().get_transaction(),
479
transactions.ReadOnlyTransaction))
480
self.get_branch().unlock()
482
def test_lock_write_acquires_write_transaction(self):
483
self.get_branch().lock_write()
484
# cannot use get_transaction as its magic
485
self.failUnless(isinstance(self.get_branch().control_files._transaction,
486
transactions.WriteTransaction))
487
self.get_branch().unlock()
490
class TestBranchPushLocations(TestCaseWithBranch):
492
def test_get_push_location_unset(self):
493
self.assertEqual(None, self.get_branch().get_push_location())
495
def test_get_push_location_exact(self):
496
from bzrlib.config import (locations_config_filename,
497
ensure_config_dir_exists)
498
ensure_config_dir_exists()
499
fn = locations_config_filename()
500
print >> open(fn, 'wt'), ("[%s]\n"
501
"push_location=foo" %
502
self.get_branch().base[:-1])
503
self.assertEqual("foo", self.get_branch().get_push_location())
505
def test_set_push_location(self):
506
from bzrlib.config import (locations_config_filename,
507
ensure_config_dir_exists)
508
ensure_config_dir_exists()
509
fn = locations_config_filename()
510
self.get_branch().set_push_location('foo')
511
self.assertFileEqual("[%s]\n"
512
"push_location = foo" % self.get_branch().base[:-1],
515
# TODO RBC 20051029 test getting a push location from a branch in a
516
# recursive section - that is, it appends the branch name.
519
class TestFormat(TestCaseWithBranch):
520
"""Tests for the format itself."""
522
def test_format_initialize_find_open(self):
523
# loopback test to check the current format initializes to itself.
524
if not self.branch_format.is_supported():
525
# unsupported formats are not loopback testable
526
# because the default open will not open them and
527
# they may not be initializable.
529
# supported formats must be able to init and open
530
t = get_transport(self.get_url())
531
readonly_t = get_transport(self.get_readonly_url())
532
made_branch = self.make_branch('.')
533
self.failUnless(isinstance(made_branch, branch.Branch))
535
# find it via bzrdir opening:
536
opened_control = bzrdir.BzrDir.open(readonly_t.base)
537
direct_opened_branch = opened_control.open_branch()
538
self.assertEqual(direct_opened_branch.__class__, made_branch.__class__)
539
self.assertEqual(opened_control, direct_opened_branch.bzrdir)
540
self.failUnless(isinstance(direct_opened_branch._format,
541
self.branch_format.__class__))
543
# find it via Branch.open
544
opened_branch = branch.Branch.open(readonly_t.base)
545
self.failUnless(isinstance(opened_branch, made_branch.__class__))
546
self.assertEqual(made_branch._format.__class__,
547
opened_branch._format.__class__)
548
# if it has a unique id string, can we probe for it ?
550
self.branch_format.get_format_string()
551
except NotImplementedError:
553
self.assertEqual(self.branch_format,
554
branch.BranchFormat.find_format(opened_control))
added
removed
bzrlib/selftest/testbranch.py
