/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/workingtree_implementations/test_locking.py

  • Committer: Jan Balster
  • Date: 2006-08-15 12:39:42 UTC
  • mfrom: (1923 +trunk)
  • mto: This revision was merged to the branch mainline in revision 1928.
  • Revision ID: jan@merlinux.de-20060815123942-22c388c6e9a8ac91
merge bzr.dev 1923

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2006 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
"""Tests for the (un)lock interfaces on all working tree implemenations."""
 
18
 
 
19
import bzrlib.branch as branch
 
20
import bzrlib.errors as errors
 
21
from bzrlib.tests.workingtree_implementations import TestCaseWithWorkingTree
 
22
 
 
23
 
 
24
class TestWorkingTreeLocking(TestCaseWithWorkingTree):
 
25
 
 
26
    def test_trivial_lock_read_unlock(self):
 
27
        """Locking and unlocking should work trivially."""
 
28
        wt = self.make_branch_and_tree('.')
 
29
 
 
30
        self.assertFalse(wt.is_locked())
 
31
        self.assertFalse(wt.branch.is_locked())
 
32
        wt.lock_read()
 
33
        try:
 
34
            self.assertTrue(wt.is_locked())
 
35
            self.assertTrue(wt.branch.is_locked())
 
36
        finally:
 
37
            wt.unlock()
 
38
        self.assertFalse(wt.is_locked())
 
39
        self.assertFalse(wt.branch.is_locked())
 
40
 
 
41
    def test_trivial_lock_write_unlock(self):
 
42
        """Locking for write and unlocking should work trivially."""
 
43
        wt = self.make_branch_and_tree('.')
 
44
 
 
45
        self.assertFalse(wt.is_locked())
 
46
        self.assertFalse(wt.branch.is_locked())
 
47
        wt.lock_write()
 
48
        try:
 
49
            self.assertTrue(wt.is_locked())
 
50
            self.assertTrue(wt.branch.is_locked())
 
51
        finally:
 
52
            wt.unlock()
 
53
        self.assertFalse(wt.is_locked())
 
54
        self.assertFalse(wt.branch.is_locked())
 
55
        
 
56
    def test_unlock_branch_failures(self):
 
57
        """If the branch unlock fails the tree must still unlock."""
 
58
        # The public interface for WorkingTree requires a branch, but
 
59
        # does not require that the working tree use the branch - its
 
60
        # implementation specific how the WorkingTree, Branch, and Repository
 
61
        # hang together.
 
62
        # in order to test that implementations which *do* unlock via the branch
 
63
        # do so correctly, we unlock the branch after locking the working tree.
 
64
        # The next unlock on working tree should trigger a LockNotHeld exception
 
65
        # from the branch object, which must be exposed to the caller. To meet 
 
66
        # our object model - where locking a tree locks its branch, and
 
67
        # unlocking a branch does not unlock a working tree, *even* for 
 
68
        # all-in-one implementations like bzr 0.6, git, and hg, implementations
 
69
        # must have some separate counter for each object, so our explicit 
 
70
        # unlock should trigger some error on all implementations, and 
 
71
        # requiring that to be LockNotHeld seems reasonable.
 
72
        #
 
73
        # we use this approach rather than decorating the Branch, because the
 
74
        # public interface of WorkingTree does not permit altering the branch
 
75
        # object - and we cannot tell which attribute might allow us to 
 
76
        # backdoor-in and change it reliably. For implementation specific tests
 
77
        # we can do such skullduggery, but not for interface specific tests.
 
78
        # And, its simpler :)
 
79
        wt = self.make_branch_and_tree('.')
 
80
 
 
81
        self.assertFalse(wt.is_locked())
 
82
        self.assertFalse(wt.branch.is_locked())
 
83
        wt.lock_write()
 
84
        self.assertTrue(wt.is_locked())
 
85
        self.assertTrue(wt.branch.is_locked())
 
86
 
 
87
        # manually unlock the branch, preparing a LockNotHeld error.
 
88
        wt.branch.unlock()
 
89
        # the branch *may* still be locked here, if its an all-in-one
 
90
        # implementation because there is a single lock object with three
 
91
        # references on it, and unlocking the branch only drops this by two
 
92
        self.assertRaises(errors.LockNotHeld, wt.unlock)
 
93
        # but now, the tree must be unlocked
 
94
        self.assertFalse(wt.is_locked())
 
95
        # and the branch too.
 
96
        self.assertFalse(wt.branch.is_locked())
 
97
 
 
98
    def test_failing_to_lock_branch_does_not_lock(self):
 
99
        """If the branch cannot be locked, dont lock the tree."""
 
100
        # Many implementations treat read-locks as non-blocking, but some
 
101
        # treat them as blocking with writes.. Accordingly we test this by
 
102
        # opening the branch twice, and locking the branch for write in the
 
103
        # second instance.  Our lock contract requires separate instances to
 
104
        # mutually exclude if a lock is exclusive at all: If we get no error
 
105
        # locking, the test still passes.
 
106
        wt = self.make_branch_and_tree('.')
 
107
        branch_copy = branch.Branch.open('.')
 
108
        branch_copy.lock_write()
 
109
        try:
 
110
            try:
 
111
                wt.lock_read()
 
112
            except errors.LockError:
 
113
                # any error here means the locks are exclusive in some 
 
114
                # manner
 
115
                self.assertFalse(wt.is_locked())
 
116
                self.assertFalse(wt.branch.is_locked())
 
117
                return
 
118
            else:
 
119
                # no error - the branch allows read locks while writes
 
120
                # are taken, just pass.
 
121
                wt.unlock()
 
122
        finally:
 
123
            branch_copy.unlock()
 
124
 
 
125
    def test_failing_to_lock_write_branch_does_not_lock(self):
 
126
        """If the branch cannot be write locked, dont lock the tree."""
 
127
        # all implementations of branch are required to treat write 
 
128
        # locks as blocking (compare to repositories which are not required
 
129
        # to do so).
 
130
        # Accordingly we test this by opening the branch twice, and locking the
 
131
        # branch for write in the second instance.  Our lock contract requires
 
132
        # separate instances to mutually exclude.
 
133
        wt = self.make_branch_and_tree('.')
 
134
        branch_copy = branch.Branch.open('.')
 
135
        branch_copy.lock_write()
 
136
        try:
 
137
            try:
 
138
                self.assertRaises(errors.LockError, wt.lock_write)
 
139
                self.assertFalse(wt.is_locked())
 
140
                self.assertFalse(wt.branch.is_locked())
 
141
            finally:
 
142
                if wt.is_locked():
 
143
                    wt.unlock()
 
144
        finally:
 
145
            branch_copy.unlock()