/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 at utoronto
  • Date: 2005-10-10 04:13:48 UTC
  • mto: (1185.25.1)
  • mto: This revision was merged to the branch mainline in revision 1460.
  • Revision ID: aaron.bentley@utoronto.ca-20051010041348-3049184277dce841
RestoredĀ oldĀ tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
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
class TestBranch(TestCaseInTempDir):
 
29
 
 
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"])
 
37
 
 
38
    def test_fetch_revisions(self):
 
39
        """Test fetch-revision operation."""
 
40
        from bzrlib.fetch import Fetcher
 
41
        os.mkdir('b1')
 
42
        os.mkdir('b2')
 
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)
 
48
 
 
49
        mutter('start fetch')
 
50
        f = Fetcher(from_branch=b1, to_branch=b2)
 
51
        eq = self.assertEquals
 
52
        eq(f.count_copied, 1)
 
53
        eq(f.last_revision, 'revision-1')
 
54
 
 
55
        rev = b2.get_revision('revision-1')
 
56
        tree = b2.revision_tree('revision-1')
 
57
        eq(tree.get_file_text('foo-id'), 'hello')
 
58
 
 
59
    def test_push_stores(self):
 
60
        """Copy the stores from one branch to another"""
 
61
        os.mkdir('a')
 
62
        br_a = Branch.initialize("a")
 
63
        file('a/b', 'wb').write('b')
 
64
        br_a.add('b')
 
65
        commit(br_a, "silly commit")
 
66
 
 
67
        os.mkdir('b')
 
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])
 
74
        for file_id in tree:
 
75
            if tree.inventory[file_id].kind == "file":
 
76
                tree.get_file(file_id).read()
 
77
        return br_a, br_b
 
78
 
 
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")
 
83
        os.mkdir('c')
 
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())
 
89
 
 
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')
 
94
        br_a.add(['one'])
 
95
        br_a.commit('commit one', rev_id='u@d-1')
 
96
        self.build_tree(['a/two'])
 
97
        br_a.add(['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'))
 
103
        
 
104
 
 
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')
 
115
 
 
116
# TODO 20051003 RBC:
 
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.
 
120
        
 
121
    def test_pending_merges(self):
 
122
        """Tracking pending-merged revisions."""
 
123
        b = Branch.initialize('.')
 
124
 
 
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(), [])
 
143
 
 
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'))
 
153
        
 
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'])
 
158
#         >>> br1.add('foo')
 
159
#         >>> br1.add('bar')
 
160
#         >>> commit(br1, "lala!", rev_id="REVISION-ID-1", verbose=False)
 
161
#         >>> br2 = ScratchBranch()
 
162
#         >>> br2.update_revisions(br1)
 
163
#         Added 2 texts.
 
164
#         Added 1 inventories.
 
165
#         Added 1 revisions.
 
166
#         >>> br2.revision_history()
 
167
#         [u'REVISION-ID-1']
 
168
#         >>> br2.update_revisions(br1)
 
169
#         Added 0 revisions.
 
170
#         >>> br1.text_store.total_size() == br2.text_store.total_size()
 
171
#         True
 
172
 
 
173
class InstrumentedTransaction(object):
 
174
 
 
175
    def finish(self):
 
176
        self.calls.append('finish')
 
177
 
 
178
    def __init__(self):
 
179
        self.calls = []
 
180
 
 
181
 
 
182
class TestBranchTransaction(TestCaseInTempDir):
 
183
 
 
184
    def setUp(self):
 
185
        super(TestBranchTransaction, self).setUp()
 
186
        self.branch = Branch.initialize('.')
 
187
        
 
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))
 
192
 
 
193
    def test__set_new_transaction(self):
 
194
        self.branch._set_transaction(transactions.ReadOnlyTransaction())
 
195
 
 
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())
 
201
 
 
202
    def test_finish_no_transaction_raises(self):
 
203
        self.assertRaises(errors.LockError, self.branch._finish_transaction)
 
204
 
 
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)
 
209
 
 
210
    def test_unlock_calls_finish(self):
 
211
        self.branch.lock_read()
 
212
        transaction = InstrumentedTransaction()
 
213
        self.branch._transaction = transaction
 
214
        self.branch.unlock()
 
215
        self.assertEqual(['finish'], transaction.calls)
 
216
 
 
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))
 
221
        self.branch.unlock()
 
222
        
 
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))
 
228
        self.branch.unlock()