/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

  • Committer: Richard Wilbur
  • Date: 2016-02-04 19:07:28 UTC
  • mto: This revision was merged to the branch mainline in revision 6618.
  • Revision ID: richard.wilbur@gmail.com-20160204190728-p0zvfii6zase0fw7
Update COPYING.txt from the original http://www.gnu.org/licenses/gpl-2.0.txt  (Only differences were in whitespace.)  Thanks to Petr Stodulka for pointing out the discrepancy.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
1
 
# Copyright (C) 2006-2012, 2016 Canonical Ltd
 
1
# Copyright (C) 2006-2011 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
19
19
import os
20
20
import time
21
21
 
22
 
import breezy
23
 
from .. import (
 
22
import bzrlib
 
23
from bzrlib import (
24
24
    config,
25
25
    errors,
26
26
    lock,
29
29
    tests,
30
30
    transport,
31
31
    )
32
 
from ..errors import (
 
32
from bzrlib.errors import (
33
33
    LockBreakMismatch,
34
34
    LockBroken,
35
35
    LockContention,
36
36
    LockFailed,
37
37
    LockNotHeld,
38
38
    )
39
 
from ..lockdir import (
 
39
from bzrlib.lockdir import (
40
40
    LockDir,
41
41
    LockHeldInfo,
42
42
    )
43
 
from ..sixish import (
44
 
    text_type,
45
 
    )
46
 
from . import (
 
43
from bzrlib.tests import (
47
44
    features,
48
45
    TestCase,
49
46
    TestCaseInTempDir,
135
132
            # a single process are not detected
136
133
            lf2.attempt_lock()
137
134
            self.fail('Failed to detect lock collision')
138
 
        except LockContention as e:
 
135
        except LockContention, e:
139
136
            self.assertEqual(e.lock, lf2)
140
137
            self.assertContainsRe(str(e),
141
138
                    r'^Could not acquire.*test_lock.*$')
151
148
        # lock is held, should get some info on it
152
149
        info1 = lf1.peek()
153
150
        self.assertEqual(set(info1.info_dict.keys()),
154
 
            {'user', 'nonce', 'hostname', 'pid', 'start_time'})
 
151
            set(['user', 'nonce', 'hostname', 'pid', 'start_time']))
155
152
        # should get the same info if we look at it through a different
156
153
        # instance
157
154
        info2 = LockDir(t, 'test_lock').peek()
202
199
            r' \(process #\d+\), acquired .* ago\.\n'
203
200
            r'Will continue to try until \d{2}:\d{2}:\d{2}, unless '
204
201
            r'you press Ctrl-C.\n'
205
 
            r'See "brz help break-lock" for more.')
 
202
            r'See "bzr help break-lock" for more.')
206
203
 
207
204
    def test_31_lock_wait_easy(self):
208
205
        """Succeed when waiting on a lock with no contention.
333
330
        ld1.lock_write()
334
331
        # do this without IO redirection to ensure it doesn't prompt.
335
332
        self.assertRaises(AssertionError, ld1.break_lock)
336
 
        orig_factory = breezy.ui.ui_factory
337
 
        breezy.ui.ui_factory = breezy.ui.CannedInputUIFactory([True])
 
333
        orig_factory = bzrlib.ui.ui_factory
 
334
        bzrlib.ui.ui_factory = bzrlib.ui.CannedInputUIFactory([True])
338
335
        try:
339
336
            ld2.break_lock()
340
337
            self.assertRaises(LockBroken, ld1.unlock)
341
338
        finally:
342
 
            breezy.ui.ui_factory = orig_factory
 
339
            bzrlib.ui.ui_factory = orig_factory
343
340
 
344
341
    def test_break_lock_corrupt_info(self):
345
342
        """break_lock works even if the info file is corrupt (and tells the UI
349
346
        ld2 = self.get_lock()
350
347
        ld.create()
351
348
        ld.lock_write()
352
 
        ld.transport.put_bytes_non_atomic('test_lock/held/info', b'\0')
 
349
        ld.transport.put_bytes_non_atomic('test_lock/held/info', '\0')
353
350
 
354
 
        class LoggingUIFactory(breezy.ui.SilentUIFactory):
 
351
        class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
355
352
            def __init__(self):
356
353
                self.prompts = []
357
354
 
360
357
                return True
361
358
 
362
359
        ui = LoggingUIFactory()
363
 
        self.overrideAttr(breezy.ui, 'ui_factory', ui)
 
360
        self.overrideAttr(bzrlib.ui, 'ui_factory', ui)
364
361
        ld2.break_lock()
365
362
        self.assertLength(1, ui.prompts)
366
363
        self.assertEqual('boolean', ui.prompts[0][0])
377
374
        ld.lock_write()
378
375
        ld.transport.delete('test_lock/held/info')
379
376
 
380
 
        class LoggingUIFactory(breezy.ui.SilentUIFactory):
 
377
        class LoggingUIFactory(bzrlib.ui.SilentUIFactory):
381
378
            def __init__(self):
382
379
                self.prompts = []
383
380
 
386
383
                return True
387
384
 
388
385
        ui = LoggingUIFactory()
389
 
        orig_factory = breezy.ui.ui_factory
390
 
        breezy.ui.ui_factory = ui
 
386
        orig_factory = bzrlib.ui.ui_factory
 
387
        bzrlib.ui.ui_factory = ui
391
388
        try:
392
389
            ld2.break_lock()
393
390
            self.assertRaises(LockBroken, ld.unlock)
394
391
            self.assertLength(0, ui.prompts)
395
392
        finally:
396
 
            breezy.ui.ui_factory = orig_factory
 
393
            bzrlib.ui.ui_factory = orig_factory
397
394
        # Suppress warnings due to ld not being unlocked
398
395
        # XXX: if lock_broken hook was invoked in this case, this hack would
399
396
        # not be necessary.  - Andrew Bennetts, 2010-09-06.
432
429
        finally:
433
430
            ld1.unlock()
434
431
        self.assertEqual(info_list['user'], u'jrandom@example.com')
435
 
        self.assertContainsRe(info_list['pid'], '^\\d+$')
436
 
        self.assertContainsRe(info_list['time_ago'], '^\\d+ seconds? ago$')
 
432
        self.assertContainsRe(info_list['pid'], '^\d+$')
 
433
        self.assertContainsRe(info_list['time_ago'], r'^\d+ seconds? ago$')
437
434
 
438
435
    def test_lock_without_email(self):
439
436
        global_config = config.GlobalStack()
478
475
        # now the original caller should succeed in unlocking
479
476
        ld1.unlock()
480
477
        # and there should be nothing left over
481
 
        self.assertEqual([], t.list_dir('test_lock'))
 
478
        self.assertEquals([], t.list_dir('test_lock'))
482
479
 
483
480
    def test_failed_lock_leaves_no_trash(self):
484
481
        # if we fail to acquire the lock, we don't leave pending directories
490
487
        t = self.get_transport().clone('test_lock')
491
488
 
492
489
        def check_dir(a):
493
 
            self.assertEqual(a, t.list_dir('.'))
 
490
            self.assertEquals(a, t.list_dir('.'))
494
491
 
495
492
        check_dir([])
496
493
        # when held, that's all we see
511
508
        t = self.get_transport()
512
509
        t.mkdir('test_lock')
513
510
        t.mkdir('test_lock/held')
514
 
        t.put_bytes('test_lock/held/info', b'')
 
511
        t.put_bytes('test_lock/held/info', '')
515
512
        lf = LockDir(t, 'test_lock')
516
513
        info = lf.peek()
517
514
        formatted_info = info.to_readable_dict()
518
 
        self.assertEqual(
 
515
        self.assertEquals(
519
516
            dict(user='<unknown>', hostname='<unknown>', pid='<unknown>',
520
517
                time_ago='(unknown)'),
521
518
            formatted_info)
529
526
        t = self.get_transport()
530
527
        t.mkdir('test_lock')
531
528
        t.mkdir('test_lock/held')
532
 
        t.put_bytes('test_lock/held/info', b'\0')
 
529
        t.put_bytes('test_lock/held/info', '\0')
533
530
        lf = LockDir(t, 'test_lock')
534
531
        self.assertRaises(errors.LockCorrupt, lf.peek)
535
532
        # Currently attempt_lock gives LockContention, but LockCorrupt would be
668
665
 
669
666
    def test_unicode(self):
670
667
        info = LockHeldInfo.for_this_process(None)
671
 
        self.assertContainsRe(text_type(info),
 
668
        self.assertContainsRe(unicode(info),
672
669
            r'held by .* on .* \(process #\d+\), acquired .* ago')
673
670
 
674
671
    def test_is_locked_by_this_process(self):