← 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: Martin Pool
  • Date: 2006-06-15 05:36:34 UTC
  • mto: This revision was merged to the branch mainline in revision 1797.
  • Revision ID: mbp@sourcefrog.net-20060615053634-4fd52ba691855659
Clean up many exception classes.

Errors indicating a user error are now shown with is_user_error on the
exception; use this rather than hardcoding a list of exceptions that should be
handled this way.

Exceptions now inherit from BzrNewException where possible to use consistent
formatting method.

Remove rather obsolete docstring test on Branch.missing_revisions.

Remove dead code from find_merge_base.


Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockable_files.py
 
1
# Copyright (C) 2005, 2006 by 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
        finally:
 
64
            self.lockable.unlock()
 
65
 
 
66
    def test_locks(self):
 
67
        self.lockable.lock_read()
 
68
        try:
 
69
            self.assertRaises(ReadOnlyError, self.lockable.put, 'foo', 
 
70
                              StringIO('bar\u1234'))
 
71
        finally:
 
72
            self.lockable.unlock()
 
73
 
 
74
    def test_transactions(self):
 
75
        self.assertIs(self.lockable.get_transaction().__class__,
 
76
                      PassThroughTransaction)
 
77
        self.lockable.lock_read()
 
78
        try:
 
79
            self.assertIs(self.lockable.get_transaction().__class__,
 
80
                          ReadOnlyTransaction)
 
81
        finally:
 
82
            self.lockable.unlock()
 
83
        self.assertIs(self.lockable.get_transaction().__class__,
 
84
                      PassThroughTransaction)
 
85
        self.lockable.lock_write()
 
86
        self.assertIs(self.lockable.get_transaction().__class__,
 
87
                      WriteTransaction)
 
88
        # check that finish is called:
 
89
        vf = DummyWeave('a')
 
90
        self.lockable.get_transaction().register_dirty(vf)
 
91
        self.lockable.unlock()
 
92
        self.assertTrue(vf.finished)
 
93
 
 
94
    def test__escape(self):
 
95
        self.assertEqual('%25', self.lockable._escape('%'))
 
96
        
 
97
    def test__escape_empty(self):
 
98
        self.assertEqual('', self.lockable._escape(''))
 
99
 
 
100
    def test_break_lock(self):
 
101
        # some locks are not breakable
 
102
        self.lockable.lock_write()
 
103
        try:
 
104
            self.assertRaises(AssertionError, self.lockable.break_lock)
 
105
        except NotImplementedError:
 
106
            # this lock cannot be broken
 
107
            self.lockable.unlock()
 
108
            return
 
109
        l2 = self.get_lockable()
 
110
        orig_factory = bzrlib.ui.ui_factory
 
111
        # silent ui - no need for stdout
 
112
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
 
113
        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
 
114
        try:
 
115
            l2.break_lock()
 
116
        finally:
 
117
            bzrlib.ui.ui_factory = orig_factory
 
118
        try:
 
119
            l2.lock_write()
 
120
            l2.unlock()
 
121
        finally:
 
122
            self.assertRaises(errors.LockBroken, self.lockable.unlock)
 
123
            self.assertFalse(self.lockable.is_locked())
 
124
 
 
125
 
 
126
# This method of adapting tests to parameters is different to 
 
127
# the TestProviderAdapters used elsewhere, but seems simpler for this 
 
128
# case.  
 
129
class TestLockableFiles_TransportLock(TestCaseInTempDir,
 
130
                                      _TestLockableFiles_mixin):
 
131
 
 
132
    def setUp(self):
 
133
        super(TestLockableFiles_TransportLock, self).setUp()
 
134
        transport = get_transport('.')
 
135
        transport.mkdir('.bzr')
 
136
        self.sub_transport = transport.clone('.bzr')
 
137
        self.lockable = self.get_lockable()
 
138
        self.lockable.create_lock()
 
139
 
 
140
    def tearDown(self):
 
141
        super(TestLockableFiles_TransportLock, self).tearDown()
 
142
        # free the subtransport so that we do not get a 5 second
 
143
        # timeout due to the SFTP connection cache.
 
144
        del self.sub_transport
 
145
 
 
146
    def get_lockable(self):
 
147
        return LockableFiles(self.sub_transport, 'my-lock', TransportLock)
 
148
        
 
149
 
 
150
class TestLockableFiles_LockDir(TestCaseInTempDir,
 
151
                              _TestLockableFiles_mixin):
 
152
    """LockableFile tests run with LockDir underneath"""
 
153
 
 
154
    def setUp(self):
 
155
        super(TestLockableFiles_LockDir, self).setUp()
 
156
        self.transport = get_transport('.')
 
157
        self.lockable = self.get_lockable()
 
158
        # the lock creation here sets mode - test_permissions on branch 
 
159
        # tests that implicitly, but it might be a good idea to factor 
 
160
        # out the mode checking logic and have it applied to loackable files
 
161
        # directly. RBC 20060418
 
162
        self.lockable.create_lock()
 
163
 
 
164
    def get_lockable(self):
 
165
        return LockableFiles(self.transport, 'my-lock', LockDir)
 
166
 
 
167
    def test_lock_created(self):
 
168
        self.assertTrue(self.transport.has('my-lock'))
 
169
        self.lockable.lock_write()
 
170
        self.assertTrue(self.transport.has('my-lock/held/info'))
 
171
        self.lockable.unlock()
 
172
        self.assertFalse(self.transport.has('my-lock/held/info'))
 
173
        self.assertTrue(self.transport.has('my-lock'))
 
174
 
 
175
 
 
176
    # TODO: Test the lockdir inherits the right file and directory permissions
 
177
    # from the LockableFiles.