Refactor common testing helpers into a separate module

miguelgrinberg · Sep 6, 2023 · 7208ec0 · 7208ec0
1 parent 5541dda
commit 7208ec0
Show file tree

Hide file tree

Showing 6 changed files with 23 additions and 93 deletions.
diff --git a/tests/asyncio/helpers.py b/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)
diff --git a/tests/asyncio/test_asyncio_client.py b/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 ')

diff --git a/tests/asyncio/test_asyncio_manager.py b/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 ')

diff --git a/tests/asyncio/test_asyncio_namespace.py b/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 ')

diff --git a/tests/asyncio/test_asyncio_pubsub_manager.py b/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 ')

diff --git a/tests/asyncio/test_asyncio_server.py b/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 ')