Refactor common testing helpers into a separate module

Author: miguelgrinberg

tests/asyncio/helpers.py

@@ -0,0 +1,18 @@
+import asyncio
+from unittest import mock
+
+
+def AsyncMock(*args, **kwargs):
+    """Return a mock asynchronous function."""
+    m = mock.MagicMock(*args, **kwargs)
+
+    async def mock_coro(*args, **kwargs):
+        return m(*args, **kwargs)
+
+    mock_coro.mock = m
+    return mock_coro
+
+
+def _run(coro):
+    """Run the given coroutine."""
+    return asyncio.get_event_loop().run_until_complete(coro)

tests/asyncio/test_asyncio_client.py

@@ -11,35 +11,7 @@
 from engineio import exceptions as engineio_exceptions
 from socketio import exceptions
 from socketio import packet
-
-
-def AsyncMock(*args, **kwargs):
-    """Return a mock asynchronous function."""
-    m = mock.MagicMock(*args, **kwargs)
-
-    async def mock_coro(*args, **kwargs):
-        return m(*args, **kwargs)
-
-    mock_coro.mock = m
-    return mock_coro
-
-
-@contextmanager
-def mock_wait_for():
-    async def fake_wait_for(coro, timeout):
-        await coro
-        await fake_wait_for._mock(timeout)
-
-    original_wait_for = asyncio.wait_for
-    asyncio.wait_for = fake_wait_for
-    fake_wait_for._mock = AsyncMock()
-    yield
-    asyncio.wait_for = original_wait_for
-
-
-def _run(coro):
-    """Run the given coroutine."""
-    return asyncio.get_event_loop().run_until_complete(coro)
+from .helpers import AsyncMock, _run
 
 
 @unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+')

tests/asyncio/test_asyncio_manager.py

@@ -5,22 +5,7 @@
 
 from socketio import asyncio_manager
 from socketio import packet
-
-
-def AsyncMock(*args, **kwargs):
-    """Return a mock asynchronous function."""
-    m = mock.MagicMock(*args, **kwargs)
-
-    async def mock_coro(*args, **kwargs):
-        return m(*args, **kwargs)
-
-    mock_coro.mock = m
-    return mock_coro
-
-
-def _run(coro):
-    """Run the given coroutine."""
-    return asyncio.get_event_loop().run_until_complete(coro)
+from .helpers import AsyncMock, _run
 
 
 @unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+')

tests/asyncio/test_asyncio_namespace.py

@@ -4,22 +4,7 @@
 from unittest import mock
 
 from socketio import asyncio_namespace
-
-
-def AsyncMock(*args, **kwargs):
-    """Return a mock asynchronous function."""
-    m = mock.MagicMock(*args, **kwargs)
-
-    async def mock_coro(*args, **kwargs):
-        return m(*args, **kwargs)
-
-    mock_coro.mock = m
-    return mock_coro
-
-
-def _run(coro):
-    """Run the given coroutine."""
-    return asyncio.get_event_loop().run_until_complete(coro)
+from .helpers import AsyncMock, _run
 
 
 @unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+')

tests/asyncio/test_asyncio_pubsub_manager.py

@@ -9,22 +9,7 @@
 from socketio import asyncio_manager
 from socketio import asyncio_pubsub_manager
 from socketio import packet
-
-
-def AsyncMock(*args, **kwargs):
-    """Return a mock asynchronous function."""
-    m = mock.MagicMock(*args, **kwargs)
-
-    async def mock_coro(*args, **kwargs):
-        return m(*args, **kwargs)
-
-    mock_coro.mock = m
-    return mock_coro
-
-
-def _run(coro):
-    """Run the given coroutine."""
-    return asyncio.get_event_loop().run_until_complete(coro)
+from .helpers import AsyncMock, _run
 
 
 @unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+')

tests/asyncio/test_asyncio_server.py

@@ -13,22 +13,7 @@
 from socketio import exceptions
 from socketio import namespace
 from socketio import packet
-
-
-def AsyncMock(*args, **kwargs):
-    """Return a mock asynchronous function."""
-    m = mock.MagicMock(*args, **kwargs)
-
-    async def mock_coro(*args, **kwargs):
-        return m(*args, **kwargs)
-
-    mock_coro.mock = m
-    return mock_coro
-
-
-def _run(coro):
-    """Run the given coroutine."""
-    return asyncio.get_event_loop().run_until_complete(coro)
+from .helpers import AsyncMock, _run
 
 
 @unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+')