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
18
from bzrlib.branch import Branch
19
from bzrlib.clone import copy_branch
20
from bzrlib.commit import commit
21
from bzrlib.errors import NoSuchRevision, UnlistableBranch
22
from bzrlib.selftest import TestCaseInTempDir
34
23
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):
26
class TestBranch(TestCaseInTempDir):
74
28
def test_append_revisions(self):
75
29
"""Test appending more than one revision"""
76
br = self.get_branch()
30
br = Branch.initialize(".")
77
31
br.append_revision("rev1")
78
32
self.assertEquals(br.revision_history(), ["rev1",])
79
33
br.append_revision("rev2", "rev3")
82
36
def test_fetch_revisions(self):
83
37
"""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)
38
from bzrlib.fetch import Fetcher
41
b1 = Branch.initialize('b1')
42
b2 = Branch.initialize('b2')
43
file(os.sep.join(['b1', 'foo']), 'w').write('hello')
44
b1.add(['foo'], ['foo-id'])
45
b1.commit('lala!', rev_id='revision-1', allow_pointless=False)
93
47
mutter('start fetch')
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')
48
f = Fetcher(from_branch=b1, to_branch=b2)
49
eq = self.assertEquals
51
eq(f.last_revision, 'revision-1')
53
rev = b2.get_revision('revision-1')
54
tree = b2.revision_tree('revision-1')
55
eq(tree.get_file_text('foo-id'), 'hello')
57
def test_push_stores(self):
58
"""Copy the stores from one branch to another"""
60
br_a = Branch.initialize("a")
120
61
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):
63
commit(br_a, "silly commit")
66
br_b = Branch.initialize("b")
67
self.assertRaises(NoSuchRevision, br_b.get_revision,
68
br_a.revision_history()[0])
69
br_a.push_stores(br_b)
70
rev = br_b.get_revision(br_a.revision_history()[0])
71
tree = br_b.revision_tree(br_a.revision_history()[0])
73
if tree.inventory[file_id].kind == "file":
74
tree.get_file(file_id).read()
77
def test_copy_branch(self):
135
78
"""Copy the stores from one branch to another"""
136
tree_a, tree_b = self.get_balanced_branch_pair()
137
tree_b.commit("silly commit")
79
br_a, br_b = self.test_push_stores()
80
commit(br_b, "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())
82
br_c = copy_branch(br_a, 'c', basis_branch=br_b)
83
self.assertEqual(br_a.revision_history(), br_c.revision_history())
84
## # basis branches currently disabled for weave format
85
## self.assertFalse(br_b.last_revision() in br_c.revision_history())
86
## br_c.get_revision(br_b.last_revision())
144
def test_clone_partial(self):
88
def test_copy_partial(self):
145
89
"""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_submit_branch(self):
193
"""Submit location can be queried and set"""
194
branch = self.make_branch('branch')
195
self.assertEqual(branch.get_submit_branch(), None)
196
branch.set_submit_branch('sftp://example.com')
197
self.assertEqual(branch.get_submit_branch(), 'sftp://example.com')
198
branch.set_submit_branch('sftp://example.net')
199
self.assertEqual(branch.get_submit_branch(), 'sftp://example.net')
90
self.build_tree(['a/', 'a/one'])
91
br_a = Branch.initialize('a')
93
br_a.commit('commit one', rev_id='u@d-1')
94
self.build_tree(['a/two'])
96
br_a.commit('commit two', rev_id='u@d-2')
97
br_b = copy_branch(br_a, 'b', revision='u@d-1')
98
self.assertEqual(br_b.last_revision(), 'u@d-1')
99
self.assertTrue(os.path.exists('b/one'))
100
self.assertFalse(os.path.exists('b/two'))
201
103
def test_record_initial_ghost_merge(self):
202
104
"""A pending merge with no revision present is still a merge."""
203
wt = self.make_branch_and_tree('.')
205
wt.add_pending_merge('non:existent@rev--ision--0--2')
206
wt.commit('pretend to merge nonexistent-revision', rev_id='first')
207
rev = branch.repository.get_revision(branch.last_revision())
105
branch = Branch.initialize('.')
106
branch.add_pending_merge('non:existent@rev--ision--0--2')
107
branch.commit('pretend to merge nonexistent-revision', rev_id='first')
108
rev = branch.get_revision(branch.last_revision())
208
109
self.assertEqual(len(rev.parent_ids), 1)
209
110
# parent_sha1s is not populated now, WTF. rbc 20051003
210
111
self.assertEqual(len(rev.parent_sha1s), 0)
211
112
self.assertEqual(rev.parent_ids[0], 'non:existent@rev--ision--0--2')
213
def test_bad_revision(self):
214
self.assertRaises(errors.InvalidRevisionId,
215
self.get_branch().repository.get_revision,
218
114
# TODO 20051003 RBC:
219
115
# compare the gpg-to-sign info for a commit with a ghost and
220
116
# an identical tree without a ghost
223
119
def test_pending_merges(self):
224
120
"""Tracking pending-merged revisions."""
225
wt = self.make_branch_and_tree('.')
227
self.assertEquals(wt.pending_merges(), [])
228
wt.add_pending_merge('foo@azkhazan-123123-abcabc')
229
self.assertEquals(wt.pending_merges(), ['foo@azkhazan-123123-abcabc'])
230
wt.add_pending_merge('foo@azkhazan-123123-abcabc')
231
self.assertEquals(wt.pending_merges(), ['foo@azkhazan-123123-abcabc'])
232
wt.add_pending_merge('wibble@fofof--20050401--1928390812')
233
self.assertEquals(wt.pending_merges(),
121
b = Branch.initialize('.')
123
self.assertEquals(b.pending_merges(), [])
124
b.add_pending_merge('foo@azkhazan-123123-abcabc')
125
self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
126
b.add_pending_merge('foo@azkhazan-123123-abcabc')
127
self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
128
b.add_pending_merge('wibble@fofof--20050401--1928390812')
129
self.assertEquals(b.pending_merges(),
234
130
['foo@azkhazan-123123-abcabc',
235
131
'wibble@fofof--20050401--1928390812'])
236
wt.commit("commit from base with two merges")
237
rev = b.repository.get_revision(b.revision_history()[0])
132
b.commit("commit from base with two merges")
133
rev = b.get_revision(b.revision_history()[0])
238
134
self.assertEquals(len(rev.parent_ids), 2)
239
135
self.assertEquals(rev.parent_ids[0],
240
136
'foo@azkhazan-123123-abcabc')
241
137
self.assertEquals(rev.parent_ids[1],
242
138
'wibble@fofof--20050401--1928390812')
243
139
# list should be cleared when we do a commit
244
self.assertEquals(wt.pending_merges(), [])
246
def test_sign_existing_revision(self):
247
wt = self.make_branch_and_tree('.')
249
wt.commit("base", allow_pointless=True, rev_id='A')
250
from bzrlib.testament import Testament
251
strategy = gpg.LoopbackGPGStrategy(None)
252
branch.repository.sign_revision('A', strategy)
253
self.assertEqual(Testament.from_revision(branch.repository,
254
'A').as_short_text(),
255
branch.repository.get_signature_text('A'))
257
def test_store_signature(self):
258
wt = self.make_branch_and_tree('.')
260
branch.repository.store_revision_signature(
261
gpg.LoopbackGPGStrategy(None), 'FOO', 'A')
262
self.assertRaises(errors.NoSuchRevision,
263
branch.repository.has_signature_for_revision_id,
265
wt.commit("base", allow_pointless=True, rev_id='A')
266
self.assertEqual('FOO',
267
branch.repository.get_signature_text('A'))
269
def test_branch_keeps_signatures(self):
270
wt = self.make_branch_and_tree('source')
271
wt.commit('A', allow_pointless=True, rev_id='A')
272
wt.branch.repository.sign_revision('A',
273
gpg.LoopbackGPGStrategy(None))
274
#FIXME: clone should work to urls,
275
# wt.clone should work to disks.
276
self.build_tree(['target/'])
277
d2 = wt.bzrdir.clone('target')
278
self.assertEqual(wt.branch.repository.get_signature_text('A'),
279
d2.open_repository().get_signature_text('A'))
281
def test_nicks(self):
282
"""Branch nicknames"""
283
t = get_transport(self.get_url())
285
branch = self.make_branch('bzr.dev')
286
self.assertEqual(branch.nick, 'bzr.dev')
287
t.move('bzr.dev', 'bzr.ab')
288
branch = Branch.open(self.get_url('bzr.ab'))
289
self.assertEqual(branch.nick, 'bzr.ab')
290
branch.nick = "Aaron's branch"
291
branch.nick = "Aaron's branch"
295
branch.control_files.controlfilename("branch.conf")
299
self.assertEqual(branch.nick, "Aaron's branch")
300
t.move('bzr.ab', 'integration')
301
branch = Branch.open(self.get_url('integration'))
302
self.assertEqual(branch.nick, "Aaron's branch")
303
branch.nick = u"\u1234"
304
self.assertEqual(branch.nick, u"\u1234")
306
def test_commit_nicks(self):
307
"""Nicknames are committed to the revision"""
308
get_transport(self.get_url()).mkdir('bzr.dev')
309
wt = self.make_branch_and_tree('bzr.dev')
311
branch.nick = "My happy branch"
312
wt.commit('My commit respect da nick.')
313
committed = branch.repository.get_revision(branch.last_revision())
314
self.assertEqual(committed.properties["branch-nick"],
317
def test_create_open_branch_uses_repository(self):
319
repo = self.make_repository('.', shared=True)
320
except errors.IncompatibleFormat:
322
repo.bzrdir.root_transport.mkdir('child')
323
child_dir = self.bzrdir_format.initialize('child')
325
child_branch = self.branch_format.initialize(child_dir)
326
except errors.UninitializableFormat:
327
# branch references are not default init'able.
329
self.assertEqual(repo.bzrdir.root_transport.base,
330
child_branch.repository.bzrdir.root_transport.base)
331
child_branch = branch.Branch.open(self.get_url('child'))
332
self.assertEqual(repo.bzrdir.root_transport.base,
333
child_branch.repository.bzrdir.root_transport.base)
335
def test_format_description(self):
336
tree = self.make_branch_and_tree('tree')
337
text = tree.branch._format.get_format_description()
338
self.failUnless(len(text))
340
def test_check_branch_report_results(self):
341
"""Checking a branch produces results which can be printed"""
342
branch = self.make_branch('.')
343
result = branch.check()
344
# reports results through logging
345
result.report_results(verbose=True)
346
result.report_results(verbose=False)
348
def test_get_commit_builder(self):
349
self.assertIsInstance(self.make_branch(".").get_commit_builder([]),
350
repository.CommitBuilder)
352
def test_generate_revision_history(self):
353
"""Create a fake revision history easily."""
354
tree = self.make_branch_and_tree('.')
355
rev1 = tree.commit('foo')
356
orig_history = tree.branch.revision_history()
357
rev2 = tree.commit('bar', allow_pointless=True)
358
tree.branch.generate_revision_history(rev1)
359
self.assertEqual(orig_history, tree.branch.revision_history())
362
class ChrootedTests(TestCaseWithBranch):
363
"""A support class that provides readonly urls outside the local namespace.
365
This is done by checking if self.transport_server is a MemoryServer. if it
366
is then we are chrooted already, if it is not then an HttpServer is used
371
super(ChrootedTests, self).setUp()
372
if not self.transport_server == MemoryServer:
373
self.transport_readonly_server = HttpServer
375
def test_open_containing(self):
376
self.assertRaises(NotBranchError, Branch.open_containing,
377
self.get_readonly_url(''))
378
self.assertRaises(NotBranchError, Branch.open_containing,
379
self.get_readonly_url('g/p/q'))
380
branch = self.make_branch('.')
381
branch, relpath = Branch.open_containing(self.get_readonly_url(''))
382
self.assertEqual('', relpath)
383
branch, relpath = Branch.open_containing(self.get_readonly_url('g/p/q'))
384
self.assertEqual('g/p/q', relpath)
387
class InstrumentedTransaction(object):
390
self.calls.append('finish')
396
class TestDecorator(object):
402
self._calls.append('lr')
404
def lock_write(self):
405
self._calls.append('lw')
408
self._calls.append('ul')
411
def do_with_read(self):
415
def except_with_read(self):
419
def do_with_write(self):
423
def except_with_write(self):
427
class TestDecorators(TestCase):
429
def test_needs_read_lock(self):
430
branch = TestDecorator()
431
self.assertEqual(1, branch.do_with_read())
432
self.assertEqual(['lr', 'ul'], branch._calls)
434
def test_excepts_in_read_lock(self):
435
branch = TestDecorator()
436
self.assertRaises(RuntimeError, branch.except_with_read)
437
self.assertEqual(['lr', 'ul'], branch._calls)
439
def test_needs_write_lock(self):
440
branch = TestDecorator()
441
self.assertEqual(2, branch.do_with_write())
442
self.assertEqual(['lw', 'ul'], branch._calls)
444
def test_excepts_in_write_lock(self):
445
branch = TestDecorator()
446
self.assertRaises(RuntimeError, branch.except_with_write)
447
self.assertEqual(['lw', 'ul'], branch._calls)
450
class TestBranchTransaction(TestCaseWithBranch):
453
super(TestBranchTransaction, self).setUp()
456
def test_default_get_transaction(self):
457
"""branch.get_transaction on a new branch should give a PassThrough."""
458
self.failUnless(isinstance(self.get_branch().get_transaction(),
459
transactions.PassThroughTransaction))
461
def test__set_new_transaction(self):
462
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
464
def test__set_over_existing_transaction_raises(self):
465
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
466
self.assertRaises(errors.LockError,
467
self.get_branch()._set_transaction,
468
transactions.ReadOnlyTransaction())
470
def test_finish_no_transaction_raises(self):
471
self.assertRaises(errors.LockError, self.get_branch()._finish_transaction)
473
def test_finish_readonly_transaction_works(self):
474
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
475
self.get_branch()._finish_transaction()
476
self.assertEqual(None, self.get_branch().control_files._transaction)
478
def test_unlock_calls_finish(self):
479
self.get_branch().lock_read()
480
transaction = InstrumentedTransaction()
481
self.get_branch().control_files._transaction = transaction
482
self.get_branch().unlock()
483
self.assertEqual(['finish'], transaction.calls)
485
def test_lock_read_acquires_ro_transaction(self):
486
self.get_branch().lock_read()
487
self.failUnless(isinstance(self.get_branch().get_transaction(),
488
transactions.ReadOnlyTransaction))
489
self.get_branch().unlock()
491
def test_lock_write_acquires_write_transaction(self):
492
self.get_branch().lock_write()
493
# cannot use get_transaction as its magic
494
self.failUnless(isinstance(self.get_branch().control_files._transaction,
495
transactions.WriteTransaction))
496
self.get_branch().unlock()
499
class TestBranchPushLocations(TestCaseWithBranch):
501
def test_get_push_location_unset(self):
502
self.assertEqual(None, self.get_branch().get_push_location())
504
def test_get_push_location_exact(self):
505
from bzrlib.config import (locations_config_filename,
506
ensure_config_dir_exists)
507
ensure_config_dir_exists()
508
fn = locations_config_filename()
509
print >> open(fn, 'wt'), ("[%s]\n"
510
"push_location=foo" %
511
self.get_branch().base[:-1])
512
self.assertEqual("foo", self.get_branch().get_push_location())
514
def test_set_push_location(self):
515
from bzrlib.config import (locations_config_filename,
516
ensure_config_dir_exists)
517
ensure_config_dir_exists()
518
fn = locations_config_filename()
519
self.get_branch().set_push_location('foo')
520
self.assertFileEqual("[%s]\n"
521
"push_location = foo" % self.get_branch().base[:-1],
524
# TODO RBC 20051029 test getting a push location from a branch in a
525
# recursive section - that is, it appends the branch name.
528
class TestFormat(TestCaseWithBranch):
529
"""Tests for the format itself."""
531
def test_format_initialize_find_open(self):
532
# loopback test to check the current format initializes to itself.
533
if not self.branch_format.is_supported():
534
# unsupported formats are not loopback testable
535
# because the default open will not open them and
536
# they may not be initializable.
538
# supported formats must be able to init and open
539
t = get_transport(self.get_url())
540
readonly_t = get_transport(self.get_readonly_url())
541
made_branch = self.make_branch('.')
542
self.failUnless(isinstance(made_branch, branch.Branch))
544
# find it via bzrdir opening:
545
opened_control = bzrdir.BzrDir.open(readonly_t.base)
546
direct_opened_branch = opened_control.open_branch()
547
self.assertEqual(direct_opened_branch.__class__, made_branch.__class__)
548
self.assertEqual(opened_control, direct_opened_branch.bzrdir)
549
self.failUnless(isinstance(direct_opened_branch._format,
550
self.branch_format.__class__))
552
# find it via Branch.open
553
opened_branch = branch.Branch.open(readonly_t.base)
554
self.failUnless(isinstance(opened_branch, made_branch.__class__))
555
self.assertEqual(made_branch._format.__class__,
556
opened_branch._format.__class__)
557
# if it has a unique id string, can we probe for it ?
559
self.branch_format.get_format_string()
560
except NotImplementedError:
562
self.assertEqual(self.branch_format,
563
branch.BranchFormat.find_format(opened_control))
140
self.assertEquals(b.pending_merges(), [])
143
# TODO: rewrite this as a regular unittest, without relying on the displayed output
144
# >>> from bzrlib.commit import commit
145
# >>> bzrlib.trace.silent = True
146
# >>> br1 = ScratchBranch(files=['foo', 'bar'])
149
# >>> commit(br1, "lala!", rev_id="REVISION-ID-1", verbose=False)
150
# >>> br2 = ScratchBranch()
151
# >>> br2.update_revisions(br1)
153
# Added 1 inventories.
155
# >>> br2.revision_history()
157
# >>> br2.update_revisions(br1)
159
# >>> br1.text_store.total_size() == br2.text_store.total_size()
added
removed
bzrlib/selftest/testbranch.py
