~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

• browse files at revision 5050.12.1
• compare with another revision
• download diff
• download tarball
• view history from revision 5050.12.1

• reverse the comparison (5050.12.1 to 5401)
• stop comparing with revision 5401

Show diffs side-by-side

added

removed

bzrlib/tests/__init__.py

34

34

import difflib

35

35

import doctest

36

36

import errno

37

 

import itertools

38

37

import logging

39

38

import math

40

39

import os

114

113

    TestUtil,

115

114

    treeshape,

116

115

    )

 

116

from bzrlib.tests.http_server import HttpServer

 

117

from bzrlib.tests.TestUtil import (

 

118

                          TestSuite,

 

119

                          TestLoader,

 

120

                          )

117

121

from bzrlib.ui import NullProgressView

118

122

from bzrlib.ui.text import TextUIFactory

119

123

import bzrlib.version_info_formats.format_custom

136

140

SUBUNIT_SEEK_CUR = 1

137

141

 

138

142

 

139

 

class ExtendedTestResult(testtools.TextTestResult):

 

143

class ExtendedTestResult(unittest._TextTestResult):

140

144

    """Accepts, reports and accumulates the results of running tests.

141

145

 

142

146

    Compared to the unittest version this class adds support for

163

167

        :param bench_history: Optionally, a writable file object to accumulate

164

168

            benchmark results.

165

169

        """

166

 

        testtools.TextTestResult.__init__(self, stream)

 

170

        unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)

167

171

        if bench_history is not None:

168

172

            from bzrlib.version import _get_bzr_source_tree

169

173

            src_tree = _get_bzr_source_tree()

196

200

        actionTaken = "Ran"

197

201

        stopTime = time.time()

198

202

        timeTaken = stopTime - self.startTime

199

 

        # GZ 2010-07-19: Seems testtools has no printErrors method, and though

200

 

        #                the parent class method is similar have to duplicate

201

 

        self._show_list('ERROR', self.errors)

202

 

        self._show_list('FAIL', self.failures)

203

 

        self.stream.write(self.sep2)

204

 

        self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,

 

203

        self.printErrors()

 

204

        self.stream.writeln(self.separator2)

 

205

        self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,

205

206

                            run, run != 1 and "s" or "", timeTaken))

 

207

        self.stream.writeln()

206

208

        if not self.wasSuccessful():

207

209

            self.stream.write("FAILED (")

208

210

            failed, errored = map(len, (self.failures, self.errors))

215

217

                if failed or errored: self.stream.write(", ")

216

218

                self.stream.write("known_failure_count=%d" %

217

219

                    self.known_failure_count)

218

 

            self.stream.write(")\n")

 

220

            self.stream.writeln(")")

219

221

        else:

220

222

            if self.known_failure_count:

221

 

                self.stream.write("OK (known_failures=%d)\n" %

 

223

                self.stream.writeln("OK (known_failures=%d)" %

222

224

                    self.known_failure_count)

223

225

            else:

224

 

                self.stream.write("OK\n")

 

226

                self.stream.writeln("OK")

225

227

        if self.skip_count > 0:

226

228

            skipped = self.skip_count

227

 

            self.stream.write('%d test%s skipped\n' %

 

229

            self.stream.writeln('%d test%s skipped' %

228

230

                                (skipped, skipped != 1 and "s" or ""))

229

231

        if self.unsupported:

230

232

            for feature, count in sorted(self.unsupported.items()):

231

 

                self.stream.write("Missing feature '%s' skipped %d tests.\n" %

 

233

                self.stream.writeln("Missing feature '%s' skipped %d tests." %

232

234

                    (feature, count))

233

235

        if self._strict:

234

236

            ok = self.wasStrictlySuccessful()

276

278

        return what

277

279

 

278

280

    def startTest(self, test):

279

 

        super(ExtendedTestResult, self).startTest(test)

 

281

        unittest.TestResult.startTest(self, test)

280

282

        if self.count == 0:

281

283

            self.startTests()

282

284

        self.report_test_start(test)

320

322

        fails with an unexpected error.

321

323

        """

322

324

        self._post_mortem()

323

 

        super(ExtendedTestResult, self).addError(test, err)

 

325

        unittest.TestResult.addError(self, test, err)

324

326

        self.error_count += 1

325

327

        self.report_error(test, err)

326

328

        if self.stop_early:

334

336

        fails because e.g. an assert() method failed.

335

337

        """

336

338

        self._post_mortem()

337

 

        super(ExtendedTestResult, self).addFailure(test, err)

 

339

        unittest.TestResult.addFailure(self, test, err)

338

340

        self.failure_count += 1

339

341

        self.report_failure(test, err)

340

342

        if self.stop_early:

354

356

                    test.id()))

355

357

        self.report_success(test)

356

358

        self._cleanupLogFile(test)

357

 

        super(ExtendedTestResult, self).addSuccess(test)

 

359

        unittest.TestResult.addSuccess(self, test)

358

360

        test._log_contents = ''

359

361

 

360

362

    def addExpectedFailure(self, test, err):

548

550

        return '%s%s' % (indent, err[1])

549

551

 

550

552

    def report_error(self, test, err):

551

 

        self.stream.write('ERROR %s\n%s\n'

 

553

        self.stream.writeln('ERROR %s\n%s'

552

554

                % (self._testTimeString(test),

553

555

                   self._error_summary(err)))

554

556

 

555

557

    def report_failure(self, test, err):

556

 

        self.stream.write(' FAIL %s\n%s\n'

 

558

        self.stream.writeln(' FAIL %s\n%s'

557

559

                % (self._testTimeString(test),

558

560

                   self._error_summary(err)))

559

561

 

560

562

    def report_known_failure(self, test, err):

561

 

        self.stream.write('XFAIL %s\n%s\n'

 

563

        self.stream.writeln('XFAIL %s\n%s'

562

564

                % (self._testTimeString(test),

563

565

                   self._error_summary(err)))

564

566

 

565

567

    def report_success(self, test):

566

 

        self.stream.write('   OK %s\n' % self._testTimeString(test))

 

568

        self.stream.writeln('   OK %s' % self._testTimeString(test))

567

569

        for bench_called, stats in getattr(test, '_benchcalls', []):

568

 

            self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)

 

570

            self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)

569

571

            stats.pprint(file=self.stream)

570

572

        # flush the stream so that we get smooth output. This verbose mode is

571

573

        # used to show the output in PQM.

572

574

        self.stream.flush()

573

575

 

574

576

    def report_skip(self, test, reason):

575

 

        self.stream.write(' SKIP %s\n%s\n'

 

577

        self.stream.writeln(' SKIP %s\n%s'

576

578

                % (self._testTimeString(test), reason))

577

579

 

578

580

    def report_not_applicable(self, test, reason):

579

 

        self.stream.write('  N/A %s\n    %s\n'

 

581

        self.stream.writeln('  N/A %s\n    %s'

580

582

                % (self._testTimeString(test), reason))

581

583

 

582

584

    def report_unsupported(self, test, feature):

583

585

        """test cannot be run because feature is missing."""

584

 

        self.stream.write("NODEP %s\n    The feature '%s' is not available.\n"

 

586

        self.stream.writeln("NODEP %s\n    The feature '%s' is not available."

585

587

                %(self._testTimeString(test), feature))

586

588

 

587

589

 

616

618

            encode = codec.encode

617

619

        stream = osutils.UnicodeOrBytesToBytesWriter(encode, stream)

618

620

        stream.encoding = new_encoding

619

 

        self.stream = stream

 

621

        self.stream = unittest._WritelnDecorator(stream)

620

622

        self.descriptions = descriptions

621

623

        self.verbosity = verbosity

622

624

        self._bench_history = bench_history

843

845

        # going away but leak one) but it seems less likely than the actual

844

846

        # false positives (the test see threads going away and does not leak).

845

847

        if leaked_threads > 0:

846

 

            if 'threads' in selftest_debug_flags:

847

 

                print '%s is leaking, active is now %d' % (self.id(), active)

848

848

            TestCase._leaking_threads_tests += 1

849

849

            if TestCase._first_thread_leaker_id is None:

850

850

                TestCase._first_thread_leaker_id = self.id()

2009

2009

 

2010

2010

    def get_bzr_path(self):

2011

2011

        """Return the path of the 'bzr' executable for this test suite."""

2012

 

        bzr_path = os.path.join(self.get_source_path(), "bzr")

 

2012

        bzr_path = self.get_source_path()+'/bzr'

2013

2013

        if not os.path.isfile(bzr_path):

2014

2014

            # We are probably installed. Assume sys.argv is the right file

2015

2015

            bzr_path = sys.argv[0]

2436

2436

 

2437

2437

    def setUp(self):

2438

2438

        super(TestCaseWithMemoryTransport, self).setUp()

2439

 

        # Ensure that ConnectedTransport doesn't leak sockets

2440

 

        def get_transport_with_cleanup(*args, **kwargs):

2441

 

            t = orig_get_transport(*args, **kwargs)

2442

 

            if isinstance(t, _mod_transport.ConnectedTransport):

2443

 

                self.addCleanup(t.disconnect)

2444

 

            return t

2445

 

 

2446

 

        orig_get_transport = self.overrideAttr(_mod_transport, 'get_transport',

2447

 

                                               get_transport_with_cleanup)

2448

2439

        self._make_test_root()

2449

2440

        self.addCleanup(os.chdir, os.getcwdu())

2450

2441

        self.makeAndChdirToTestDir()

2742

2733

    """

2743

2734

 

2744

2735

    def setUp(self):

2745

 

        from bzrlib.tests import http_server

2746

2736

        super(ChrootedTestCase, self).setUp()

2747

2737

        if not self.vfs_transport_factory == memory.MemoryServer:

2748

 

            self.transport_readonly_server = http_server.HttpServer

 

2738

            self.transport_readonly_server = HttpServer

2749

2739

 

2750

2740

 

2751

2741

def condition_id_re(pattern):

3072

3062

    return suite

3073

3063

 

3074

3064

 

3075

 

class TestDecorator(TestUtil.TestSuite):

 

3065

class TestDecorator(TestSuite):

3076

3066

    """A decorator for TestCase/TestSuite objects.

3077

3067

    

3078

3068

    Usually, subclasses should override __iter__(used when flattening test

3081

3071

    """

3082

3072

 

3083

3073

    def __init__(self, suite):

3084

 

        TestUtil.TestSuite.__init__(self)

 

3074

        TestSuite.__init__(self)

3085

3075

        self.addTest(suite)

3086

3076

 

3087

3077

    def countTestCases(self):

3206

3196

 

3207

3197

def partition_tests(suite, count):

3208

3198

    """Partition suite into count lists of tests."""

3209

 

    # This just assigns tests in a round-robin fashion.  On one hand this

3210

 

    # splits up blocks of related tests that might run faster if they shared

3211

 

    # resources, but on the other it avoids assigning blocks of slow tests to

3212

 

    # just one partition.  So the slowest partition shouldn't be much slower

3213

 

    # than the fastest.

3214

 

    partitions = [list() for i in range(count)]

3215

 

    tests = iter_suite_tests(suite)

3216

 

    for partition, test in itertools.izip(itertools.cycle(partitions), tests):

3217

 

        partition.append(test)

3218

 

    return partitions

 

3199

    result = []

 

3200

    tests = list(iter_suite_tests(suite))

 

3201

    tests_per_process = int(math.ceil(float(len(tests)) / count))

 

3202

    for block in range(count):

 

3203

        low_test = block * tests_per_process

 

3204

        high_test = low_test + tests_per_process

 

3205

        process_tests = tests[low_test:high_test]

 

3206

        result.append(process_tests)

 

3207

    return result

3219

3208

 

3220

3209

 

3221

3210

def workaround_zealous_crypto_random():

3255

3244

 

3256

3245

    test_blocks = partition_tests(suite, concurrency)

3257

3246

    for process_tests in test_blocks:

3258

 

        process_suite = TestUtil.TestSuite()

 

3247

        process_suite = TestSuite()

3259

3248

        process_suite.addTests(process_tests)

3260

3249

        c2pread, c2pwrite = os.pipe()

3261

3250

        pid = os.fork()

3416

3405

#                           rather than failing tests. And no longer raise

3417

3406

#                           LockContention when fctnl locks are not being used

3418

3407

#                           with proper exclusion rules.

3419

 

#   -Ethreads               Will display thread ident at creation/join time to

3420

 

#                           help track thread leaks

3421

3408

selftest_debug_flags = set()

3422

3409

 

3423

3410

 

3656

3643

        'bzrlib.doc',

3657

3644

        'bzrlib.tests.blackbox',

3658

3645

        'bzrlib.tests.commands',

3659

 

        'bzrlib.tests.doc_generate',

3660

3646

        'bzrlib.tests.per_branch',

3661

 

        'bzrlib.tests.per_controldir',

3662

 

        'bzrlib.tests.per_controldir_colo',

 

3647

        'bzrlib.tests.per_bzrdir',

 

3648

        'bzrlib.tests.per_bzrdir_colo',

3663

3649

        'bzrlib.tests.per_foreign_vcs',

3664

3650

        'bzrlib.tests.per_interrepository',

3665

3651

        'bzrlib.tests.per_intertree',

3678

3664

        'bzrlib.tests.per_workingtree',

3679

3665

        'bzrlib.tests.test__annotator',

3680

3666

        'bzrlib.tests.test__bencode',

3681

 

        'bzrlib.tests.test__btree_serializer',

3682

3667

        'bzrlib.tests.test__chk_map',

3683

3668

        'bzrlib.tests.test__dirstate_helpers',

3684

3669

        'bzrlib.tests.test__groupcompress',

3817

3802

        'bzrlib.tests.test_switch',

3818

3803

        'bzrlib.tests.test_symbol_versioning',

3819

3804

        'bzrlib.tests.test_tag',

3820

 

        'bzrlib.tests.test_test_server',

3821

3805

        'bzrlib.tests.test_testament',

3822

3806

        'bzrlib.tests.test_textfile',

3823

3807

        'bzrlib.tests.test_textmerge',

3839

3823

        'bzrlib.tests.test_urlutils',

3840

3824

        'bzrlib.tests.test_version',

3841

3825

        'bzrlib.tests.test_version_info',

3842

 

        'bzrlib.tests.test_versionedfile',

3843

3826

        'bzrlib.tests.test_weave',

3844

3827

        'bzrlib.tests.test_whitebox',

3845

3828

        'bzrlib.tests.test_win32utils',

4014

3997

    ...     bzrlib.tests.test_sampler.DemoTest('test_nothing'),

4015

3998

    ...     [('one', dict(param=1)),

4016

3999

    ...      ('two', dict(param=2))],

4017

 

    ...     TestUtil.TestSuite())

 

4000

    ...     TestSuite())

4018

4001

    >>> tests = list(iter_suite_tests(r))

4019

4002

    >>> len(tests)

4020

4003

    2

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1