← 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: 2006-09-20 17:01:14 UTC
  • mto: This revision was merged to the branch mainline in revision 2050.
  • Revision ID: john@arbash-meinel.com-20060920170114-20251f6bb2e7b7f7
Change default timeouts, and report differently the first failure

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_lockdir.py
182
182
            lf1.unlock()
183
183
        lock_base = lf2.transport.abspath(lf2.path)
184
184
        self.assertEqual(1, len(self._logged_reports))
185
 
        self.assertEqual('Unable to obtain %s\n'
 
185
        self.assertEqual('%s %s\n'
186
186
                         '%s\n%s\n'
187
187
                         'Will continue to try for %s seconds\n',
188
188
                         self._logged_reports[0][0])
189
189
        args = self._logged_reports[0][1]
190
 
        self.assertEqual('lock %s' % (lock_base,), args[0])
191
 
        self.assertStartsWith(args[1], 'held by ')
192
 
        self.assertStartsWith(args[2], 'locked ')
193
 
        self.assertEndsWith(args[2], ' ago')
 
190
        self.assertEqual('Unable to obtain', args[0])
 
191
        self.assertEqual('lock %s' % (lock_base,), args[1])
 
192
        self.assertStartsWith(args[2], 'held by ')
 
193
        self.assertStartsWith(args[3], 'locked ')
 
194
        self.assertEndsWith(args[3], ' ago')
194
195
 
195
196
    def test_31_lock_wait_easy(self):
196
197
        """Succeed when waiting on a lock with no contention.
239
240
        # wait for a while
240
241
        lock_base = lf2.transport.abspath(lf2.path)
241
242
        self.assertEqual(1, len(self._logged_reports))
242
 
        self.assertEqual('Unable to obtain %s\n'
 
243
        self.assertEqual('%s %s\n'
243
244
                         '%s\n%s\n'
244
245
                         'Will continue to try for %s seconds\n',
245
246
                         self._logged_reports[0][0])
246
247
        args = self._logged_reports[0][1]
247
 
        self.assertEqual('lock %s' % (lock_base,), args[0])
248
 
        self.assertStartsWith(args[1], 'held by ')
249
 
        self.assertStartsWith(args[2], 'locked ')
250
 
        self.assertEndsWith(args[2], ' ago')
 
248
        self.assertEqual('Unable to obtain', args[0])
 
249
        self.assertEqual('lock %s' % (lock_base,), args[1])
 
250
        self.assertStartsWith(args[2], 'held by ')
 
251
        self.assertStartsWith(args[3], 'locked ')
 
252
        self.assertEndsWith(args[3], ' ago')
251
253
 
252
254
    def test_33_wait(self):
253
255
        """Succeed when waiting on a lock that gets released
295
297
            # wait and then lock
296
298
            lf2.lock_write()
297
299
            after = time.time()
298
 
            self.assertTrue(after - before <= 1.0)
299
300
        finally:
300
301
            unlocker.join()
301
302
 
303
304
        # wait for a while
304
305
        lock_base = lf2.transport.abspath(lf2.path)
305
306
        self.assertEqual(1, len(self._logged_reports))
306
 
        self.assertEqual('Unable to obtain %s\n'
 
307
        self.assertEqual('%s %s\n'
307
308
                         '%s\n%s\n'
308
309
                         'Will continue to try for %s seconds\n',
309
310
                         self._logged_reports[0][0])
310
311
        args = self._logged_reports[0][1]
311
 
        self.assertEqual('lock %s' % (lock_base,), args[0])
312
 
        self.assertStartsWith(args[1], 'held by ')
313
 
        self.assertStartsWith(args[2], 'locked ')
314
 
        self.assertEndsWith(args[2], ' ago')
 
312
        self.assertEqual('Unable to obtain', args[0])
 
313
        self.assertEqual('lock %s' % (lock_base,), args[1])
 
314
        self.assertStartsWith(args[2], 'held by ')
 
315
        self.assertStartsWith(args[3], 'locked ')
 
316
        self.assertEndsWith(args[3], ' ago')
315
317
 
316
318
    def test_35_wait_lock_changing(self):
317
319
        """LockDir.wait_lock() will report if the lock changes underneath.
400
402
        unlocker.start()
401
403
        try:
402
404
            # Start the other thread
403
 
            time.sleep(0.1)
 
405
            time.sleep(0.2)
404
406
            # wait and then lock
405
 
            lf2.wait_lock(timeout=1.0, poll=0.1)
 
407
            lf2.wait_lock(timeout=2.0, poll=0.1)
406
408
        finally:
407
409
            unlocker.join()
408
410
        lf2.unlock()
410
412
        # There should be 2 reports, because the lock changed
411
413
        lock_base = lf2.transport.abspath(lf2.path)
412
414
        self.assertEqual(2, len(self._logged_reports))
413
 
        def check_one(index):
414
 
            self.assertEqual('Unable to obtain %s\n'
 
415
        def check_one(index, msg):
 
416
            self.assertEqual('%s %s\n'
415
417
                             '%s\n%s\n'
416
418
                             'Will continue to try for %s seconds\n',
417
419
                             self._logged_reports[index][0])
418
420
            args = self._logged_reports[index][1]
419
 
            self.assertEqual('lock %s' % (lock_base,), args[0])
420
 
            self.assertStartsWith(args[1], 'held by ')
421
 
            self.assertStartsWith(args[2], 'locked ')
422
 
            self.assertEndsWith(args[2], ' ago')
423
 
        check_one(0)
424
 
        check_one(1)
 
421
            self.assertEqual(msg, args[0])
 
422
            self.assertEqual('lock %s' % (lock_base,), args[1])
 
423
            self.assertStartsWith(args[2], 'held by ')
 
424
            self.assertStartsWith(args[3], 'locked ')
 
425
            self.assertEndsWith(args[3], ' ago')
 
426
        check_one(0, 'Unable to obtain')
 
427
        check_one(1, 'Lock owner changed for')
425
428
 
426
429
    def test_40_confirm_easy(self):
427
430
        """Confirm a lock that's already held"""