From 6ce1d7da2f9dce6a3d806253c73cec0d98542a6d Mon Sep 17 00:00:00 2001 From: phg Date: Fri, 4 Sep 2026 21:47:35 +0200 Subject: [PATCH] Fix unawaited coroutine test warnings --- pyproject.toml | 6 +++ tests/test_dns.py | 115 +++++++++++++++++++++++++++------------------- 2 files changed, 75 insertions(+), 46 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d49d9ba..a87509a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", +] diff --git a/tests/test_dns.py b/tests/test_dns.py index 9c1a3d8..8c90cea 100644 --- a/tests/test_dns.py +++ b/tests/test_dns.py @@ -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: