← Back to branch summary

~amanica/bzr/320119-log_exclusive_lower_bound

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Jelmer Vernooij
  • Date: 2009-04-06 02:54:14 UTC
  • mfrom: (4253 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4255.
  • Revision ID: jelmer@samba.org-20090406025414-65tpjwcmjp5wa5oj
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/__init__.py
545
545
            actionTaken = "Listed"
546
546
        else:
547
547
            try:
548
 
                from testtools import ThreadsafeForwardingResult
 
548
                import testtools
549
549
            except ImportError:
550
550
                test.run(result)
551
551
            else:
552
 
                if type(result) == ThreadsafeForwardingResult:
 
552
                if isinstance(test, testtools.ConcurrentTestSuite):
 
553
                    # We need to catch bzr specific behaviors
553
554
                    test.run(BZRTransformingResult(result))
554
555
                else:
555
556
                    test.run(result)
674
675
    Allows get_password to be tested without real tty attached.
675
676
    """
676
677
 
677
 
    def __init__(self,
678
 
                 stdout=None,
679
 
                 stderr=None,
680
 
                 stdin=None):
681
 
        super(TestUIFactory, self).__init__()
 
678
    def __init__(self, stdout=None, stderr=None, stdin=None):
682
679
        if stdin is not None:
683
680
            # We use a StringIOWrapper to be able to test various
684
681
            # encodings, but the user is still responsible to
685
682
            # encode the string and to set the encoding attribute
686
683
            # of StringIOWrapper.
687
 
            self.stdin = StringIOWrapper(stdin)
688
 
        if stdout is None:
689
 
            self.stdout = sys.stdout
690
 
        else:
691
 
            self.stdout = stdout
692
 
        if stderr is None:
693
 
            self.stderr = sys.stderr
694
 
        else:
695
 
            self.stderr = stderr
 
684
            stdin = StringIOWrapper(stdin)
 
685
        super(TestUIFactory, self).__init__(stdin, stdout, stderr)
696
686
 
697
687
    def clear(self):
698
688
        """See progress.ProgressBar.clear()."""
700
690
    def clear_term(self):
701
691
        """See progress.ProgressBar.clear_term()."""
702
692
 
703
 
    def clear_term(self):
704
 
        """See progress.ProgressBar.clear_term()."""
705
 
 
706
693
    def finished(self):
707
694
        """See progress.ProgressBar.finished()."""
708
695
 
719
706
    def update(self, message, count=None, total=None):
720
707
        """See progress.ProgressBar.update()."""
721
708
 
722
 
    def get_non_echoed_password(self, prompt):
 
709
    def get_non_echoed_password(self):
723
710
        """Get password from stdin without trying to handle the echo mode"""
724
 
        if prompt:
725
 
            self.stdout.write(prompt.encode(self.stdout.encoding, 'replace'))
726
711
        password = self.stdin.readline()
727
712
        if not password:
728
713
            raise EOFError
794
779
        import pdb
795
780
        pdb.Pdb().set_trace(sys._getframe().f_back)
796
781
 
797
 
    def exc_info(self):
798
 
        absent_attr = object()
799
 
        exc_info = getattr(self, '_exc_info', absent_attr)
800
 
        if exc_info is absent_attr:
801
 
            exc_info = getattr(self, '_TestCase__exc_info')
802
 
        return exc_info()
803
 
 
804
782
    def _check_leaked_threads(self):
805
783
        active = threading.activeCount()
806
784
        leaked_threads = active - TestCase._active_threads
1280
1258
            'NO_PROXY': None,
1281
1259
            'all_proxy': None,
1282
1260
            'ALL_PROXY': None,
1283
 
            # Nobody cares about these ones AFAIK. So far at
 
1261
            # Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
1284
1262
            # least. If you do (care), please update this comment
1285
 
            # -- vila 20061212
 
1263
            # -- vila 20080401
1286
1264
            'ftp_proxy': None,
1287
1265
            'FTP_PROXY': None,
1288
1266
            'BZR_REMOTE_PATH': None,
1315
1293
    def _do_skip(self, result, reason):
1316
1294
        addSkip = getattr(result, 'addSkip', None)
1317
1295
        if not callable(addSkip):
1318
 
            result.addError(self, self.exc_info())
 
1296
            result.addError(self, sys.exc_info())
1319
1297
        else:
1320
1298
            addSkip(self, reason)
1321
1299
 
1354
1332
                        self.tearDown()
1355
1333
                        return
1356
1334
                    except:
1357
 
                        result.addError(self, self.exc_info())
 
1335
                        result.addError(self, sys.exc_info())
1358
1336
                        return
1359
1337
 
1360
1338
                    ok = False
1362
1340
                        testMethod()
1363
1341
                        ok = True
1364
1342
                    except self.failureException:
1365
 
                        result.addFailure(self, self.exc_info())
 
1343
                        result.addFailure(self, sys.exc_info())
1366
1344
                    except TestSkipped, e:
1367
1345
                        if not e.args:
1368
1346
                            reason = "No reason given."
1372
1350
                    except KeyboardInterrupt:
1373
1351
                        raise
1374
1352
                    except:
1375
 
                        result.addError(self, self.exc_info())
 
1353
                        result.addError(self, sys.exc_info())
1376
1354
 
1377
1355
                    try:
1378
1356
                        self.tearDown()
1383
1361
                    except KeyboardInterrupt:
1384
1362
                        raise
1385
1363
                    except:
1386
 
                        result.addError(self, self.exc_info())
 
1364
                        result.addError(self, sys.exc_info())
1387
1365
                        ok = False
1388
1366
                    if ok: result.addSuccess(self)
1389
1367
                finally:
2683
2661
 
2684
2662
# A registry where get() returns a suite decorator.
2685
2663
parallel_registry = registry.Registry()
 
2664
 
 
2665
 
2686
2666
def fork_decorator(suite):
2687
2667
    concurrency = local_concurrency()
2688
2668
    if concurrency == 1:
2690
2670
    from testtools import ConcurrentTestSuite
2691
2671
    return ConcurrentTestSuite(suite, fork_for_tests)
2692
2672
parallel_registry.register('fork', fork_decorator)
 
2673
 
 
2674
 
2693
2675
def subprocess_decorator(suite):
2694
2676
    concurrency = local_concurrency()
2695
2677
    if concurrency == 1:
2881
2863
    """Take suite and start up one runner per CPU by forking()
2882
2864
 
2883
2865
    :return: An iterable of TestCase-like objects which can each have
2884
 
        run(result) called on them to feed tests to result, and
2885
 
        cleanup() called on them to stop them/kill children/end threads.
 
2866
        run(result) called on them to feed tests to result.
2886
2867
    """
2887
2868
    concurrency = local_concurrency()
2888
2869
    result = []
2898
2879
                ProtocolTestCase.run(self, result)
2899
2880
            finally:
2900
2881
                os.waitpid(self.pid, os.WNOHANG)
2901
 
            # print "pid %d finished" % finished_process
2902
2882
 
2903
2883
    test_blocks = partition_tests(suite, concurrency)
2904
2884
    for process_tests in test_blocks:
2912
2892
                # Leave stderr and stdout open so we can see test noise
2913
2893
                # Close stdin so that the child goes away if it decides to
2914
2894
                # read from stdin (otherwise its a roulette to see what
2915
 
                # child actually gets keystrokes for pdb etc.
 
2895
                # child actually gets keystrokes for pdb etc).
2916
2896
                sys.stdin.close()
2917
2897
                sys.stdin = None
2918
 
                stream = os.fdopen(c2pwrite, 'wb', 0)
 
2898
                stream = os.fdopen(c2pwrite, 'wb', 1)
2919
2899
                subunit_result = TestProtocolClient(stream)
2920
2900
                process_suite.run(subunit_result)
2921
2901
            finally:
2932
2912
    """Take suite and start up one runner per CPU using subprocess().
2933
2913
 
2934
2914
    :return: An iterable of TestCase-like objects which can each have
2935
 
        run(result) called on them to feed tests to result, and
2936
 
        cleanup() called on them to stop them/kill children/end threads.
 
2915
        run(result) called on them to feed tests to result.
2937
2916
    """
2938
2917
    concurrency = local_concurrency()
2939
2918
    result = []
3324
3303
                   'bzrlib.tests.test_directory_service',
3325
3304
                   'bzrlib.tests.test_dirstate',
3326
3305
                   'bzrlib.tests.test_email_message',
 
3306
                   'bzrlib.tests.test_eol_filters',
3327
3307
                   'bzrlib.tests.test_errors',
3328
3308
                   'bzrlib.tests.test_export',
3329
3309
                   'bzrlib.tests.test_extract',
3349
3329
                   'bzrlib.tests.test_index',
3350
3330
                   'bzrlib.tests.test_info',
3351
3331
                   'bzrlib.tests.test_inv',
 
3332
                   'bzrlib.tests.test_inventory_delta',
3352
3333
                   'bzrlib.tests.test_knit',
3353
3334
                   'bzrlib.tests.test_lazy_import',
3354
3335
                   'bzrlib.tests.test_lazy_regex',
3393
3374
                   'bzrlib.tests.test_rules',
3394
3375
                   'bzrlib.tests.test_sampler',
3395
3376
                   'bzrlib.tests.test_selftest',
 
3377
                   'bzrlib.tests.test_serializer',
3396
3378
                   'bzrlib.tests.test_setup',
3397
3379
                   'bzrlib.tests.test_sftp_transport',
3398
3380
                   'bzrlib.tests.test_shelf',
3588
3570
    the scenario name at the end of its id(), and updating the test object's
3589
3571
    __dict__ with the scenario_param_dict.
3590
3572
 
 
3573
    >>> import bzrlib.tests.test_sampler
3591
3574
    >>> r = multiply_tests(
3592
3575
    ...     bzrlib.tests.test_sampler.DemoTest('test_nothing'),
3593
3576
    ...     [('one', dict(param=1)),