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]
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 asyncio
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
from datetime import datetime, timedelta
import socket
@ -94,72 +94,83 @@ class TestResolveHostname:
async def test_successful_resolution(self):
"""Test successful hostname resolution."""
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 successful getaddrinfo result
mock_result = [
(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.resolved_ip == "192.0.2.1"
assert resolution.status == DNSResolutionStatus.RESOLVED
assert resolution.error_message is None
assert resolution.is_success() is True
assert resolution.hostname == "example.com"
assert resolution.resolved_ip == "192.0.2.1"
assert resolution.status == DNSResolutionStatus.RESOLVED
assert resolution.error_message is None
assert resolution.is_success() is True
@pytest.mark.asyncio
async def test_timeout_resolution(self):
"""Test hostname resolution timeout."""
async def mock_wait_for(*args, **kwargs):
async def mock_wait_for(awaitable, **kwargs):
await awaitable
raise asyncio.TimeoutError()
with patch("asyncio.wait_for", side_effect=mock_wait_for) as mock_wait_for:
resolution = await resolve_hostname("slow.example", timeout=1.0)
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=[])
assert resolution.hostname == "slow.example"
assert resolution.resolved_ip is None
assert resolution.status == DNSResolutionStatus.RESOLUTION_FAILED
assert resolution.error_message is not None
assert "Timeout after 1.0s" in resolution.error_message
assert resolution.is_success() is False
with patch("asyncio.wait_for", side_effect=mock_wait_for):
resolution = await resolve_hostname("slow.example", timeout=1.0)
assert resolution.hostname == "slow.example"
assert resolution.resolved_ip is None
assert resolution.status == DNSResolutionStatus.RESOLUTION_FAILED
assert resolution.error_message is not None
assert "Timeout after 1.0s" in resolution.error_message
assert resolution.is_success() is False
@pytest.mark.asyncio
async def test_dns_error_resolution(self):
"""Test hostname resolution with DNS error."""
with patch("asyncio.wait_for", side_effect=socket.gaierror("Name not found")):
resolution = await resolve_hostname("nonexistent.example")
assert resolution.hostname == "nonexistent.example"
assert resolution.resolved_ip is None
assert resolution.status == DNSResolutionStatus.RESOLUTION_FAILED
assert resolution.error_message == "Name not found"
assert resolution.is_success() is False
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")
assert resolution.hostname == "nonexistent.example"
assert resolution.resolved_ip is None
assert resolution.status == DNSResolutionStatus.RESOLUTION_FAILED
assert resolution.error_message == "Name not found"
assert resolution.is_success() is False
@pytest.mark.asyncio
async def test_empty_result_resolution(self):
"""Test hostname resolution with empty result."""
async def mock_wait_for(*args, **kwargs):
return []
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_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.resolved_ip is None
assert resolution.status == DNSResolutionStatus.RESOLUTION_FAILED
assert resolution.error_message == "No address found"
assert resolution.is_success() is False
assert resolution.hostname == "empty.example"
assert resolution.resolved_ip is None
assert resolution.status == DNSResolutionStatus.RESOLUTION_FAILED
assert resolution.error_message == "No address found"
assert resolution.is_success() is False
class TestResolveHostnamesBatch:
@ -240,6 +251,8 @@ class TestResolveHostnamesBatch:
# Create a mock that returns the expected results
async def mock_gather(*tasks, return_exceptions=True):
for task in tasks:
await task
return [
DNSResolution(
hostname="example.com",
@ -250,15 +263,25 @@ class TestResolveHostnamesBatch:
Exception("Network error"),
]
with patch("asyncio.gather", side_effect=mock_gather):
resolutions = await resolve_hostnames_batch(hostnames)
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):
resolutions = await resolve_hostnames_batch(hostnames)
assert len(resolutions) == 2
assert resolutions[0].is_success() is True
assert resolutions[1].hostname == "error.example"
assert resolutions[1].is_success() is False
assert resolutions[1].error_message is not None
assert "Network error" in resolutions[1].error_message
assert len(resolutions) == 2
assert resolutions[0].is_success() is True
assert resolutions[1].hostname == "error.example"
assert resolutions[1].is_success() is False
assert resolutions[1].error_message is not None
assert "Network error" in resolutions[1].error_message
class TestDNSService: