← Back to branch summary

~ubuntu-branches/ubuntu/raring/subunit/raring-proposed

« back to all changes in this revision

Viewing changes to python/subunit/__init__.py

  • Committer: Bazaar Package Importer
  • Author(s): Robert Collins
  • Date: 2010-07-04 21:24:12 UTC
  • mfrom: (1.1.5 upstream) (3.2.5 sid)
  • Revision ID: james.westby@ubuntu.com-20100704212412-203f450zuuwoolyt
Tags: 0.0.6-1
New upstream release.

Show diffs side-by-side

added added

removed removed

Lines of Context:
python/subunit/__init__.py
133
133
except ImportError:
134
134
    raise ImportError ("testtools.testresult.real does not contain "
135
135
        "_StringException, check your version.")
136
 
 
137
 
 
138
 
from testtools.testresult.real import _StringException
 
136
from testtools import testresult
139
137
 
140
138
import chunked, details, test_results
141
139
 
244
242
 
245
243
    def lostConnection(self):
246
244
        """Connection lost."""
247
 
        self.parser._lostConnectionInTest('unknown state of ')
 
245
        self.parser._lostConnectionInTest(u'unknown state of ')
248
246
 
249
247
    def startTest(self, offset, line):
250
248
        """A test start command received."""
324
322
 
325
323
    def lostConnection(self):
326
324
        """Connection lost."""
327
 
        self.parser._lostConnectionInTest('')
 
325
        self.parser._lostConnectionInTest(u'')
328
326
 
329
327
 
330
328
class _OutSideTest(_ParserState):
359
357
 
360
358
    def lostConnection(self):
361
359
        """Connection lost."""
362
 
        self.parser._lostConnectionInTest('%s report of ' %
 
360
        self.parser._lostConnectionInTest(u'%s report of ' %
363
361
            self._outcome_label())
364
362
 
365
363
    def _outcome_label(self):
501
499
        self._state.lineReceived(line)
502
500
 
503
501
    def _lostConnectionInTest(self, state_string):
504
 
        error_string = "lost connection during %stest '%s'" % (
 
502
        error_string = u"lost connection during %stest '%s'" % (
505
503
            state_string, self.current_test_description)
506
504
        self.client.addError(self._current_test, RemoteError(error_string))
507
505
        self.client.stopTest(self._current_test)
531
529
        self._stream.write(line)
532
530
 
533
531
 
534
 
class TestProtocolClient(unittest.TestResult):
 
532
class TestProtocolClient(testresult.TestResult):
535
533
    """A TestResult which generates a subunit stream for a test run.
536
534
    
537
535
    # Get a TestSuite or TestCase to run
550
548
    """
551
549
 
552
550
    def __init__(self, stream):
553
 
        unittest.TestResult.__init__(self)
 
551
        testresult.TestResult.__init__(self)
554
552
        self._stream = stream
 
553
        _make_stream_binary(stream)
555
554
 
556
555
    def addError(self, test, error=None, details=None):
557
556
        """Report an error in test test.
618
617
            raise ValueError
619
618
        if error is not None:
620
619
            self._stream.write(" [\n")
621
 
            for line in self._exc_info_to_string(error, test).splitlines():
622
 
                self._stream.write("%s\n" % line)
 
620
            # XXX: this needs to be made much stricter, along the lines of
 
621
            # Martin[gz]'s work in testtools. Perhaps subunit can use that?
 
622
            for line in self._exc_info_to_unicode(error, test).splitlines():
 
623
                self._stream.write(("%s\n" % line).encode('utf8'))
623
624
        else:
624
625
            self._write_details(details)
625
626
        self._stream.write("]\n")
704
705
        """Obey the testtools result.done() interface."""
705
706
 
706
707
 
707
 
def RemoteError(description=""):
 
708
def RemoteError(description=u""):
708
709
    return (_StringException, _StringException(description), None)
709
710
 
710
711
 
754
755
    def run(self, result=None):
755
756
        if result is None: result = self.defaultTestResult()
756
757
        result.startTest(self)
757
 
        result.addError(self, RemoteError("Cannot run RemotedTestCases.\n"))
 
758
        result.addError(self, RemoteError(u"Cannot run RemotedTestCases.\n"))
758
759
        result.stopTest(self)
759
760
 
760
761
    def _strclass(self):
784
785
 
785
786
    def debug(self):
786
787
        """Run the test without collecting errors in a TestResult"""
787
 
        self._run(unittest.TestResult())
 
788
        self._run(testresult.TestResult())
788
789
 
789
790
    def _run(self, result):
790
791
        protocol = TestProtocolServer(result)
816
817
    """
817
818
 
818
819
    def run(self, result=None):
819
 
        if result is None: result = unittest.TestResult()
 
820
        if result is None: result = testresult.TestResult()
820
821
        run_isolated(unittest.TestSuite, self, result)
821
822
 
822
823
 
913
914
                    subunit.write("]\n")
914
915
                continue
915
916
        # not a plan line, or have seen one before
916
 
        match = re.match("(ok|not ok)(?:\s+(\d+)?)?(?:\s+([^#]*[^#\s]+)\s*)?(?:\s+#\s+(TODO|SKIP)(?:\s+(.*))?)?\n", line)
 
917
        match = re.match("(ok|not ok)(?:\s+(\d+)?)?(?:\s+([^#]*[^#\s]+)\s*)?(?:\s+#\s+(TODO|SKIP|skip|todo)(?:\s+(.*))?)?\n", line)
917
918
        if match:
918
919
            # new test, emit current one.
919
920
            _emit_test()
927
928
            else:
928
929
                description = ' ' + description
929
930
            if directive is not None:
930
 
                if directive == 'TODO':
 
931
                if directive.upper() == 'TODO':
931
932
                    result = 'xfail'
932
 
                elif directive == 'SKIP':
 
933
                elif directive.upper() == 'SKIP':
933
934
                    result = 'skip'
934
935
                if directive_comment is not None:
935
936
                    log.append(directive_comment)
1045
1046
            subunit input is not forwarded.
1046
1047
        """
1047
1048
        self._stream = stream
 
1049
        _make_stream_binary(stream)
1048
1050
        self._passthrough = passthrough
1049
1051
        self._forward = forward
 
1052
        _make_stream_binary(forward)
1050
1053
 
1051
1054
    def __call__(self, result=None):
1052
1055
        return self.run(result)
1062
1065
        protocol.lostConnection()
1063
1066
 
1064
1067
 
1065
 
class TestResultStats(unittest.TestResult):
 
1068
class TestResultStats(testresult.TestResult):
1066
1069
    """A pyunit TestResult interface implementation for making statistics.
1067
1070
    
1068
1071
    :ivar total_tests: The total tests seen.
1073
1076
 
1074
1077
    def __init__(self, stream):
1075
1078
        """Create a TestResultStats which outputs to stream."""
1076
 
        unittest.TestResult.__init__(self)
 
1079
        testresult.TestResult.__init__(self)
1077
1080
        self._stream = stream
1078
1081
        self.failed_tests = 0
1079
1082
        self.skipped_tests = 0
1124
1127
    else:
1125
1128
        return sys.stdout
1126
1129
 
 
1130
 
 
1131
def _make_stream_binary(stream):
 
1132
    """Ensure that a stream will be binary safe. See _make_binary_on_windows."""
 
1133
    if getattr(stream, 'fileno', None) is not None:
 
1134
        _make_binary_on_windows(stream.fileno())
 
1135
 
 
1136
def _make_binary_on_windows(fileno):
 
1137
    """Win32 mangles \r\n to \n and that breaks streams. See bug lp:505078."""
 
1138
    if sys.platform == "win32":
 
1139
        import msvcrt
 
1140
        msvcrt.setmode(fileno, os.O_BINARY)