Support unexpected success outcomes.

2 files changed, 177 insertions, 15 deletions

diff --git a/python/subunit/__init__.py b/python/subunit/__init__.py
index 807b605..9d5d075 100644
--- a/python/subunit/__init__.py
+++ b/python/subunit/__init__.py
@@ -201,6 +201,7 @@ class _ParserState(object):
         self._tags_sym = (_b('tags'),)
         self._time_sym = (_b('time'),)
         self._xfail_sym = (_b('xfail'),)
+        self._uxsuccess_sym = (_b('uxsuccess'),)
         self._start_simple = _u(" [")
         self._start_multipart = _u(" [ multipart")
 
@@ -251,6 +252,8 @@ class _ParserState(object):
                 self.parser.subunitLineReceived(line)
             elif cmd in self._xfail_sym:
                 self.addExpectedFail(offset, line)
+            elif cmd in self._uxsuccess_sym:
+                self.addUnexpectedSuccess(offset, line)
             else:
                 self.parser.stdOutLineReceived(line)
         else:
@@ -314,6 +317,14 @@ class _InTest(_ParserState):
         self._outcome(offset, line, self._xfail,
             self.parser._reading_xfail_details)
 
+    def _uxsuccess(self):
+        self.parser.client.addUnexpectedSuccess(self.parser._current_test)
+
+    def addUnexpectedSuccess(self, offset, line):
+        """A 'uxsuccess:' directive has been read."""
+        self._outcome(offset, line, self._uxsuccess,
+            self.parser._reading_uxsuccess_details)
+
     def _failure(self):
         self.parser.client.addFailure(self.parser._current_test, details={})
 
@@ -425,6 +436,17 @@ class _ReadingExpectedFailureDetails(_ReadingDetails):
         return "xfail"
 
 
+class _ReadingUnexpectedSuccessDetails(_ReadingDetails):
+    """State for the subunit parser when reading uxsuccess details."""
+
+    def _report_outcome(self):
+        self.parser.client.addUnexpectedSuccess(self.parser._current_test,
+            details=self.details_parser.get_details())
+
+    def _outcome_label(self):
+        return "uxsuccess"
+
+
 class _ReadingSkipDetails(_ReadingDetails):
     """State for the subunit parser when reading skip details."""
 
@@ -481,6 +503,7 @@ class TestProtocolServer(object):
         self._reading_skip_details = _ReadingSkipDetails(self)
         self._reading_success_details = _ReadingSuccessDetails(self)
         self._reading_xfail_details = _ReadingExpectedFailureDetails(self)
+        self._reading_uxsuccess_details = _ReadingUnexpectedSuccessDetails(self)
         # start with outside test.
         self._state = self._outside_test
         # Avoid casts on every call
@@ -632,7 +655,8 @@ class TestProtocolClient(testresult.TestResult):
         """
         self._addOutcome("failure", test, error=error, details=details)
 
-    def _addOutcome(self, outcome, test, error=None, details=None):
+    def _addOutcome(self, outcome, test, error=None, details=None,
+        error_permitted=True):
         """Report a failure in test test.
 
         Only one of error and details should be provided: conceptually there
@@ -646,19 +670,28 @@ class TestProtocolClient(testresult.TestResult):
             exc_info tuple.
         :param details: New Testing-in-python drafted API; a dict from string
             to subunit.Content objects.
-        """
+        :param error_permitted: If True then one and only one of error or
+            details must be supplied. If False then error must not be supplied
+            and details is still optional.  """
         self._stream.write(_b("%s: %s" % (outcome, test.id())))
-        if error is None and details is None:
-            raise ValueError
+        if error_permitted:
+            if error is None and details is None:
+                raise ValueError
+        else:
+            if error is not None:
+                raise ValueError
         if error is not None:
             self._stream.write(self._start_simple)
             # XXX: this needs to be made much stricter, along the lines of
             # Martin[gz]'s work in testtools. Perhaps subunit can use that?
             for line in self._exc_info_to_unicode(error, test).splitlines():
                 self._stream.write(("%s\n" % line).encode('utf8'))
-        else:
+        elif details is not None:
             self._write_details(details)
-        self._stream.write(self._end_simple)
+        else:
+            self._stream.write(_b("\n"))
+        if details or error:
+            self._stream.write(self._end_simple)
 
     def addSkip(self, test, reason=None, details=None):
         """Report a skipped test."""
@@ -671,13 +704,21 @@ class TestProtocolClient(testresult.TestResult):
 
     def addSuccess(self, test, details=None):
         """Report a success in a test."""
-        self._stream.write(_b("successful: %s" % test.id()))
-        if not details:
-            self._stream.write(_b("\n"))
-        else:
-            self._write_details(details)
-            self._stream.write(self._end_simple)
-    addUnexpectedSuccess = addSuccess
+        self._addOutcome("successful", test, details=details, error_permitted=False)
+
+    def addUnexpectedSuccess(self, test, details=None):
+        """Report an unexpected success in test test.
+
+        Details can optionally be provided: conceptually there
+        are two separate methods:
+            addError(self, test)
+            addError(self, test, details)
+
+        :param details: New Testing-in-python drafted API; a dict from string
+            to subunit.Content objects.
+        """
+        self._addOutcome("uxsuccess", test, details=details,
+            error_permitted=False)
 
     def startTest(self, test):
         """Mark a test as starting its test run."""
diff --git a/python/subunit/tests/test_test_protocol.py b/python/subunit/tests/test_test_protocol.py
index 939b642..03d921a 100644
--- a/python/subunit/tests/test_test_protocol.py
+++ b/python/subunit/tests/test_test_protocol.py
@@ -370,6 +370,12 @@ 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):
 
@@ -611,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.
 
@@ -1160,13 +1281,13 @@ class TestTestProtocolClient(unittest.TestCase):
         """Test addUnexpectedSuccess on a TestProtocolClient."""
         self.protocol.addUnexpectedSuccess(self.test)
         self.assertEqual(
-            self.io.getvalue(), _b("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(), _b("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()))