You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
54 lines
1.7 KiB
54 lines
1.7 KiB
import asyncio
|
|
import sys
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
if sys.version_info >= (3, 8):
|
|
from unittest.mock import AsyncMock
|
|
else:
|
|
from asynctest import CoroutineMock as AsyncMock
|
|
|
|
import pytest
|
|
|
|
from announce_server.decorator import _announce_server, announce_server
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
# @patch('announce_server.decorator.announce_server', new=MagicMock())
|
|
async def test_announce_server_decorator(mocked_announce_server, event_loop):
|
|
# Sample function to be decorated
|
|
async def sample_async_function():
|
|
await asyncio.sleep(1)
|
|
return "Hello, world!"
|
|
|
|
# Mock the _announce_server function to prevent actual connections
|
|
mocked_announce_server.return_value = lambda x: x
|
|
|
|
# Decorate the sample function with announce_server
|
|
decorated_function = announce_server(
|
|
name="test_server",
|
|
ip="127.0.0.1",
|
|
port=8000,
|
|
host_ip="127.0.0.1",
|
|
host_port=5000,
|
|
loop=event_loop, # Pass the current event loop
|
|
)(sample_async_function)
|
|
|
|
# Run the decorated function
|
|
coro = asyncio.to_thread(decorated_function)
|
|
task = await asyncio.gather(coro)
|
|
await asyncio.sleep(1.1) # Sleep slightly longer than sample_async_function
|
|
task.cancel() # Cancel the task
|
|
|
|
# Check if the _announce_server function was called with the correct arguments
|
|
mocked_announce_server.assert_called_once_with(
|
|
name="test_server",
|
|
ip="127.0.0.1",
|
|
port=8000,
|
|
host_ip="127.0.0.1",
|
|
host_port=5000,
|
|
loop=event_loop,
|
|
)
|
|
|
|
# Check if the decorated function returns the expected result
|
|
result = await sample_async_function()
|
|
assert result == "Hello, world!"
|