← 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: Martin Pool
  • Date: 2010-01-29 14:09:05 UTC
  • mto: This revision was merged to the branch mainline in revision 4992.
  • Revision ID: mbp@sourcefrog.net-20100129140905-2uiarb6p8di1ywsr
Correction to url

from review: https://code.edge.launchpad.net/~mbp/bzr/doc/+merge/18250

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006, 2007, 2008, 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)
194
195
        self.assertEqual(1, len(self._logged_reports))
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}')
 
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
                         'See "bzr help break-lock" for more.',
 
202
                         self._logged_reports[0][0])
 
203
        args = self._logged_reports[0][1]
 
204
        self.assertEqual('Unable to obtain', args[0])
 
205
        self.assertEqual('lock %s' % (lock_base,), args[1])
 
206
        self.assertStartsWith(args[2], 'held by ')
 
207
        self.assertStartsWith(args[3], 'locked ')
 
208
        self.assertEndsWith(args[3], ' ago')
 
209
        self.assertContainsRe(args[4], r'\d\d:\d\d:\d\d')
209
210
 
210
211
    def test_31_lock_wait_easy(self):
211
212
        """Succeed when waiting on a lock with no contention.
596
597
            info_list = ld1._format_lock_info(ld1.peek())
597
598
        finally:
598
599
            ld1.unlock()
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
 
600
        self.assertEqual('lock %s' % (ld1.transport.abspath(ld1.path),),
 
601
                         info_list[0])
 
602
        self.assertContainsRe(info_list[1],
 
603
                              r'^held by .* on host .* \[process #\d*\]$')
 
604
        self.assertContainsRe(info_list[2], r'locked \d+ seconds? ago$')
603
605
 
604
606
    def test_lock_without_email(self):
605
607
        global_config = config.GlobalConfig()
664
666
        # no kibble
665
667
        check_dir(['held'])
666
668
 
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
 
 
684
669
 
685
670
class TestLockDirHooks(TestCaseWithTransport):
686
671