--- /dev/null
+#
+# subunit: extensions to python unittest to get test results from subprocesses.
+# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
+# Copyright (C) 2007 Jelmer Vernooij <jelmer@samba.org>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+#
+
+import os
+from StringIO import StringIO
+import sys
+import unittest
+
+def test_suite():
+ import subunit.tests
+ return subunit.tests.test_suite()
+
+
+def join_dir(base_path, path):
+ """
+ Returns an absolute path to C{path}, calculated relative to the parent
+ of C{base_path}.
+
+ @param base_path: A path to a file or directory.
+ @param path: An absolute path, or a path relative to the containing
+ directory of C{base_path}.
+
+ @return: An absolute path to C{path}.
+ """
+ return os.path.join(os.path.dirname(os.path.abspath(base_path)), path)
+
+
+class TestProtocolServer(object):
+ """A class for receiving results from a TestProtocol client."""
+
+ OUTSIDE_TEST = 0
+ TEST_STARTED = 1
+ READING_FAILURE = 2
+ READING_ERROR = 3
+
+ def __init__(self, client, stream=sys.stdout):
+ """Create a TestProtocol server instance.
+
+ client should be an object that provides
+ - startTest
+ - addSuccess
+ - addFailure
+ - addError
+ - stopTest
+ methods, i.e. a TestResult.
+ """
+ self.state = TestProtocolServer.OUTSIDE_TEST
+ self.client = client
+ self._stream = stream
+
+ def _addError(self, offset, line):
+ if (self.state == TestProtocolServer.TEST_STARTED and
+ self.current_test_description == line[offset:-1]):
+ self.state = TestProtocolServer.OUTSIDE_TEST
+ self.current_test_description = None
+ self.client.addError(self._current_test, RemoteError(""))
+ self.client.stopTest(self._current_test)
+ self._current_test = None
+ elif (self.state == TestProtocolServer.TEST_STARTED and
+ self.current_test_description + " [" == line[offset:-1]):
+ self.state = TestProtocolServer.READING_ERROR
+ self._message = ""
+ else:
+ self.stdOutLineReceived(line)
+
+ def _addFailure(self, offset, line):
+ if (self.state == TestProtocolServer.TEST_STARTED and
+ self.current_test_description == line[offset:-1]):
+ self.state = TestProtocolServer.OUTSIDE_TEST
+ self.current_test_description = None
+ self.client.addFailure(self._current_test, RemoteError())
+ self.client.stopTest(self._current_test)
+ elif (self.state == TestProtocolServer.TEST_STARTED and
+ self.current_test_description + " [" == line[offset:-1]):
+ self.state = TestProtocolServer.READING_FAILURE
+ self._message = ""
+ else:
+ self.stdOutLineReceived(line)
+
+ def _addSuccess(self, offset, line):
+ if (self.state == TestProtocolServer.TEST_STARTED and
+ self.current_test_description == line[offset:-1]):
+ self.client.addSuccess(self._current_test)
+ self.client.stopTest(self._current_test)
+ self.current_test_description = None
+ self._current_test = None
+ self.state = TestProtocolServer.OUTSIDE_TEST
+ else:
+ self.stdOutLineReceived(line)
+
+ def _appendMessage(self, line):
+ if line[0:2] == " ]":
+ # quoted ] start
+ self._message += line[1:]
+ else:
+ self._message += line
+
+ def endQuote(self, line):
+ if self.state == TestProtocolServer.READING_FAILURE:
+ self.state = TestProtocolServer.OUTSIDE_TEST
+ self.current_test_description = None
+ self.client.addFailure(self._current_test,
+ RemoteError(self._message))
+ self.client.stopTest(self._current_test)
+ elif self.state == TestProtocolServer.READING_ERROR:
+ self.state = TestProtocolServer.OUTSIDE_TEST
+ self.current_test_description = None
+ self.client.addError(self._current_test,
+ RemoteError(self._message))
+ self.client.stopTest(self._current_test)
+ else:
+ self.stdOutLineReceived(line)
+
+ def lineReceived(self, line):
+ """Call the appropriate local method for the received line."""
+ if line == "]\n":
+ self.endQuote(line)
+ elif (self.state == TestProtocolServer.READING_FAILURE or
+ self.state == TestProtocolServer.READING_ERROR):
+ self._appendMessage(line)
+ else:
+ parts = line.split(None, 1)
+ if len(parts) == 2:
+ cmd, rest = parts
+ offset = len(cmd) + 1
+ cmd = cmd.strip(':')
+ if cmd in ('test', 'testing'):
+ self._startTest(offset, line)
+ elif cmd == 'error':
+ self._addError(offset, line)
+ elif cmd == 'failure':
+ self._addFailure(offset, line)
+ elif cmd in ('success', 'successful'):
+ self._addSuccess(offset, line)
+ else:
+ self.stdOutLineReceived(line)
+ else:
+ self.stdOutLineReceived(line)
+
+ def lostConnection(self):
+ """The input connection has finished."""
+ if self.state == TestProtocolServer.TEST_STARTED:
+ self.client.addError(self._current_test,
+ RemoteError("lost connection during test '%s'"
+ % self.current_test_description))
+ self.client.stopTest(self._current_test)
+ elif self.state == TestProtocolServer.READING_ERROR:
+ self.client.addError(self._current_test,
+ RemoteError("lost connection during "
+ "error report of test "
+ "'%s'" %
+ self.current_test_description))
+ self.client.stopTest(self._current_test)
+ elif self.state == TestProtocolServer.READING_FAILURE:
+ self.client.addError(self._current_test,
+ RemoteError("lost connection during "
+ "failure report of test "
+ "'%s'" %
+ self.current_test_description))
+ self.client.stopTest(self._current_test)
+
+ def readFrom(self, pipe):
+ for line in pipe.readlines():
+ self.lineReceived(line)
+ self.lostConnection()
+
+ def _startTest(self, offset, line):
+ """Internal call to change state machine. Override startTest()."""
+ if self.state == TestProtocolServer.OUTSIDE_TEST:
+ self.state = TestProtocolServer.TEST_STARTED
+ self._current_test = RemotedTestCase(line[offset:-1])
+ self.current_test_description = line[offset:-1]
+ self.client.startTest(self._current_test)
+ else:
+ self.stdOutLineReceived(line)
+
+ def stdOutLineReceived(self, line):
+ self._stream.write(line)
+
+
+class RemoteException(Exception):
+ """An exception that occured remotely to python."""
+
+ def __eq__(self, other):
+ try:
+ return self.args == other.args
+ except AttributeError:
+ return False
+
+
+class TestProtocolClient(unittest.TestResult):
+ """A class that looks like a TestResult and informs a TestProtocolServer."""
+
+ def __init__(self, stream):
+ super(TestProtocolClient, self).__init__()
+ self._stream = stream
+
+ def addError(self, test, error):
+ """Report an error in test test."""
+ self._stream.write("error: %s [\n" % (test.shortDescription() or str(test)))
+ for line in self._exc_info_to_string(error, test).splitlines():
+ self._stream.write("%s\n" % line)
+ self._stream.write("]\n")
+ super(TestProtocolClient, self).addError(test, error)
+
+ def addFailure(self, test, error):
+ """Report a failure in test test."""
+ self._stream.write("failure: %s [\n" % (test.shortDescription() or str(test)))
+ for line in self._exc_info_to_string(error, test).splitlines():
+ self._stream.write("%s\n" % line)
+ self._stream.write("]\n")
+ super(TestProtocolClient, self).addFailure(test, error)
+
+ def addSuccess(self, test):
+ """Report a success in a test."""
+ self._stream.write("successful: %s\n" % (test.shortDescription() or str(test)))
+ super(TestProtocolClient, self).addSuccess(test)
+
+ def startTest(self, test):
+ """Mark a test as starting its test run."""
+ self._stream.write("test: %s\n" % (test.shortDescription() or str(test)))
+ super(TestProtocolClient, self).startTest(test)
+
+
+def RemoteError(description=""):
+ if description == "":
+ description = "\n"
+ return (RemoteException, RemoteException(description), None)
+
+
+class RemotedTestCase(unittest.TestCase):
+ """A class to represent test cases run in child processes."""
+
+ def __eq__ (self, other):
+ try:
+ return self.__description == other.__description
+ except AttributeError:
+ return False
+
+ def __init__(self, description):
+ """Create a psuedo test case with description description."""
+ self.__description = description
+
+ def error(self, label):
+ raise NotImplementedError("%s on RemotedTestCases is not permitted." %
+ label)
+
+ def setUp(self):
+ self.error("setUp")
+
+ def tearDown(self):
+ self.error("tearDown")
+
+ def shortDescription(self):
+ return self.__description
+
+ def id(self):
+ return "%s.%s" % (self._strclass(), self.__description)
+
+ def __str__(self):
+ return "%s (%s)" % (self.__description, self._strclass())
+
+ def __repr__(self):
+ return "<%s description='%s'>" % \
+ (self._strclass(), self.__description)
+
+ def run(self, result=None):
+ if result is None: result = self.defaultTestResult()
+ result.startTest(self)
+ result.addError(self, RemoteError("Cannot run RemotedTestCases.\n"))
+ result.stopTest(self)
+
+ def _strclass(self):
+ cls = self.__class__
+ return "%s.%s" % (cls.__module__, cls.__name__)
+
+
+class ExecTestCase(unittest.TestCase):
+ """A test case which runs external scripts for test fixtures."""
+
+ def __init__(self, methodName='runTest'):
+ """Create an instance of the class that will use the named test
+ method when executed. Raises a ValueError if the instance does
+ not have a method with the specified name.
+ """
+ unittest.TestCase.__init__(self, methodName)
+ testMethod = getattr(self, methodName)
+ self.script = join_dir(sys.modules[self.__class__.__module__].__file__,
+ testMethod.__doc__)
+
+ def countTestCases(self):
+ return 1
+
+ def run(self, result=None):
+ if result is None: result = self.defaultTestResult()
+ self._run(result)
+
+ def debug(self):
+ """Run the test without collecting errors in a TestResult"""
+ self._run(unittest.TestResult())
+
+ def _run(self, result):
+ protocol = TestProtocolServer(result)
+ output = os.popen(self.script, mode='r')
+ protocol.readFrom(output)
+
+
+class IsolatedTestCase(unittest.TestCase):
+ """A TestCase which runs its tests in a forked process."""
+
+ def run(self, result=None):
+ if result is None: result = self.defaultTestResult()
+ run_isolated(unittest.TestCase, self, result)
+
+
+class IsolatedTestSuite(unittest.TestSuite):
+ """A TestCase which runs its tests in a forked process."""
+
+ def run(self, result=None):
+ if result is None: result = unittest.TestResult()
+ run_isolated(unittest.TestSuite, self, result)
+
+
+def run_isolated(klass, self, result):
+ """Run a test suite or case in a subprocess, using the run method on klass.
+ """
+ c2pread, c2pwrite = os.pipe()
+ # fixme - error -> result
+ # now fork
+ pid = os.fork()
+ if pid == 0:
+ # Child
+ # Close parent's pipe ends
+ os.close(c2pread)
+ # Dup fds for child
+ os.dup2(c2pwrite, 1)
+ # Close pipe fds.
+ os.close(c2pwrite)
+
+ # at this point, sys.stdin is redirected, now we want
+ # to filter it to escape ]'s.
+ ### XXX: test and write that bit.
+
+ result = TestProtocolClient(sys.stdout)
+ klass.run(self, result)
+ sys.stdout.flush()
+ sys.stderr.flush()
+ # exit HARD, exit NOW.
+ os._exit(0)
+ else:
+ # Parent
+ # Close child pipe ends
+ os.close(c2pwrite)
+ # hookup a protocol engine
+ protocol = TestProtocolServer(result)
+ protocol.readFrom(os.fdopen(c2pread, 'rU'))
+ os.waitpid(pid, 0)
+ # TODO return code evaluation.
+ return result
+
+
+class SubunitTestRunner(object):
+ def __init__(self, stream=sys.stdout):
+ self.stream = stream
+
+ def run(self, test):
+ "Run the given test case or test suite."
+ result = TestProtocolClient(self.stream)
+ test(result)
+ return result
+
git.samba.org / nivanova / samba-autobuild / .git / blobdiff