L1L`XPL0IT

✏️ Edit: runner.py

📁 /snap/google-cloud-cli/501/platform/bundledpythonunix/lib/python3.14/unittest/runner.py

"""Running tests"""

import sys
import time
import warnings

from _colorize import get_theme

from . import result
from .case import _SubTest
from .signals import registerResult

__unittest = True

class _WritelnDecorator(object):
    """Used to decorate file-like objects with a handy 'writeln' method"""
    def __init__(self, stream):
        self.stream = stream

def __getattr__(self, attr):
        if attr in ('stream', '__getstate__'):
            raise AttributeError(attr)
        return getattr(self.stream, attr)

def writeln(self, arg=None):
        if arg:
            self.write(arg)
        self.write('\n')  # text-mode streams translate to \r\n if needed

class TextTestResult(result.TestResult):
    """A test result class that can print formatted text results to a stream.

Used by TextTestRunner.
    """
    separator1 = '=' * 70
    separator2 = '-' * 70

def __init__(self, stream, descriptions, verbosity, *, durations=None):
        """Construct a TextTestResult. Subclasses should accept **kwargs
        to ensure compatibility as the interface changes."""
        super(TextTestResult, self).__init__(stream, descriptions, verbosity)
        self.stream = stream
        self.showAll = verbosity > 1
        self.dots = verbosity == 1
        self.descriptions = descriptions
        self._theme = get_theme(tty_file=stream).unittest
        self._newline = True
        self.durations = durations

def getDescription(self, test):
        doc_first_line = test.shortDescription()
        if self.descriptions and doc_first_line:
            return '\n'.join((str(test), doc_first_line))
        else:
            return str(test)

def startTest(self, test):
        super(TextTestResult, self).startTest(test)
        if self.showAll:
            self.stream.write(self.getDescription(test))
            self.stream.write(" ... ")
            self.stream.flush()
            self._newline = False

def _write_status(self, test, status):
        is_subtest = isinstance(test, _SubTest)
        if is_subtest or self._newline:
            if not self._newline:
                self.stream.writeln()
            if is_subtest:
                self.stream.write("  ")
            self.stream.write(self.getDescription(test))
            self.stream.write(" ... ")
        self.stream.writeln(status)
        self.stream.flush()
        self._newline = True

def addSubTest(self, test, subtest, err):
        if err is not None:
            t = self._theme
            if self.showAll:
                if issubclass(err[0], subtest.failureException):
                    self._write_status(subtest, f"{t.fail}FAIL{t.reset}")
                else:
                    self._write_status(subtest, f"{t.fail}ERROR{t.reset}")
            elif self.dots:
                if issubclass(err[0], subtest.failureException):
                    self.stream.write(f"{t.fail}F{t.reset}")
                else:
                    self.stream.write(f"{t.fail}E{t.reset}")
                self.stream.flush()
        super(TextTestResult, self).addSubTest(test, subtest, err)

def addSuccess(self, test):
        super(TextTestResult, self).addSuccess(test)
        t = self._theme
        if self.showAll:
            self._write_status(test, f"{t.passed}ok{t.reset}")
        elif self.dots:
            self.stream.write(f"{t.passed}.{t.reset}")
            self.stream.flush()

def addError(self, test, err):
        super(TextTestResult, self).addError(test, err)
        t = self._theme
        if self.showAll:
            self._write_status(test, f"{t.fail}ERROR{t.reset}")
        elif self.dots:
            self.stream.write(f"{t.fail}E{t.reset}")
            self.stream.flush()

def addFailure(self, test, err):
        super(TextTestResult, self).addFailure(test, err)
        t = self._theme
        if self.showAll:
            self._write_status(test, f"{t.fail}FAIL{t.reset}")
        elif self.dots:
            self.stream.write(f"{t.fail}F{t.reset}")
            self.stream.flush()

def addSkip(self, test, reason):
        super(TextTestResult, self).addSkip(test, reason)
        t = self._theme
        if self.showAll:
            self._write_status(test, f"{t.warn}skipped{t.reset} {reason!r}")
        elif self.dots:
            self.stream.write(f"{t.warn}s{t.reset}")
            self.stream.flush()

def addExpectedFailure(self, test, err):
        super(TextTestResult, self).addExpectedFailure(test, err)
        t = self._theme
        if self.showAll:
            self.stream.writeln(f"{t.warn}expected failure{t.reset}")
            self.stream.flush()
        elif self.dots:
            self.stream.write(f"{t.warn}x{t.reset}")
            self.stream.flush()

def addUnexpectedSuccess(self, test):
        super(TextTestResult, self).addUnexpectedSuccess(test)
        t = self._theme
        if self.showAll:
            self.stream.writeln(f"{t.fail}unexpected success{t.reset}")
            self.stream.flush()
        elif self.dots:
            self.stream.write(f"{t.fail}u{t.reset}")
            self.stream.flush()

def printErrors(self):
        t = self._theme
        if self.dots or self.showAll:
            self.stream.writeln()
            self.stream.flush()
        self.printErrorList(f"{t.fail}ERROR{t.reset}", self.errors)
        self.printErrorList(f"{t.fail}FAIL{t.reset}", self.failures)
        unexpectedSuccesses = getattr(self, "unexpectedSuccesses", ())
        if unexpectedSuccesses:
            self.stream.writeln(self.separator1)
            for test in unexpectedSuccesses:
                self.stream.writeln(
                    f"{t.fail}UNEXPECTED SUCCESS{t.fail_info}: "
                    f"{self.getDescription(test)}{t.reset}"
                )
            self.stream.flush()

def printErrorList(self, flavour, errors):
        t = self._theme
        for test, err in errors:
            self.stream.writeln(self.separator1)
            self.stream.writeln(
                f"{flavour}{t.fail_info}: {self.getDescription(test)}{t.reset}"
            )
            self.stream.writeln(self.separator2)
            self.stream.writeln("%s" % err)
            self.stream.flush()

class TextTestRunner(object):
    """A test runner class that displays results in textual form.

It prints out the names of tests as they are run, errors as they
    occur, and a summary of the results at the end of the test run.
    """
    resultclass = TextTestResult

def __init__(self, stream=None, descriptions=True, verbosity=1,
                 failfast=False, buffer=False, resultclass=None, warnings=None,
                 *, tb_locals=False, durations=None):
        """Construct a TextTestRunner.

Subclasses should accept **kwargs to ensure compatibility as the
        interface changes.
        """
        if stream is None:
            stream = sys.stderr
        self.stream = _WritelnDecorator(stream)
        self.descriptions = descriptions
        self.verbosity = verbosity
        self.failfast = failfast
        self.buffer = buffer
        self.tb_locals = tb_locals
        self.durations = durations
        self.warnings = warnings
        if resultclass is not None:
            self.resultclass = resultclass

def _makeResult(self):
        try:
            return self.resultclass(self.stream, self.descriptions,
                                    self.verbosity, durations=self.durations)
        except TypeError:
            # didn't accept the durations argument
            return self.resultclass(self.stream, self.descriptions,
                                    self.verbosity)

def _printDurations(self, result):
        if not result.collectedDurations:
            return
        ls = sorted(result.collectedDurations, key=lambda x: x[1],
                    reverse=True)
        if self.durations > 0:
            ls = ls[:self.durations]
        self.stream.writeln("Slowest test durations")
        if hasattr(result, 'separator2'):
            self.stream.writeln(result.separator2)
        hidden = False
        for test, elapsed in ls:
            if self.verbosity < 2 and elapsed < 0.001:
                hidden = True
                continue
            self.stream.writeln("%-10s %s" % ("%.3fs" % elapsed, test))
        if hidden:
            self.stream.writeln("\n(durations < 0.001s were hidden; "
                                "use -v to show these durations)")
        else:
            self.stream.writeln("")

def run(self, test):
        "Run the given test case or test suite."
        result = self._makeResult()
        registerResult(result)
        result.failfast = self.failfast
        result.buffer = self.buffer
        result.tb_locals = self.tb_locals
        with warnings.catch_warnings():
            if self.warnings:
                # if self.warnings is set, use it to filter all the warnings
                warnings.simplefilter(self.warnings)
            start_time = time.perf_counter()
            startTestRun = getattr(result, 'startTestRun', None)
            if startTestRun is not None:
                startTestRun()
            try:
                test(result)
            finally:
                stopTestRun = getattr(result, 'stopTestRun', None)
                if stopTestRun is not None:
                    stopTestRun()
            stop_time = time.perf_counter()
        time_taken = stop_time - start_time
        result.printErrors()
        if self.durations is not None:
            self._printDurations(result)

if hasattr(result, 'separator2'):
            self.stream.writeln(result.separator2)

run = result.testsRun
        self.stream.writeln("Ran %d test%s in %.3fs" %
                            (run, run != 1 and "s" or "", time_taken))
        self.stream.writeln()

expected_fails = unexpected_successes = skipped = 0
        try:
            results = map(len, (result.expectedFailures,
                                result.unexpectedSuccesses,
                                result.skipped))
        except AttributeError:
            pass
        else:
            expected_fails, unexpected_successes, skipped = results

infos = []
        t = get_theme(tty_file=self.stream).unittest

if not result.wasSuccessful():
            self.stream.write(f"{t.fail_info}FAILED{t.reset}")
            failed, errored = len(result.failures), len(result.errors)
            if failed:
                infos.append(f"{t.fail_info}failures={failed}{t.reset}")
            if errored:
                infos.append(f"{t.fail_info}errors={errored}{t.reset}")
        elif run == 0 and not skipped:
            self.stream.write(f"{t.warn}NO TESTS RAN{t.reset}")
        else:
            self.stream.write(f"{t.passed}OK{t.reset}")
        if skipped:
            infos.append(f"{t.warn}skipped={skipped}{t.reset}")
        if expected_fails:
            infos.append(f"{t.warn}expected failures={expected_fails}{t.reset}")
        if unexpected_successes:
            infos.append(
                f"{t.fail}unexpected successes={unexpected_successes}{t.reset}"
            )
        if infos:
            self.stream.writeln(" (%s)" % (", ".join(infos),))
        else:
            self.stream.write("\n")
        self.stream.flush()
        return result

Cancel

💻 Terminal

📤 Upload

📁 /snap/google-cloud-cli/501/platform/bundledpythonunix/lib/python3.14/unittest

//snap/google-cloud-cli/501/platform/bundledpythonunix/lib/python3.14/unittest

Name	Size	Perm	Modified	Actions
📁 __pycache__/	-	0775	2026-09-22 12:33	🗑 Delete
__init__.py	3.2K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
__main__.py	472B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
_log.py	2.7K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
async_case.py	5.9K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
case.py	62.7K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
loader.py	20.7K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
main.py	11.4K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
mock.py	109.7K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
result.py	9.1K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
runner.py	11.1K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
signals.py	2.3K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
suite.py	13.2K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
util.py	5.1K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del