Throw exception in init_qt5_reactor() if other reactor already installed

Author: altendky

pytest_twisted.py

@@ -1,5 +1,3 @@
-import sys
-
 import decorator
 import greenlet
 import pytest
@@ -9,6 +7,10 @@
 from twisted.python import failure
 
 
+class WrongReactorAlreadyInstalledError(Exception):
+    pass
+
+
 class _instances:
     gr_twisted = None
     reactor = None
@@ -126,11 +128,14 @@ def init_qt5_reactor(qapp):
     try:
         qt5reactor.install()
     except error.ReactorAlreadyInstalledError:
-        if not isinstance(_instances.reactor, qt5reactor.QtReactor):
-            stop_twisted_greenlet()
-            _instances.gr_twisted = None
-            del sys.modules['twisted.internet.reactor']
-            qt5reactor.install()
+        import twisted.internet.reactor
+        if not isinstance(twisted.internet.reactor, qt5reactor.QtReactor):
+            raise WrongReactorAlreadyInstalledError(
+                'expected {0} but found {1}'.format(
+                    qt5reactor.QtReactor,
+                    type(twisted.internet.reactor),
+                )
+            )
     init_reactor()
 
 

testing/test_basic.py

@@ -2,6 +2,19 @@
 
 import sys
 
+import pytest
+
+
+def skip_if_reactor_not(expected_reactor):
+    actual_reactor = pytest.config.getoption('reactor')
+    return pytest.mark.skipif(
+        actual_reactor != expected_reactor,
+        reason='reactor is {0} not {1}'.format(
+            actual_reactor,
+            expected_reactor,
+        ),
+    )
+
 
 def test_fail_later(testdir):
     testdir.makepyfile("""
@@ -106,6 +119,7 @@ def test_MAIN():
     assert outcomes.get("passed") == 1
 
 
+@skip_if_reactor_not('default')
 def test_blockon_in_hook(testdir):
     testdir.makeconftest("""
         import pytest_twisted as pt
@@ -130,6 +144,54 @@ def test_succeed():
     assert outcomes.get("passed") == 1
 
 
+@skip_if_reactor_not('qt5reactor')
+def test_blockon_in_hook_with_qt5reactor(testdir):
+    testdir.makeconftest("""
+    import pytest_twisted as pt
+    import pytestqt
+    from twisted.internet import defer
+
+
+    def pytest_configure(config):
+        qapp = pytestqt.plugin.qapp(pytestqt.plugin.qapp_args())
+
+        pt.init_qt5_reactor(qapp)
+        d = defer.Deferred()
+
+        from twisted.internet import reactor
+        reactor.callLater(0.01, d.callback, 1)
+        pt.blockon(d)
+    """)
+    testdir.makepyfile("""
+        from twisted.internet import reactor, defer
+
+        def test_succeed():
+            d = defer.Deferred()
+            reactor.callLater(0.01, d.callback, 1)
+            return d
+    """)
+    rr = testdir.run(sys.executable, "-m", "pytest", "-v")
+    outcomes = rr.parseoutcomes()
+    assert outcomes.get("passed") == 1
+
+
+@skip_if_reactor_not('qt5reactor')
+def test_wrong_reactor_with_qt5reactor(testdir):
+    testdir.makepyfile("""
+        import twisted.internet.default
+        twisted.internet.default.install()
+
+        def test_succeed():
+            pass
+    """)
+    rr = testdir.run(
+        sys.executable, "-m", "pytest", "-v", "--reactor=qt5reactor"
+    )
+    outcomes = rr.parseoutcomes()
+    assert 'WrongReactorAlreadyInstalledError' in rr.stdout.str()
+    assert outcomes.get("error") == 1
+
+
 def test_pytest_from_reactor_thread(testdir):
     testdir.makepyfile("""
         import pytest