← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_lockdir.py

  • Committer: John Arbash Meinel
  • Date: 2010-07-13 07:44:02 UTC
  • mto: This revision was merged to the branch mainline in revision 5342.
  • Revision ID: john@arbash-meinel.com-20100713074402-wp3oh7cyx76fkvmm
Bump trunk to 2.3-dev1

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
1
 
# Copyright (C) 2006, 2007, 2008 Canonical Ltd
 
1
# Copyright (C) 2006-2010 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
191
191
                    "took %f seconds to detect lock contention" % (after - before))
192
192
        finally:
193
193
            lf1.unlock()
194
 
        lock_base = lf2.transport.abspath(lf2.path)
195
194
        self.assertEqual(1, len(self._logged_reports))
196
 
        lock_url = lf2.transport.abspath(lf2.path)
197
 
        self.assertEqual('%s %s\n'
198
 
                         '%s\n%s\n'
199
 
                         'Will continue to try until %s, unless '
200
 
                         'you press Ctrl-C\n'
201
 
                         'If you\'re sure that it\'s not being '
202
 
                         'modified, use bzr break-lock %s',
203
 
                         self._logged_reports[0][0])
204
 
        args = self._logged_reports[0][1]
205
 
        self.assertEqual('Unable to obtain', args[0])
206
 
        self.assertEqual('lock %s' % (lock_base,), args[1])
207
 
        self.assertStartsWith(args[2], 'held by ')
208
 
        self.assertStartsWith(args[3], 'locked ')
209
 
        self.assertEndsWith(args[3], ' ago')
210
 
        self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
 
195
        self.assertEqual(self._logged_reports[0][0],
 
196
            '%s lock %s held by %s\n'
 
197
            'at %s [process #%s], acquired %s.\n'
 
198
            'Will continue to try until %s, unless '
 
199
            'you press Ctrl-C.\n'
 
200
            'See "bzr help break-lock" for more.')
 
201
        start, lock_url, user, hostname, pid, time_ago, deadline_str = \
 
202
            self._logged_reports[0][1]
 
203
        self.assertEqual(start, u'Unable to obtain')
 
204
        self.assertEqual(user, u'jrandom@example.com')
 
205
        # skip hostname
 
206
        self.assertContainsRe(pid, r'\d+')
 
207
        self.assertContainsRe(time_ago, r'.* ago')
 
208
        self.assertContainsRe(deadline_str, r'\d{2}:\d{2}:\d{2}')
211
209
 
212
210
    def test_31_lock_wait_easy(self):
213
211
        """Succeed when waiting on a lock with no contention.
420
418
        self.assertEqual('%s %s\n'
421
419
                         '%s\n%s\n'
422
420
                         'Will continue to try until %s, unless '
423
 
                         'you press Ctrl-C\n'
424
 
                         'If you\'re sure that it\'s not being '
425
 
                         'modified, use bzr break-lock %s',
 
421
                         'you press Ctrl-C.\n'
 
422
                         'See "bzr help break-lock" for more.',
426
423
                         self._logged_reports[0][0])
427
424
        args = self._logged_reports[0][1]
428
425
        self.assertEqual('Unable to obtain', args[0])
435
432
        self.assertEqual('%s %s\n'
436
433
                         '%s\n%s\n'
437
434
                         'Will continue to try until %s, unless '
438
 
                         'you press Ctrl-C\n'
439
 
                         'If you\'re sure that it\'s not being '
440
 
                         'modified, use bzr break-lock %s',
 
435
                         'you press Ctrl-C.\n'
 
436
                         'See "bzr help break-lock" for more.',
441
437
                         self._logged_reports[1][0])
442
438
        args = self._logged_reports[1][1]
443
439
        self.assertEqual('Lock owner changed for', args[0])
600
596
            info_list = ld1._format_lock_info(ld1.peek())
601
597
        finally:
602
598
            ld1.unlock()
603
 
        self.assertEqual('lock %s' % (ld1.transport.abspath(ld1.path),),
604
 
                         info_list[0])
605
 
        self.assertContainsRe(info_list[1],
606
 
                              r'^held by .* on host .* \[process #\d*\]$')
607
 
        self.assertContainsRe(info_list[2], r'locked \d+ seconds? ago$')
 
599
        self.assertEqual(info_list[0], u'jrandom@example.com')
 
600
        # info_list[1] is hostname. we skip this.
 
601
        self.assertContainsRe(info_list[2], '^\d+$') # pid
 
602
        self.assertContainsRe(info_list[3], r'^\d+ seconds? ago$') # time_ago
608
603
 
609
604
    def test_lock_without_email(self):
610
605
        global_config = config.GlobalConfig()
669
664
        # no kibble
670
665
        check_dir(['held'])
671
666
 
 
667
    def test_no_lockdir_info(self):
 
668
        """We can cope with empty info files."""
 
669
        # This seems like a fairly common failure case - see
 
670
        # <https://bugs.launchpad.net/bzr/+bug/185103> and all its dupes.
 
671
        # Processes are often interrupted after opening the file
 
672
        # before the actual contents are committed.
 
673
        t = self.get_transport()
 
674
        t.mkdir('test_lock')
 
675
        t.mkdir('test_lock/held')
 
676
        t.put_bytes('test_lock/held/info', '')
 
677
        lf = LockDir(t, 'test_lock')
 
678
        info = lf.peek()
 
679
        formatted_info = lf._format_lock_info(info)
 
680
        self.assertEquals(
 
681
            ['<unknown>', '<unknown>', '<unknown>', '(unknown)'],
 
682
            formatted_info)
 
683
 
672
684
 
673
685
class TestLockDirHooks(TestCaseWithTransport):
674
686