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
import bzrlib.branch as branch
20
23
from bzrlib.branch import Branch, needs_read_lock, needs_write_lock
21
24
from bzrlib.commit import commit
22
25
import bzrlib.errors as errors
23
from bzrlib.errors import NoSuchRevision, UnlistableBranch, NotBranchError
26
from bzrlib.errors import (FileExists,
29
UninitializableFormat,
25
33
from bzrlib.osutils import getcwd
26
from bzrlib.tests import TestCase, TestCaseInTempDir
27
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
34
from bzrlib.revision import NULL_REVISION
35
from bzrlib.tests import TestCase, TestCaseWithTransport, TestSkipped
28
36
from bzrlib.trace import mutter
29
37
import bzrlib.transactions as transactions
30
from bzrlib.revision import NULL_REVISION
38
from bzrlib.transport import get_transport
39
from bzrlib.transport.http import HttpServer
40
from bzrlib.transport.memory import MemoryServer
41
from bzrlib.upgrade import upgrade
42
from bzrlib.workingtree import WorkingTree
32
44
# TODO: Make a branch using basis branch, and check that it
33
45
# doesn't request any files that could have been avoided, by
34
46
# hooking into the Transport.
36
class TestBranch(TestCaseInTempDir):
49
class TestCaseWithBranch(TestCaseWithTransport):
52
super(TestCaseWithBranch, self).setUp()
56
if self.branch is None:
57
self.branch = self.make_branch(None)
60
def make_branch(self, relpath):
62
url = self.get_url(relpath)
63
segments = url.split('/')
64
if segments and segments[-1] not in ('', '.'):
65
parent = '/'.join(segments[:-1])
66
t = get_transport(parent)
71
return self.branch_format.initialize(url)
72
except UninitializableFormat:
73
raise TestSkipped("Format %s is not initializable.")
76
class TestBranch(TestCaseWithBranch):
38
78
def test_append_revisions(self):
39
79
"""Test appending more than one revision"""
40
br = Branch.initialize(u".")
80
br = self.get_branch()
41
81
br.append_revision("rev1")
42
82
self.assertEquals(br.revision_history(), ["rev1",])
43
83
br.append_revision("rev2", "rev3")
46
86
def test_fetch_revisions(self):
47
87
"""Test fetch-revision operation."""
48
88
from bzrlib.fetch import Fetcher
51
b1 = Branch.initialize('b1')
52
b2 = Branch.initialize('b2')
89
get_transport(self.get_url()).mkdir('b1')
90
get_transport(self.get_url()).mkdir('b2')
91
b1 = self.make_branch('b1')
92
b2 = self.make_branch('b2')
93
wt = WorkingTree.create(b1, 'b1')
53
94
file('b1/foo', 'w').write('hello')
54
b1.working_tree().add(['foo'], ['foo-id'])
55
b1.working_tree().commit('lala!', rev_id='revision-1', allow_pointless=False)
95
wt.add(['foo'], ['foo-id'])
96
wt.commit('lala!', rev_id='revision-1', allow_pointless=False)
57
98
mutter('start fetch')
58
99
f = Fetcher(from_branch=b1, to_branch=b2)
65
106
eq(tree.get_file_text('foo-id'), 'hello')
67
108
def test_revision_tree(self):
68
b1 = Branch.initialize(u'.')
69
b1.working_tree().commit('lala!', rev_id='revision-1', allow_pointless=True)
109
b1 = self.get_branch()
110
wt = WorkingTree.create(b1, '.')
111
wt.commit('lala!', rev_id='revision-1', allow_pointless=True)
70
112
tree = b1.repository.revision_tree('revision-1')
71
113
tree = b1.repository.revision_tree(None)
72
114
self.assertEqual(len(tree.list_files()), 0)
73
115
tree = b1.repository.revision_tree(NULL_REVISION)
74
116
self.assertEqual(len(tree.list_files()), 0)
76
def get_unbalanced_branch_pair(self):
118
def get_unbalanced_tree_pair(self):
77
119
"""Return two branches, a and b, with one file in a."""
79
br_a = Branch.initialize("a")
120
get_transport(self.get_url()).mkdir('a')
121
br_a = self.make_branch('a')
122
tree_a = WorkingTree.create(br_a, 'a')
80
123
file('a/b', 'wb').write('b')
81
br_a.working_tree().add('b')
82
commit(br_a, "silly commit", rev_id='A')
84
br_b = Branch.initialize("b")
125
tree_a.commit("silly commit", rev_id='A')
127
get_transport(self.get_url()).mkdir('b')
128
br_b = self.make_branch('b')
129
tree_b = WorkingTree.create(br_b, 'b')
130
return tree_a, tree_b
87
132
def get_balanced_branch_pair(self):
88
133
"""Returns br_a, br_b as with one commit in a, and b has a's stores."""
89
br_a, br_b = self.get_unbalanced_branch_pair()
90
br_a.push_stores(br_b)
134
tree_a, tree_b = self.get_unbalanced_tree_pair()
135
tree_a.branch.push_stores(tree_b.branch)
136
return tree_a, tree_b
93
138
def test_push_stores(self):
94
139
"""Copy the stores from one branch to another"""
95
br_a, br_b = self.get_unbalanced_branch_pair()
140
tree_a, tree_b = self.get_unbalanced_tree_pair()
96
143
# ensure the revision is missing.
97
144
self.assertRaises(NoSuchRevision, br_b.repository.get_revision,
98
145
br_a.revision_history()[0])
103
150
for file_id in tree:
104
151
if tree.inventory[file_id].kind == "file":
105
152
tree.get_file(file_id).read()
108
154
def test_clone_branch(self):
109
155
"""Copy the stores from one branch to another"""
110
br_a, br_b = self.get_balanced_branch_pair()
111
commit(br_b, "silly commit")
156
tree_a, tree_b = self.get_balanced_branch_pair()
157
tree_b.commit("silly commit")
113
br_c = br_a.clone('c', basis_branch=br_b)
114
self.assertEqual(br_a.revision_history(), br_c.revision_history())
159
br_c = tree_a.branch.clone('c', basis_branch=tree_b.branch)
160
self.assertEqual(tree_a.branch.revision_history(),
161
br_c.revision_history())
116
163
def test_clone_partial(self):
117
164
"""Copy only part of the history of a branch."""
118
self.build_tree(['a/', 'a/one'])
119
br_a = Branch.initialize('a')
120
br_a.working_tree().add(['one'])
121
br_a.working_tree().commit('commit one', rev_id='u@d-1')
165
get_transport(self.get_url()).mkdir('a')
166
br_a = self.make_branch('a')
167
wt = WorkingTree.create(br_a, "a")
168
self.build_tree(['a/one'])
170
wt.commit('commit one', rev_id='u@d-1')
122
171
self.build_tree(['a/two'])
123
br_a.working_tree().add(['two'])
124
br_a.working_tree().commit('commit two', rev_id='u@d-2')
173
wt.commit('commit two', rev_id='u@d-2')
125
174
br_b = br_a.clone('b', revision='u@d-1')
126
175
self.assertEqual(br_b.last_revision(), 'u@d-1')
127
176
self.assertTrue(os.path.exists('b/one'))
130
179
def test_record_initial_ghost_merge(self):
131
180
"""A pending merge with no revision present is still a merge."""
132
branch = Branch.initialize(u'.')
133
branch.working_tree().add_pending_merge('non:existent@rev--ision--0--2')
134
branch.working_tree().commit('pretend to merge nonexistent-revision', rev_id='first')
181
branch = self.get_branch()
182
wt = WorkingTree.create(branch, ".")
183
wt.add_pending_merge('non:existent@rev--ision--0--2')
184
wt.commit('pretend to merge nonexistent-revision', rev_id='first')
135
185
rev = branch.repository.get_revision(branch.last_revision())
136
186
self.assertEqual(len(rev.parent_ids), 1)
137
187
# parent_sha1s is not populated now, WTF. rbc 20051003
172
222
self.assertEquals(wt.pending_merges(), [])
174
224
def test_sign_existing_revision(self):
175
branch = Branch.initialize(u'.')
176
branch.working_tree().commit("base", allow_pointless=True, rev_id='A')
225
branch = self.get_branch()
226
wt = WorkingTree.create(branch, ".")
227
wt.commit("base", allow_pointless=True, rev_id='A')
177
228
from bzrlib.testament import Testament
178
229
strategy = bzrlib.gpg.LoopbackGPGStrategy(None)
179
230
branch.repository.sign_revision('A', strategy)
185
236
def test_store_signature(self):
186
branch = Branch.initialize('.')
237
branch = self.get_branch()
187
238
branch.repository.store_revision_signature(
188
239
bzrlib.gpg.LoopbackGPGStrategy(None), 'FOO', 'A')
189
240
self.assertEqual('FOO',
190
241
branch.repository.revision_store.get('A',
244
def test_branch_keeps_signatures(self):
245
wt = self.make_branch_and_tree('source')
246
wt.commit('A', allow_pointless=True, rev_id='A')
247
wt.branch.repository.sign_revision('A',
248
bzrlib.gpg.LoopbackGPGStrategy(None))
249
#FIXME: clone should work to urls,
250
# wt.clone should work to disks.
251
self.build_tree(['target/'])
252
b2 = wt.branch.clone('target')
253
self.assertEqual(wt.branch.repository.revision_store.get('A',
255
b2.repository.revision_store.get('A',
258
def test_upgrade_preserves_signatures(self):
259
# this is in the current test format
260
wt = self.make_branch_and_tree('source')
261
wt.commit('A', allow_pointless=True, rev_id='A')
262
wt.branch.repository.sign_revision('A',
263
bzrlib.gpg.LoopbackGPGStrategy(None))
264
old_signature = wt.branch.repository.revision_store.get('A',
267
wt = WorkingTree(wt.basedir)
268
new_signature = wt.branch.repository.revision_store.get('A',
270
self.assertEqual(old_signature, new_signature)
193
272
def test_nicks(self):
194
273
"""Branch nicknames"""
196
branch = Branch.initialize('bzr.dev')
274
t = get_transport(self.get_url())
276
branch = self.make_branch('bzr.dev')
197
277
self.assertEqual(branch.nick, 'bzr.dev')
198
os.rename('bzr.dev', 'bzr.ab')
199
branch = Branch.open('bzr.ab')
278
t.move('bzr.dev', 'bzr.ab')
279
branch = Branch.open(self.get_url('bzr.ab'))
200
280
self.assertEqual(branch.nick, 'bzr.ab')
201
281
branch.nick = "Aaron's branch"
202
282
branch.nick = "Aaron's branch"
203
self.failUnlessExists(branch.control_files.controlfilename("branch.conf"))
286
branch.control_files.controlfilename("branch.conf")
204
290
self.assertEqual(branch.nick, "Aaron's branch")
205
os.rename('bzr.ab', 'integration')
206
branch = Branch.open('integration')
291
t.move('bzr.ab', 'integration')
292
branch = Branch.open(self.get_url('integration'))
207
293
self.assertEqual(branch.nick, "Aaron's branch")
208
294
branch.nick = u"\u1234"
209
295
self.assertEqual(branch.nick, u"\u1234")
211
297
def test_commit_nicks(self):
212
298
"""Nicknames are committed to the revision"""
214
branch = Branch.initialize('bzr.dev')
299
get_transport(self.get_url()).mkdir('bzr.dev')
300
branch = self.make_branch('bzr.dev')
215
301
branch.nick = "My happy branch"
216
branch.working_tree().commit('My commit respect da nick.')
302
WorkingTree.create(branch, 'bzr.dev').commit('My commit respect da nick.')
217
303
committed = branch.repository.get_revision(branch.last_revision())
218
304
self.assertEqual(committed.properties["branch-nick"],
219
305
"My happy branch")
221
307
def test_no_ancestry_weave(self):
222
308
# We no longer need to create the ancestry.weave file
223
309
# since it is *never* used.
224
branch = Branch.initialize(u'.')
310
branch = Branch.create('.')
225
311
self.failIfExists('.bzr/ancestry.weave')
228
class TestRemote(TestCaseWithWebserver):
314
class ChrootedTests(TestCaseWithBranch):
315
"""A support class that provides readonly urls outside the local namespace.
317
This is done by checking if self.transport_server is a MemoryServer. if it
318
is then we are chrooted already, if it is not then an HttpServer is used
323
super(ChrootedTests, self).setUp()
324
if not self.transport_server == MemoryServer:
325
self.transport_readonly_server = HttpServer
230
327
def test_open_containing(self):
231
328
self.assertRaises(NotBranchError, Branch.open_containing,
232
self.get_remote_url(''))
329
self.get_readonly_url(''))
233
330
self.assertRaises(NotBranchError, Branch.open_containing,
234
self.get_remote_url('g/p/q'))
235
b = Branch.initialize(u'.')
236
branch, relpath = Branch.open_containing(self.get_remote_url(''))
331
self.get_readonly_url('g/p/q'))
333
branch = self.branch_format.initialize(self.get_url())
334
except UninitializableFormat:
335
raise TestSkipped("Format %s is not initializable.")
336
branch, relpath = Branch.open_containing(self.get_readonly_url(''))
237
337
self.assertEqual('', relpath)
238
branch, relpath = Branch.open_containing(self.get_remote_url('g/p/q'))
338
branch, relpath = Branch.open_containing(self.get_readonly_url('g/p/q'))
239
339
self.assertEqual('g/p/q', relpath)
241
341
# TODO: rewrite this as a regular unittest, without relying on the displayed output
320
420
self.assertEqual(['lw', 'ul'], branch._calls)
323
class TestBranchTransaction(TestCaseInTempDir):
423
class TestBranchTransaction(TestCaseWithBranch):
326
426
super(TestBranchTransaction, self).setUp()
327
self.branch = Branch.initialize('.')
329
429
def test_default_get_transaction(self):
330
430
"""branch.get_transaction on a new branch should give a PassThrough."""
331
self.failUnless(isinstance(self.branch.get_transaction(),
431
self.failUnless(isinstance(self.get_branch().get_transaction(),
332
432
transactions.PassThroughTransaction))
334
434
def test__set_new_transaction(self):
335
self.branch._set_transaction(transactions.ReadOnlyTransaction())
435
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
337
437
def test__set_over_existing_transaction_raises(self):
338
self.branch._set_transaction(transactions.ReadOnlyTransaction())
438
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
339
439
self.assertRaises(errors.LockError,
340
self.branch._set_transaction,
440
self.get_branch()._set_transaction,
341
441
transactions.ReadOnlyTransaction())
343
443
def test_finish_no_transaction_raises(self):
344
self.assertRaises(errors.LockError, self.branch._finish_transaction)
444
self.assertRaises(errors.LockError, self.get_branch()._finish_transaction)
346
446
def test_finish_readonly_transaction_works(self):
347
self.branch._set_transaction(transactions.ReadOnlyTransaction())
348
self.branch._finish_transaction()
349
self.assertEqual(None, self.branch.control_files._transaction)
447
self.get_branch()._set_transaction(transactions.ReadOnlyTransaction())
448
self.get_branch()._finish_transaction()
449
self.assertEqual(None, self.get_branch().control_files._transaction)
351
451
def test_unlock_calls_finish(self):
352
self.branch.lock_read()
452
self.get_branch().lock_read()
353
453
transaction = InstrumentedTransaction()
354
self.branch.control_files._transaction = transaction
454
self.get_branch().control_files._transaction = transaction
455
self.get_branch().unlock()
356
456
self.assertEqual(['finish'], transaction.calls)
358
458
def test_lock_read_acquires_ro_transaction(self):
359
self.branch.lock_read()
360
self.failUnless(isinstance(self.branch.get_transaction(),
459
self.get_branch().lock_read()
460
self.failUnless(isinstance(self.get_branch().get_transaction(),
361
461
transactions.ReadOnlyTransaction))
462
self.get_branch().unlock()
364
464
def test_lock_write_acquires_passthrough_transaction(self):
365
self.branch.lock_write()
465
self.get_branch().lock_write()
366
466
# cannot use get_transaction as its magic
367
self.failUnless(isinstance(self.branch.control_files._transaction,
467
self.failUnless(isinstance(self.get_branch().control_files._transaction,
368
468
transactions.PassThroughTransaction))
372
class TestBranchPushLocations(TestCaseInTempDir):
375
super(TestBranchPushLocations, self).setUp()
376
self.branch = Branch.initialize(u'.')
469
self.get_branch().unlock()
472
class TestBranchPushLocations(TestCaseWithBranch):
378
474
def test_get_push_location_unset(self):
379
self.assertEqual(None, self.branch.get_push_location())
475
self.assertEqual(None, self.get_branch().get_push_location())
381
477
def test_get_push_location_exact(self):
382
478
from bzrlib.config import (branches_config_filename,
385
481
fn = branches_config_filename()
386
482
print >> open(fn, 'wt'), ("[%s]\n"
387
483
"push_location=foo" %
389
self.assertEqual("foo", self.branch.get_push_location())
484
self.get_branch().base[:-1])
485
self.assertEqual("foo", self.get_branch().get_push_location())
391
487
def test_set_push_location(self):
392
488
from bzrlib.config import (branches_config_filename,
393
489
ensure_config_dir_exists)
394
490
ensure_config_dir_exists()
395
491
fn = branches_config_filename()
396
self.branch.set_push_location('foo')
492
self.get_branch().set_push_location('foo')
397
493
self.assertFileEqual("[%s]\n"
398
"push_location = foo" % getcwd(),
494
"push_location = foo" % self.get_branch().base[:-1],
401
497
# TODO RBC 20051029 test getting a push location from a branch in a
402
498
# recursive section - that is, it appends the branch name.
501
class TestFormat(TestCaseWithBranch):
502
"""Tests for the format itself."""
504
def test_format_initialize_find_open(self):
505
# loopback test to check the current format initializes to itself.
506
if not self.branch_format.is_supported():
507
# unsupported formats are not loopback testable
508
# because the default open will not open them and
509
# they may not be initializable.
511
# supported formats must be able to init and open
512
t = get_transport(self.get_url())
513
readonly_t = get_transport(self.get_readonly_url())
514
made_branch = self.branch_format.initialize(t.base)
515
self.failUnless(isinstance(made_branch, branch.Branch))
516
self.assertEqual(self.branch_format,
517
branch.BzrBranchFormat.find_format(readonly_t))
518
direct_opened_branch = self.branch_format.open(readonly_t)
519
opened_branch = branch.Branch.open(t.base)
520
self.assertEqual(made_branch._branch_format,
521
opened_branch._branch_format)
522
self.assertEqual(direct_opened_branch._branch_format,
523
opened_branch._branch_format)
524
self.failUnless(isinstance(opened_branch, branch.Branch))
526
def test_open_not_branch(self):
527
self.assertRaises(NoSuchFile,
528
self.branch_format.open,
529
get_transport(self.get_readonly_url()))