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