unexpected output on stdout -> stdout.
exit w/0 or last test -> error
-rough structure of the server:
-SubprocessTestCase(unittest.TestCase):
-
+TODO:
def run:
do a fork,
this process runs server
child runs client and calls self.run() with a SubprocessTestResult
-
-developing - write a server and feed it chosen strings DONE. a mock child process etc.
-write a client and check the right things come back for the various permutations
-check that stdout output in the client is passed to the real stdout.
-
-
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
+import os
from StringIO import StringIO
import subprocess
import sys
return False
+class TestProtocolClient(unittest.TestResult):
+ """A class that looks like a TestResult and informs a TestProtocolServer."""
+
+ def __init__(self, stream):
+ unittest.TestResult.__init__(self)
+ self._stream = stream
+
+ def addError(self, test, error):
+ """Report an error in test test."""
+ self._stream.write("error: %s [\n" % test.shortDescription())
+ for line in self._exc_info_to_string(error, test).split():
+ self._stream.write("%s\n" % line)
+ self._stream.write("]\n")
+
+ def addFailure(self, test, error):
+ """Report a failure in test test."""
+ self._stream.write("failure: %s [\n" % test.shortDescription())
+ for line in self._exc_info_to_string(error, test).split():
+ self._stream.write("%s\n" % line)
+ self._stream.write("]\n")
+
+ def addSuccess(self, test):
+ """Report a success in a test."""
+ self._stream.write("successful: %s\n" % test.shortDescription())
+
+ def startTest(self, test):
+ """Mark a test as starting its test run."""
+ self._stream.write("test: %s\n" % test.shortDescription())
+
+
def RemoteError(description=""):
if description == "":
description = "\n"
testMethod = getattr(self, methodName)
self.script = testMethod.__doc__
- def setUp(self):
- "Hook method for setting up the test fixture before exercising it."
- pass
-
- def tearDown(self):
- "Hook method for deconstructing the test fixture after testing it."
- pass
-
def countTestCases(self):
return 1
- def defaultTestResult(self):
- return TestResult()
-
def run(self, result=None):
if result is None: result = self.defaultTestResult()
self._run(result)
stdout=subprocess.PIPE).communicate()[0]
protocol.readFrom(StringIO(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()
+ 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)
+ unittest.TestCase.run(self, result)
+ # exit HARD, exit NOW.
+ os._exit()
+ else:
+ # Parent
+ # Close child pipe ends
+ os.close(c2pwrite)
+ # hookup a protocol engine
+ protocol = TestProtocolServer(result)
+ protocol.readFrom(StringIO(os.fdopen(c2pread, 'rU').read()))
+
+
+#
+# def debug(self):
+# """Run the test without collecting errors in a TestResult"""
+# self._run(unittest.TestCase.debug, unittest.TestResult())
+#
+# def _run(self, base_method, result):
+# protocol = TestProtocolServer(result)
+# output = subprocess.Popen([self.script],
+# stdout=subprocess.PIPE).communicate()[0]
+# protocol.readFrom(StringIO(output))
+#
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_empty_constructor(self):
self.assertEqual(subunit.RemoteError(), subunit.RemoteError(""))
+
class TestExecTestCase(unittest.TestCase):
class SampleExecTestCase(subunit.ExecTestCase):
def test_count_test_cases(self):
"""TODO run the child process and count responses to determine the count."""
+
class DoExecTestCase(subunit.ExecTestCase):
def test_working_script(self):
"""./lib/subunit/tests/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 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: Test startTest on a "
+ "TestProtocolClient.\n")
+
+ 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: Test startTest on a "
+ "TestProtocolClient.\n")
+
+ def test_add_failure(self):
+ """Test addFailure on a TestProtocolClient."""
+ self.protocol.addFailure(self.test, subunit.RemoteError("boo"))
+ self.assertEqual(self.io.getvalue(), "failure: Test startTest on a "
+ "TestProtocolClient. [\n"
+ "RemoteError:\n"
+ "boo\n"
+ "]\n")
+
+ def test_add_error(self):
+ """Test stopTest on a TestProtocolClient."""
+ self.protocol.addError(self.test, subunit.RemoteError("phwoar"))
+ self.assertEqual(self.io.getvalue(), "error: Test startTest on a "
+ "TestProtocolClient. [\n"
+ "RemoteError:\n"
+ "phwoar\n"
+ "]\n")
+
+
def test_suite():
loader = subunit.tests.TestUtil.TestLoader()
result = loader.loadTestsFromName(__name__)
git.samba.org / third_party / subunit / commitdiff