Matcher for no result

testtools/_deferredmatchers.py

@@ -0,0 +1,35 @@
+# Copyright (c) testtools developers. See LICENSE for details.
+
+"""Matchers that operate on Deferreds.
+
+Depends on Twisted.
+"""
+
+from testtools.compat import _u
+from testtools.matchers import Mismatch
+
+
+class _NoResult(object):
+    """Matches a Deferred that has not yet fired."""
+
+    def match(self, deferred):
+        """Match ``deferred`` if it hasn't fired."""
+        result = []
+
+        def callback(x):
+            result.append(x)
+            # XXX: assertNoResult returns `x` here, but then swallows it if
+            # it's a failure. Not 100% sure why that's the case. I guess maybe
+            # to handle the case where you assert that there's no result but
+            # then later make more assertions / callbacks?
+            return x
+        deferred.addBoth(callback)
+        if result:
+            return Mismatch(
+                _u('%r has already fired with %r' % (deferred, result[0])))
+
+
+# XXX: Maybe just a constant, rather than a function?
+def no_result():
+    """Match a Deferred that has not yet fired."""
+    return _NoResult()

testtools/tests/__init__.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2008-2013 testtools developers. See LICENSE for details.
+# Copyright (c) 2008-2015 testtools developers. See LICENSE for details.
 
 """Tests for testtools itself."""
 
@@ -14,6 +14,7 @@ def test_suite():
         test_compat,
         test_content,
         test_content_type,
+        test_deferredmatchers,
         test_deferredruntest,
         test_distutilscmd,
         test_fixturesupport,
@@ -34,6 +35,7 @@ def test_suite():
         test_compat,
         test_content,
         test_content_type,
+        test_deferredmatchers,
         test_deferredruntest,
         test_distutilscmd,
         test_fixturesupport,

testtools/tests/test_deferredmatchers.py

@@ -0,0 +1,110 @@
+# Copyright (c) testtools developers. See LICENSE for details.
+
+"""Tests for Deferred matchers."""
+
+from extras import try_import
+
+from testtools.compat import _u
+from testtools._deferredmatchers import (
+    no_result,
+)
+from testtools.matchers import (
+    AfterPreprocessing,
+    Equals,
+    Is,
+    MatchesDict,
+)
+from testtools.tests.test_spinner import NeedsTwistedTestCase
+
+
+defer = try_import('twisted.internet.defer')
+failure = try_import('twisted.python.failure')
+
+
+def mismatches(description, details=None):
+    """Match a ``Mismatch`` object."""
+    if details is None:
+        details = Equals({})
+
+    matcher = MatchesDict({
+        'description': description,
+        'details': details,
+    })
+
+    def get_mismatch_info(mismatch):
+        return {
+            'description': mismatch.describe(),
+            'details': mismatch.get_details(),
+        }
+
+    return AfterPreprocessing(get_mismatch_info, matcher)
+
+
+def make_failure(exc_value):
+    """Raise ``exc_value`` and return the failure."""
+    try:
+        raise exc_value
+    except:
+        return failure.Failure()
+
+
+class NoResultTests(NeedsTwistedTestCase):
+    """
+    Tests for ``no_result``.
+    """
+
+    def match(self, thing):
+        return no_result().match(thing)
+
+    def test_unfired_matches(self):
+        # A Deferred that hasn't fired matches no_result.
+        self.assertThat(self.match(defer.Deferred()), Is(None))
+
+    def test_successful_does_no_match(self):
+        # A Deferred that's fired successfully does not match no_result.
+        result = None
+        deferred = defer.succeed(result)
+        mismatch = self.match(deferred)
+        self.assertThat(
+            mismatch, mismatches(Equals(_u(
+                '%r has already fired with %r' % (deferred, result)))))
+
+    def test_failed_does_not_match(self):
+        # A Deferred that's failed does not match no_result.
+        fail = make_failure(RuntimeError('arbitrary failure'))
+        deferred = defer.fail(fail)
+        # Suppress unhandled error in Deferred.
+        self.addCleanup(deferred.addErrback, lambda _: None)
+        mismatch = self.match(deferred)
+        self.assertThat(
+            mismatch, mismatches(Equals(_u(
+                '%r has already fired with %r' % (deferred, fail)))))
+
+    def test_success_after_assertion(self):
+        # We can create a Deferred, assert that it hasn't fired, then fire it
+        # and collect the result.
+        deferred = defer.Deferred()
+        self.assertThat(deferred, no_result())
+        results = []
+        deferred.addCallback(results.append)
+        marker = object()
+        deferred.callback(marker)
+        self.assertThat(results, Equals([marker]))
+
+    def test_failure_after_assertion(self):
+        # We can create a Deferred, assert that it hasn't fired, then fire it
+        # with a failure and collect the result.
+
+        # XXX: Ask Jean-Paul about whether this is good behaviour.
+        deferred = defer.Deferred()
+        self.assertThat(deferred, no_result())
+        results = []
+        deferred.addErrback(results.append)
+        fail = make_failure(RuntimeError('arbitrary failure'))
+        deferred.errback(fail)
+        self.assertThat(results, Equals([fail]))
+
+
+def test_suite():
+    from unittest2 import TestLoader, TestSuite
+    return TestLoader().loadTestsFromName(__name__)