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
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)
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
# TODO: Make a branch using basis branch, and check that it
29
# doesn't request any files that could have been avoided, by
30
# hooking into the Transport.
32
class TestBranch(TestCaseInTempDir):
34
def test_append_revisions(self):
35
"""Test appending more than one revision"""
36
br = Branch.initialize(".")
26
37
br.append_revision("rev1")
27
38
self.assertEquals(br.revision_history(), ["rev1",])
28
39
br.append_revision("rev2", "rev3")
29
40
self.assertEquals(br.revision_history(), ["rev1", "rev2", "rev3"])
42
def test_fetch_revisions(self):
43
"""Test fetch-revision operation."""
44
from bzrlib.fetch import Fetcher
47
b1 = Branch.initialize('b1')
48
b2 = Branch.initialize('b2')
49
file(os.sep.join(['b1', 'foo']), 'w').write('hello')
50
b1.add(['foo'], ['foo-id'])
51
b1.commit('lala!', rev_id='revision-1', allow_pointless=False)
54
f = Fetcher(from_branch=b1, to_branch=b2)
55
eq = self.assertEquals
57
eq(f.last_revision, 'revision-1')
59
rev = b2.get_revision('revision-1')
60
tree = b2.revision_tree('revision-1')
61
eq(tree.get_file_text('foo-id'), 'hello')
63
def test_push_stores(self):
64
"""Copy the stores from one branch to another"""
66
br_a = Branch.initialize("a")
67
file('a/b', 'wb').write('b')
69
commit(br_a, "silly commit")
72
br_b = Branch.initialize("b")
73
self.assertRaises(NoSuchRevision, br_b.get_revision,
74
br_a.revision_history()[0])
75
br_a.push_stores(br_b)
76
rev = br_b.get_revision(br_a.revision_history()[0])
77
tree = br_b.revision_tree(br_a.revision_history()[0])
79
if tree.inventory[file_id].kind == "file":
80
tree.get_file(file_id).read()
83
def test_copy_branch(self):
84
"""Copy the stores from one branch to another"""
85
br_a, br_b = self.test_push_stores()
86
commit(br_b, "silly commit")
88
br_c = copy_branch(br_a, 'c', basis_branch=br_b)
89
self.assertEqual(br_a.revision_history(), br_c.revision_history())
91
def test_copy_partial(self):
92
"""Copy only part of the history of a branch."""
93
self.build_tree(['a/', 'a/one'])
94
br_a = Branch.initialize('a')
96
br_a.commit('commit one', rev_id='u@d-1')
97
self.build_tree(['a/two'])
99
br_a.commit('commit two', rev_id='u@d-2')
100
br_b = copy_branch(br_a, 'b', revision='u@d-1')
101
self.assertEqual(br_b.last_revision(), 'u@d-1')
102
self.assertTrue(os.path.exists('b/one'))
103
self.assertFalse(os.path.exists('b/two'))
106
def test_record_initial_ghost_merge(self):
107
"""A pending merge with no revision present is still a merge."""
108
branch = Branch.initialize('.')
109
branch.add_pending_merge('non:existent@rev--ision--0--2')
110
branch.commit('pretend to merge nonexistent-revision', rev_id='first')
111
rev = branch.get_revision(branch.last_revision())
112
self.assertEqual(len(rev.parent_ids), 1)
113
# parent_sha1s is not populated now, WTF. rbc 20051003
114
self.assertEqual(len(rev.parent_sha1s), 0)
115
self.assertEqual(rev.parent_ids[0], 'non:existent@rev--ision--0--2')
118
# compare the gpg-to-sign info for a commit with a ghost and
119
# an identical tree without a ghost
120
# fetch missing should rewrite the TOC of weaves to list newly available parents.
122
def test_pending_merges(self):
123
"""Tracking pending-merged revisions."""
124
b = Branch.initialize('.')
126
self.assertEquals(b.pending_merges(), [])
127
b.add_pending_merge('foo@azkhazan-123123-abcabc')
128
self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
129
b.add_pending_merge('foo@azkhazan-123123-abcabc')
130
self.assertEquals(b.pending_merges(), ['foo@azkhazan-123123-abcabc'])
131
b.add_pending_merge('wibble@fofof--20050401--1928390812')
132
self.assertEquals(b.pending_merges(),
133
['foo@azkhazan-123123-abcabc',
134
'wibble@fofof--20050401--1928390812'])
135
b.commit("commit from base with two merges")
136
rev = b.get_revision(b.revision_history()[0])
137
self.assertEquals(len(rev.parent_ids), 2)
138
self.assertEquals(rev.parent_ids[0],
139
'foo@azkhazan-123123-abcabc')
140
self.assertEquals(rev.parent_ids[1],
141
'wibble@fofof--20050401--1928390812')
142
# list should be cleared when we do a commit
143
self.assertEquals(b.pending_merges(), [])
145
def test_sign_existing_revision(self):
147
branch = Branch.initialize('.')
148
branch.commit("base", allow_pointless=True, rev_id='A')
149
from bzrlib.testament import Testament
150
branch.sign_revision('A', bzrlib.gpg.LoopbackGPGStrategy(None))
151
self.assertEqual(Testament.from_revision(branch, 'A').as_short_text(),
152
branch.revision_store.get('A', 'sig').read())
155
class TestRemote(TestCaseWithWebserver):
157
def test_open_containing(self):
158
self.assertRaises(NotBranchError, Branch.open_containing,
159
self.get_remote_url(''))
160
self.assertRaises(NotBranchError, Branch.open_containing,
161
self.get_remote_url('g/p/q'))
162
b = Branch.initialize('.')
163
Branch.open_containing(self.get_remote_url(''))
164
Branch.open_containing(self.get_remote_url('g/p/q'))
166
# TODO: rewrite this as a regular unittest, without relying on the displayed output
167
# >>> from bzrlib.commit import commit
168
# >>> bzrlib.trace.silent = True
169
# >>> br1 = ScratchBranch(files=['foo', 'bar'])
172
# >>> commit(br1, "lala!", rev_id="REVISION-ID-1", verbose=False)
173
# >>> br2 = ScratchBranch()
174
# >>> br2.update_revisions(br1)
176
# Added 1 inventories.
178
# >>> br2.revision_history()
180
# >>> br2.update_revisions(br1)
182
# >>> br1.text_store.total_size() == br2.text_store.total_size()
185
class InstrumentedTransaction(object):
188
self.calls.append('finish')
194
class TestBranchTransaction(TestCaseInTempDir):
197
super(TestBranchTransaction, self).setUp()
198
self.branch = Branch.initialize('.')
200
def test_default_get_transaction(self):
201
"""branch.get_transaction on a new branch should give a PassThrough."""
202
self.failUnless(isinstance(self.branch.get_transaction(),
203
transactions.PassThroughTransaction))
205
def test__set_new_transaction(self):
206
self.branch._set_transaction(transactions.ReadOnlyTransaction())
208
def test__set_over_existing_transaction_raises(self):
209
self.branch._set_transaction(transactions.ReadOnlyTransaction())
210
self.assertRaises(errors.LockError,
211
self.branch._set_transaction,
212
transactions.ReadOnlyTransaction())
214
def test_finish_no_transaction_raises(self):
215
self.assertRaises(errors.LockError, self.branch._finish_transaction)
217
def test_finish_readonly_transaction_works(self):
218
self.branch._set_transaction(transactions.ReadOnlyTransaction())
219
self.branch._finish_transaction()
220
self.assertEqual(None, self.branch._transaction)
222
def test_unlock_calls_finish(self):
223
self.branch.lock_read()
224
transaction = InstrumentedTransaction()
225
self.branch._transaction = transaction
227
self.assertEqual(['finish'], transaction.calls)
229
def test_lock_read_acquires_ro_transaction(self):
230
self.branch.lock_read()
231
self.failUnless(isinstance(self.branch.get_transaction(),
232
transactions.ReadOnlyTransaction))
235
def test_lock_write_acquires_passthrough_transaction(self):
236
self.branch.lock_write()
237
# cannot use get_transaction as its magic
238
self.failUnless(isinstance(self.branch._transaction,
239
transactions.PassThroughTransaction))
added
removed
bzrlib/selftest/testbranch.py
