Server IP : 172.67.216.182 / Your IP : 162.158.108.76 Web Server : Apache System : Linux krdc-ubuntu-s-2vcpu-4gb-amd-blr1-01.localdomain 5.15.0-142-generic #152-Ubuntu SMP Mon May 19 10:54:31 UTC 2025 x86_64 User : www ( 1000) PHP Version : 7.4.33 Disable Function : passthru,exec,system,putenv,chroot,chgrp,chown,shell_exec,popen,proc_open,pcntl_exec,ini_alter,ini_restore,dl,openlog,syslog,readlink,symlink,popepassthru,pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,imap_open,apache_setenv MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /usr/lib/python3/dist-packages/twisted/trial/_dist/ |
Upload File : |
# -*- test-case-name: twisted.trial._dist.test.test_distreporter -*- # # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ The reporter is not made to support concurrent test running, so we will hold test results in here and only send them to the reporter once the test is over. @since: 12.3 """ from zope.interface import implementer from twisted.python.components import proxyForInterface from twisted.trial.itrial import IReporter @implementer(IReporter) class DistReporter(proxyForInterface(IReporter)): # type: ignore[misc] """ See module docstring. """ def __init__(self, original): super().__init__(original) self.running = {} def startTest(self, test): """ Queue test starting. """ self.running[test.id()] = [] self.running[test.id()].append((self.original.startTest, test)) def addFailure(self, test, fail): """ Queue adding a failure. """ self.running[test.id()].append((self.original.addFailure, test, fail)) def addError(self, test, error): """ Queue error adding. """ self.running[test.id()].append((self.original.addError, test, error)) def addSkip(self, test, reason): """ Queue adding a skip. """ self.running[test.id()].append((self.original.addSkip, test, reason)) def addUnexpectedSuccess(self, test, todo=None): """ Queue adding an unexpected success. """ self.running[test.id()].append((self.original.addUnexpectedSuccess, test, todo)) def addExpectedFailure(self, test, error, todo=None): """ Queue adding an unexpected failure. """ self.running[test.id()].append( (self.original.addExpectedFailure, test, error, todo) ) def addSuccess(self, test): """ Queue adding a success. """ self.running[test.id()].append((self.original.addSuccess, test)) def stopTest(self, test): """ Queue stopping the test, then unroll the queue. """ self.running[test.id()].append((self.original.stopTest, test)) for step in self.running[test.id()]: step[0](*step[1:]) del self.running[test.id()]