/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testbranch.py

  • Committer: Aaron Bentley
  • Date: 2005-10-18 18:48:27 UTC
  • mto: (1185.25.1)
  • mto: This revision was merged to the branch mainline in revision 1474.
  • Revision ID: abentley@panoramicfeedback.com-20051018184827-2cc69376beb1cdf3
Switched to ConfigObj

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testbranch.py
 
1
# (C) 2005 Canonical Ltd
 
2
 
 
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.
 
7
 
 
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.
 
12
 
 
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
 
16
 
 
17
import os
 
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
 
27
 
 
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.
 
31
 
 
32
class TestBranch(TestCaseInTempDir):
 
33
 
 
34
    def test_append_revisions(self):
 
35
        """Test appending more than one revision"""
 
36
        br = Branch.initialize(".")
 
37
        br.append_revision("rev1")
 
38
        self.assertEquals(br.revision_history(), ["rev1",])
 
39
        br.append_revision("rev2", "rev3")
 
40
        self.assertEquals(br.revision_history(), ["rev1", "rev2", "rev3"])
 
41
 
 
42
    def test_fetch_revisions(self):
 
43
        """Test fetch-revision operation."""
 
44
        from bzrlib.fetch import Fetcher
 
45
        os.mkdir('b1')
 
46
        os.mkdir('b2')
 
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)
 
52
 
 
53
        mutter('start fetch')
 
54
        f = Fetcher(from_branch=b1, to_branch=b2)
 
55
        eq = self.assertEquals
 
56
        eq(f.count_copied, 1)
 
57
        eq(f.last_revision, 'revision-1')
 
58
 
 
59
        rev = b2.get_revision('revision-1')
 
60
        tree = b2.revision_tree('revision-1')
 
61
        eq(tree.get_file_text('foo-id'), 'hello')
 
62
 
 
63
    def test_push_stores(self):
 
64
        """Copy the stores from one branch to another"""
 
65
        os.mkdir('a')
 
66
        br_a = Branch.initialize("a")
 
67
        file('a/b', 'wb').write('b')
 
68
        br_a.add('b')
 
69
        commit(br_a, "silly commit")
 
70
 
 
71
        os.mkdir('b')
 
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])
 
78
        for file_id in tree:
 
79
            if tree.inventory[file_id].kind == "file":
 
80
                tree.get_file(file_id).read()
 
81
        return br_a, br_b
 
82
 
 
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")
 
87
        os.mkdir('c')
 
88
        br_c = copy_branch(br_a, 'c', basis_branch=br_b)
 
89
        self.assertEqual(br_a.revision_history(), br_c.revision_history())
 
90
 
 
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')
 
95
        br_a.add(['one'])
 
96
        br_a.commit('commit one', rev_id='u@d-1')
 
97
        self.build_tree(['a/two'])
 
98
        br_a.add(['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'))
 
104
        
 
105
 
 
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')
 
116
 
 
117
# TODO 20051003 RBC:
 
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.
 
121
        
 
122
    def test_pending_merges(self):
 
123
        """Tracking pending-merged revisions."""
 
124
        b = Branch.initialize('.')
 
125
 
 
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(), [])
 
144
 
 
145
 
 
146
class TestRemote(TestCaseWithWebserver):
 
147
 
 
148
    def test_open_containing(self):
 
149
        self.assertRaises(NotBranchError, Branch.open_containing,
 
150
                          self.get_remote_url(''))
 
151
        self.assertRaises(NotBranchError, Branch.open_containing,
 
152
                          self.get_remote_url('g/p/q'))
 
153
        b = Branch.initialize('.')
 
154
        Branch.open_containing(self.get_remote_url(''))
 
155
        Branch.open_containing(self.get_remote_url('g/p/q'))
 
156
        
 
157
# TODO: rewrite this as a regular unittest, without relying on the displayed output        
 
158
#         >>> from bzrlib.commit import commit
 
159
#         >>> bzrlib.trace.silent = True
 
160
#         >>> br1 = ScratchBranch(files=['foo', 'bar'])
 
161
#         >>> br1.add('foo')
 
162
#         >>> br1.add('bar')
 
163
#         >>> commit(br1, "lala!", rev_id="REVISION-ID-1", verbose=False)
 
164
#         >>> br2 = ScratchBranch()
 
165
#         >>> br2.update_revisions(br1)
 
166
#         Added 2 texts.
 
167
#         Added 1 inventories.
 
168
#         Added 1 revisions.
 
169
#         >>> br2.revision_history()
 
170
#         [u'REVISION-ID-1']
 
171
#         >>> br2.update_revisions(br1)
 
172
#         Added 0 revisions.
 
173
#         >>> br1.text_store.total_size() == br2.text_store.total_size()
 
174
#         True
 
175
 
 
176
class InstrumentedTransaction(object):
 
177
 
 
178
    def finish(self):
 
179
        self.calls.append('finish')
 
180
 
 
181
    def __init__(self):
 
182
        self.calls = []
 
183
 
 
184
 
 
185
class TestBranchTransaction(TestCaseInTempDir):
 
186
 
 
187
    def setUp(self):
 
188
        super(TestBranchTransaction, self).setUp()
 
189
        self.branch = Branch.initialize('.')
 
190
        
 
191
    def test_default_get_transaction(self):
 
192
        """branch.get_transaction on a new branch should give a PassThrough."""
 
193
        self.failUnless(isinstance(self.branch.get_transaction(),
 
194
                                   transactions.PassThroughTransaction))
 
195
 
 
196
    def test__set_new_transaction(self):
 
197
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
198
 
 
199
    def test__set_over_existing_transaction_raises(self):
 
200
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
201
        self.assertRaises(errors.LockError,
 
202
                          self.branch._set_transaction,
 
203
                          transactions.ReadOnlyTransaction())
 
204
 
 
205
    def test_finish_no_transaction_raises(self):
 
206
        self.assertRaises(errors.LockError, self.branch._finish_transaction)
 
207
 
 
208
    def test_finish_readonly_transaction_works(self):
 
209
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
210
        self.branch._finish_transaction()
 
211
        self.assertEqual(None, self.branch._transaction)
 
212
 
 
213
    def test_unlock_calls_finish(self):
 
214
        self.branch.lock_read()
 
215
        transaction = InstrumentedTransaction()
 
216
        self.branch._transaction = transaction
 
217
        self.branch.unlock()
 
218
        self.assertEqual(['finish'], transaction.calls)
 
219
 
 
220
    def test_lock_read_acquires_ro_transaction(self):
 
221
        self.branch.lock_read()
 
222
        self.failUnless(isinstance(self.branch.get_transaction(),
 
223
                                   transactions.ReadOnlyTransaction))
 
224
        self.branch.unlock()
 
225
        
 
226
    def test_lock_write_acquires_passthrough_transaction(self):
 
227
        self.branch.lock_write()
 
228
        # cannot use get_transaction as its magic
 
229
        self.failUnless(isinstance(self.branch._transaction,
 
230
                                   transactions.PassThroughTransaction))
 
231
        self.branch.unlock()