191
191
"took %f seconds to detect lock contention" % (after - before))
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'
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')
195
self.assertEqual(self._logged_reports[0][0],
196
'%s lock 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, 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')
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
210
def test_31_lock_wait_easy(self):
212
211
"""Succeed when waiting on a lock with no contention.
597
596
info_list = ld1._format_lock_info(ld1.peek())
600
self.assertEqual('lock %s' % (ld1.transport.abspath(ld1.path),),
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$')
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
606
604
def test_lock_without_email(self):
607
605
global_config = config.GlobalConfig()
681
679
formatted_info = lf._format_lock_info(info)
682
680
self.assertEquals(
683
['lock %s' % t.abspath('test_lock'),
684
'held by <unknown> on host <unknown> [process #<unknown>]',
681
['<unknown>', '<unknown>', '<unknown>', '(unknown)'],