From de23db91358a9bcd86d74ba3b9487dad903e33af Mon Sep 17 00:00:00 2001
From: Kyle Altendorf <sda@fstab.net>
Date: Wed, 27 Feb 2019 23:26:00 -0500
Subject: [PATCH] First pass at unhandled errback reporting

---
 pytest_twisted.py     | 56 +++++++++++++++++++++++++++++++++++
 testing/test_basic.py | 68 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 124 insertions(+)

diff --git a/pytest_twisted.py b/pytest_twisted.py
index 4cc289b..770c6bc 100644
--- a/pytest_twisted.py
+++ b/pytest_twisted.py
@@ -1,4 +1,5 @@
 import functools
+import gc
 import inspect
 import warnings
 
@@ -8,6 +9,7 @@
 
 from twisted.internet import error, defer
 from twisted.internet.threads import blockingCallFromThread
+from twisted.logger import globalLogPublisher
 from twisted.python import failure
 
 
@@ -77,6 +79,60 @@ def block_from_thread(d):
     return blockingCallFromThread(_instances.reactor, lambda x: x, d)
 
 
+class _Observer:
+    def __init__(self):
+        self.failures = []
+        self.asserted = False
+
+    def register(self):
+        globalLogPublisher.addObserver(self)
+
+    def __call__(self, event_dict):
+        is_error = event_dict.get('isError')
+        s = 'Unhandled error in Deferred'.lower()
+        log_format = event_dict.get('log_format')
+        if log_format is None:
+            log_format = ''
+        is_unhandled = s in log_format.lower()
+
+        if is_error and is_unhandled:
+            self.failures.append(event_dict)
+
+    def assert_empty(self):
+        self.asserted = True
+
+        gc.collect()
+        globalLogPublisher.removeObserver(self)
+
+        assert self.failures == []
+
+
+@pytest.fixture(scope='function')
+def unhandled_errback_observer():
+    observer = _Observer()
+    observer.register()
+
+    yield observer
+
+    if not observer.asserted:
+        observer.assert_empty()
+
+
+def assert_on_unhandled_errbacks(f):
+    @functools.wraps(f)
+    def wrapped(*args, **kwargs):
+        observer = _Observer()
+        observer.register()
+
+        result = f(*args, **kwargs)
+
+        observer.assert_empty()
+
+        return result
+
+    return wrapped
+
+
 @decorator.decorator
 def inlineCallbacks(fun, *args, **kw):
     return defer.inlineCallbacks(fun)(*args, **kw)
diff --git a/testing/test_basic.py b/testing/test_basic.py
index 6f2e949..f0c1c81 100755
--- a/testing/test_basic.py
+++ b/testing/test_basic.py
@@ -447,3 +447,71 @@ def main():
     assert_outcomes(rr, {"passed": 1, "failed": 1})
     # test embedded mode:
     assert testdir.run(sys.executable, "runner.py").ret == 0
+
+
+def test_no_unhandled_errbacks_to_assert(testdir, cmd_opts):
+    test_file = """
+    def test_fail(unhandled_errback_observer):
+        unhandled_errback_observer.assert_empty()
+    """
+    testdir.makepyfile(test_file)
+    rr = testdir.run(sys.executable, "-m", "pytest", *cmd_opts)
+    assert_outcomes(rr, {"passed": 1, "error": None})
+
+
+def test_an_unhandled_errback_to_auto_assert(testdir, cmd_opts):
+    test_file = """
+    from twisted.internet.defer import Deferred
+
+    def test_fail(unhandled_errback_observer):
+        d = Deferred()
+        d.errback(42)
+    """
+    testdir.makepyfile(test_file)
+    rr = testdir.run(sys.executable, "-m", "pytest", *cmd_opts)
+    assert_outcomes(rr, {"passed": 1, "error": 1})
+
+
+def test_an_unhandled_errback_to_manually_assert(testdir, cmd_opts):
+    test_file = """
+    from twisted.internet.defer import Deferred
+
+    def test_fail(unhandled_errback_observer):
+        d = Deferred()
+        d.errback(42)
+        
+        del d
+        unhandled_errback_observer.assert_empty()
+    """
+    testdir.makepyfile(test_file)
+    rr = testdir.run(sys.executable, "-m", "pytest", *cmd_opts)
+    assert_outcomes(rr, {"failed": 1})
+
+
+def test_no_unhandled_errbacks_to_assert_decorator(testdir, cmd_opts):
+    test_file = """
+    import pytest_twisted
+
+    @pytest_twisted.assert_on_unhandled_errbacks
+    def test_fail():
+        pass
+    """
+    testdir.makepyfile(test_file)
+    rr = testdir.run(sys.executable, "-m", "pytest", *cmd_opts)
+    assert_outcomes(rr, {"passed": 1})
+
+
+def test_an_unhandled_errback_to_assert_decorator(testdir, cmd_opts):
+    test_file = """
+    from twisted.internet.defer import Deferred
+
+    import pytest_twisted
+
+    @pytest_twisted.assert_on_unhandled_errbacks
+    def test_fail():
+        d = Deferred()
+        d.errback(42)
+    """
+    testdir.makepyfile(test_file)
+    rr = testdir.run(sys.executable, "-m", "pytest", *cmd_opts)
+    assert_outcomes(rr, {"failed": 1})