diff options
Diffstat (limited to 'src/third_party/wiredtiger/test/3rdparty/python-subunit-0.0.16/python/subunit/tests/test_test_protocol.py')
| -rw-r--r-- | src/third_party/wiredtiger/test/3rdparty/python-subunit-0.0.16/python/subunit/tests/test_test_protocol.py | 1362 |
1 files changed, 1362 insertions, 0 deletions
diff --git a/src/third_party/wiredtiger/test/3rdparty/python-subunit-0.0.16/python/subunit/tests/test_test_protocol.py b/src/third_party/wiredtiger/test/3rdparty/python-subunit-0.0.16/python/subunit/tests/test_test_protocol.py new file mode 100644 index 00000000000..c6008f42eb2 --- /dev/null +++ b/src/third_party/wiredtiger/test/3rdparty/python-subunit-0.0.16/python/subunit/tests/test_test_protocol.py @@ -0,0 +1,1362 @@ +# +# subunit: extensions to Python unittest to get test results from subprocesses. +# Copyright (C) 2005 Robert Collins <robertc@robertcollins.net> +# +# Licensed under either the Apache License, Version 2.0 or the BSD 3-clause +# 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 +# license you chose for the specific language governing permissions and +# limitations under that license. +# + +import datetime +import unittest +import os + +from testtools import PlaceHolder, skipIf, TestCase, TestResult +from testtools.compat import _b, _u, BytesIO +from testtools.content import Content, TracebackContent, text_content +from testtools.content_type import ContentType +try: + from testtools.testresult.doubles import ( + Python26TestResult, + Python27TestResult, + ExtendedTestResult, + ) +except ImportError: + from testtools.tests.helpers import ( + Python26TestResult, + Python27TestResult, + ExtendedTestResult, + ) +from testtools.matchers import Contains + +import subunit +from subunit.tests import ( + _remote_exception_repr, + _remote_exception_str, + _remote_exception_str_chunked, + ) +import subunit.iso8601 as iso8601 + + +def details_to_str(details): + return TestResult()._err_details_to_string(None, details=details) + + +class TestTestImports(unittest.TestCase): + + def test_imports(self): + from subunit import DiscardStream + from subunit import TestProtocolServer + from subunit import RemotedTestCase + from subunit import RemoteError + from subunit import ExecTestCase + from subunit import IsolatedTestCase + from subunit import TestProtocolClient + from subunit import ProtocolTestCase + + +class TestDiscardStream(unittest.TestCase): + + def test_write(self): + subunit.DiscardStream().write("content") + + +class TestProtocolServerForward(unittest.TestCase): + + def test_story(self): + client = unittest.TestResult() + out = BytesIO() + protocol = subunit.TestProtocolServer(client, forward_stream=out) + pipe = BytesIO(_b("test old mcdonald\n" + "success old mcdonald\n")) + protocol.readFrom(pipe) + self.assertEqual(client.testsRun, 1) + self.assertEqual(pipe.getvalue(), out.getvalue()) + + def test_not_command(self): + client = unittest.TestResult() + out = BytesIO() + protocol = subunit.TestProtocolServer(client, + stream=subunit.DiscardStream(), forward_stream=out) + pipe = BytesIO(_b("success old mcdonald\n")) + protocol.readFrom(pipe) + self.assertEqual(client.testsRun, 0) + self.assertEqual(_b(""), out.getvalue()) + + +class TestTestProtocolServerPipe(unittest.TestCase): + + def test_story(self): + client = unittest.TestResult() + protocol = subunit.TestProtocolServer(client) + traceback = "foo.c:53:ERROR invalid state\n" + pipe = BytesIO(_b("test old mcdonald\n" + "success old mcdonald\n" + "test bing crosby\n" + "failure bing crosby [\n" + + traceback + + "]\n" + "test an error\n" + "error an error\n")) + protocol.readFrom(pipe) + bing = subunit.RemotedTestCase("bing crosby") + an_error = subunit.RemotedTestCase("an error") + self.assertEqual(client.errors, + [(an_error, _remote_exception_repr + '\n')]) + self.assertEqual( + client.failures, + [(bing, _remote_exception_repr + ": " + + details_to_str({'traceback': text_content(traceback)}) + "\n")]) + self.assertEqual(client.testsRun, 3) + + def test_non_test_characters_forwarded_immediately(self): + pass + + +class TestTestProtocolServerStartTest(unittest.TestCase): + + def setUp(self): + self.client = Python26TestResult() + self.stream = BytesIO() + self.protocol = subunit.TestProtocolServer(self.client, self.stream) + + def test_start_test(self): + 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(_b("testing old mcdonald\n")) + self.assertEqual(self.client._events, + [('startTest', subunit.RemotedTestCase("old mcdonald"))]) + + def test_start_test_colon(self): + 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): + 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(_b("testing: old mcdonald\n")) + self.assertEqual(self.client._events, + [('startTest', subunit.RemotedTestCase("old mcdonald"))]) + + +class TestTestProtocolServerPassThrough(unittest.TestCase): + + def setUp(self): + 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(_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" + "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._events, []) + + def test_keywords_after_error(self): + 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), + ('addError', self.test, {}), + ('stopTest', self.test), + ], self.client._events) + + def test_keywords_after_failure(self): + 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), + ('addFailure', self.test, {}), + ('stopTest', self.test), + ]) + + def test_keywords_after_success(self): + 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), + ('addSuccess', self.test), + ('stopTest', self.test), + ], self.client._events) + + def test_keywords_after_test(self): + 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" + "error: a\n" + "success a\n" + "success: a\n" + "successful a\n" + "successful: a\n" + "]\n")) + self.assertEqual(self.client._events, [ + ('startTest', self.test), + ('addFailure', self.test, {}), + ('stopTest', self.test), + ]) + + def test_keywords_during_failure(self): + # A smoke test to make sure that the details parsers have control + # appropriately. + 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:[_b( + "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._events, [ + ('startTest', self.test), + ('addFailure', self.test, details), + ('stopTest', self.test), + ]) + + def test_stdout_passthrough(self): + """Lines received which cannot be interpreted as any protocol action + should be passed through to sys.stdout. + """ + bytes = _b("randombytes\n") + self.protocol.lineReceived(bytes) + self.assertEqual(self.stdout.getvalue(), bytes) + + +class TestTestProtocolServerLostConnection(unittest.TestCase): + + def setUp(self): + self.client = Python26TestResult() + 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._events) + + def test_lost_connection_after_start(self): + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lostConnection() + failure = subunit.RemoteError( + _u("lost connection during test 'old mcdonald'")) + self.assertEqual([ + ('startTest', self.test), + ('addError', self.test, failure), + ('stopTest', self.test), + ], self.client._events) + + def test_lost_connected_after_error(self): + 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(""))), + ('stopTest', self.test), + ], self.client._events) + + def do_connection_lost(self, 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'") % + outcome) + self.assertEqual([ + ('startTest', self.test), + ('addError', self.test, failure), + ('stopTest', self.test), + ], self.client._events) + + def test_lost_connection_during_error(self): + self.do_connection_lost("error", "[\n") + + def test_lost_connection_during_error_details(self): + self.do_connection_lost("error", "[ multipart\n") + + def test_lost_connected_after_failure(self): + 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(""))), + ('stopTest', self.test), + ], self.client._events) + + def test_lost_connection_during_failure(self): + self.do_connection_lost("failure", "[\n") + + def test_lost_connection_during_failure_details(self): + self.do_connection_lost("failure", "[ multipart\n") + + def test_lost_connection_after_success(self): + self.protocol.lineReceived(_b("test old mcdonald\n")) + self.protocol.lineReceived(_b("success old mcdonald\n")) + self.protocol.lostConnection() + self.assertEqual([ + ('startTest', self.test), + ('addSuccess', self.test), + ('stopTest', self.test), + ], self.client._events) + + def test_lost_connection_during_success(self): + self.do_connection_lost("success", "[\n") + + def test_lost_connection_during_success_details(self): + self.do_connection_lost("success", "[ multipart\n") + + def test_lost_connection_during_skip(self): + self.do_connection_lost("skip", "[\n") + + def test_lost_connection_during_skip_details(self): + self.do_connection_lost("skip", "[ multipart\n") + + def test_lost_connection_during_xfail(self): + self.do_connection_lost("xfail", "[\n") + + 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(_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, _b("mcdonalds farm [ multipart\n"), + None, self.protocol._reading_success_details) + parser = self.protocol._reading_success_details.details_parser + self.assertNotEqual(None, parser) + self.assertTrue(isinstance(parser, + subunit.details.MultipartDetailsParser)) + + +class TestTestProtocolServerAddError(unittest.TestCase): + + def setUp(self): + self.client = ExtendedTestResult() + self.protocol = subunit.TestProtocolServer(self.client) + self.protocol.lineReceived(_b("test mcdonalds farm\n")) + self.test = subunit.RemotedTestCase("mcdonalds farm") + + def simple_error_keyword(self, keyword): + self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) + details = {} + self.assertEqual([ + ('startTest', self.test), + ('addError', self.test, details), + ('stopTest', self.test), + ], self.client._events) + + 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(_b("error mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) + details = {} + details['traceback'] = Content(ContentType("text", "x-traceback", + {'charset': 'utf8'}), lambda:[_b("")]) + self.assertEqual([ + ('startTest', self.test), + ('addError', self.test, details), + ('stopTest', self.test), + ], self.client._events) + + def error_quoted_bracket(self, keyword): + 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:[_b("]\n")]) + self.assertEqual([ + ('startTest', self.test), + ('addError', self.test, details), + ('stopTest', self.test), + ], self.client._events) + + 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 = ExtendedTestResult() + self.protocol = subunit.TestProtocolServer(self.client) + self.protocol.lineReceived(_b("test mcdonalds farm\n")) + self.test = subunit.RemotedTestCase("mcdonalds farm") + + def assertFailure(self, details): + self.assertEqual([ + ('startTest', self.test), + ('addFailure', self.test, details), + ('stopTest', self.test), + ], self.client._events) + + def simple_failure_keyword(self, keyword): + self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) + details = {} + self.assertFailure(details) + + 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(_b("failure mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) + details = {} + details['traceback'] = Content(ContentType("text", "x-traceback", + {'charset': 'utf8'}), lambda:[_b("")]) + self.assertFailure(details) + + def failure_quoted_bracket(self, keyword): + 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:[_b("]\n")]) + self.assertFailure(details) + + def test_failure_quoted_bracket(self): + self.failure_quoted_bracket("failure") + + def test_failure_colon_quoted_bracket(self): + self.failure_quoted_bracket("failure:") + + +class TestTestProtocolServerAddxFail(unittest.TestCase): + """Tests for the xfail keyword. + + In Python this can thunk through to Success due to stdlib limitations (see + README). + """ + + 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_xfail_keyword(self, keyword, as_success): + 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): + if as_success: + self.assertEqual([ + ('startTest', self.test), + ('addSuccess', self.test), + ('stopTest', self.test), + ], self.client._events) + else: + 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: + if error_message is not None: + value = subunit.RemoteError(details_to_str(details)) + else: + value = subunit.RemoteError() + self.assertEqual([ + ('startTest', self.test), + ('addExpectedFailure', self.test, value), + ('stopTest', self.test), + ], self.client._events) + + def test_simple_xfail(self): + self.setup_python26() + self.simple_xfail_keyword("xfail", True) + self.setup_python27() + self.simple_xfail_keyword("xfail", False) + self.setup_python_ex() + self.simple_xfail_keyword("xfail", False) + + def test_simple_xfail_colon(self): + self.setup_python26() + self.simple_xfail_keyword("xfail:", True) + self.setup_python27() + self.simple_xfail_keyword("xfail:", False) + self.setup_python_ex() + self.simple_xfail_keyword("xfail:", False) + + def test_xfail_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_success, error_message="\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(_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): + self.setup_python26() + self.xfail_quoted_bracket("xfail", True) + self.setup_python27() + self.xfail_quoted_bracket("xfail", False) + self.setup_python_ex() + self.xfail_quoted_bracket("xfail", False) + + def test_xfail_colon_quoted_bracket(self): + self.setup_python26() + self.xfail_quoted_bracket("xfail:", True) + self.setup_python27() + self.xfail_quoted_bracket("xfail:", False) + self.setup_python_ex() + 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. + + In Python this meets the testtools extended TestResult contract. + (See https://launchpad.net/testtools). + """ + + def setUp(self): + """Setup a test object ready to be skipped.""" + self.client = ExtendedTestResult() + self.protocol = subunit.TestProtocolServer(self.client) + self.protocol.lineReceived(_b("test mcdonalds farm\n")) + self.test = self.client._events[-1][-1] + + def assertSkip(self, reason): + details = {} + if reason is not None: + details['reason'] = Content( + ContentType("text", "plain"), lambda:[reason]) + self.assertEqual([ + ('startTest', self.test), + ('addSkip', self.test, details), + ('stopTest', self.test), + ], self.client._events) + + def simple_skip_keyword(self, keyword): + self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) + self.assertSkip(None) + + def test_simple_skip(self): + self.simple_skip_keyword("skip") + + def test_simple_skip_colon(self): + self.simple_skip_keyword("skip:") + + def test_skip_empty_message(self): + 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(_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") + + def test_skip_colon_quoted_bracket(self): + self.skip_quoted_bracket("skip:") + + +class TestTestProtocolServerAddSuccess(unittest.TestCase): + + def setUp(self): + self.client = ExtendedTestResult() + self.protocol = subunit.TestProtocolServer(self.client) + self.protocol.lineReceived(_b("test mcdonalds farm\n")) + self.test = subunit.RemotedTestCase("mcdonalds farm") + + def simple_success_keyword(self, keyword): + self.protocol.lineReceived(_b("%s mcdonalds farm\n" % keyword)) + self.assertEqual([ + ('startTest', self.test), + ('addSuccess', self.test), + ('stopTest', self.test), + ], self.client._events) + + def test_simple_success(self): + self.simple_success_keyword("successful") + + def test_simple_success_colon(self): + self.simple_success_keyword("successful:") + + def assertSuccess(self, details): + self.assertEqual([ + ('startTest', self.test), + ('addSuccess', self.test, details), + ('stopTest', self.test), + ], self.client._events) + + def test_success_empty_message(self): + self.protocol.lineReceived(_b("success mcdonalds farm [\n")) + self.protocol.lineReceived(_b("]\n")) + details = {} + details['message'] = Content(ContentType("text", "plain"), + 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(_b("%s mcdonalds farm [\n" % keyword)) + self.protocol.lineReceived(_b(" ]\n")) + self.protocol.lineReceived(_b("]\n")) + details = {} + details['message'] = Content(ContentType("text", "plain"), + lambda:[_b("]\n")]) + self.assertSuccess(details) + + def test_success_quoted_bracket(self): + self.success_quoted_bracket("success") + + def test_success_colon_quoted_bracket(self): + self.success_quoted_bracket("success:") + + +class TestTestProtocolServerProgress(unittest.TestCase): + """Test receipt of progress: directives.""" + + def test_progress_accepted_stdlib(self): + self.result = Python26TestResult() + self.stream = BytesIO() + self.protocol = subunit.TestProtocolServer(self.result, + stream=self.stream) + 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 = BytesIO() + self.protocol = subunit.TestProtocolServer(self.result, + stream=self.stream) + 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), + ('progress', -2, subunit.PROGRESS_CUR), + ('progress', None, subunit.PROGRESS_POP), + ('progress', 4, subunit.PROGRESS_CUR), + ], self.result._events) + + +class TestTestProtocolServerStreamTags(unittest.TestCase): + """Test managing tags on the protocol level.""" + + def setUp(self): + self.client = ExtendedTestResult() + self.protocol = subunit.TestProtocolServer(self.client) + + def test_initial_tags(self): + 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(_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(_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(_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(_b("test mcdonalds farm\n")) + test = self.client._events[-1][-1] + self.protocol.lineReceived(_b("tags: foo bar\n")) + self.protocol.lineReceived(_b("success mcdonalds farm\n")) + self.assertEqual(None, getattr(test, 'tags', None)) + + +class TestTestProtocolServerStreamTime(unittest.TestCase): + """Test managing time information at the protocol level.""" + + def test_time_accepted_stdlib(self): + self.result = Python26TestResult() + self.stream = BytesIO() + self.protocol = subunit.TestProtocolServer(self.result, + stream=self.stream) + 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 = BytesIO() + self.protocol = subunit.TestProtocolServer(self.result, + stream=self.stream) + 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())) + ], self.result._events) + + +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("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, _remote_exception_repr + ": " + "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(_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(""))) + + +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_sample_method_args(self): + """sample-script.py foo""" + # sample that will run just one test. + + def test_construct(self): + test = self.SampleExecTestCase("test_sample_method") + self.assertEqual(test.script, + subunit.join_dir(__file__, 'sample-script.py')) + + def test_args(self): + result = unittest.TestResult() + test = self.SampleExecTestCase("test_sample_method_args") + test.run(result) + self.assertEqual(1, result.testsRun) + + def test_run(self): + result = ExtendedTestResult() + test = self.SampleExecTestCase("test_sample_method") + test.run(result) + mcdonald = subunit.RemotedTestCase("old mcdonald") + bing = subunit.RemotedTestCase("bing crosby") + bing_details = {} + bing_details['traceback'] = Content(ContentType("text", "x-traceback", + {'charset': 'utf8'}), lambda:[_b("foo.c:53:ERROR invalid state\n")]) + an_error = subunit.RemotedTestCase("an error") + error_details = {} + self.assertEqual([ + ('startTest', mcdonald), + ('addSuccess', mcdonald), + ('stopTest', mcdonald), + ('startTest', bing), + ('addFailure', bing, bing_details), + ('stopTest', bing), + ('startTest', an_error), + ('addError', an_error, error_details), + ('stopTest', an_error), + ], result._events) + + 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') + filedir = os.path.abspath(os.path.dirname(__file__)) + expected = os.path.join(filedir, 'foo') + self.assertEqual(sibling, expected) + + +class DoExecTestCase(subunit.ExecTestCase): + + def test_working_script(self): + """sample-two-script.py""" + + +class TestIsolatedTestCase(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): + 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") + 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(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): + subunit.IsolatedTestSuite() + + @skipIf(os.name != "posix", "Need a posix system for forking tests") + 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(TestCase): + + def setUp(self): + super(TestTestProtocolClient, self).setUp() + self.io = BytesIO() + self.protocol = subunit.TestProtocolClient(self.io) + self.unicode_test = PlaceHolder(_u('\u2603')) + self.test = TestTestProtocolClient("test_start_test") + self.sample_details = {'something':Content( + 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) + + def test_start_test(self): + """Test startTest on a TestProtocolClient.""" + self.protocol.startTest(self.test) + self.assertEqual(self.io.getvalue(), _b("test: %s\n" % self.test.id())) + + def test_start_test_unicode_id(self): + """Test startTest on a TestProtocolClient.""" + self.protocol.startTest(self.unicode_test) + expected = _b("test: ") + _u('\u2603').encode('utf8') + _b("\n") + self.assertEqual(expected, self.io.getvalue()) + + def test_stop_test(self): + # stopTest doesn't output anything. + self.protocol.stopTest(self.test) + 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(), _b("successful: %s\n" % self.test.id())) + + def test_add_outcome_unicode_id(self): + """Test addSuccess on a TestProtocolClient.""" + self.protocol.addSuccess(self.unicode_test) + expected = _b("successful: ") + _u('\u2603').encode('utf8') + _b("\n") + self.assertEqual(expected, self.io.getvalue()) + + 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(), _b("successful: %s [ multipart\n" + "Content-Type: text/plain\n" + "something\n" + "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.assertEqual( + self.io.getvalue(), + _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.""" + self.protocol.addFailure( + self.test, details=self.sample_tb_details) + self.assertThat([ + _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" + _remote_exception_str_chunked + + "]\n") % self.test.id()), + _b(("failure: %s [ multipart\n" + "Content-Type: text/plain\n" + "something\n" + "F\r\nserialised\nform0\r\n" + "Content-Type: text/x-traceback;language=python,charset=utf8\n" + "traceback\n" + _remote_exception_str_chunked + + "]\n") % self.test.id()), + ], + Contains(self.io.getvalue())), + + def test_add_error(self): + """Test stopTest on a TestProtocolClient.""" + self.protocol.addError( + self.test, subunit.RemoteError(_u("phwoar crikey"))) + self.assertEqual( + self.io.getvalue(), + _b(('error: %s [\n' + + _remote_exception_str + ": phwoar crikey\n" + "]\n") % self.test.id())) + + def test_add_error_details(self): + """Test stopTest on a TestProtocolClient with details.""" + self.protocol.addError( + self.test, details=self.sample_tb_details) + self.assertThat([ + _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" + _remote_exception_str_chunked + + "]\n") % self.test.id()), + _b(("error: %s [ multipart\n" + "Content-Type: text/plain\n" + "something\n" + "F\r\nserialised\nform0\r\n" + "Content-Type: text/x-traceback;language=python,charset=utf8\n" + "traceback\n" + _remote_exception_str_chunked + + "]\n") % self.test.id()), + ], + Contains(self.io.getvalue())), + + def test_add_expected_failure(self): + """Test addExpectedFailure on a TestProtocolClient.""" + self.protocol.addExpectedFailure( + self.test, subunit.RemoteError(_u("phwoar crikey"))) + self.assertEqual( + self.io.getvalue(), + _b(('xfail: %s [\n' + + _remote_exception_str + ": phwoar crikey\n" + "]\n") % self.test.id())) + + def test_add_expected_failure_details(self): + """Test addExpectedFailure on a TestProtocolClient with details.""" + self.protocol.addExpectedFailure( + self.test, details=self.sample_tb_details) + self.assertThat([ + _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" + _remote_exception_str_chunked + + "]\n") % self.test.id()), + _b(("xfail: %s [ multipart\n" + "Content-Type: text/plain\n" + "something\n" + "F\r\nserialised\nform0\r\n" + "Content-Type: text/x-traceback;language=python,charset=utf8\n" + "traceback\n" + _remote_exception_str_chunked + + "]\n") % self.test.id()), + ], + Contains(self.io.getvalue())), + + def test_add_skip(self): + """Test addSkip on a TestProtocolClient.""" + self.protocol.addSkip( + self.test, "Has it really?") + self.assertEqual( + self.io.getvalue(), + _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:[_b('Has it really?')])} + self.protocol.addSkip(self.test, details=details) + self.assertEqual( + self.io.getvalue(), + _b("skip: %s [ multipart\n" + "Content-Type: text/plain\n" + "reason\n" + "E\r\nHas it really?0\r\n" + "]\n" % self.test.id())) + + def test_progress_set(self): + self.protocol.progress(23, subunit.PROGRESS_SET) + 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(), _b('progress: -23\n')) + + def test_progress_pos_cur(self): + self.protocol.progress(23, subunit.PROGRESS_CUR) + 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(), _b('progress: pop\n')) + + def test_progress_push(self): + self.protocol.progress(1234, subunit.PROGRESS_PUSH) + 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( + _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(), _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(), _b("uxsuccess: %s [ multipart\n" + "Content-Type: text/plain\n" + "something\n" + "F\r\nserialised\nform0\r\n]\n" % self.test.id())) + + def test_tags_empty(self): + self.protocol.tags(set(), set()) + self.assertEqual(_b(""), self.io.getvalue()) + + def test_tags_add(self): + self.protocol.tags(set(['foo']), set()) + self.assertEqual(_b("tags: foo\n"), self.io.getvalue()) + + def test_tags_both(self): + self.protocol.tags(set(['quux']), set(['bar'])) + self.assertThat( + [b"tags: quux -bar\n", b"tags: -bar quux\n"], + Contains(self.io.getvalue())) + + def test_tags_gone(self): + self.protocol.tags(set(), set(['bar'])) + self.assertEqual(_b("tags: -bar\n"), self.io.getvalue()) |