diff options
Diffstat (limited to 'lib/subunit/python/subunit/tests/test_test_protocol.py')
| -rw-r--r-- | lib/subunit/python/subunit/tests/test_test_protocol.py | 555 |
1 files changed, 336 insertions, 219 deletions
diff --git a/lib/subunit/python/subunit/tests/test_test_protocol.py b/lib/subunit/python/subunit/tests/test_test_protocol.py index e1287b6c81..03d921abf1 100644 --- a/lib/subunit/python/subunit/tests/test_test_protocol.py +++ b/lib/subunit/python/subunit/tests/test_test_protocol.py @@ -6,7 +6,7 @@ # license at the users choice. A copy of both licenses are available in the # project source as Apache-2.0 and BSD. You may not use this file except in # compliance with one of these two licences. -# +# # Unless required by applicable law or agreed to in writing, software # distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the @@ -16,10 +16,10 @@ import datetime import unittest -from StringIO import StringIO import os -import sys +from testtools import skipIf, TestCase +from testtools.compat import _b, _u, BytesIO, StringIO from testtools.content import Content, TracebackContent from testtools.content_type import ContentType from testtools.tests.helpers import ( @@ -29,7 +29,7 @@ from testtools.tests.helpers import ( ) import subunit -from subunit import _remote_exception_str +from subunit import _remote_exception_str, _remote_exception_str_chunked import subunit.iso8601 as iso8601 @@ -56,41 +56,39 @@ class TestProtocolServerForward(unittest.TestCase): def test_story(self): client = unittest.TestResult() - out = StringIO() + out = BytesIO() protocol = subunit.TestProtocolServer(client, forward_stream=out) - pipe = StringIO("test old mcdonald\n" - "success old mcdonald\n") + pipe = BytesIO(_b("test old mcdonald\n" + "success old mcdonald\n")) protocol.readFrom(pipe) - mcdonald = subunit.RemotedTestCase("old mcdonald") self.assertEqual(client.testsRun, 1) self.assertEqual(pipe.getvalue(), out.getvalue()) def test_not_command(self): client = unittest.TestResult() - out = StringIO() + out = BytesIO() protocol = subunit.TestProtocolServer(client, stream=subunit.DiscardStream(), forward_stream=out) - pipe = StringIO("success old mcdonald\n") + pipe = BytesIO(_b("success old mcdonald\n")) protocol.readFrom(pipe) self.assertEqual(client.testsRun, 0) - self.assertEqual("", out.getvalue()) - + self.assertEqual(_b(""), out.getvalue()) + class TestTestProtocolServerPipe(unittest.TestCase): def test_story(self): client = unittest.TestResult() protocol = subunit.TestProtocolServer(client) - pipe = StringIO("test old mcdonald\n" + pipe = BytesIO(_b("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") + "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, @@ -110,29 +108,32 @@ class TestTestProtocolServerStartTest(unittest.TestCase): def setUp(self): self.client = Python26TestResult() - self.protocol = subunit.TestProtocolServer(self.client) + self.stream = BytesIO() + self.protocol = subunit.TestProtocolServer(self.client, self.stream) def test_start_test(self): - self.protocol.lineReceived("test old mcdonald\n") + self.protocol.lineReceived(_b("test old mcdonald\n")) self.assertEqual(self.client._events, [('startTest', subunit.RemotedTestCase("old mcdonald"))]) def test_start_testing(self): - self.protocol.lineReceived("testing old mcdonald\n") + self.protocol.lineReceived(_b("testing old mcdonald\n")) self.assertEqual(self.client._events, [('startTest', subunit.RemotedTestCase("old mcdonald"))]) def test_start_test_colon(self): - self.protocol.lineReceived("test: old mcdonald\n") + self.protocol.lineReceived(_b("test: old mcdonald\n")) self.assertEqual(self.client._events, [('startTest', subunit.RemotedTestCase("old mcdonald"))]) def test_indented_test_colon_ignored(self): - self.protocol.lineReceived(" test: old mcdonald\n") + ignored_line = _b(" test: old mcdonald\n") + self.protocol.lineReceived(ignored_line) self.assertEqual([], self.client._events) + self.assertEqual(self.stream.getvalue(), ignored_line) def test_start_testing_colon(self): - self.protocol.lineReceived("testing: old mcdonald\n") + self.protocol.lineReceived(_b("testing: old mcdonald\n")) self.assertEqual(self.client._events, [('startTest', subunit.RemotedTestCase("old mcdonald"))]) @@ -140,22 +141,22 @@ class TestTestProtocolServerStartTest(unittest.TestCase): class TestTestProtocolServerPassThrough(unittest.TestCase): def setUp(self): - self.stdout = StringIO() + self.stdout = BytesIO() self.test = subunit.RemotedTestCase("old mcdonald") self.client = ExtendedTestResult() 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" + self.protocol.lineReceived(_b("failure a\n")) + self.protocol.lineReceived(_b("failure: a\n")) + self.protocol.lineReceived(_b("error a\n")) + self.protocol.lineReceived(_b("error: a\n")) + self.protocol.lineReceived(_b("success a\n")) + self.protocol.lineReceived(_b("success: a\n")) + self.protocol.lineReceived(_b("successful a\n")) + self.protocol.lineReceived(_b("successful: a\n")) + self.protocol.lineReceived(_b("]\n")) + self.assertEqual(self.stdout.getvalue(), _b("failure a\n" "failure: a\n" "error a\n" "error: a\n" @@ -163,15 +164,15 @@ class TestTestProtocolServerPassThrough(unittest.TestCase): "success: a\n" "successful a\n" "successful: a\n" - "]\n") + "]\n")) def test_keywords_before_test(self): self.keywords_before_test() self.assertEqual(self.client._events, []) def test_keywords_after_error(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("error old mcdonald\n") + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("error old mcdonald\n")) self.keywords_before_test() self.assertEqual([ ('startTest', self.test), @@ -180,8 +181,8 @@ class TestTestProtocolServerPassThrough(unittest.TestCase): ], self.client._events) def test_keywords_after_failure(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("failure old mcdonald\n") + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("failure old mcdonald\n")) self.keywords_before_test() self.assertEqual(self.client._events, [ ('startTest', self.test), @@ -190,8 +191,8 @@ class TestTestProtocolServerPassThrough(unittest.TestCase): ]) def test_keywords_after_success(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("success old mcdonald\n") + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("success old mcdonald\n")) self.keywords_before_test() self.assertEqual([ ('startTest', self.test), @@ -200,19 +201,19 @@ class TestTestProtocolServerPassThrough(unittest.TestCase): ], self.client._events) 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" + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("failure a\n")) + self.protocol.lineReceived(_b("failure: a\n")) + self.protocol.lineReceived(_b("error a\n")) + self.protocol.lineReceived(_b("error: a\n")) + self.protocol.lineReceived(_b("success a\n")) + self.protocol.lineReceived(_b("success: a\n")) + self.protocol.lineReceived(_b("successful a\n")) + self.protocol.lineReceived(_b("successful: a\n")) + self.protocol.lineReceived(_b("]\n")) + self.protocol.lineReceived(_b("failure old mcdonald\n")) + self.assertEqual(self.stdout.getvalue(), _b("test old mcdonald\n" "failure a\n" "failure: a\n" "error a\n" @@ -221,7 +222,7 @@ class TestTestProtocolServerPassThrough(unittest.TestCase): "success: a\n" "successful a\n" "successful: a\n" - "]\n") + "]\n")) self.assertEqual(self.client._events, [ ('startTest', self.test), ('addFailure', self.test, {}), @@ -231,24 +232,24 @@ class TestTestProtocolServerPassThrough(unittest.TestCase): def test_keywords_during_failure(self): # A smoke test to make sure that the details parsers have control # appropriately. - 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.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("failure: old mcdonald [\n")) + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("failure a\n")) + self.protocol.lineReceived(_b("failure: a\n")) + self.protocol.lineReceived(_b("error a\n")) + self.protocol.lineReceived(_b("error: a\n")) + self.protocol.lineReceived(_b("success a\n")) + self.protocol.lineReceived(_b("success: a\n")) + self.protocol.lineReceived(_b("successful a\n")) + self.protocol.lineReceived(_b("successful: a\n")) + self.protocol.lineReceived(_b(" ]\n")) + self.protocol.lineReceived(_b("]\n")) + self.assertEqual(self.stdout.getvalue(), _b("")) details = {} details['traceback'] = Content(ContentType("text", "x-traceback", {'charset': 'utf8'}), - lambda:[ + lambda:[_b( "test old mcdonald\n" "failure a\n" "failure: a\n" @@ -258,7 +259,7 @@ class TestTestProtocolServerPassThrough(unittest.TestCase): "success: a\n" "successful a\n" "successful: a\n" - "]\n"]) + "]\n")]) self.assertEqual(self.client._events, [ ('startTest', self.test), ('addFailure', self.test, details), @@ -269,7 +270,7 @@ class TestTestProtocolServerPassThrough(unittest.TestCase): """Lines received which cannot be interpreted as any protocol action should be passed through to sys.stdout. """ - bytes = "randombytes\n" + bytes = _b("randombytes\n") self.protocol.lineReceived(bytes) self.assertEqual(self.stdout.getvalue(), bytes) @@ -286,10 +287,10 @@ class TestTestProtocolServerLostConnection(unittest.TestCase): self.assertEqual([], self.client._events) def test_lost_connection_after_start(self): - self.protocol.lineReceived("test old mcdonald\n") + self.protocol.lineReceived(_b("test old mcdonald\n")) self.protocol.lostConnection() failure = subunit.RemoteError( - u"lost connection during test 'old mcdonald'") + _u("lost connection during test 'old mcdonald'")) self.assertEqual([ ('startTest', self.test), ('addError', self.test, failure), @@ -297,21 +298,21 @@ class TestTestProtocolServerLostConnection(unittest.TestCase): ], self.client._events) def test_lost_connected_after_error(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("error old mcdonald\n") + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("error old mcdonald\n")) self.protocol.lostConnection() self.assertEqual([ ('startTest', self.test), - ('addError', self.test, subunit.RemoteError(u"")), + ('addError', self.test, subunit.RemoteError(_u(""))), ('stopTest', self.test), ], self.client._events) def do_connection_lost(self, outcome, opening): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("%s old mcdonald %s" % (outcome, opening)) + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("%s old mcdonald %s" % (outcome, opening))) self.protocol.lostConnection() failure = subunit.RemoteError( - u"lost connection during %s report of test 'old mcdonald'" % + _u("lost connection during %s report of test 'old mcdonald'") % outcome) self.assertEqual([ ('startTest', self.test), @@ -326,12 +327,12 @@ class TestTestProtocolServerLostConnection(unittest.TestCase): self.do_connection_lost("error", "[ multipart\n") def test_lost_connected_after_failure(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("failure old mcdonald\n") + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("failure old mcdonald\n")) self.protocol.lostConnection() self.assertEqual([ ('startTest', self.test), - ('addFailure', self.test, subunit.RemoteError(u"")), + ('addFailure', self.test, subunit.RemoteError(_u(""))), ('stopTest', self.test), ], self.client._events) @@ -342,8 +343,8 @@ class TestTestProtocolServerLostConnection(unittest.TestCase): self.do_connection_lost("failure", "[ multipart\n") def test_lost_connection_after_success(self): - self.protocol.lineReceived("test old mcdonald\n") - self.protocol.lineReceived("success old mcdonald\n") + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("success old mcdonald\n")) self.protocol.lostConnection() self.assertEqual([ ('startTest', self.test), @@ -369,18 +370,24 @@ class TestTestProtocolServerLostConnection(unittest.TestCase): def test_lost_connection_during_xfail_details(self): self.do_connection_lost("xfail", "[ multipart\n") + def test_lost_connection_during_uxsuccess(self): + self.do_connection_lost("uxsuccess", "[\n") + + def test_lost_connection_during_uxsuccess_details(self): + self.do_connection_lost("uxsuccess", "[ multipart\n") + class TestInTestMultipart(unittest.TestCase): def setUp(self): self.client = ExtendedTestResult() self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") - self.test = subunit.RemotedTestCase("mcdonalds farm") + self.protocol.lineReceived(_b("test mcdonalds farm\n")) + self.test = subunit.RemotedTestCase(_u("mcdonalds farm")) def test__outcome_sets_details_parser(self): self.protocol._reading_success_details.details_parser = None - self.protocol._state._outcome(0, "mcdonalds farm [ multipart\n", + self.protocol._state._outcome(0, _b("mcdonalds farm [ multipart\n"), None, self.protocol._reading_success_details) parser = self.protocol._reading_success_details.details_parser self.assertNotEqual(None, parser) @@ -393,11 +400,11 @@ class TestTestProtocolServerAddError(unittest.TestCase): def setUp(self): self.client = ExtendedTestResult() self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") + self.protocol.lineReceived(_b("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.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) details = {} self.assertEqual([ ('startTest', self.test), @@ -412,11 +419,11 @@ class TestTestProtocolServerAddError(unittest.TestCase): self.simple_error_keyword("error:") def test_error_empty_message(self): - self.protocol.lineReceived("error mcdonalds farm [\n") - self.protocol.lineReceived("]\n") + self.protocol.lineReceived(_b("error mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) details = {} details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:[""]) + {'charset': 'utf8'}), lambda:[_b("")]) self.assertEqual([ ('startTest', self.test), ('addError', self.test, details), @@ -424,12 +431,12 @@ class TestTestProtocolServerAddError(unittest.TestCase): ], self.client._events) def error_quoted_bracket(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword) - self.protocol.lineReceived(" ]\n") - self.protocol.lineReceived("]\n") + self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) + self.protocol.lineReceived(_b(" ]\n")) + self.protocol.lineReceived(_b("]\n")) details = {} details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:["]\n"]) + {'charset': 'utf8'}), lambda:[_b("]\n")]) self.assertEqual([ ('startTest', self.test), ('addError', self.test, details), @@ -448,7 +455,7 @@ class TestTestProtocolServerAddFailure(unittest.TestCase): def setUp(self): self.client = ExtendedTestResult() self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") + self.protocol.lineReceived(_b("test mcdonalds farm\n")) self.test = subunit.RemotedTestCase("mcdonalds farm") def assertFailure(self, details): @@ -459,7 +466,7 @@ class TestTestProtocolServerAddFailure(unittest.TestCase): ], self.client._events) def simple_failure_keyword(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm\n" % keyword) + self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) details = {} self.assertFailure(details) @@ -470,20 +477,20 @@ class TestTestProtocolServerAddFailure(unittest.TestCase): self.simple_failure_keyword("failure:") def test_failure_empty_message(self): - self.protocol.lineReceived("failure mcdonalds farm [\n") - self.protocol.lineReceived("]\n") + self.protocol.lineReceived(_b("failure mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) details = {} details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:[""]) + {'charset': 'utf8'}), lambda:[_b("")]) self.assertFailure(details) def failure_quoted_bracket(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword) - self.protocol.lineReceived(" ]\n") - self.protocol.lineReceived("]\n") + self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) + self.protocol.lineReceived(_b(" ]\n")) + self.protocol.lineReceived(_b("]\n")) details = {} details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:["]\n"]) + {'charset': 'utf8'}), lambda:[_b("]\n")]) self.assertFailure(details) def test_failure_quoted_bracket(self): @@ -521,11 +528,11 @@ class TestTestProtocolServerAddxFail(unittest.TestCase): def setup_protocol(self): """Setup the protocol based on self.client.""" self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") + self.protocol.lineReceived(_b("test mcdonalds farm\n")) self.test = self.client._events[-1][-1] def simple_xfail_keyword(self, keyword, as_success): - self.protocol.lineReceived("%s mcdonalds farm\n" % keyword) + self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) self.check_success_or_xfail(as_success) def check_success_or_xfail(self, as_success, error_message=None): @@ -540,13 +547,14 @@ class TestTestProtocolServerAddxFail(unittest.TestCase): if error_message is not None: details['traceback'] = Content( ContentType("text", "x-traceback", {'charset': 'utf8'}), - lambda:[error_message]) + lambda:[_b(error_message)]) if isinstance(self.client, ExtendedTestResult): value = details else: if error_message is not None: - value = subunit.RemoteError(u'Text attachment: traceback\n' - '------------\n' + error_message + '------------\n') + value = subunit.RemoteError(_u("Text attachment: traceback\n" + "------------\n") + _u(error_message) + + _u("------------\n")) else: value = subunit.RemoteError() self.assertEqual([ @@ -580,16 +588,16 @@ class TestTestProtocolServerAddxFail(unittest.TestCase): self.empty_message(False, error_message="") def empty_message(self, as_success, error_message="\n"): - self.protocol.lineReceived("xfail mcdonalds farm [\n") - self.protocol.lineReceived("]\n") + self.protocol.lineReceived(_b("xfail mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) self.check_success_or_xfail(as_success, error_message) def xfail_quoted_bracket(self, keyword, as_success): # This tests it is accepted, but cannot test it is used today, because # of not having a way to expose it in Python so far. - self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword) - self.protocol.lineReceived(" ]\n") - self.protocol.lineReceived("]\n") + self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) + self.protocol.lineReceived(_b(" ]\n")) + self.protocol.lineReceived(_b("]\n")) self.check_success_or_xfail(as_success, "]\n") def test_xfail_quoted_bracket(self): @@ -609,6 +617,121 @@ class TestTestProtocolServerAddxFail(unittest.TestCase): self.xfail_quoted_bracket("xfail:", False) +class TestTestProtocolServerAddunexpectedSuccess(TestCase): + """Tests for the uxsuccess keyword.""" + + def capture_expected_failure(self, test, err): + self._events.append((test, err)) + + def setup_python26(self): + """Setup a test object ready to be xfailed and thunk to success.""" + self.client = Python26TestResult() + self.setup_protocol() + + def setup_python27(self): + """Setup a test object ready to be xfailed.""" + self.client = Python27TestResult() + self.setup_protocol() + + def setup_python_ex(self): + """Setup a test object ready to be xfailed with details.""" + self.client = ExtendedTestResult() + self.setup_protocol() + + def setup_protocol(self): + """Setup the protocol based on self.client.""" + self.protocol = subunit.TestProtocolServer(self.client) + self.protocol.lineReceived(_b("test mcdonalds farm\n")) + self.test = self.client._events[-1][-1] + + def simple_uxsuccess_keyword(self, keyword, as_fail): + self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) + self.check_fail_or_uxsuccess(as_fail) + + def check_fail_or_uxsuccess(self, as_fail, error_message=None): + details = {} + if error_message is not None: + details['traceback'] = Content( + ContentType("text", "x-traceback", {'charset': 'utf8'}), + lambda:[_b(error_message)]) + if isinstance(self.client, ExtendedTestResult): + value = details + else: + value = None + if as_fail: + self.client._events[1] = self.client._events[1][:2] + # The value is generated within the extended to original decorator: + # todo use the testtools matcher to check on this. + self.assertEqual([ + ('startTest', self.test), + ('addFailure', self.test), + ('stopTest', self.test), + ], self.client._events) + elif value: + self.assertEqual([ + ('startTest', self.test), + ('addUnexpectedSuccess', self.test, value), + ('stopTest', self.test), + ], self.client._events) + else: + self.assertEqual([ + ('startTest', self.test), + ('addUnexpectedSuccess', self.test), + ('stopTest', self.test), + ], self.client._events) + + def test_simple_uxsuccess(self): + self.setup_python26() + self.simple_uxsuccess_keyword("uxsuccess", True) + self.setup_python27() + self.simple_uxsuccess_keyword("uxsuccess", False) + self.setup_python_ex() + self.simple_uxsuccess_keyword("uxsuccess", False) + + def test_simple_uxsuccess_colon(self): + self.setup_python26() + self.simple_uxsuccess_keyword("uxsuccess:", True) + self.setup_python27() + self.simple_uxsuccess_keyword("uxsuccess:", False) + self.setup_python_ex() + self.simple_uxsuccess_keyword("uxsuccess:", False) + + def test_uxsuccess_empty_message(self): + self.setup_python26() + self.empty_message(True) + self.setup_python27() + self.empty_message(False) + self.setup_python_ex() + self.empty_message(False, error_message="") + + def empty_message(self, as_fail, error_message="\n"): + self.protocol.lineReceived(_b("uxsuccess mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) + self.check_fail_or_uxsuccess(as_fail, error_message) + + def uxsuccess_quoted_bracket(self, keyword, as_fail): + self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) + self.protocol.lineReceived(_b(" ]\n")) + self.protocol.lineReceived(_b("]\n")) + self.check_fail_or_uxsuccess(as_fail, "]\n") + + def test_uxsuccess_quoted_bracket(self): + self.setup_python26() + self.uxsuccess_quoted_bracket("uxsuccess", True) + self.setup_python27() + self.uxsuccess_quoted_bracket("uxsuccess", False) + self.setup_python_ex() + self.uxsuccess_quoted_bracket("uxsuccess", False) + + def test_uxsuccess_colon_quoted_bracket(self): + self.setup_python26() + self.uxsuccess_quoted_bracket("uxsuccess:", True) + self.setup_python27() + self.uxsuccess_quoted_bracket("uxsuccess:", False) + self.setup_python_ex() + self.uxsuccess_quoted_bracket("uxsuccess:", False) + + class TestTestProtocolServerAddSkip(unittest.TestCase): """Tests for the skip keyword. @@ -620,7 +743,7 @@ class TestTestProtocolServerAddSkip(unittest.TestCase): """Setup a test object ready to be skipped.""" self.client = ExtendedTestResult() self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") + self.protocol.lineReceived(_b("test mcdonalds farm\n")) self.test = self.client._events[-1][-1] def assertSkip(self, reason): @@ -635,7 +758,7 @@ class TestTestProtocolServerAddSkip(unittest.TestCase): ], self.client._events) def simple_skip_keyword(self, keyword): - self.protocol.lineReceived("%s mcdonalds farm\n" % keyword) + self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) self.assertSkip(None) def test_simple_skip(self): @@ -645,17 +768,17 @@ class TestTestProtocolServerAddSkip(unittest.TestCase): self.simple_skip_keyword("skip:") def test_skip_empty_message(self): - self.protocol.lineReceived("skip mcdonalds farm [\n") - self.protocol.lineReceived("]\n") - self.assertSkip("") + self.protocol.lineReceived(_b("skip mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) + self.assertSkip(_b("")) def skip_quoted_bracket(self, keyword): # This tests it is accepted, but cannot test it is used today, because # of not having a way to expose it in Python so far. - self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword) - self.protocol.lineReceived(" ]\n") - self.protocol.lineReceived("]\n") - self.assertSkip("]\n") + self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) + self.protocol.lineReceived(_b(" ]\n")) + self.protocol.lineReceived(_b("]\n")) + self.assertSkip(_b("]\n")) def test_skip_quoted_bracket(self): self.skip_quoted_bracket("skip") @@ -669,11 +792,11 @@ class TestTestProtocolServerAddSuccess(unittest.TestCase): def setUp(self): self.client = ExtendedTestResult() self.protocol = subunit.TestProtocolServer(self.client) - self.protocol.lineReceived("test mcdonalds farm\n") + self.protocol.lineReceived(_b("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.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) self.assertEqual([ ('startTest', self.test), ('addSuccess', self.test), @@ -681,12 +804,6 @@ class TestTestProtocolServerAddSuccess(unittest.TestCase): ], self.client._events) 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): @@ -700,22 +817,22 @@ class TestTestProtocolServerAddSuccess(unittest.TestCase): ], self.client._events) def test_success_empty_message(self): - self.protocol.lineReceived("success mcdonalds farm [\n") - self.protocol.lineReceived("]\n") + self.protocol.lineReceived(_b("success mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) details = {} details['message'] = Content(ContentType("text", "plain"), - lambda:[""]) + lambda:[_b("")]) self.assertSuccess(details) def success_quoted_bracket(self, keyword): # This tests it is accepted, but cannot test it is used today, because # of not having a way to expose it in Python so far. - self.protocol.lineReceived("%s mcdonalds farm [\n" % keyword) - self.protocol.lineReceived(" ]\n") - self.protocol.lineReceived("]\n") + self.protocol.lineReceived(_b("%s mcdonalds farm [\n" % keyword)) + self.protocol.lineReceived(_b(" ]\n")) + self.protocol.lineReceived(_b("]\n")) details = {} details['message'] = Content(ContentType("text", "plain"), - lambda:["]\n"]) + lambda:[_b("]\n")]) self.assertSuccess(details) def test_success_quoted_bracket(self): @@ -730,26 +847,26 @@ class TestTestProtocolServerProgress(unittest.TestCase): def test_progress_accepted_stdlib(self): self.result = Python26TestResult() - self.stream = StringIO() + self.stream = BytesIO() self.protocol = subunit.TestProtocolServer(self.result, stream=self.stream) - self.protocol.lineReceived("progress: 23") - self.protocol.lineReceived("progress: -2") - self.protocol.lineReceived("progress: +4") - self.assertEqual("", self.stream.getvalue()) + self.protocol.lineReceived(_b("progress: 23")) + self.protocol.lineReceived(_b("progress: -2")) + self.protocol.lineReceived(_b("progress: +4")) + self.assertEqual(_b(""), self.stream.getvalue()) def test_progress_accepted_extended(self): # With a progress capable TestResult, progress events are emitted. self.result = ExtendedTestResult() - self.stream = StringIO() + self.stream = BytesIO() self.protocol = subunit.TestProtocolServer(self.result, stream=self.stream) - self.protocol.lineReceived("progress: 23") - self.protocol.lineReceived("progress: push") - self.protocol.lineReceived("progress: -2") - self.protocol.lineReceived("progress: pop") - self.protocol.lineReceived("progress: +4") - self.assertEqual("", self.stream.getvalue()) + self.protocol.lineReceived(_b("progress: 23")) + self.protocol.lineReceived(_b("progress: push")) + self.protocol.lineReceived(_b("progress: -2")) + self.protocol.lineReceived(_b("progress: pop")) + self.protocol.lineReceived(_b("progress: +4")) + self.assertEqual(_b(""), self.stream.getvalue()) self.assertEqual([ ('progress', 23, subunit.PROGRESS_SET), ('progress', None, subunit.PROGRESS_PUSH), @@ -767,33 +884,33 @@ class TestTestProtocolServerStreamTags(unittest.TestCase): self.protocol = subunit.TestProtocolServer(self.client) def test_initial_tags(self): - self.protocol.lineReceived("tags: foo bar:baz quux\n") + self.protocol.lineReceived(_b("tags: foo bar:baz quux\n")) self.assertEqual([ ('tags', set(["foo", "bar:baz", "quux"]), set()), ], self.client._events) def test_minus_removes_tags(self): - self.protocol.lineReceived("tags: -bar quux\n") + self.protocol.lineReceived(_b("tags: -bar quux\n")) self.assertEqual([ ('tags', set(["quux"]), set(["bar"])), ], self.client._events) def test_tags_do_not_get_set_on_test(self): - self.protocol.lineReceived("test mcdonalds farm\n") + self.protocol.lineReceived(_b("test mcdonalds farm\n")) test = self.client._events[0][-1] self.assertEqual(None, getattr(test, 'tags', None)) def test_tags_do_not_get_set_on_global_tags(self): - self.protocol.lineReceived("tags: foo bar\n") - self.protocol.lineReceived("test mcdonalds farm\n") + self.protocol.lineReceived(_b("tags: foo bar\n")) + self.protocol.lineReceived(_b("test mcdonalds farm\n")) test = self.client._events[-1][-1] self.assertEqual(None, getattr(test, 'tags', None)) def test_tags_get_set_on_test_tags(self): - self.protocol.lineReceived("test mcdonalds farm\n") + self.protocol.lineReceived(_b("test mcdonalds farm\n")) test = self.client._events[-1][-1] - self.protocol.lineReceived("tags: foo bar\n") - self.protocol.lineReceived("success mcdonalds farm\n") + self.protocol.lineReceived(_b("tags: foo bar\n")) + self.protocol.lineReceived(_b("success mcdonalds farm\n")) self.assertEqual(None, getattr(test, 'tags', None)) @@ -802,19 +919,19 @@ class TestTestProtocolServerStreamTime(unittest.TestCase): def test_time_accepted_stdlib(self): self.result = Python26TestResult() - self.stream = StringIO() + self.stream = BytesIO() self.protocol = subunit.TestProtocolServer(self.result, stream=self.stream) - self.protocol.lineReceived("time: 2001-12-12 12:59:59Z\n") - self.assertEqual("", self.stream.getvalue()) + self.protocol.lineReceived(_b("time: 2001-12-12 12:59:59Z\n")) + self.assertEqual(_b(""), self.stream.getvalue()) def test_time_accepted_extended(self): self.result = ExtendedTestResult() - self.stream = StringIO() + self.stream = BytesIO() self.protocol = subunit.TestProtocolServer(self.result, stream=self.stream) - self.protocol.lineReceived("time: 2001-12-12 12:59:59Z\n") - self.assertEqual("", self.stream.getvalue()) + self.protocol.lineReceived(_b("time: 2001-12-12 12:59:59Z\n")) + self.assertEqual(_b(""), self.stream.getvalue()) self.assertEqual([ ('time', datetime.datetime(2001, 12, 12, 12, 59, 59, 0, iso8601.Utc())) @@ -850,15 +967,15 @@ class TestRemotedTestCase(unittest.TestCase): class TestRemoteError(unittest.TestCase): def test_eq(self): - error = subunit.RemoteError(u"Something went wrong") - another_error = subunit.RemoteError(u"Something went wrong") - different_error = subunit.RemoteError(u"boo!") + error = subunit.RemoteError(_u("Something went wrong")) + another_error = subunit.RemoteError(_u("Something went wrong")) + different_error = subunit.RemoteError(_u("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(u"")) + self.assertEqual(subunit.RemoteError(), subunit.RemoteError(_u(""))) class TestExecTestCase(unittest.TestCase): @@ -893,7 +1010,7 @@ class TestExecTestCase(unittest.TestCase): bing = subunit.RemotedTestCase("bing crosby") bing_details = {} bing_details['traceback'] = Content(ContentType("text", "x-traceback", - {'charset': 'utf8'}), lambda:["foo.c:53:ERROR invalid state\n"]) + {'charset': 'utf8'}), lambda:[_b("foo.c:53:ERROR invalid state\n")]) an_error = subunit.RemotedTestCase("an error") error_details = {} self.assertEqual([ @@ -917,7 +1034,8 @@ class TestExecTestCase(unittest.TestCase): def test_join_dir(self): sibling = subunit.join_dir(__file__, 'foo') - expected = '%s/foo' % (os.path.split(__file__)[0],) + filedir = os.path.abspath(os.path.dirname(__file__)) + expected = os.path.join(filedir, 'foo') self.assertEqual(sibling, expected) @@ -927,7 +1045,7 @@ class DoExecTestCase(subunit.ExecTestCase): """sample-two-script.py""" -class TestIsolatedTestCase(unittest.TestCase): +class TestIsolatedTestCase(TestCase): class SampleIsolatedTestCase(subunit.IsolatedTestCase): @@ -946,8 +1064,9 @@ class TestIsolatedTestCase(unittest.TestCase): def test_construct(self): - test = self.SampleIsolatedTestCase("test_sets_global_state") + self.SampleIsolatedTestCase("test_sets_global_state") + @skipIf(os.name != "posix", "Need a posix system for forking tests") def test_run(self): result = unittest.TestResult() test = self.SampleIsolatedTestCase("test_sets_global_state") @@ -963,7 +1082,7 @@ class TestIsolatedTestCase(unittest.TestCase): #test.debug() -class TestIsolatedTestSuite(unittest.TestCase): +class TestIsolatedTestSuite(TestCase): class SampleTestToIsolate(unittest.TestCase): @@ -982,8 +1101,9 @@ class TestIsolatedTestSuite(unittest.TestCase): def test_construct(self): - suite = subunit.IsolatedTestSuite() + subunit.IsolatedTestSuite() + @skipIf(os.name != "posix", "Need a posix system for forking tests") def test_run(self): result = unittest.TestResult() suite = subunit.IsolatedTestSuite() @@ -1002,48 +1122,48 @@ class TestIsolatedTestSuite(unittest.TestCase): class TestTestProtocolClient(unittest.TestCase): def setUp(self): - self.io = StringIO() + self.io = BytesIO() self.protocol = subunit.TestProtocolClient(self.io) self.test = TestTestProtocolClient("test_start_test") self.sample_details = {'something':Content( - ContentType('text', 'plain'), lambda:['serialised\nform'])} + ContentType('text', 'plain'), lambda:[_b('serialised\nform')])} self.sample_tb_details = dict(self.sample_details) self.sample_tb_details['traceback'] = TracebackContent( - subunit.RemoteError(u"boo qux"), self.test) + subunit.RemoteError(_u("boo qux")), self.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()) + self.assertEqual(self.io.getvalue(), _b("test: %s\n" % self.test.id())) def test_stop_test(self): # stopTest doesn't output anything. self.protocol.stopTest(self.test) - self.assertEqual(self.io.getvalue(), "") + self.assertEqual(self.io.getvalue(), _b("")) 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()) + self.io.getvalue(), _b("successful: %s\n" % self.test.id())) def test_add_success_details(self): """Test addSuccess on a TestProtocolClient with details.""" self.protocol.addSuccess(self.test, details=self.sample_details) self.assertEqual( - self.io.getvalue(), "successful: %s [ multipart\n" + self.io.getvalue(), _b("successful: %s [ multipart\n" "Content-Type: text/plain\n" "something\n" - "F\r\nserialised\nform0\r\n]\n" % self.test.id()) + "F\r\nserialised\nform0\r\n]\n" % self.test.id())) def test_add_failure(self): """Test addFailure on a TestProtocolClient.""" self.protocol.addFailure( - self.test, subunit.RemoteError(u"boo qux")) + self.test, subunit.RemoteError(_u("boo qux"))) self.assertEqual( self.io.getvalue(), - ('failure: %s [\n' + _remote_exception_str + ': boo qux\n]\n') - % self.test.id()) + _b(('failure: %s [\n' + _remote_exception_str + ': boo qux\n]\n') + % self.test.id())) def test_add_failure_details(self): """Test addFailure on a TestProtocolClient with details.""" @@ -1051,24 +1171,23 @@ class TestTestProtocolClient(unittest.TestCase): self.test, details=self.sample_tb_details) self.assertEqual( self.io.getvalue(), - ("failure: %s [ multipart\n" + _b(("failure: %s [ multipart\n" "Content-Type: text/plain\n" "something\n" "F\r\nserialised\nform0\r\n" "Content-Type: text/x-traceback;charset=utf8,language=python\n" - "traceback\n" - "1A\r\n" + _remote_exception_str + ": boo qux\n0\r\n" - "]\n") % self.test.id()) + "traceback\n" + _remote_exception_str_chunked + ": boo qux\n0\r\n" + "]\n") % self.test.id())) def test_add_error(self): """Test stopTest on a TestProtocolClient.""" self.protocol.addError( - self.test, subunit.RemoteError(u"phwoar crikey")) + self.test, subunit.RemoteError(_u("phwoar crikey"))) self.assertEqual( self.io.getvalue(), - ('error: %s [\n' + + _b(('error: %s [\n' + _remote_exception_str + ": phwoar crikey\n" - "]\n") % self.test.id()) + "]\n") % self.test.id())) def test_add_error_details(self): """Test stopTest on a TestProtocolClient with details.""" @@ -1076,24 +1195,23 @@ class TestTestProtocolClient(unittest.TestCase): self.test, details=self.sample_tb_details) self.assertEqual( self.io.getvalue(), - ("error: %s [ multipart\n" + _b(("error: %s [ multipart\n" "Content-Type: text/plain\n" "something\n" "F\r\nserialised\nform0\r\n" "Content-Type: text/x-traceback;charset=utf8,language=python\n" - "traceback\n" - "1A\r\n" + _remote_exception_str + ": boo qux\n0\r\n" - "]\n") % self.test.id()) + "traceback\n" + _remote_exception_str_chunked + ": boo qux\n0\r\n" + "]\n") % self.test.id())) def test_add_expected_failure(self): """Test addExpectedFailure on a TestProtocolClient.""" self.protocol.addExpectedFailure( - self.test, subunit.RemoteError(u"phwoar crikey")) + self.test, subunit.RemoteError(_u("phwoar crikey"))) self.assertEqual( self.io.getvalue(), - ('xfail: %s [\n' + + _b(('xfail: %s [\n' + _remote_exception_str + ": phwoar crikey\n" - "]\n") % self.test.id()) + "]\n") % self.test.id())) def test_add_expected_failure_details(self): """Test addExpectedFailure on a TestProtocolClient with details.""" @@ -1101,14 +1219,14 @@ class TestTestProtocolClient(unittest.TestCase): self.test, details=self.sample_tb_details) self.assertEqual( self.io.getvalue(), - ("xfail: %s [ multipart\n" + _b(("xfail: %s [ multipart\n" "Content-Type: text/plain\n" "something\n" "F\r\nserialised\nform0\r\n" "Content-Type: text/x-traceback;charset=utf8,language=python\n" - "traceback\n" - "1A\r\n"+ _remote_exception_str + ": boo qux\n0\r\n" - "]\n") % self.test.id()) + "traceback\n" + _remote_exception_str_chunked + ": boo qux\n0\r\n" + "]\n") % self.test.id())) + def test_add_skip(self): """Test addSkip on a TestProtocolClient.""" @@ -1116,64 +1234,63 @@ class TestTestProtocolClient(unittest.TestCase): self.test, "Has it really?") self.assertEqual( self.io.getvalue(), - 'skip: %s [\nHas it really?\n]\n' % self.test.id()) - + _b('skip: %s [\nHas it really?\n]\n' % self.test.id())) + def test_add_skip_details(self): """Test addSkip on a TestProtocolClient with details.""" details = {'reason':Content( - ContentType('text', 'plain'), lambda:['Has it really?'])} - self.protocol.addSkip( - self.test, details=details) + ContentType('text', 'plain'), lambda:[_b('Has it really?')])} + self.protocol.addSkip(self.test, details=details) self.assertEqual( self.io.getvalue(), - "skip: %s [ multipart\n" + _b("skip: %s [ multipart\n" "Content-Type: text/plain\n" "reason\n" "E\r\nHas it really?0\r\n" - "]\n" % self.test.id()) + "]\n" % self.test.id())) def test_progress_set(self): self.protocol.progress(23, subunit.PROGRESS_SET) - self.assertEqual(self.io.getvalue(), 'progress: 23\n') + self.assertEqual(self.io.getvalue(), _b('progress: 23\n')) def test_progress_neg_cur(self): self.protocol.progress(-23, subunit.PROGRESS_CUR) - self.assertEqual(self.io.getvalue(), 'progress: -23\n') + self.assertEqual(self.io.getvalue(), _b('progress: -23\n')) def test_progress_pos_cur(self): self.protocol.progress(23, subunit.PROGRESS_CUR) - self.assertEqual(self.io.getvalue(), 'progress: +23\n') + self.assertEqual(self.io.getvalue(), _b('progress: +23\n')) def test_progress_pop(self): self.protocol.progress(1234, subunit.PROGRESS_POP) - self.assertEqual(self.io.getvalue(), 'progress: pop\n') + self.assertEqual(self.io.getvalue(), _b('progress: pop\n')) def test_progress_push(self): self.protocol.progress(1234, subunit.PROGRESS_PUSH) - self.assertEqual(self.io.getvalue(), 'progress: push\n') + self.assertEqual(self.io.getvalue(), _b('progress: push\n')) def test_time(self): # Calling time() outputs a time signal immediately. self.protocol.time( datetime.datetime(2009,10,11,12,13,14,15, iso8601.Utc())) self.assertEqual( - "time: 2009-10-11 12:13:14.000015Z\n", + _b("time: 2009-10-11 12:13:14.000015Z\n"), self.io.getvalue()) def test_add_unexpected_success(self): """Test addUnexpectedSuccess on a TestProtocolClient.""" self.protocol.addUnexpectedSuccess(self.test) self.assertEqual( - self.io.getvalue(), "successful: %s\n" % self.test.id()) + self.io.getvalue(), _b("uxsuccess: %s\n" % self.test.id())) def test_add_unexpected_success_details(self): """Test addUnexpectedSuccess on a TestProtocolClient with details.""" self.protocol.addUnexpectedSuccess(self.test, details=self.sample_details) self.assertEqual( - self.io.getvalue(), "successful: %s [ multipart\n" + self.io.getvalue(), _b("uxsuccess: %s [ multipart\n" "Content-Type: text/plain\n" "something\n" - "F\r\nserialised\nform0\r\n]\n" % self.test.id()) + "F\r\nserialised\nform0\r\n]\n" % self.test.id())) def test_suite(): |