git.samba.org - third_party/subunit/blob - lib/subunit/__init__.py

   1 #
   2 #  subunit: extensions to python unittest to get test results from subprocesses.
   3 #  Copyright (C) 2005  Robert Collins <robertc@robertcollins.net>
   4 #
   5 #  This program is free software; you can redistribute it and/or modify
   6 #  it under the terms of the GNU General Public License as published by
   7 #  the Free Software Foundation; either version 2 of the License, or
   8 #  (at your option) any later version.
   9 #
  10 #  This program is distributed in the hope that it will be useful,
  11 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #  GNU General Public License for more details.
  14 #
  15 #  You should have received a copy of the GNU General Public License
  16 #  along with this program; if not, write to the Free Software
  17 #  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  18 #
  19
  20 import sys
  21 import unittest
  22
  23 def test_suite():
  24     import subunit.tests
  25     return subunit.tests.test_suite()
  26
  27 class TestProtocolServer(object):
  28     """A class for recieving results from a TestProtocol client."""
  29
  30     OUTSIDE_TEST = 0
  31     TEST_STARTED = 1
  32     READING_FAILURE = 2
  33     READING_ERROR = 3
  34
  35     def __init__(self):
  36         self.state = TestProtocolServer.OUTSIDE_TEST
  37
  38     def _addError(self, offset, line):
  39         if (self.state == TestProtocolServer.TEST_STARTED and
  40             self.current_test_description == line[offset:-1]):
  41             self.state = TestProtocolServer.OUTSIDE_TEST
  42             self.current_test_description = None
  43             self.addError(self._current_test, RemoteError(""))
  44         elif (self.state == TestProtocolServer.TEST_STARTED and
  45             self.current_test_description + " [" == line[offset:-1]):
  46             self.state = TestProtocolServer.READING_ERROR
  47             self._message = ""
  48         else:
  49             self.stdOutLineRecieved(line)
  50
  51     def _addFailure(self, offset, line):
  52         if (self.state == TestProtocolServer.TEST_STARTED and
  53             self.current_test_description == line[offset:-1]):
  54             self.state = TestProtocolServer.OUTSIDE_TEST
  55             self.current_test_description = None
  56             self.addFailure(self._current_test, RemoteError())
  57         elif (self.state == TestProtocolServer.TEST_STARTED and
  58             self.current_test_description + " [" == line[offset:-1]):
  59             self.state = TestProtocolServer.READING_FAILURE
  60             self._message = ""
  61         else:
  62             self.stdOutLineRecieved(line)
  63
  64     def _addSuccess(self, offset, line):
  65         if (self.state == TestProtocolServer.TEST_STARTED and
  66             self.current_test_description == line[offset:-1]):
  67             self.addSuccess(self._current_test)
  68             self.current_test_description = None
  69             self._current_test = None
  70             self.state = TestProtocolServer.OUTSIDE_TEST
  71         else:
  72             self.stdOutLineRecieved(line)
  73
  74     def _appendMessage(self, line):
  75         if line[0:2] == " ]":
  76             # quoted ] start
  77             self._message += line[1:]
  78         else:
  79             self._message += line
  80
  81     def endQuote(self, line):
  82         if self.state == TestProtocolServer.READING_FAILURE:
  83             self.state = TestProtocolServer.OUTSIDE_TEST
  84             self.current_test_description = None
  85             self.addFailure(self._current_test, RemoteError(self._message))
  86         elif self.state == TestProtocolServer.READING_ERROR:
  87             self.state = TestProtocolServer.OUTSIDE_TEST
  88             self.current_test_description = None
  89             self.addError(self._current_test, RemoteError(self._message))
  90         else:
  91             self.stdOutLineRecieved(line)
  92
  93     def lineReceived(self, line):
  94         """Call the appropriate local method for the recieved line."""
  95         if line == "]\n":
  96             self.endQuote(line)
  97         elif (self.state == TestProtocolServer.READING_FAILURE or
  98               self.state == TestProtocolServer.READING_ERROR):
  99             self._appendMessage(line)
 100         elif line.startswith("test:"):
 101             self._startTest(6, line)
 102         elif line.startswith("testing:"):
 103             self._startTest(9, line)
 104         elif line.startswith("testing"):
 105             self._startTest(8, line)
 106         elif line.startswith("test"):
 107             self._startTest(5, line)
 108         elif line.startswith("error:"):
 109             self._addError(7, line)
 110         elif line.startswith("error"):
 111             self._addError(6, line)
 112         elif line.startswith("failure:"):
 113             self._addFailure(9, line)
 114         elif line.startswith("failure"):
 115             self._addFailure(8, line)
 116         elif line.startswith("successful:"):
 117             self._addSuccess(12, line)
 118         elif line.startswith("successful"):
 119             self._addSuccess(11, line)
 120         elif line.startswith("success:"):
 121             self._addSuccess(9, line)
 122         elif line.startswith("success"):
 123             self._addSuccess(8, line)
 124         else:
 125             self.stdOutLineRecieved(line)
 126
 127     def lostConnection(self):
 128         """The input connection has finished."""
 129         if self.state == TestProtocolServer.TEST_STARTED:
 130             self.addError(self._current_test,
 131                           RemoteError("lost connection during test '%s'"
 132                                       % self.current_test_description))
 133         elif self.state == TestProtocolServer.READING_ERROR:
 134             self.addError(self._current_test,
 135                           RemoteError("lost connection during "
 136                                       "error report of test "
 137                                       "'%s'" % self.current_test_description))
 138         elif self.state == TestProtocolServer.READING_FAILURE:
 139             self.addError(self._current_test,
 140                           RemoteError("lost connection during "
 141                                       "failure report of test "
 142                                       "'%s'" % self.current_test_description))
 143
 144     def readFrom(self, pipe):
 145         for line in pipe.readlines(pipe):
 146             self.lineReceived(line)
 147         self.lostConnection()
 148
 149     def _startTest(self, offset, line):
 150         """Internal call to change state machine. Override startTest()."""
 151         if self.state == TestProtocolServer.OUTSIDE_TEST:
 152             self.state = TestProtocolServer.TEST_STARTED
 153             self._current_test = RemotedTestCase(line[offset:-1])
 154             self.current_test_description = line[offset:-1]
 155             self.startTest(self._current_test)
 156         else:
 157             self.stdOutLineRecieved(line)
 158
 159     def stdOutLineRecieved(self, line):
 160         sys.stdout.write(line)
 161
 162
 163 class RemoteException(Exception):
 164     """An exception that occured remotely to python."""
 165
 166     def __eq__(self, other):
 167         try:
 168             return self.args == other.args
 169         except AttributeError:
 170             return False
 171
 172
 173 def RemoteError(description=""):
 174     if description == "":
 175         description = "\n"
 176     return (RemoteException("RemoteError:\n%s" % description), None, None)
 177
 178
 179 class RemotedTestCase(unittest.TestCase):
 180     """A class to represent test cases run in child processes."""
 181
 182     def __eq__ (self, other):
 183         try:
 184             return self.__description == other.__description
 185         except AttributeError:
 186             return False
 187
 188     def __init__(self, description):
 189         """Create a psuedo test case with description description."""
 190         self.__description = description
 191
 192     def error(self, label):
 193         raise NotImplementedError("%s on RemotedTestCases is not permitted." %
 194             label)
 195
 196     def setUp(self):
 197         self.error("setUp")
 198
 199     def tearDown(self):
 200         self.error("tearDown")
 201
 202     def shortDescription(self):
 203         return self.__description
 204
 205     def id(self):
 206         return "%s.%s" % (self._strclass(), self.__description)
 207
 208     def __str__(self):
 209         return "%s (%s)" % (self.__description, self._strclass())
 210
 211     def __repr__(self):
 212         return "<%s description='%s'>" % \
 213                (self._strclass(), self.__description)
 214
 215     def run(self, result=None):
 216         if result is None: result = self.defaultTestResult()
 217         result.startTest(self)
 218         result.addError(self, RemoteError("Cannot run RemotedTestCases.\n"))
 219         result.stopTest(self)
 220
 221     def _strclass(self):
 222         cls = self.__class__
 223         return "%s.%s" % (cls.__module__, cls.__name__)