← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockable_files.py

  • Committer: Robert Collins
  • Date: 2005-10-06 22:15:52 UTC
  • mfrom: (1185.13.2)
  • mto: This revision was merged to the branch mainline in revision 1420.
  • Revision ID: robertc@robertcollins.net-20051006221552-9b15c96fa504e0ad
merge from upstream

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockable_files.py
1
 
# Copyright (C) 2005, 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
 
from StringIO import StringIO
18
 
 
19
 
import bzrlib
20
 
from bzrlib.branch import Branch
21
 
import bzrlib.errors as errors
22
 
from bzrlib.errors import BzrBadParameterNotString, NoSuchFile, ReadOnlyError
23
 
from bzrlib.lockable_files import LockableFiles, TransportLock
24
 
from bzrlib.lockdir import LockDir
25
 
from bzrlib.tests import TestCaseInTempDir
26
 
from bzrlib.tests.test_transactions import DummyWeave
27
 
from bzrlib.transactions import (PassThroughTransaction,
28
 
                                 ReadOnlyTransaction,
29
 
                                 WriteTransaction,
30
 
                                 )
31
 
from bzrlib.transport import get_transport
32
 
 
33
 
 
34
 
# these tests are applied in each parameterized suite for LockableFiles
35
 
class _TestLockableFiles_mixin(object):
36
 
 
37
 
    def test_read_write(self):
38
 
        self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
39
 
        self.assertRaises(NoSuchFile, self.lockable.get_utf8, 'foo')
40
 
        self.lockable.lock_write()
41
 
        try:
42
 
            unicode_string = u'bar\u1234'
43
 
            self.assertEqual(4, len(unicode_string))
44
 
            byte_string = unicode_string.encode('utf-8')
45
 
            self.assertEqual(6, len(byte_string))
46
 
            self.assertRaises(UnicodeEncodeError, self.lockable.put, 'foo', 
47
 
                              StringIO(unicode_string))
48
 
            self.lockable.put('foo', StringIO(byte_string))
49
 
            self.assertEqual(byte_string,
50
 
                             self.lockable.get('foo').read())
51
 
            self.assertEqual(unicode_string,
52
 
                             self.lockable.get_utf8('foo').read())
53
 
            self.assertRaises(BzrBadParameterNotString,
54
 
                              self.lockable.put_utf8,
55
 
                              'bar',
56
 
                              StringIO(unicode_string)
57
 
                              )
58
 
            self.lockable.put_utf8('bar', unicode_string)
59
 
            self.assertEqual(unicode_string,
60
 
                             self.lockable.get_utf8('bar').read())
61
 
            self.assertEqual(byte_string,
62
 
                             self.lockable.get('bar').read())
63
 
            self.lockable.put_bytes('raw', 'raw\xffbytes')
64
 
            self.assertEqual('raw\xffbytes',
65
 
                             self.lockable.get('raw').read())
66
 
        finally:
67
 
            self.lockable.unlock()
68
 
 
69
 
    def test_locks(self):
70
 
        self.lockable.lock_read()
71
 
        try:
72
 
            self.assertRaises(ReadOnlyError, self.lockable.put, 'foo', 
73
 
                              StringIO('bar\u1234'))
74
 
        finally:
75
 
            self.lockable.unlock()
76
 
 
77
 
    def test_transactions(self):
78
 
        self.assertIs(self.lockable.get_transaction().__class__,
79
 
                      PassThroughTransaction)
80
 
        self.lockable.lock_read()
81
 
        try:
82
 
            self.assertIs(self.lockable.get_transaction().__class__,
83
 
                          ReadOnlyTransaction)
84
 
        finally:
85
 
            self.lockable.unlock()
86
 
        self.assertIs(self.lockable.get_transaction().__class__,
87
 
                      PassThroughTransaction)
88
 
        self.lockable.lock_write()
89
 
        self.assertIs(self.lockable.get_transaction().__class__,
90
 
                      WriteTransaction)
91
 
        # check that finish is called:
92
 
        vf = DummyWeave('a')
93
 
        self.lockable.get_transaction().register_dirty(vf)
94
 
        self.lockable.unlock()
95
 
        self.assertTrue(vf.finished)
96
 
 
97
 
    def test__escape(self):
98
 
        self.assertEqual('%25', self.lockable._escape('%'))
99
 
        
100
 
    def test__escape_empty(self):
101
 
        self.assertEqual('', self.lockable._escape(''))
102
 
 
103
 
    def test_break_lock(self):
104
 
        # some locks are not breakable
105
 
        self.lockable.lock_write()
106
 
        try:
107
 
            self.assertRaises(AssertionError, self.lockable.break_lock)
108
 
        except NotImplementedError:
109
 
            # this lock cannot be broken
110
 
            self.lockable.unlock()
111
 
            return
112
 
        l2 = self.get_lockable()
113
 
        orig_factory = bzrlib.ui.ui_factory
114
 
        # silent ui - no need for stdout
115
 
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
116
 
        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
117
 
        try:
118
 
            l2.break_lock()
119
 
        finally:
120
 
            bzrlib.ui.ui_factory = orig_factory
121
 
        try:
122
 
            l2.lock_write()
123
 
            l2.unlock()
124
 
        finally:
125
 
            self.assertRaises(errors.LockBroken, self.lockable.unlock)
126
 
            self.assertFalse(self.lockable.is_locked())
127
 
 
128
 
 
129
 
# This method of adapting tests to parameters is different to 
130
 
# the TestProviderAdapters used elsewhere, but seems simpler for this 
131
 
# case.  
132
 
class TestLockableFiles_TransportLock(TestCaseInTempDir,
133
 
                                      _TestLockableFiles_mixin):
134
 
 
135
 
    def setUp(self):
136
 
        super(TestLockableFiles_TransportLock, self).setUp()
137
 
        transport = get_transport('.')
138
 
        transport.mkdir('.bzr')
139
 
        self.sub_transport = transport.clone('.bzr')
140
 
        self.lockable = self.get_lockable()
141
 
        self.lockable.create_lock()
142
 
 
143
 
    def tearDown(self):
144
 
        super(TestLockableFiles_TransportLock, self).tearDown()
145
 
        # free the subtransport so that we do not get a 5 second
146
 
        # timeout due to the SFTP connection cache.
147
 
        del self.sub_transport
148
 
 
149
 
    def get_lockable(self):
150
 
        return LockableFiles(self.sub_transport, 'my-lock', TransportLock)
151
 
        
152
 
 
153
 
class TestLockableFiles_LockDir(TestCaseInTempDir,
154
 
                              _TestLockableFiles_mixin):
155
 
    """LockableFile tests run with LockDir underneath"""
156
 
 
157
 
    def setUp(self):
158
 
        super(TestLockableFiles_LockDir, self).setUp()
159
 
        self.transport = get_transport('.')
160
 
        self.lockable = self.get_lockable()
161
 
        # the lock creation here sets mode - test_permissions on branch 
162
 
        # tests that implicitly, but it might be a good idea to factor 
163
 
        # out the mode checking logic and have it applied to loackable files
164
 
        # directly. RBC 20060418
165
 
        self.lockable.create_lock()
166
 
 
167
 
    def get_lockable(self):
168
 
        return LockableFiles(self.transport, 'my-lock', LockDir)
169
 
 
170
 
    def test_lock_created(self):
171
 
        self.assertTrue(self.transport.has('my-lock'))
172
 
        self.lockable.lock_write()
173
 
        self.assertTrue(self.transport.has('my-lock/held/info'))
174
 
        self.lockable.unlock()
175
 
        self.assertFalse(self.transport.has('my-lock/held/info'))
176
 
        self.assertTrue(self.transport.has('my-lock'))
177
 
 
178
 
 
179
 
    # TODO: Test the lockdir inherits the right file and directory permissions
180
 
    # from the LockableFiles.