Added test for limit_twisted_callback_hell

Author: rednafi

patterns/inequal_producer_consumer.py

@@ -46,7 +46,7 @@ async def consumer(
                 sys.exit(0)
 
 
-async def main() -> None:
+async def orchestrator() -> None:
     result_queue = asyncio.Queue()  # type: asyncio.Queue[int]
     event = asyncio.Event()  # type: asyncio.Event
     limit = asyncio.Semaphore(5)  # type: asyncio.Semaphore
@@ -63,4 +63,4 @@ async def main() -> None:
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(orchestrator())

patterns/twisted_callback_hell.py

@@ -1,34 +1,43 @@
 from twisted.internet import defer, reactor
 
 
-def get_dummy_data(input_data):
+def get_dummy_number(
+    input_number: int, defer_duration: float = 2, multiplier: float = 3
+) -> defer.Deferred:
     """
     This function is a dummy which simulates a delayed result and
-    returns a Deferred which will fire with that result. Don't try too
-    hard to understand this.
+    returns a Deferred which will fire with that result.
     """
-    print("get_dummy_data called")
+
+    print("get_dummy_number called")
+
     deferred = defer.Deferred()
-    # simulate a delayed result by asking the reactor to fire the
-    # Deferred in 2 seconds time with the result inputData * 3
-    reactor.callLater(2, deferred.callback, input_data * 3)
+    # Simulate a delayed result by asking the reactor to fire the
+    # Deferred in '2 seconds' time with the result 'input_number * 3'.
+    reactor.callLater(defer_duration, deferred.callback, input_number * multiplier)
     return deferred
 
 
-def cb_print_data(result):
+def cb_print_number(result: int) -> None:
     """
     Data handling function to be added as a callback: handles the
     data by printing the result
     """
     print(f"Result received: {result}")
 
 
-deferred = get_dummy_data(3)
-deferred.addCallback(cb_print_data)
+def orchestrator(stop_after: float = 4) -> None:
+    deferred = get_dummy_number(3)
+    deferred.addCallback(cb_print_number)
+
+    # Manually set up the end of the process by asking the reactor to
+    # stop itself in 4 seconds time.
+    reactor.callLater(stop_after, reactor.stop)
+
+    # Start up the Twisted reactor (event loop handler) manually.
+    print("Starting the reactor")
+    reactor.run()
+
 
-# manually set up the end of the process by asking the reactor to
-# stop itself in 4 seconds time
-reactor.callLater(4, reactor.stop)
-# start up the Twisted reactor (event loop handler) manually
-print("Starting the reactor")
-reactor.run()
+if __name__ == "__main__":
+    orchestrator()

requirements.in

@@ -2,4 +2,5 @@ aiobotocore
 aioredis
 httpx
 rich
+twisted
 uvloop

requirements.txt

@@ -19,7 +19,12 @@ async-timeout==3.0.1
     #   aiohttp
     #   aioredis
 attrs==21.2.0
-    # via aiohttp
+    # via
+    #   aiohttp
+    #   automat
+    #   twisted
+automat==20.2.0
+    # via twisted
 botocore==1.20.106
     # via aiobotocore
 certifi==2021.10.8
@@ -32,17 +37,24 @@ colorama==0.4.4
     # via rich
 commonmark==0.9.1
     # via rich
+constantly==15.1.0
+    # via twisted
 h11==0.12.0
     # via httpcore
 httpcore==0.13.7
     # via httpx
 httpx==0.20.0
     # via -r requirements.in
+hyperlink==21.0.0
+    # via twisted
 idna==3.3
     # via
     #   anyio
+    #   hyperlink
     #   rfc3986
     #   yarl
+incremental==21.3.0
+    # via twisted
 jmespath==0.10.0
     # via botocore
 multidict==5.2.0
@@ -58,16 +70,21 @@ rfc3986[idna2008]==1.5.0
 rich==10.12.0
     # via -r requirements.in
 six==1.16.0
-    # via python-dateutil
+    # via
+    #   automat
+    #   python-dateutil
 sniffio==1.2.0
     # via
     #   anyio
     #   httpcore
     #   httpx
+twisted==21.7.0
+    # via -r requirements.in
 typing-extensions==3.10.0.2
     # via
     #   aiohttp
     #   aioredis
+    #   twisted
 urllib3==1.26.7
     # via botocore
 uvloop==0.16.0
@@ -76,3 +93,8 @@ wrapt==1.13.2
     # via aiobotocore
 yarl==1.7.0
     # via aiohttp
+zope.interface==5.4.0
+    # via twisted
+
+# The following packages are considered to be unsafe in a requirements file:
+# setuptools

tests/test_inequal_producer_consumer.py

@@ -80,7 +80,7 @@ async def test_consumer(*args):
 @patch("patterns.inequal_producer_consumer.asyncio.Queue", autospec=True)
 @patch("patterns.inequal_producer_consumer.consumer", autospec=True)
 @patch("patterns.inequal_producer_consumer.producer", new_callable=Mock())
-async def test_main(
+async def test_orchestrator(
     mock_producer,
     mock_consumer,
     mock_asyncio_queue,
@@ -90,7 +90,7 @@ async def test_main(
 ):
 
     # Call the 'main' function.
-    await main.main()
+    await main.orchestrator()
 
     mock_asyncio_event.assert_called_once()
     mock_asyncio_queue.assert_called_once()

tests/test_twisted_callback_hell.py

@@ -0,0 +1,52 @@
+from unittest.mock import patch
+
+import pytest
+from twisted.internet import defer, reactor
+
+import patterns.twisted_callback_hell as main
+
+
+def test_get_dummy_number(capsys):
+    # Call 'get_dummy_number'.
+    main.get_dummy_number(input_number=10, defer_duration=0, multiplier=20)
+
+    # Assert.
+    out, err = capsys.readouterr()
+    assert err == ""
+    assert "get_dummy_number called" in out
+
+
+def test_cb_print_number(capsys):
+    input_number = 11
+
+    # Call 'cb_print_number'.
+    main.cb_print_number(result=input_number)
+
+    # Assert.
+    out, err = capsys.readouterr()
+    assert err == ""
+    assert f"Result received: {input_number}" in out
+
+
+@patch(
+    "patterns.twisted_callback_hell.get_dummy_number",
+    autospec=True,
+    return_value=main.get_dummy_number(
+        input_number=1,
+        defer_duration=0.1,
+        multiplier=2,
+    ),
+)
+def test_orchestrator(mock_get_dummy_number, capsys):
+
+    # Call 'orchestrator'.
+    # On Python3.10, twisted raises:
+    # DeprecationWarning: currentThread() is deprecated, use current_thread() instead
+    with pytest.warns(None):
+        main.orchestrator(stop_after=0.01)
+
+    # Assert.
+    out, err = capsys.readouterr()
+    assert err == ""
+    assert "Result received: 2" in out
+    mock_get_dummy_number.assert_called_once()