/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/test_lockdir.py

Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
1
1
# Copyright (C) 2006 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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
 
 
7
#
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
12
 
 
 
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for LockDir"""
18
18
 
 
19
from cStringIO import StringIO
19
20
from threading import Thread
20
21
import time
21
22
 
 
23
import bzrlib
22
24
from bzrlib.errors import (
23
25
        LockBreakMismatch,
24
26
        LockContention, LockError, UnlockableTransport,
56
58
        lf = LockDir(self.get_transport(), 'test_lock')
57
59
        self.assertEqual(lf.peek(), None)
58
60
 
 
61
    def get_lock(self):
 
62
        return LockDir(self.get_transport(), 'test_lock')
 
63
 
 
64
    def test_unlock_after_break_raises(self):
 
65
        ld = self.get_lock()
 
66
        ld2 = self.get_lock()
 
67
        ld.create()
 
68
        ld.attempt_lock()
 
69
        ld2.force_break(ld2.peek())
 
70
        self.assertRaises(LockBroken, ld.unlock)
 
71
 
59
72
    def test_03_readonly_peek(self):
60
73
        lf = LockDir(self.get_readonly_transport(), 'test_lock')
61
74
        self.assertEqual(lf.peek(), None)
149
162
            self.assertRaises(LockContention, lf2.wait_lock,
150
163
                              timeout=0.4, poll=0.1)
151
164
            after = time.time()
152
 
            self.assertTrue(after - before <= 1.0)
 
165
            # it should only take about 0.4 seconds, but we allow more time in
 
166
            # case the machine is heavily loaded
 
167
            self.assertTrue(after - before <= 8.0, 
 
168
                    "took %f seconds to detect lock contention" % (after - before))
153
169
        finally:
154
170
            lf1.unlock()
155
171
 
318
334
        self.assertTrue(t.has('test_lock/held/info'))
319
335
        lf1.unlock()
320
336
        self.assertFalse(t.has('test_lock/held/info'))
 
337
 
 
338
    def test_break_lock(self):
 
339
        # the ui based break_lock routine should Just Work (tm)
 
340
        ld1 = self.get_lock()
 
341
        ld2 = self.get_lock()
 
342
        ld1.create()
 
343
        ld1.lock_write()
 
344
        # do this without IO redirection to ensure it doesn't prompt.
 
345
        self.assertRaises(AssertionError, ld1.break_lock)
 
346
        orig_factory = bzrlib.ui.ui_factory
 
347
        # silent ui - no need for stdout
 
348
        bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
 
349
        bzrlib.ui.ui_factory.stdin = StringIO("y\n")
 
350
        try:
 
351
            ld2.break_lock()
 
352
            self.assertRaises(LockBroken, ld1.unlock)
 
353
        finally:
 
354
            bzrlib.ui.ui_factory = orig_factory