1
# (C) 2005 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
from bzrlib.branch import Branch
19
from bzrlib.clone import copy_branch
20
from bzrlib.commit import commit
21
import bzrlib.errors as errors
22
from bzrlib.errors import NoSuchRevision, UnlistableBranch, NotBranchError
23
from bzrlib.selftest import TestCaseInTempDir
24
from bzrlib.trace import mutter
25
import bzrlib.transactions as transactions
26
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
28
class TestBranch(TestCaseInTempDir):
30
def test_append_revisions(self):
31
"""Test appending more than one revision"""
32
br = Branch.initialize(".")
33
br.append_revision("rev1")
34
self.assertEquals(br.revision_history(), ["rev1",])
35
br.append_revision("rev2", "rev3")
36
self.assertEquals(br.revision_history(), ["rev1", "rev2", "rev3"])
38
def test_fetch_revisions(self):
39
"""Test fetch-revision operation."""
40
from bzrlib.fetch import Fetcher
43
b1 = Branch.initialize('b1')
44
b2 = Branch.initialize('b2')
45
file(os.sep.join(['b1', 'foo']), 'w').write('hello')
46
b1.add(['foo'], ['foo-id'])
47
b1.commit('lala!', rev_id='revision-1', allow_pointless=False)
50
f = Fetcher(from_branch=b1, to_branch=b2)
51
eq = self.assertEquals
53
eq(f.last_revision, 'revision-1')
55
rev = b2.get_revision('revision-1')
56
tree = b2.revision_tree('revision-1')
57
eq(tree.get_file_text('foo-id'), 'hello')
59
def test_push_stores(self):
60
"""Copy the stores from one branch to another"""
62
br_a = Branch.initialize("a")
63
file('a/b', 'wb').write('b')
65
commit(br_a, "silly commit")
68
br_b = Branch.initialize("b")
69
self.assertRaises(NoSuchRevision, br_b.get_revision,
70
br_a.revision_history()[0])
71
br_a.push_stores(br_b)
72
rev = br_b.get_revision(br_a.revision_history()[0])
73
tree = br_b.revision_tree(br_a.revision_history()[0])
75
if tree.inventory[file_id].kind == "file":
76
tree.get_file(file_id).read()
79
def test_copy_branch(self):
80
"""Copy the stores from one branch to another"""
81
br_a, br_b = self.test_push_stores()
82
commit(br_b, "silly commit")
84
br_c = copy_branch(br_a, 'c', basis_branch=br_b)
85
self.assertEqual(br_a.revision_history(), br_c.revision_history())
86
## # basis branches currently disabled for weave format
87
## self.assertFalse(br_b.last_revision() in br_c.revision_history())
88
## br_c.get_revision(br_b.last_revision())
90
def test_copy_partial(self):
91
"""Copy only part of the history of a branch."""
92
self.build_tree(['a/', 'a/one'])
93
br_a = Branch.initialize('a')
95
br_a.commit('commit one', rev_id='u@d-1')
96
self.build_tree(['a/two'])
98
br_a.commit('commit two', rev_id='u@d-2')
99
br_b = copy_branch(br_a, 'b', revision='u@d-1')
100
self.assertEqual(br_b.last_revision(), 'u@d-1')
101
self.assertTrue(os.path.exists('b/one'))
102
self.assertFalse(os.path.exists('b/two'))
105
def test_record_initial_ghost_merge(self):
106
"""A pending merge with no revision present is still a merge."""
107
branch = Branch.initialize('.')
108
branch.add_pending_merge('non:existent@rev--ision--0--2')
109
branch.commit('pretend to merge nonexistent-revision', rev_id='first')
110
rev = branch.get_revision(branch.last_revision())
111
self.assertEqual(len(rev.parent_ids), 1)
112
# parent_sha1s is not populated now, WTF. rbc 20051003
113
self.assertEqual(len(rev.parent_sha1s), 0)
114
self.assertEqual(rev.parent_ids[0], 'non:existent@rev--ision--0--2')
117
# compare the gpg-to-sign info for a commit with a ghost and
118
# an identical tree without a ghost
119
# fetch missing should rewrite the TOC of weaves to list newly available parents.
121
def test_pending_merges(self):
122
"""Tracking pending-merged revisions."""
123
b = Branch.initialize('.')
125
self.assertEquals(b.pending_merges(), [])
126
b.add_pending_merge('foo@azkhazan-123123-abcabc')
127
self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
128
b.add_pending_merge('foo@azkhazan-123123-abcabc')
129
self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
130
b.add_pending_merge('wibble@fofof--20050401--1928390812')
131
self.assertEquals(b.pending_merges(),
132
['foo@azkhazan-123123-abcabc',
133
'wibble@fofof--20050401--1928390812'])
134
b.commit("commit from base with two merges")
135
rev = b.get_revision(b.revision_history()[0])
136
self.assertEquals(len(rev.parent_ids), 2)
137
self.assertEquals(rev.parent_ids[0],
138
'foo@azkhazan-123123-abcabc')
139
self.assertEquals(rev.parent_ids[1],
140
'wibble@fofof--20050401--1928390812')
141
# list should be cleared when we do a commit
142
self.assertEquals(b.pending_merges(), [])
144
class TestRemote(TestCaseWithWebserver):
145
def test_open_containing(self):
146
self.assertRaises(NotBranchError, Branch.open_containing,
147
self.get_remote_url(''))
148
self.assertRaises(NotBranchError, Branch.open_containing,
149
self.get_remote_url('g/p/q'))
150
b = Branch.initialize('.')
151
Branch.open_containing(self.get_remote_url(''))
152
Branch.open_containing(self.get_remote_url('g/p/q'))
154
# TODO: rewrite this as a regular unittest, without relying on the displayed output
155
# >>> from bzrlib.commit import commit
156
# >>> bzrlib.trace.silent = True
157
# >>> br1 = ScratchBranch(files=['foo', 'bar'])
160
# >>> commit(br1, "lala!", rev_id="REVISION-ID-1", verbose=False)
161
# >>> br2 = ScratchBranch()
162
# >>> br2.update_revisions(br1)
164
# Added 1 inventories.
166
# >>> br2.revision_history()
168
# >>> br2.update_revisions(br1)
170
# >>> br1.text_store.total_size() == br2.text_store.total_size()
173
class InstrumentedTransaction(object):
176
self.calls.append('finish')
182
class TestBranchTransaction(TestCaseInTempDir):
185
super(TestBranchTransaction, self).setUp()
186
self.branch = Branch.initialize('.')
188
def test_default_get_transaction(self):
189
"""branch.get_transaction on a new branch should give a PassThrough."""
190
self.failUnless(isinstance(self.branch.get_transaction(),
191
transactions.PassThroughTransaction))
193
def test__set_new_transaction(self):
194
self.branch._set_transaction(transactions.ReadOnlyTransaction())
196
def test__set_over_existing_transaction_raises(self):
197
self.branch._set_transaction(transactions.ReadOnlyTransaction())
198
self.assertRaises(errors.LockError,
199
self.branch._set_transaction,
200
transactions.ReadOnlyTransaction())
202
def test_finish_no_transaction_raises(self):
203
self.assertRaises(errors.LockError, self.branch._finish_transaction)
205
def test_finish_readonly_transaction_works(self):
206
self.branch._set_transaction(transactions.ReadOnlyTransaction())
207
self.branch._finish_transaction()
208
self.assertEqual(None, self.branch._transaction)
210
def test_unlock_calls_finish(self):
211
self.branch.lock_read()
212
transaction = InstrumentedTransaction()
213
self.branch._transaction = transaction
215
self.assertEqual(['finish'], transaction.calls)
217
def test_lock_read_acquires_ro_transaction(self):
218
self.branch.lock_read()
219
self.failUnless(isinstance(self.branch.get_transaction(),
220
transactions.ReadOnlyTransaction))
223
def test_lock_write_acquires_passthrough_transaction(self):
224
self.branch.lock_write()
225
# cannot use get_transaction as its magic
226
self.failUnless(isinstance(self.branch._transaction,
227
transactions.PassThroughTransaction))