Fix unawaited coroutine test warnings

This commit is contained in:
Philip Henning 2026-09-04 21:47:35 +02:00
parent e268794564
commit 6ce1d7da2f
2 changed files with 75 additions and 46 deletions

View file

@ -28,3 +28,9 @@ build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel] [tool.hatch.build.targets.wheel]
packages = ["src/hosts"] packages = ["src/hosts"]
[tool.pytest.ini_options]
filterwarnings = [
"error:coroutine .* was never awaited:RuntimeWarning",
"error::pytest.PytestUnraisableExceptionWarning",
]

View file

@ -7,7 +7,7 @@ and integration with hosts entries.
import pytest import pytest
import asyncio import asyncio
from unittest.mock import AsyncMock, patch from unittest.mock import AsyncMock, MagicMock, patch
from datetime import datetime, timedelta from datetime import datetime, timedelta
import socket import socket
@ -94,16 +94,15 @@ class TestResolveHostname:
async def test_successful_resolution(self): async def test_successful_resolution(self):
"""Test successful hostname resolution.""" """Test successful hostname resolution."""
with patch("asyncio.get_event_loop") as mock_loop: with patch("asyncio.get_event_loop") as mock_loop:
mock_event_loop = AsyncMock() mock_event_loop = MagicMock()
mock_loop.return_value = mock_event_loop mock_loop.return_value = mock_event_loop
# Mock successful getaddrinfo result # Mock successful getaddrinfo result
mock_result = [ mock_result = [
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.1", 80)) (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("192.0.2.1", 80))
] ]
mock_event_loop.getaddrinfo.return_value = mock_result mock_event_loop.getaddrinfo = AsyncMock(return_value=mock_result)
with patch("asyncio.wait_for", return_value=mock_result):
resolution = await resolve_hostname("example.com") resolution = await resolve_hostname("example.com")
assert resolution.hostname == "example.com" assert resolution.hostname == "example.com"
@ -116,10 +115,16 @@ class TestResolveHostname:
async def test_timeout_resolution(self): async def test_timeout_resolution(self):
"""Test hostname resolution timeout.""" """Test hostname resolution timeout."""
async def mock_wait_for(*args, **kwargs): async def mock_wait_for(awaitable, **kwargs):
await awaitable
raise asyncio.TimeoutError() raise asyncio.TimeoutError()
with patch("asyncio.wait_for", side_effect=mock_wait_for) as mock_wait_for: with patch("asyncio.get_event_loop") as mock_loop:
mock_event_loop = MagicMock()
mock_loop.return_value = mock_event_loop
mock_event_loop.getaddrinfo = AsyncMock(return_value=[])
with patch("asyncio.wait_for", side_effect=mock_wait_for):
resolution = await resolve_hostname("slow.example", timeout=1.0) resolution = await resolve_hostname("slow.example", timeout=1.0)
assert resolution.hostname == "slow.example" assert resolution.hostname == "slow.example"
@ -132,7 +137,17 @@ class TestResolveHostname:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_dns_error_resolution(self): async def test_dns_error_resolution(self):
"""Test hostname resolution with DNS error.""" """Test hostname resolution with DNS error."""
with patch("asyncio.wait_for", side_effect=socket.gaierror("Name not found")):
async def mock_wait_for(awaitable, **kwargs):
await awaitable
raise socket.gaierror("Name not found")
with patch("asyncio.get_event_loop") as mock_loop:
mock_event_loop = MagicMock()
mock_loop.return_value = mock_event_loop
mock_event_loop.getaddrinfo = AsyncMock(return_value=[])
with patch("asyncio.wait_for", side_effect=mock_wait_for):
resolution = await resolve_hostname("nonexistent.example") resolution = await resolve_hostname("nonexistent.example")
assert resolution.hostname == "nonexistent.example" assert resolution.hostname == "nonexistent.example"
@ -144,15 +159,11 @@ class TestResolveHostname:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_empty_result_resolution(self): async def test_empty_result_resolution(self):
"""Test hostname resolution with empty result.""" """Test hostname resolution with empty result."""
async def mock_wait_for(*args, **kwargs):
return []
with patch("asyncio.get_event_loop") as mock_loop: with patch("asyncio.get_event_loop") as mock_loop:
mock_event_loop = AsyncMock() mock_event_loop = MagicMock()
mock_loop.return_value = mock_event_loop mock_loop.return_value = mock_event_loop
mock_event_loop.getaddrinfo = AsyncMock(return_value=[])
with patch("asyncio.wait_for", side_effect=mock_wait_for):
resolution = await resolve_hostname("empty.example") resolution = await resolve_hostname("empty.example")
assert resolution.hostname == "empty.example" assert resolution.hostname == "empty.example"
@ -240,6 +251,8 @@ class TestResolveHostnamesBatch:
# Create a mock that returns the expected results # Create a mock that returns the expected results
async def mock_gather(*tasks, return_exceptions=True): async def mock_gather(*tasks, return_exceptions=True):
for task in tasks:
await task
return [ return [
DNSResolution( DNSResolution(
hostname="example.com", hostname="example.com",
@ -250,6 +263,16 @@ class TestResolveHostnamesBatch:
Exception("Network error"), Exception("Network error"),
] ]
with patch(
"src.hosts.core.dns.resolve_hostname",
new_callable=AsyncMock,
) as mock_resolve:
mock_resolve.return_value = DNSResolution(
hostname="ignored.example",
resolved_ip=None,
status=DNSResolutionStatus.RESOLUTION_FAILED,
resolved_at=datetime.now(),
)
with patch("asyncio.gather", side_effect=mock_gather): with patch("asyncio.gather", side_effect=mock_gather):
resolutions = await resolve_hostnames_batch(hostnames) resolutions = await resolve_hostnames_batch(hostnames)