← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_branch.py

  • Committer: Martin Pool
  • Date: 2008-10-27 08:02:47 UTC
  • mfrom: (3795 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3798.
  • Revision ID: mbp@sourcefrog.net-20081027080247-0al6nrx2v8u1dcci
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_branch.py
41
41
    BzrBranchFormat5,
42
42
    BzrBranchFormat6,
43
43
    PullResult,
 
44
    _run_with_write_locked_target,
44
45
    )
45
46
from bzrlib.bzrdir import (BzrDirMetaFormat1, BzrDirMeta1, 
46
47
                           BzrDir, BzrDirFormat)
438
439
        # it's still supported
439
440
        a = "%d revisions pulled" % r
440
441
        self.assertEqual(a, "10 revisions pulled")
 
442
 
 
443
 
 
444
 
 
445
class _StubLockable(object):
 
446
    """Helper for TestRunWithWriteLockedTarget."""
 
447
 
 
448
    def __init__(self, calls, unlock_exc=None):
 
449
        self.calls = calls
 
450
        self.unlock_exc = unlock_exc
 
451
 
 
452
    def lock_write(self):
 
453
        self.calls.append('lock_write')
 
454
    
 
455
    def unlock(self):
 
456
        self.calls.append('unlock')
 
457
        if self.unlock_exc is not None:
 
458
            raise self.unlock_exc
 
459
 
 
460
 
 
461
class _ErrorFromCallable(Exception):
 
462
    """Helper for TestRunWithWriteLockedTarget."""
 
463
 
 
464
 
 
465
class _ErrorFromUnlock(Exception):
 
466
    """Helper for TestRunWithWriteLockedTarget."""
 
467
 
 
468
 
 
469
class TestRunWithWriteLockedTarget(TestCase):
 
470
    """Tests for _run_with_write_locked_target."""
 
471
 
 
472
    def setUp(self):
 
473
        self._calls = []
 
474
 
 
475
    def func_that_returns_ok(self):
 
476
        self._calls.append('func called')
 
477
        return 'ok'
 
478
 
 
479
    def func_that_raises(self):
 
480
        self._calls.append('func called')
 
481
        raise _ErrorFromCallable()
 
482
 
 
483
    def test_success_unlocks(self):
 
484
        lockable = _StubLockable(self._calls)
 
485
        result = _run_with_write_locked_target(
 
486
            lockable, self.func_that_returns_ok)
 
487
        self.assertEqual('ok', result)
 
488
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
489
 
 
490
    def test_exception_unlocks_and_propagates(self):
 
491
        lockable = _StubLockable(self._calls)
 
492
        self.assertRaises(_ErrorFromCallable,
 
493
            _run_with_write_locked_target, lockable, self.func_that_raises)
 
494
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
495
 
 
496
    def test_callable_succeeds_but_error_during_unlock(self):
 
497
        lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
 
498
        self.assertRaises(_ErrorFromUnlock,
 
499
            _run_with_write_locked_target, lockable, self.func_that_returns_ok)
 
500
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
501
 
 
502
    def test_error_during_unlock_does_not_mask_original_error(self):
 
503
        lockable = _StubLockable(self._calls, unlock_exc=_ErrorFromUnlock())
 
504
        self.assertRaises(_ErrorFromCallable,
 
505
            _run_with_write_locked_target, lockable, self.func_that_raises)
 
506
        self.assertEqual(['lock_write', 'func called', 'unlock'], self._calls)
 
507
 
 
508