← 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

first cut at merge from integration.

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
 
1
# Copyright (C) 2005 by Canonical Ltd
2
2
 
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
16
16
 
17
17
from StringIO import StringIO
18
18
 
19
 
import bzrlib
20
19
from bzrlib.branch import Branch
21
 
import bzrlib.errors as errors
22
20
from bzrlib.errors import BzrBadParameterNotString, NoSuchFile, ReadOnlyError
23
 
from bzrlib.lockable_files import LockableFiles, TransportLock
24
 
from bzrlib.lockdir import LockDir
 
21
from bzrlib.lockable_files import LockableFiles
25
22
from bzrlib.tests import TestCaseInTempDir
26
 
from bzrlib.tests.test_transactions import DummyWeave
27
 
from bzrlib.transactions import (PassThroughTransaction,
28
 
                                 ReadOnlyTransaction,
29
 
                                 WriteTransaction,
30
 
                                 )
 
23
from bzrlib.transactions import PassThroughTransaction, ReadOnlyTransaction
31
24
from bzrlib.transport import get_transport
32
25
 
 
26
class TestLockableFiles(TestCaseInTempDir):
33
27
 
34
 
# these tests are applied in each parameterized suite for LockableFiles
35
 
class _TestLockableFiles_mixin(object):
 
28
    def setUp(self):
 
29
        super(TestLockableFiles, self).setUp()
 
30
        transport = get_transport('.')
 
31
        transport.mkdir('.bzr')
 
32
        transport.put('.bzr/my-lock', StringIO(''))
 
33
        self.lockable = LockableFiles(transport.clone('.bzr'), 'my-lock')
36
34
 
37
35
    def test_read_write(self):
38
36
        self.assertRaises(NoSuchFile, self.lockable.get, 'foo')
84
82
                      PassThroughTransaction)
85
83
        self.lockable.lock_write()
86
84
        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)
 
85
                      PassThroughTransaction)
91
86
        self.lockable.unlock()
92
 
        self.assertTrue(vf.finished)
93
87
 
94
88
    def test__escape(self):
95
89
        self.assertEqual('%25', self.lockable._escape('%'))
97
91
    def test__escape_empty(self):
98
92
        self.assertEqual('', self.lockable._escape(''))
99
93
 
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.