Skip to content

Commit

Permalink
Refactor common testing helpers into a separate module
Browse files Browse the repository at this point in the history
  • Loading branch information
miguelgrinberg committed Sep 6, 2023
1 parent 5541dda commit 7208ec0
Show file tree
Hide file tree
Showing 6 changed files with 23 additions and 93 deletions.
18 changes: 18 additions & 0 deletions tests/asyncio/helpers.py
Original file line number Diff line number Diff line change
@@ -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)
30 changes: 1 addition & 29 deletions tests/asyncio/test_asyncio_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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+')
Expand Down
17 changes: 1 addition & 16 deletions tests/asyncio/test_asyncio_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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+')
Expand Down
17 changes: 1 addition & 16 deletions tests/asyncio/test_asyncio_namespace.py
Original file line number Diff line number Diff line change
Expand Up @@ -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+')
Expand Down
17 changes: 1 addition & 16 deletions tests/asyncio/test_asyncio_pubsub_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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+')
Expand Down
17 changes: 1 addition & 16 deletions tests/asyncio/test_asyncio_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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+')
Expand Down

0 comments on commit 7208ec0

Please sign in to comment.