# 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
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 (
)
import subunit
-from subunit import _remote_exception_str
+from subunit import _remote_exception_str, _remote_exception_str_chunked
import subunit.iso8601 as iso8601
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,
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"))])
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"
"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),
], 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),
])
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),
], 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"
"success: a\n"
"successful a\n"
"successful: a\n"
- "]\n")
+ "]\n"))
self.assertEqual(self.client._events, [
('startTest', self.test),
('addFailure', self.test, {}),
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"
"success: a\n"
"successful a\n"
"successful: a\n"
- "]\n"])
+ "]\n")])
self.assertEqual(self.client._events, [
('startTest', self.test),
('addFailure', self.test, details),
"""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)
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),
], 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),
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)
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),
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)
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),
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),
], 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),
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):
], 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)
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):
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):
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([
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):
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.
"""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):
], 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):
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")
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),
('stopTest', self.test),
], 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")
], 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):
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),
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))
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()))
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):
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([
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)
"""sample-two-script.py"""
-class TestIsolatedTestCase(unittest.TestCase):
+class TestIsolatedTestCase(TestCase):
class SampleIsolatedTestCase(subunit.IsolatedTestCase):
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")
#test.debug()
-class TestIsolatedTestSuite(unittest.TestCase):
+class TestIsolatedTestSuite(TestCase):
class SampleTestToIsolate(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()
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."""
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."""
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."""
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."""
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():
git.samba.org / samba.git / blobdiff