~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-05-01 04:42:04 UTC
  • mfrom: (4314.2.3 integration)
  • Revision ID: pqm@pqm.ubuntu.com-20090501044204-tvp4oeoj89zr9fby
(robertc) Add debugging of lock activity during tests. (Robert
        Collins)

Show diffs side-by-side

added added

removed removed

Lines of Context:
55
55
    debug,
56
56
    errors,
57
57
    hooks,
 
58
    lock as _mod_lock,
58
59
    memorytree,
59
60
    osutils,
60
61
    progress,
798
799
        self._benchcalls = []
799
800
        self._benchtime = None
800
801
        self._clear_hooks()
 
802
        # Track locks - needs to be called before _clear_debug_flags.
 
803
        self._track_locks()
801
804
        self._clear_debug_flags()
802
805
        TestCase._active_threads = threading.activeCount()
803
806
        self.addCleanup(self._check_leaked_threads)
850
853
        ui.ui_factory = ui.SilentUIFactory()
851
854
        self.addCleanup(_restore)
852
855
 
 
856
    def _check_locks(self):
 
857
        """Check that all lock take/release actions have been paired."""
 
858
        # once we have fixed all the current lock problems, we can change the
 
859
        # following code to always check for mismatched locks, but only do
 
860
        # traceback showing with -Dlock (self._lock_check_thorough is True).
 
861
        # For now, because the test suite will fail, we only assert that lock
 
862
        # matching has occured with -Dlock.
 
863
        # unhook:
 
864
        acquired_locks = [lock for action, lock in self._lock_actions
 
865
            if action == 'acquired']
 
866
        released_locks = [lock for action, lock in self._lock_actions
 
867
            if action == 'released']
 
868
        # trivially, given the tests for lock acquistion and release, if we
 
869
        # have as many in each list, it should be ok.
 
870
        if len(acquired_locks) != len(released_locks):
 
871
            message = \
 
872
                ("Different number of acquired and released locks. (%s, %s)" %
 
873
                (acquired_locks, released_locks))
 
874
            if not self._lock_check_thorough:
 
875
                # Rather than fail, just warn
 
876
                print "Broken test %s: %s" % (self, message)
 
877
                return
 
878
            self.fail(message)
 
879
 
 
880
    def _track_locks(self):
 
881
        """Track lock activity during tests."""
 
882
        self._lock_actions = []
 
883
        self._lock_check_thorough = 'lock' in debug.debug_flags
 
884
        self.addCleanup(self._check_locks)
 
885
        _mod_lock.Lock.hooks.install_named_hook('lock_acquired', self._lock_acquired, None)
 
886
        _mod_lock.Lock.hooks.install_named_hook('lock_released', self._lock_released, None)
 
887
 
 
888
    def _lock_acquired(self, result):
 
889
        self._lock_actions.append(('acquired', result))
 
890
 
 
891
    def _lock_released(self, result):
 
892
        self._lock_actions.append(('released', result))
 
893
 
853
894
    def _ndiff_strings(self, a, b):
854
895
        """Return ndiff between two strings containing lines.
855
896
 
1331
1372
                else:
1332
1373
                    result.addSuccess(self)
1333
1374
                result.stopTest(self)
1334
 
                return
 
1375
                return result
1335
1376
        try:
1336
1377
            try:
1337
1378
                result.startTest(self)
1350
1391
                                "test setUp did not invoke "
1351
1392
                                "bzrlib.tests.TestCase's setUp")
1352
1393
                    except KeyboardInterrupt:
 
1394
                        self._runCleanups()
1353
1395
                        raise
1354
1396
                    except TestSkipped, e:
1355
1397
                        self._do_skip(result, e.args[0])
1356
1398
                        self.tearDown()
1357
 
                        return
 
1399
                        return result
1358
1400
                    except:
1359
1401
                        result.addError(self, sys.exc_info())
1360
 
                        return
 
1402
                        self._runCleanups()
 
1403
                        return result
1361
1404
 
1362
1405
                    ok = False
1363
1406
                    try:
1372
1415
                            reason = e.args[0]
1373
1416
                        self._do_skip(result, reason)
1374
1417
                    except KeyboardInterrupt:
 
1418
                        self._runCleanups()
1375
1419
                        raise
1376
1420
                    except:
1377
1421
                        result.addError(self, sys.exc_info())
1383
1427
                                "test tearDown did not invoke "
1384
1428
                                "bzrlib.tests.TestCase's tearDown")
1385
1429
                    except KeyboardInterrupt:
 
1430
                        self._runCleanups()
1386
1431
                        raise
1387
1432
                    except:
1388
1433
                        result.addError(self, sys.exc_info())
 
1434
                        self._runCleanups()
1389
1435
                        ok = False
1390
1436
                    if ok: result.addSuccess(self)
1391
1437
                finally:
1392
1438
                    result.stopTest(self)
1393
 
                return
 
1439
                return result
1394
1440
            except TestNotApplicable:
1395
1441
                # Not moved from the result [yet].
 
1442
                self._runCleanups()
1396
1443
                raise
1397
1444
            except KeyboardInterrupt:
 
1445
                self._runCleanups()
1398
1446
                raise
1399
1447
        finally:
1400
1448
            saved_attrs = {}
1406
1454
            self.__dict__ = saved_attrs
1407
1455
 
1408
1456
    def tearDown(self):
1409
 
        self._bzr_test_tearDown_run = True
1410
1457
        self._runCleanups()
1411
1458
        self._log_contents = ''
 
1459
        self._bzr_test_tearDown_run = True
1412
1460
        unittest.TestCase.tearDown(self)
1413
1461
 
1414
1462
    def time(self, callable, *args, **kwargs):