/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

Fix formatting, remove catch-all for exceptions when opening local repositories.

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