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.
51 lines
1.5 KiB
51 lines
1.5 KiB
2 years ago
|
import asyncio
|
||
|
|
import sys
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
if sys.version_info >= (3, 8):
|
||
|
|
from unittest.mock import AsyncMock
|
||
|
|
else:
|
||
|
|
from asynctest import CoroutineMock as AsyncMock
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from announce_server import _announce_server, announce_server
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_announce_server_decorator(event_loop, mocker):
|
||
|
|
# Mock the _announce_server function to prevent actual connections
|
||
|
|
mocker.patch("announce_server._announce_server")
|
||
|
|
|
||
|
|
# Sample function to be decorated
|
||
|
|
async def sample_async_function():
|
||
|
|
await asyncio.sleep(1)
|
||
|
|
return "Hello, world!"
|
||
|
|
|
||
|
|
# 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
|
||
|
|
task = await decorated_function()
|
||
|
|
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
|
||
|
|
announce_server._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,
|
||
|
|
)
|
||
|
|
|
||
|
|
# Check if the decorated function returns the expected result
|
||
|
|
result = await sample_async_function()
|
||
|
|
assert result == "Hello, world!"
|