diff options
Diffstat (limited to 'source4/scripting/python/subunit')
6 files changed, 0 insertions, 1241 deletions
diff --git a/source4/scripting/python/subunit/__init__.py b/source4/scripting/python/subunit/__init__.py deleted file mode 100644 index 406cd8765b..0000000000 --- a/source4/scripting/python/subunit/__init__.py +++ /dev/null @@ -1,388 +0,0 @@ -# -# 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 - diff --git a/source4/scripting/python/subunit/tests/TestUtil.py b/source4/scripting/python/subunit/tests/TestUtil.py deleted file mode 100644 index 1b5ba9c293..0000000000 --- a/source4/scripting/python/subunit/tests/TestUtil.py +++ /dev/null @@ -1,80 +0,0 @@ -# Copyright (c) 2004 Canonical Limited -# Author: Robert Collins <robert.collins@canonical.com> -# -# 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 2 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 sys -import logging -import unittest - - -class LogCollector(logging.Handler): - def __init__(self): - logging.Handler.__init__(self) - self.records=[] - def emit(self, record): - self.records.append(record.getMessage()) - - -def makeCollectingLogger(): - """I make a logger instance that collects its logs for programmatic analysis - -> (logger, collector)""" - logger=logging.Logger("collector") - handler=LogCollector() - handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) - logger.addHandler(handler) - return logger, handler - - -def visitTests(suite, visitor): - """A foreign method for visiting the tests in a test suite.""" - for test in suite._tests: - #Abusing types to avoid monkey patching unittest.TestCase. - # Maybe that would be better? - try: - test.visit(visitor) - except AttributeError: - if isinstance(test, unittest.TestCase): - visitor.visitCase(test) - elif isinstance(test, unittest.TestSuite): - visitor.visitSuite(test) - visitTests(test, visitor) - else: - print "unvisitable non-unittest.TestCase element %r (%r)" % (test, test.__class__) - - -class TestSuite(unittest.TestSuite): - """I am an extended TestSuite with a visitor interface. - This is primarily to allow filtering of tests - and suites or - more in the future. An iterator of just tests wouldn't scale...""" - - def visit(self, visitor): - """visit the composite. Visiting is depth-first. - current callbacks are visitSuite and visitCase.""" - visitor.visitSuite(self) - visitTests(self, visitor) - - -class TestLoader(unittest.TestLoader): - """Custome TestLoader to set the right TestSuite class.""" - suiteClass = TestSuite - -class TestVisitor(object): - """A visitor for Tests""" - def visitSuite(self, aTestSuite): - pass - def visitCase(self, aTestCase): - pass diff --git a/source4/scripting/python/subunit/tests/__init__.py b/source4/scripting/python/subunit/tests/__init__.py deleted file mode 100644 index 544d0e704f..0000000000 --- a/source4/scripting/python/subunit/tests/__init__.py +++ /dev/null @@ -1,25 +0,0 @@ -# -# subunit: extensions to python unittest to get test results from subprocesses. -# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net> -# -# 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 2 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 -# - -from subunit.tests import TestUtil, test_test_protocol - -def test_suite(): - result = TestUtil.TestSuite() - result.addTest(test_test_protocol.test_suite()) - return result diff --git a/source4/scripting/python/subunit/tests/sample-script.py b/source4/scripting/python/subunit/tests/sample-script.py deleted file mode 100755 index 223d2f5d9f..0000000000 --- a/source4/scripting/python/subunit/tests/sample-script.py +++ /dev/null @@ -1,11 +0,0 @@ -#!/usr/bin/env python -import sys -print "test old mcdonald" -print "success old mcdonald" -print "test bing crosby" -print "failure bing crosby [" -print "foo.c:53:ERROR invalid state" -print "]" -print "test an error" -print "error an error" -sys.exit(0) diff --git a/source4/scripting/python/subunit/tests/sample-two-script.py b/source4/scripting/python/subunit/tests/sample-two-script.py deleted file mode 100755 index d5550842bf..0000000000 --- a/source4/scripting/python/subunit/tests/sample-two-script.py +++ /dev/null @@ -1,7 +0,0 @@ -#!/usr/bin/env python -import sys -print "test old mcdonald" -print "success old mcdonald" -print "test bing crosby" -print "success bing crosby" -sys.exit(0) diff --git a/source4/scripting/python/subunit/tests/test_test_protocol.py b/source4/scripting/python/subunit/tests/test_test_protocol.py deleted file mode 100644 index af31584a97..0000000000 --- a/source4/scripting/python/subunit/tests/test_test_protocol.py +++ /dev/null @@ -1,730 +0,0 @@ -# -# subunit: extensions to python unittest to get test results from subprocesses. -# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net> -# -# 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 2 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 unittest -from StringIO import StringIO -import os -import subunit -import sys - -try: - class MockTestProtocolServerClient(object): - """A mock protocol server client to test callbacks.""" - - def __init__(self): - self.end_calls = [] - self.error_calls = [] - self.failure_calls = [] - self.start_calls = [] - self.success_calls = [] - super(MockTestProtocolServerClient, self).__init__() - - def addError(self, test, error): - self.error_calls.append((test, error)) - - def addFailure(self, test, error): - self.failure_calls.append((test, error)) - - def addSuccess(self, test): - self.success_calls.append(test) - - def stopTest(self, test): - self.end_calls.append(test) - - def startTest(self, test): - self.start_calls.append(test) - -except AttributeError: - MockTestProtocolServer = None - - -class TestMockTestProtocolServer(unittest.TestCase): - - def test_start_test(self): - protocol = MockTestProtocolServerClient() - protocol.startTest(subunit.RemotedTestCase("test old mcdonald")) - self.assertEqual(protocol.start_calls, - [subunit.RemotedTestCase("test old mcdonald")]) - self.assertEqual(protocol.end_calls, []) - self.assertEqual(protocol.error_calls, []) - self.assertEqual(protocol.failure_calls, []) - self.assertEqual(protocol.success_calls, []) - - def test_add_error(self): - protocol = MockTestProtocolServerClient() - protocol.addError(subunit.RemotedTestCase("old mcdonald"), - subunit.RemoteError("omg it works")) - self.assertEqual(protocol.start_calls, []) - self.assertEqual(protocol.end_calls, []) - self.assertEqual(protocol.error_calls, [( - subunit.RemotedTestCase("old mcdonald"), - subunit.RemoteError("omg it works"))]) - self.assertEqual(protocol.failure_calls, []) - self.assertEqual(protocol.success_calls, []) - - def test_add_failure(self): - protocol = MockTestProtocolServerClient() - protocol.addFailure(subunit.RemotedTestCase("old mcdonald"), - subunit.RemoteError("omg it works")) - self.assertEqual(protocol.start_calls, []) - self.assertEqual(protocol.end_calls, []) - self.assertEqual(protocol.error_calls, []) - self.assertEqual(protocol.failure_calls, [ - (subunit.RemotedTestCase("old mcdonald"), - subunit.RemoteError("omg it works"))]) - self.assertEqual(protocol.success_calls, []) - - def test_add_success(self): - protocol = MockTestProtocolServerClient() - protocol.addSuccess(subunit.RemotedTestCase("test old mcdonald")) - self.assertEqual(protocol.start_calls, []) - self.assertEqual(protocol.end_calls, []) - self.assertEqual(protocol.error_calls, []) - self.assertEqual(protocol.failure_calls, []) - self.assertEqual(protocol.success_calls, - [subunit.RemotedTestCase("test old mcdonald")]) - - def test_end_test(self): - protocol = MockTestProtocolServerClient() - protocol.stopTest(subunit.RemotedTestCase("test old mcdonald")) - self.assertEqual(protocol.end_calls, - [subunit.RemotedTestCase("test old mcdonald")]) - self.assertEqual(protocol.error_calls, []) - self.assertEqual(protocol.failure_calls, []) - self.assertEqual(protocol.success_calls, []) - self.assertEqual(protocol.start_calls, []) - - -class TestTestImports(unittest.TestCase): - - def test_imports(self): - from subunit import TestProtocolServer - from subunit import RemotedTestCase - from subunit import RemoteError - from subunit import ExecTestCase - from subunit import IsolatedTestCase - from subunit import TestProtocolClient - - -class TestTestProtocolServerPipe(unittest.TestCase): - - def test_story(self): - client = unittest.TestResult() - protocol = subunit.TestProtocolServer(client) - pipe = StringIO("test old mcdonald\n" - "success old mcdonald\n" - "test bing crosby\n" - "failure bing crosby [\n" - "foo.c:53:ERROR invalid state\n" - "]\n" - "test an error\n" - "error an error\n") - protocol.readFrom(pipe) - mcdonald = subunit.RemotedTestCase("old mcdonald") - bing = subunit.RemotedTestCase("bing crosby") - an_error = subunit.RemotedTestCase("an error") - self.assertEqual(client.errors, - [(an_error, 'RemoteException: \n\n')]) - self.assertEqual( - client.failures, - [(bing, "RemoteException: foo.c:53:ERROR invalid state\n\n")]) - self.assertEqual(client.testsRun, 3) - - -class TestTestProtocolServerStartTest(unittest.TestCase): - - def setUp(self): - self.client = MockTestProtocolServerClient() - self.protocol = subunit.TestProtocolServer(self.client) - - def test_start_test(self): - self.protocol.lineReceived("test old mcdonald\n") - self.assertEqual(self.client.start_calls, - [subunit.RemotedTestCase("old mcdonald")]) - - def test_start_testing(self): - self.protocol.lineReceived("testing old mcdonald\n") - self.assertEqual(self.client.start_calls, - [subunit.RemotedTestCase("old mcdonald")]) - - def test_start_test_colon(self): - self.protocol.lineReceived("test: old mcdonald\n") - self.assertEqual(self.client.start_calls, - [subunit.RemotedTestCase("old mcdonald")]) - - def test_start_testing_colon(self): - self.protocol.lineReceived("testing: old mcdonald\n") - self.assertEqual(self.client.start_calls, - [subunit.RemotedTestCase("old mcdonald")]) - - -class TestTestProtocolServerPassThrough(unittest.TestCase): - - def setUp(self): - from StringIO import StringIO - self.stdout = StringIO() - self.test = subunit.RemotedTestCase("old mcdonald") - self.client = MockTestProtocolServerClient() - self.protocol = subunit.TestProtocolServer(self.client, self.stdout) - - def keywords_before_test(self): - self.protocol.lineReceived("failure a\n") - self.protocol.lineReceived("failure: a\n") - self.protocol.lineReceived("error a\n") - self.protocol.lineReceived("error: a\n") - self.protocol.lineReceived("success a\n") - self.protocol.lineReceived("success: a\n") - self.protocol.lineReceived("successful a\n") - self.protocol.lineReceived("successful: a\n") - self.protocol.lineReceived("]\n") - self.assertEqual(self.stdout.getvalue(), "failure a\n" - "failure: a\n" - "error a\n" - "error: a\n" - "success a\n" - "success: a\n" - "successful a\n" - "successful: a\n" - "]\n") - - def test_keywords_before_test(self): - self.keywords_before_test() - self.assertEqual(self.client.start_calls, []) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.success_calls, []) - - def test_keywords_after_error(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("error old mcdonald\n") - self.keywords_before_test() - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, - [(self.test, subunit.RemoteError(""))]) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.success_calls, []) - - def test_keywords_after_failure(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("failure old mcdonald\n") - self.keywords_before_test() - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, - [(self.test, subunit.RemoteError())]) - self.assertEqual(self.client.success_calls, []) - - def test_keywords_after_success(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("success old mcdonald\n") - self.keywords_before_test() - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.success_calls, [self.test]) - - def test_keywords_after_test(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("failure a\n") - self.protocol.lineReceived("failure: a\n") - self.protocol.lineReceived("error a\n") - self.protocol.lineReceived("error: a\n") - self.protocol.lineReceived("success a\n") - self.protocol.lineReceived("success: a\n") - self.protocol.lineReceived("successful a\n") - self.protocol.lineReceived("successful: a\n") - self.protocol.lineReceived("]\n") - self.protocol.lineReceived("failure old mcdonald\n") - self.assertEqual(self.stdout.getvalue(), "test old mcdonald\n" - "failure a\n" - "failure: a\n" - "error a\n" - "error: a\n" - "success a\n" - "success: a\n" - "successful a\n" - "successful: a\n" - "]\n") - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.failure_calls, - [(self.test, subunit.RemoteError())]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.success_calls, []) - - def test_keywords_during_failure(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("failure: old mcdonald [\n") - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("failure a\n") - self.protocol.lineReceived("failure: a\n") - self.protocol.lineReceived("error a\n") - self.protocol.lineReceived("error: a\n") - self.protocol.lineReceived("success a\n") - self.protocol.lineReceived("success: a\n") - self.protocol.lineReceived("successful a\n") - self.protocol.lineReceived("successful: a\n") - self.protocol.lineReceived(" ]\n") - self.protocol.lineReceived("]\n") - self.assertEqual(self.stdout.getvalue(), "") - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.failure_calls, - [(self.test, subunit.RemoteError("test old mcdonald\n" - "failure a\n" - "failure: a\n" - "error a\n" - "error: a\n" - "success a\n" - "success: a\n" - "successful a\n" - "successful: a\n" - "]\n"))]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.success_calls, []) - - def test_stdout_passthrough(self): - """Lines received which cannot be interpreted as any protocol action - should be passed through to sys.stdout. - """ - bytes = "randombytes\n" - self.protocol.lineReceived(bytes) - self.assertEqual(self.stdout.getvalue(), bytes) - - -class TestTestProtocolServerLostConnection(unittest.TestCase): - - def setUp(self): - self.client = MockTestProtocolServerClient() - self.protocol = subunit.TestProtocolServer(self.client) - self.test = subunit.RemotedTestCase("old mcdonald") - - def test_lost_connection_no_input(self): - self.protocol.lostConnection() - self.assertEqual(self.client.start_calls, []) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.success_calls, []) - - def test_lost_connection_after_start(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lostConnection() - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, [ - (self.test, subunit.RemoteError("lost connection during " - "test 'old mcdonald'"))]) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.success_calls, []) - - def test_lost_connected_after_error(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("error old mcdonald\n") - self.protocol.lostConnection() - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, [ - (self.test, subunit.RemoteError(""))]) - self.assertEqual(self.client.success_calls, []) - - def test_lost_connection_during_error(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("error old mcdonald [\n") - self.protocol.lostConnection() - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, [ - (self.test, subunit.RemoteError("lost connection during error " - "report of test 'old mcdonald'"))]) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.success_calls, []) - - def test_lost_connected_after_failure(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("failure old mcdonald\n") - self.protocol.lostConnection() - test = subunit.RemotedTestCase("old mcdonald") - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, - [(self.test, subunit.RemoteError())]) - self.assertEqual(self.client.success_calls, []) - - def test_lost_connection_during_failure(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("failure old mcdonald [\n") - self.protocol.lostConnection() - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, - [(self.test, - subunit.RemoteError("lost connection during " - "failure report" - " of test 'old mcdonald'"))]) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.success_calls, []) - - def test_lost_connection_after_success(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("success old mcdonald\n") - self.protocol.lostConnection() - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, []) - self.assertEqual(self.client.success_calls, [self.test]) - - -class TestTestProtocolServerAddError(unittest.TestCase): - - def setUp(self): - self.client = MockTestProtocolServerClient() - self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") - self.test = subunit.RemotedTestCase("mcdonalds farm") - - def simple_error_keyword(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm\n" % keyword) - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, [ - (self.test, subunit.RemoteError(""))]) - self.assertEqual(self.client.failure_calls, []) - - def test_simple_error(self): - self.simple_error_keyword("error") - - def test_simple_error_colon(self): - self.simple_error_keyword("error:") - - def test_error_empty_message(self): - self.protocol.lineReceived("error mcdonalds farm [\n") - self.protocol.lineReceived("]\n") - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, [ - (self.test, subunit.RemoteError(""))]) - self.assertEqual(self.client.failure_calls, []) - - def error_quoted_bracket(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword) - self.protocol.lineReceived(" ]\n") - self.protocol.lineReceived("]\n") - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, [ - (self.test, subunit.RemoteError("]\n"))]) - self.assertEqual(self.client.failure_calls, []) - - def test_error_quoted_bracket(self): - self.error_quoted_bracket("error") - - def test_error_colon_quoted_bracket(self): - self.error_quoted_bracket("error:") - - -class TestTestProtocolServerAddFailure(unittest.TestCase): - - def setUp(self): - self.client = MockTestProtocolServerClient() - self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") - self.test = subunit.RemotedTestCase("mcdonalds farm") - - def simple_failure_keyword(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm\n" % keyword) - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, - [(self.test, subunit.RemoteError())]) - - def test_simple_failure(self): - self.simple_failure_keyword("failure") - - def test_simple_failure_colon(self): - self.simple_failure_keyword("failure:") - - def test_failure_empty_message(self): - self.protocol.lineReceived("failure mcdonalds farm [\n") - self.protocol.lineReceived("]\n") - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, - [(self.test, subunit.RemoteError())]) - - def failure_quoted_bracket(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword) - self.protocol.lineReceived(" ]\n") - self.protocol.lineReceived("]\n") - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.failure_calls, - [(self.test, subunit.RemoteError("]\n"))]) - - def test_failure_quoted_bracket(self): - self.failure_quoted_bracket("failure") - - def test_failure_colon_quoted_bracket(self): - self.failure_quoted_bracket("failure:") - - -class TestTestProtocolServerAddSuccess(unittest.TestCase): - - def setUp(self): - self.client = MockTestProtocolServerClient() - self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") - self.test = subunit.RemotedTestCase("mcdonalds farm") - - def simple_success_keyword(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm\n" % keyword) - self.assertEqual(self.client.start_calls, [self.test]) - self.assertEqual(self.client.end_calls, [self.test]) - self.assertEqual(self.client.error_calls, []) - self.assertEqual(self.client.success_calls, [self.test]) - - def test_simple_success(self): - self.simple_success_keyword("failure") - - def test_simple_success_colon(self): - self.simple_success_keyword("failure:") - - def test_simple_success(self): - self.simple_success_keyword("successful") - - def test_simple_success_colon(self): - self.simple_success_keyword("successful:") - - -class TestRemotedTestCase(unittest.TestCase): - - def test_simple(self): - test = subunit.RemotedTestCase("A test description") - self.assertRaises(NotImplementedError, test.setUp) - self.assertRaises(NotImplementedError, test.tearDown) - self.assertEqual("A test description", - test.shortDescription()) - self.assertEqual("subunit.RemotedTestCase.A test description", - test.id()) - self.assertEqual("A test description (subunit.RemotedTestCase)", "%s" % test) - self.assertEqual("<subunit.RemotedTestCase description=" - "'A test description'>", "%r" % test) - result = unittest.TestResult() - test.run(result) - self.assertEqual([(test, "RemoteException: " - "Cannot run RemotedTestCases.\n\n")], - result.errors) - self.assertEqual(1, result.testsRun) - another_test = subunit.RemotedTestCase("A test description") - self.assertEqual(test, another_test) - different_test = subunit.RemotedTestCase("ofo") - self.assertNotEqual(test, different_test) - self.assertNotEqual(another_test, different_test) - - -class TestRemoteError(unittest.TestCase): - - def test_eq(self): - error = subunit.RemoteError("Something went wrong") - another_error = subunit.RemoteError("Something went wrong") - different_error = subunit.RemoteError("boo!") - self.assertEqual(error, another_error) - self.assertNotEqual(error, different_error) - self.assertNotEqual(different_error, another_error) - - def test_empty_constructor(self): - self.assertEqual(subunit.RemoteError(), subunit.RemoteError("")) - - -class TestExecTestCase(unittest.TestCase): - - class SampleExecTestCase(subunit.ExecTestCase): - - def test_sample_method(self): - """sample-script.py""" - # the sample script runs three tests, one each - # that fails, errors and succeeds - - - def test_construct(self): - test = self.SampleExecTestCase("test_sample_method") - self.assertEqual(test.script, - subunit.join_dir(__file__, 'sample-script.py')) - - def test_run(self): - runner = MockTestProtocolServerClient() - test = self.SampleExecTestCase("test_sample_method") - test.run(runner) - mcdonald = subunit.RemotedTestCase("old mcdonald") - bing = subunit.RemotedTestCase("bing crosby") - an_error = subunit.RemotedTestCase("an error") - self.assertEqual(runner.error_calls, - [(an_error, subunit.RemoteError())]) - self.assertEqual(runner.failure_calls, - [(bing, - subunit.RemoteError( - "foo.c:53:ERROR invalid state\n"))]) - self.assertEqual(runner.start_calls, [mcdonald, bing, an_error]) - self.assertEqual(runner.end_calls, [mcdonald, bing, an_error]) - - def test_debug(self): - test = self.SampleExecTestCase("test_sample_method") - test.debug() - - def test_count_test_cases(self): - """TODO run the child process and count responses to determine the count.""" - - def test_join_dir(self): - sibling = subunit.join_dir(__file__, 'foo') - expected = '%s/foo' % (os.path.split(__file__)[0],) - self.assertEqual(sibling, expected) - - -class DoExecTestCase(subunit.ExecTestCase): - - def test_working_script(self): - """sample-two-script.py""" - - -class TestIsolatedTestCase(unittest.TestCase): - - class SampleIsolatedTestCase(subunit.IsolatedTestCase): - - SETUP = False - TEARDOWN = False - TEST = False - - def setUp(self): - TestIsolatedTestCase.SampleIsolatedTestCase.SETUP = True - - def tearDown(self): - TestIsolatedTestCase.SampleIsolatedTestCase.TEARDOWN = True - - def test_sets_global_state(self): - TestIsolatedTestCase.SampleIsolatedTestCase.TEST = True - - - def test_construct(self): - test = self.SampleIsolatedTestCase("test_sets_global_state") - - def test_run(self): - result = unittest.TestResult() - test = self.SampleIsolatedTestCase("test_sets_global_state") - test.run(result) - self.assertEqual(result.testsRun, 1) - self.assertEqual(self.SampleIsolatedTestCase.SETUP, False) - self.assertEqual(self.SampleIsolatedTestCase.TEARDOWN, False) - self.assertEqual(self.SampleIsolatedTestCase.TEST, False) - - def test_debug(self): - pass - #test = self.SampleExecTestCase("test_sample_method") - #test.debug() - - -class TestIsolatedTestSuite(unittest.TestCase): - - class SampleTestToIsolate(unittest.TestCase): - - SETUP = False - TEARDOWN = False - TEST = False - - def setUp(self): - TestIsolatedTestSuite.SampleTestToIsolate.SETUP = True - - def tearDown(self): - TestIsolatedTestSuite.SampleTestToIsolate.TEARDOWN = True - - def test_sets_global_state(self): - TestIsolatedTestSuite.SampleTestToIsolate.TEST = True - - - def test_construct(self): - suite = subunit.IsolatedTestSuite() - - def test_run(self): - result = unittest.TestResult() - suite = subunit.IsolatedTestSuite() - sub_suite = unittest.TestSuite() - sub_suite.addTest(self.SampleTestToIsolate("test_sets_global_state")) - sub_suite.addTest(self.SampleTestToIsolate("test_sets_global_state")) - suite.addTest(sub_suite) - suite.addTest(self.SampleTestToIsolate("test_sets_global_state")) - suite.run(result) - self.assertEqual(result.testsRun, 3) - self.assertEqual(self.SampleTestToIsolate.SETUP, False) - self.assertEqual(self.SampleTestToIsolate.TEARDOWN, False) - self.assertEqual(self.SampleTestToIsolate.TEST, False) - - -class TestTestProtocolClient(unittest.TestCase): - - def setUp(self): - self.io = StringIO() - self.protocol = subunit.TestProtocolClient(self.io) - self.test = TestTestProtocolClient("test_start_test") - - - def test_start_test(self): - """Test startTest on a TestProtocolClient.""" - self.protocol.startTest(self.test) - self.assertEqual(self.io.getvalue(), "test: %s\n" % self.test.id()) - - def test_stop_test(self): - """Test stopTest on a TestProtocolClient.""" - self.protocol.stopTest(self.test) - self.assertEqual(self.io.getvalue(), "") - - def test_add_success(self): - """Test addSuccess on a TestProtocolClient.""" - self.protocol.addSuccess(self.test) - self.assertEqual( - self.io.getvalue(), "successful: %s\n" % self.test.id()) - - def test_add_failure(self): - """Test addFailure on a TestProtocolClient.""" - self.protocol.addFailure(self.test, subunit.RemoteError("boo")) - self.assertEqual( - self.io.getvalue(), - 'failure: %s [\nRemoteException: boo\n]\n' % self.test.id()) - - def test_add_error(self): - """Test stopTest on a TestProtocolClient.""" - self.protocol.addError(self.test, subunit.RemoteError("phwoar")) - self.assertEqual( - self.io.getvalue(), - 'error: %s [\n' - "RemoteException: phwoar\n" - "]\n" % self.test.id()) - - -def test_suite(): - loader = subunit.tests.TestUtil.TestLoader() - result = loader.loadTestsFromName(__name__) - return result |