/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/tests/test_branch.py

  • Committer: Vincent Ladeuil
  • Date: 2010-02-05 15:26:17 UTC
  • mto: (5013.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5014.
  • Revision ID: v.ladeuil+lp@free.fr-20100205152617-6p9rf9b0m4ucvn0u
Fix test_smart_transport.py imports.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2005, 2006, 2007 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
"""Tests for the Branch facility that are not interface  tests.
 
18
 
 
19
For interface tests see tests/per_branch/*.py.
 
20
 
 
21
For concrete class tests see this file, and for meta-branch tests
 
22
also see this file.
 
23
"""
 
24
 
 
25
from StringIO import StringIO
 
26
 
 
27
from bzrlib import (
 
28
    branch as _mod_branch,
 
29
    bzrdir,
 
30
    config,
 
31
    errors,
 
32
    trace,
 
33
    urlutils,
 
34
    )
 
35
from bzrlib.branch import (
 
36
    Branch,
 
37
    BranchHooks,
 
38
    BranchFormat,
 
39
    BranchReferenceFormat,
 
40
    BzrBranch5,
 
41
    BzrBranchFormat5,
 
42
    BzrBranchFormat6,
 
43
    BzrBranchFormat7,
 
44
    PullResult,
 
45
    _run_with_write_locked_target,
 
46
    )
 
47
from bzrlib.bzrdir import (BzrDirMetaFormat1, BzrDirMeta1,
 
48
                           BzrDir, BzrDirFormat)
 
49
from bzrlib.transport import get_transport
 
50
 
 
51
 
 
52
class TestDefaultFormat(tests.TestCase):
 
53
 
 
54
    def test_default_format(self):
 
55
        # update this if you change the default branch format
 
56
        self.assertIsInstance(BranchFormat.get_default_format(),
 
57
                BzrBranchFormat7)
 
58
 
 
59
    def test_default_format_is_same_as_bzrdir_default(self):
 
60
        # XXX: it might be nice if there was only one place the default was
 
61
        # set, but at the moment that's not true -- mbp 20070814 --
 
62
        # https://bugs.launchpad.net/bzr/+bug/132376
 
63
        self.assertEqual(BranchFormat.get_default_format(),
 
64
                BzrDirFormat.get_default_format().get_branch_format())
 
65
 
 
66
    def test_get_set_default_format(self):
 
67
        # set the format and then set it back again
 
68
        old_format = BranchFormat.get_default_format()
 
69
        BranchFormat.set_default_format(SampleBranchFormat())
 
70
        try:
 
71
            # the default branch format is used by the meta dir format
 
72
            # which is not the default bzrdir format at this point
 
73
            dir = BzrDirMetaFormat1().initialize('memory:///')
 
74
            result = dir.create_branch()
 
75
            self.assertEqual(result, 'A branch')
 
76
        finally:
 
77
            BranchFormat.set_default_format(old_format)
 
78
        self.assertEqual(old_format, BranchFormat.get_default_format())
 
79
 
 
80
 
 
81
class TestBranchFormat5(tests.TestCaseWithTransport):
 
82
    """Tests specific to branch format 5"""
 
83
 
 
84
    def test_branch_format_5_uses_lockdir(self):
 
85
        url = self.get_url()
 
86
        bzrdir = BzrDirMetaFormat1().initialize(url)
 
87
        bzrdir.create_repository()
 
88
        branch = bzrdir.create_branch()
 
89
        t = self.get_transport()
 
90
        self.log("branch instance is %r" % branch)
 
91
        self.assert_(isinstance(branch, BzrBranch5))
 
92
        self.assertIsDirectory('.', t)
 
93
        self.assertIsDirectory('.bzr/branch', t)
 
94
        self.assertIsDirectory('.bzr/branch/lock', t)
 
95
        branch.lock_write()
 
96
        try:
 
97
            self.assertIsDirectory('.bzr/branch/lock/held', t)
 
98
        finally:
 
99
            branch.unlock()
 
100
 
 
101
    def test_set_push_location(self):
 
102
        from bzrlib.config import (locations_config_filename,
 
103
                                   ensure_config_dir_exists)
 
104
        ensure_config_dir_exists()
 
105
        fn = locations_config_filename()
 
106
        # write correct newlines to locations.conf
 
107
        # by default ConfigObj uses native line-endings for new files
 
108
        # but uses already existing line-endings if file is not empty
 
109
        f = open(fn, 'wb')
 
110
        try:
 
111
            f.write('# comment\n')
 
112
        finally:
 
113
            f.close()
 
114
 
 
115
        branch = self.make_branch('.', format='knit')
 
116
        branch.set_push_location('foo')
 
117
        local_path = urlutils.local_path_from_url(branch.base[:-1])
 
118
        self.assertFileEqual("# comment\n"
 
119
                             "[%s]\n"
 
120
                             "push_location = foo\n"
 
121
                             "push_location:policy = norecurse\n" % local_path,
 
122
                             fn)
 
123
 
 
124
    # TODO RBC 20051029 test getting a push location from a branch in a
 
125
    # recursive section - that is, it appends the branch name.
 
126
 
 
127
 
 
128
class SampleBranchFormat(BranchFormat):
 
129
    """A sample format
 
130
 
 
131
    this format is initializable, unsupported to aid in testing the
 
132
    open and open_downlevel routines.
 
133
    """
 
134
 
 
135
    def get_format_string(self):
 
136
        """See BzrBranchFormat.get_format_string()."""
 
137
        return "Sample branch format."
 
138
 
 
139
    def initialize(self, a_bzrdir):
 
140
        """Format 4 branches cannot be created."""
 
141
        t = a_bzrdir.get_branch_transport(self)
 
142
        t.put_bytes('format', self.get_format_string())
 
143
        return 'A branch'
 
144
 
 
145
    def is_supported(self):
 
146
        return False
 
147
 
 
148
    def open(self, transport, _found=False, ignore_fallbacks=False):
 
149
        return "opened branch."
 
150
 
 
151
 
 
152
class TestBzrBranchFormat(tests.TestCaseWithTransport):
 
153
    """Tests for the BzrBranchFormat facility."""
 
154
 
 
155
    def test_find_format(self):
 
156
        # is the right format object found for a branch?
 
157
        # create a branch with a few known format objects.
 
158
        # this is not quite the same as
 
159
        self.build_tree(["foo/", "bar/"])
 
160
        def check_format(format, url):
 
161
            dir = format._matchingbzrdir.initialize(url)
 
162
            dir.create_repository()
 
163
            format.initialize(dir)
 
164
            found_format = BranchFormat.find_format(dir)
 
165
            self.failUnless(isinstance(found_format, format.__class__))
 
166
        check_format(BzrBranchFormat5(), "bar")
 
167
 
 
168
    def test_find_format_not_branch(self):
 
169
        dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
 
170
        self.assertRaises(errors.NotBranchError,
 
171
                          BranchFormat.find_format,
 
172
                          dir)
 
173
 
 
174
    def test_find_format_unknown_format(self):
 
175
        dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
 
176
        SampleBranchFormat().initialize(dir)
 
177
        self.assertRaises(errors.UnknownFormatError,
 
178
                          BranchFormat.find_format,
 
179
                          dir)
 
180
 
 
181
    def test_register_unregister_format(self):
 
182
        format = SampleBranchFormat()
 
183
        # make a control dir
 
184
        dir = bzrdir.BzrDirMetaFormat1().initialize(self.get_url())
 
185
        # make a branch
 
186
        format.initialize(dir)
 
187
        # register a format for it.
 
188
        BranchFormat.register_format(format)
 
189
        # which branch.Open will refuse (not supported)
 
190
        self.assertRaises(errors.UnsupportedFormatError,
 
191
                          Branch.open, self.get_url())
 
192
        self.make_branch_and_tree('foo')
 
193
        # but open_downlevel will work
 
194
        self.assertEqual(
 
195
            format.open(dir),
 
196
            bzrdir.BzrDir.open(self.get_url()).open_branch(unsupported=True))
 
197
        # unregister the format
 
198
        BranchFormat.unregister_format(format)
 
199
        self.make_branch_and_tree('bar')
 
200
 
 
201
 
 
202
class TestBranch67(object):
 
203
    """Common tests for both branch 6 and 7 which are mostly the same."""
 
204
 
 
205
    def get_format_name(self):
 
206
        raise NotImplementedError(self.get_format_name)
 
207
 
 
208
    def get_format_name_subtree(self):
 
209
        raise NotImplementedError(self.get_format_name)
 
210
 
 
211
    def get_class(self):
 
212
        raise NotImplementedError(self.get_class)
 
213
 
 
214
    def test_creation(self):
 
215
        format = BzrDirMetaFormat1()
 
216
        format.set_branch_format(_mod_branch.BzrBranchFormat6())
 
217
        branch = self.make_branch('a', format=format)
 
218
        self.assertIsInstance(branch, self.get_class())
 
219
        branch = self.make_branch('b', format=self.get_format_name())
 
220
        self.assertIsInstance(branch, self.get_class())
 
221
        branch = _mod_branch.Branch.open('a')
 
222
        self.assertIsInstance(branch, self.get_class())
 
223
 
 
224
    def test_layout(self):
 
225
        branch = self.make_branch('a', format=self.get_format_name())
 
226
        self.failUnlessExists('a/.bzr/branch/last-revision')
 
227
        self.failIfExists('a/.bzr/branch/revision-history')
 
228
        self.failIfExists('a/.bzr/branch/references')
 
229
 
 
230
    def test_config(self):
 
231
        """Ensure that all configuration data is stored in the branch"""
 
232
        branch = self.make_branch('a', format=self.get_format_name())
 
233
        branch.set_parent('http://bazaar-vcs.org')
 
234
        self.failIfExists('a/.bzr/branch/parent')
 
235
        self.assertEqual('http://bazaar-vcs.org', branch.get_parent())
 
236
        branch.set_push_location('sftp://bazaar-vcs.org')
 
237
        config = branch.get_config()._get_branch_data_config()
 
238
        self.assertEqual('sftp://bazaar-vcs.org',
 
239
                         config.get_user_option('push_location'))
 
240
        branch.set_bound_location('ftp://bazaar-vcs.org')
 
241
        self.failIfExists('a/.bzr/branch/bound')
 
242
        self.assertEqual('ftp://bazaar-vcs.org', branch.get_bound_location())
 
243
 
 
244
    def test_set_revision_history(self):
 
245
        builder = self.make_branch_builder('.', format=self.get_format_name())
 
246
        builder.build_snapshot('foo', None,
 
247
            [('add', ('', None, 'directory', None))],
 
248
            message='foo')
 
249
        builder.build_snapshot('bar', None, [], message='bar')
 
250
        branch = builder.get_branch()
 
251
        branch.lock_write()
 
252
        self.addCleanup(branch.unlock)
 
253
        branch.set_revision_history(['foo', 'bar'])
 
254
        branch.set_revision_history(['foo'])
 
255
        self.assertRaises(errors.NotLefthandHistory,
 
256
                          branch.set_revision_history, ['bar'])
 
257
 
 
258
    def do_checkout_test(self, lightweight=False):
 
259
        tree = self.make_branch_and_tree('source',
 
260
            format=self.get_format_name_subtree())
 
261
        subtree = self.make_branch_and_tree('source/subtree',
 
262
            format=self.get_format_name_subtree())
 
263
        subsubtree = self.make_branch_and_tree('source/subtree/subsubtree',
 
264
            format=self.get_format_name_subtree())
 
265
        self.build_tree(['source/subtree/file',
 
266
                         'source/subtree/subsubtree/file'])
 
267
        subsubtree.add('file')
 
268
        subtree.add('file')
 
269
        subtree.add_reference(subsubtree)
 
270
        tree.add_reference(subtree)
 
271
        tree.commit('a revision')
 
272
        subtree.commit('a subtree file')
 
273
        subsubtree.commit('a subsubtree file')
 
274
        tree.branch.create_checkout('target', lightweight=lightweight)
 
275
        self.failUnlessExists('target')
 
276
        self.failUnlessExists('target/subtree')
 
277
        self.failUnlessExists('target/subtree/file')
 
278
        self.failUnlessExists('target/subtree/subsubtree/file')
 
279
        subbranch = _mod_branch.Branch.open('target/subtree/subsubtree')
 
280
        if lightweight:
 
281
            self.assertEndsWith(subbranch.base, 'source/subtree/subsubtree/')
 
282
        else:
 
283
            self.assertEndsWith(subbranch.base, 'target/subtree/subsubtree/')
 
284
 
 
285
    def test_checkout_with_references(self):
 
286
        self.do_checkout_test()
 
287
 
 
288
    def test_light_checkout_with_references(self):
 
289
        self.do_checkout_test(lightweight=True)
 
290
 
 
291
    def test_set_push(self):
 
292
        branch = self.make_branch('source', format=self.get_format_name())
 
293
        branch.get_config().set_user_option('push_location', 'old',
 
294
            store=config.STORE_LOCATION)
 
295
        warnings = []
 
296
        def warning(*args):
 
297
            warnings.append(args[0] % args[1:])
 
298
        _warning = trace.warning
 
299
        trace.warning = warning
 
300
        try:
 
301
            branch.set_push_location('new')
 
302
        finally:
 
303
            trace.warning = _warning
 
304
        self.assertEqual(warnings[0], 'Value "new" is masked by "old" from '
 
305
                         'locations.conf')
 
306
 
 
307
 
 
308
class TestBranch6(TestBranch67, tests.TestCaseWithTransport):
 
309
 
 
310
    def get_class(self):
 
311
        return _mod_branch.BzrBranch6
 
312
 
 
313
    def get_format_name(self):
 
314
        return "dirstate-tags"
 
315
 
 
316
    def get_format_name_subtree(self):
 
317
        return "dirstate-with-subtree"
 
318
 
 
319
    def test_set_stacked_on_url_errors(self):
 
320
        branch = self.make_branch('a', format=self.get_format_name())
 
321
        self.assertRaises(errors.UnstackableBranchFormat,
 
322
            branch.set_stacked_on_url, None)
 
323
 
 
324
    def test_default_stacked_location(self):
 
325
        branch = self.make_branch('a', format=self.get_format_name())
 
326
        self.assertRaises(errors.UnstackableBranchFormat, branch.get_stacked_on_url)
 
327
 
 
328
 
 
329
class TestBranch7(TestBranch67, tests.TestCaseWithTransport):
 
330
 
 
331
    def get_class(self):
 
332
        return _mod_branch.BzrBranch7
 
333
 
 
334
    def get_format_name(self):
 
335
        return "1.9"
 
336
 
 
337
    def get_format_name_subtree(self):
 
338
        return "development-subtree"
 
339
 
 
340
    def test_set_stacked_on_url_unstackable_repo(self):
 
341
        repo = self.make_repository('a', format='dirstate-tags')
 
342
        control = repo.bzrdir
 
343
        branch = _mod_branch.BzrBranchFormat7().initialize(control)
 
344
        target = self.make_branch('b')
 
345
        self.assertRaises(errors.UnstackableRepositoryFormat,
 
346
            branch.set_stacked_on_url, target.base)
 
347
 
 
348
    def test_clone_stacked_on_unstackable_repo(self):
 
349
        repo = self.make_repository('a', format='dirstate-tags')
 
350
        control = repo.bzrdir
 
351
        branch = _mod_branch.BzrBranchFormat7().initialize(control)
 
352
        # Calling clone should not raise UnstackableRepositoryFormat.
 
353
        cloned_bzrdir = control.clone('cloned')
 
354
 
 
355
    def _test_default_stacked_location(self):
 
356
        branch = self.make_branch('a', format=self.get_format_name())
 
357
        self.assertRaises(errors.NotStacked, branch.get_stacked_on_url)
 
358
 
 
359
    def test_stack_and_unstack(self):
 
360
        branch = self.make_branch('a', format=self.get_format_name())
 
361
        target = self.make_branch_and_tree('b', format=self.get_format_name())
 
362
        branch.set_stacked_on_url(target.branch.base)
 
363
        self.assertEqual(target.branch.base, branch.get_stacked_on_url())
 
364
        revid = target.commit('foo')
 
365
        self.assertTrue(branch.repository.has_revision(revid))
 
366
        branch.set_stacked_on_url(None)
 
367
        self.assertRaises(errors.NotStacked, branch.get_stacked_on_url)
 
368
        self.assertFalse(branch.repository.has_revision(revid))
 
369
 
 
370
    def test_open_opens_stacked_reference(self):
 
371
        branch = self.make_branch('a', format=self.get_format_name())
 
372
        target = self.make_branch_and_tree('b', format=self.get_format_name())
 
373
        branch.set_stacked_on_url(target.branch.base)
 
374
        branch = branch.bzrdir.open_branch()
 
375
        revid = target.commit('foo')
 
376
        self.assertTrue(branch.repository.has_revision(revid))
 
377
 
 
378
 
 
379
class BzrBranch8(tests.TestCaseWithTransport):
 
380
 
 
381
    def make_branch(self, location, format=None):
 
382
        if format is None:
 
383
            format = bzrdir.format_registry.make_bzrdir('1.9')
 
384
            format.set_branch_format(_mod_branch.BzrBranchFormat8())
 
385
        return tests.TestCaseWithTransport.make_branch(
 
386
            self, location, format=format)
 
387
 
 
388
    def create_branch_with_reference(self):
 
389
        branch = self.make_branch('branch')
 
390
        branch._set_all_reference_info({'file-id': ('path', 'location')})
 
391
        return branch
 
392
 
 
393
    @staticmethod
 
394
    def instrument_branch(branch, gets):
 
395
        old_get = branch._transport.get
 
396
        def get(*args, **kwargs):
 
397
            gets.append((args, kwargs))
 
398
            return old_get(*args, **kwargs)
 
399
        branch._transport.get = get
 
400
 
 
401
    def test_reference_info_caching_read_locked(self):
 
402
        gets = []
 
403
        branch = self.create_branch_with_reference()
 
404
        branch.lock_read()
 
405
        self.addCleanup(branch.unlock)
 
406
        self.instrument_branch(branch, gets)
 
407
        branch.get_reference_info('file-id')
 
408
        branch.get_reference_info('file-id')
 
409
        self.assertEqual(1, len(gets))
 
410
 
 
411
    def test_reference_info_caching_read_unlocked(self):
 
412
        gets = []
 
413
        branch = self.create_branch_with_reference()
 
414
        self.instrument_branch(branch, gets)
 
415
        branch.get_reference_info('file-id')
 
416
        branch.get_reference_info('file-id')
 
417
        self.assertEqual(2, len(gets))
 
418
 
 
419
    def test_reference_info_caching_write_locked(self):
 
420
        gets = []
 
421
        branch = self.make_branch('branch')
 
422
        branch.lock_write()
 
423
        self.instrument_branch(branch, gets)
 
424
        self.addCleanup(branch.unlock)
 
425
        branch._set_all_reference_info({'file-id': ('path2', 'location2')})
 
426
        path, location = branch.get_reference_info('file-id')
 
427
        self.assertEqual(0, len(gets))
 
428
        self.assertEqual('path2', path)
 
429
        self.assertEqual('location2', location)
 
430
 
 
431
    def test_reference_info_caches_cleared(self):
 
432
        branch = self.make_branch('branch')
 
433
        branch.lock_write()
 
434
        branch.set_reference_info('file-id', 'path2', 'location2')
 
435
        branch.unlock()
 
436
        doppelganger = Branch.open('branch')
 
437
        doppelganger.set_reference_info('file-id', 'path3', 'location3')
 
438
        self.assertEqual(('path3', 'location3'),
 
439
                         branch.get_reference_info('file-id'))
 
440
 
 
441
class TestBranchReference(tests.TestCaseWithTransport):
 
442
    """Tests for the branch reference facility."""
 
443
 
 
444
    def test_create_open_reference(self):
 
445
        bzrdirformat = bzrdir.BzrDirMetaFormat1()
 
446
        t = get_transport(self.get_url('.'))
 
447
        t.mkdir('repo')
 
448
        dir = bzrdirformat.initialize(self.get_url('repo'))
 
449
        dir.create_repository()
 
450
        target_branch = dir.create_branch()
 
451
        t.mkdir('branch')
 
452
        branch_dir = bzrdirformat.initialize(self.get_url('branch'))
 
453
        made_branch = BranchReferenceFormat().initialize(branch_dir, target_branch)
 
454
        self.assertEqual(made_branch.base, target_branch.base)
 
455
        opened_branch = branch_dir.open_branch()
 
456
        self.assertEqual(opened_branch.base, target_branch.base)
 
457
 
 
458
    def test_get_reference(self):
 
459
        """For a BranchReference, get_reference should reutrn the location."""
 
460
        branch = self.make_branch('target')
 
461
        checkout = branch.create_checkout('checkout', lightweight=True)
 
462
        reference_url = branch.bzrdir.root_transport.abspath('') + '/'
 
463
        # if the api for create_checkout changes to return different checkout types
 
464
        # then this file read will fail.
 
465
        self.assertFileEqual(reference_url, 'checkout/.bzr/branch/location')
 
466
        self.assertEqual(reference_url,
 
467
            _mod_branch.BranchReferenceFormat().get_reference(checkout.bzrdir))
 
468
 
 
469
 
 
470
class TestHooks(tests.TestCase):
 
471
 
 
472
    def test_constructor(self):
 
473
        """Check that creating a BranchHooks instance has the right defaults."""
 
474
        hooks = BranchHooks()
 
475
        self.assertTrue("set_rh" in hooks, "set_rh not in %s" % hooks)
 
476
        self.assertTrue("post_push" in hooks, "post_push not in %s" % hooks)
 
477
        self.assertTrue("post_commit" in hooks, "post_commit not in %s" % hooks)
 
478
        self.assertTrue("pre_commit" in hooks, "pre_commit not in %s" % hooks)
 
479
        self.assertTrue("post_pull" in hooks, "post_pull not in %s" % hooks)
 
480
        self.assertTrue("post_uncommit" in hooks, "post_uncommit not in %s" % hooks)
 
481
        self.assertTrue("post_change_branch_tip" in hooks,
 
482
                        "post_change_branch_tip not in %s" % hooks)
 
483
 
 
484
    def test_installed_hooks_are_BranchHooks(self):
 
485
        """The installed hooks object should be a BranchHooks."""
 
486
        # the installed hooks are saved in self._preserved_hooks.
 
487
        self.assertIsInstance(self._preserved_hooks[_mod_branch.Branch][1],
 
488
            BranchHooks)
 
489
 
 
490
 
 
491
class TestPullResult(tests.TestCase):
 
492
 
 
493
    def test_pull_result_to_int(self):
 
494
        # to support old code, the pull result can be used as an int
 
495
        r = PullResult()
 
496
        r.old_revno = 10
 
497
        r.new_revno = 20
 
498
        # this usage of results is not recommended for new code (because it
 
499
        # doesn't describe very well what happened), but for api stability
 
500
        # it's still supported
 
501
        a = "%d revisions pulled" % r
 
502
        self.assertEqual(a, "10 revisions pulled")
 
503
 
 
504
    def test_report_changed(self):
 
505
        r = PullResult()
 
506
        r.old_revid = "old-revid"
 
507
        r.old_revno = 10
 
508
        r.new_revid = "new-revid"
 
509
        r.new_revno = 20
 
510
        f = StringIO()
 
511
        r.report(f)
 
512
        self.assertEqual("Now on revision 20.\n", f.getvalue())
 
513
 
 
514
    def test_report_unchanged(self):
 
515
        r = PullResult()
 
516
        r.old_revid = "same-revid"
 
517
        r.new_revid = "same-revid"
 
518
        f = StringIO()
 
519
        r.report(f)
 
520
        self.assertEqual("No revisions to pull.\n", f.getvalue())
 
521
 
 
522
 
 
523
class _StubLockable(object):
 
524
    """Helper for TestRunWithWriteLockedTarget."""
 
525
 
 
526
    def __init__(self, calls, unlock_exc=None):
 
527
        self.calls = calls
 
528
        self.unlock_exc = unlock_exc
 
529
 
 
530
    def lock_write(self):
 
531
        self.calls.append('lock_write')
 
532
 
 
533
    def unlock(self):
 
534
        self.calls.append('unlock')
 
535
        if self.unlock_exc is not None:
 
536
            raise self.unlock_exc
 
537
 
 
538
 
 
539
class _ErrorFromCallable(Exception):
 
540
    """Helper for TestRunWithWriteLockedTarget."""
 
541
 
 
542
 
 
543
class _ErrorFromUnlock(Exception):
 
544
    """Helper for TestRunWithWriteLockedTarget."""
 
545
 
 
546
 
 
547
class TestRunWithWriteLockedTarget(tests.TestCase):
 
548
    """Tests for _run_with_write_locked_target."""
 
549
 
 
550
    def setUp(self):
 
551
        tests.TestCase.setUp(self)
 
552
        self._calls = []
 
553
 
 
554
    def func_that_returns_ok(self):
 
555
        self._calls.append('func called')
 
556
        return 'ok'
 
557
 
 
558
    def func_that_raises(self):
 
559
        self._calls.append('func called')
 
560
        raise _ErrorFromCallable()
 
561
 
 
562
    def test_success_unlocks(self):
 
563
        lockable = _StubLockable(self._calls)
 
564
        result = _run_with_write_locked_target(
 
565
            lockable, self.func_that_returns_ok)
 
566
        self.assertEqual('ok', result)
 
567
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
568
 
 
569
    def test_exception_unlocks_and_propagates(self):
 
570
        lockable = _StubLockable(self._calls)
 
571
        self.assertRaises(_ErrorFromCallable,
 
572
            _run_with_write_locked_target, lockable, self.func_that_raises)
 
573
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
574
 
 
575
    def test_callable_succeeds_but_error_during_unlock(self):
 
576
        lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
 
577
        self.assertRaises(_ErrorFromUnlock,
 
578
            _run_with_write_locked_target, lockable, self.func_that_returns_ok)
 
579
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
580
 
 
581
    def test_error_during_unlock_does_not_mask_original_error(self):
 
582
        lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
 
583
        self.assertRaises(_ErrorFromCallable,
 
584
            _run_with_write_locked_target, lockable, self.func_that_raises)
 
585
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
586
 
 
587