From: Robert Collins Date: Sun, 28 Aug 2005 12:02:22 +0000 (+1000) Subject: implement an IsolatedTestCase that runs itself in a subprocess X-Git-Url: http://git.samba.org/samba.git/?a=commitdiff_plain;h=c129af1656ab1927af2ab713763541677c9a8ec7;p=third_party%2Fsubunit implement an IsolatedTestCase that runs itself in a subprocess --- diff --git a/README b/README index 7a5c41f..575eae0 100644 --- a/README +++ b/README @@ -67,16 +67,8 @@ error: test label [ 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. - - diff --git a/lib/subunit/__init__.py b/lib/subunit/__init__.py index 02e7eb8..e388350 100644 --- a/lib/subunit/__init__.py +++ b/lib/subunit/__init__.py @@ -17,6 +17,7 @@ # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # +import os from StringIO import StringIO import subprocess import sys @@ -196,6 +197,36 @@ class RemoteException(Exception): 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" @@ -260,20 +291,9 @@ class ExecTestCase(unittest.TestCase): 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) @@ -288,3 +308,50 @@ class ExecTestCase(unittest.TestCase): 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)) +# diff --git a/lib/subunit/tests/test_test_protocol.py b/lib/subunit/tests/test_test_protocol.py index f800ae9..ecd1cdf 100644 --- a/lib/subunit/tests/test_test_protocol.py +++ b/lib/subunit/tests/test_test_protocol.py @@ -116,6 +116,8 @@ class TestTestImports(unittest.TestCase): from subunit import RemotedTestCase from subunit import RemoteError from subunit import ExecTestCase + from subunit import IsolatedTestCase + from subunit import TestProtocolClient class TestTestProtocolServerPipe(unittest.TestCase): @@ -553,6 +555,7 @@ class TestRemoteError(unittest.TestCase): def test_empty_constructor(self): self.assertEqual(subunit.RemoteError(), subunit.RemoteError("")) + class TestExecTestCase(unittest.TestCase): class SampleExecTestCase(subunit.ExecTestCase): @@ -590,12 +593,93 @@ class TestExecTestCase(unittest.TestCase): 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__)