Refactor DNS resolution tests to use AsyncMock and improve timeout handling
This commit is contained in:
parent
d7ca9cc87f
commit
8d99cfe53c
3 changed files with 50 additions and 27 deletions
|
@ -115,7 +115,10 @@ class TestResolveHostname:
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_timeout_resolution(self):
|
async def test_timeout_resolution(self):
|
||||||
"""Test hostname resolution timeout."""
|
"""Test hostname resolution timeout."""
|
||||||
with patch("asyncio.wait_for", side_effect=asyncio.TimeoutError()):
|
async def mock_wait_for(*args, **kwargs):
|
||||||
|
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)
|
resolution = await resolve_hostname("slow.example", timeout=1.0)
|
||||||
|
|
||||||
assert resolution.hostname == "slow.example"
|
assert resolution.hostname == "slow.example"
|
||||||
|
@ -161,7 +164,7 @@ class TestResolveHostnamesBatch:
|
||||||
"""Test successful batch hostname resolution."""
|
"""Test successful batch hostname resolution."""
|
||||||
hostnames = ["example.com", "test.example"]
|
hostnames = ["example.com", "test.example"]
|
||||||
|
|
||||||
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
with patch("src.hosts.core.dns.resolve_hostname", new_callable=AsyncMock) as mock_resolve:
|
||||||
# Mock successful resolutions
|
# Mock successful resolutions
|
||||||
mock_resolve.side_effect = [
|
mock_resolve.side_effect = [
|
||||||
DNSResolution(
|
DNSResolution(
|
||||||
|
@ -191,23 +194,26 @@ class TestResolveHostnamesBatch:
|
||||||
"""Test batch resolution with mixed success/failure."""
|
"""Test batch resolution with mixed success/failure."""
|
||||||
hostnames = ["example.com", "nonexistent.example"]
|
hostnames = ["example.com", "nonexistent.example"]
|
||||||
|
|
||||||
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
with patch("src.hosts.core.dns.resolve_hostname", new_callable=AsyncMock) as mock_resolve:
|
||||||
# Mock mixed results
|
# Mock mixed results - use side_effect as a proper async function
|
||||||
mock_resolve.side_effect = [
|
async def mock_side_effect(hostname, timeout=5.0):
|
||||||
DNSResolution(
|
if hostname == "example.com":
|
||||||
hostname="example.com",
|
return DNSResolution(
|
||||||
resolved_ip="192.0.2.1",
|
hostname="example.com",
|
||||||
status=DNSResolutionStatus.RESOLVED,
|
resolved_ip="192.0.2.1",
|
||||||
resolved_at=datetime.now(),
|
status=DNSResolutionStatus.RESOLVED,
|
||||||
),
|
resolved_at=datetime.now(),
|
||||||
DNSResolution(
|
)
|
||||||
hostname="nonexistent.example",
|
else:
|
||||||
resolved_ip=None,
|
return DNSResolution(
|
||||||
status=DNSResolutionStatus.RESOLUTION_FAILED,
|
hostname="nonexistent.example",
|
||||||
resolved_at=datetime.now(),
|
resolved_ip=None,
|
||||||
error_message="Name not found",
|
status=DNSResolutionStatus.RESOLUTION_FAILED,
|
||||||
),
|
resolved_at=datetime.now(),
|
||||||
]
|
error_message="Name not found",
|
||||||
|
)
|
||||||
|
|
||||||
|
mock_resolve.side_effect = mock_side_effect
|
||||||
|
|
||||||
resolutions = await resolve_hostnames_batch(hostnames)
|
resolutions = await resolve_hostnames_batch(hostnames)
|
||||||
|
|
||||||
|
@ -270,7 +276,7 @@ class TestDNSService:
|
||||||
"""Test async resolution when service is enabled."""
|
"""Test async resolution when service is enabled."""
|
||||||
service = DNSService(enabled=True)
|
service = DNSService(enabled=True)
|
||||||
|
|
||||||
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
with patch("src.hosts.core.dns.resolve_hostname", new_callable=AsyncMock) as mock_resolve:
|
||||||
mock_resolution = DNSResolution(
|
mock_resolution = DNSResolution(
|
||||||
hostname="example.com",
|
hostname="example.com",
|
||||||
resolved_ip="192.0.2.1",
|
resolved_ip="192.0.2.1",
|
||||||
|
@ -301,7 +307,7 @@ class TestDNSService:
|
||||||
"""Test manual entry refresh."""
|
"""Test manual entry refresh."""
|
||||||
service = DNSService(enabled=True)
|
service = DNSService(enabled=True)
|
||||||
|
|
||||||
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
with patch("src.hosts.core.dns.resolve_hostname", new_callable=AsyncMock) as mock_resolve:
|
||||||
mock_resolution = DNSResolution(
|
mock_resolution = DNSResolution(
|
||||||
hostname="example.com",
|
hostname="example.com",
|
||||||
resolved_ip="192.0.2.1",
|
resolved_ip="192.0.2.1",
|
||||||
|
|
|
@ -421,7 +421,7 @@ class TestEntryFilter:
|
||||||
|
|
||||||
# Apply filters and check preset name is preserved
|
# Apply filters and check preset name is preserved
|
||||||
sample_entry = HostEntry("192.168.1.1", ["test.com"], "Test", True)
|
sample_entry = HostEntry("192.168.1.1", ["test.com"], "Test", True)
|
||||||
result = entry_filter.apply_filters([sample_entry], preset_options)
|
entry_filter.apply_filters([sample_entry], preset_options)
|
||||||
|
|
||||||
# The original preset name should be accessible
|
# The original preset name should be accessible
|
||||||
assert preset_options.preset_name == "Active Only"
|
assert preset_options.preset_name == "Active Only"
|
||||||
|
|
|
@ -833,11 +833,20 @@ class TestHostsManagerApp:
|
||||||
|
|
||||||
app.query_one = mock_query_one
|
app.query_one = mock_query_one
|
||||||
app.edit_handler.handle_entry_type_change = Mock()
|
app.edit_handler.handle_entry_type_change = Mock()
|
||||||
|
|
||||||
|
# Mock the set_timer method to avoid event loop issues in tests
|
||||||
|
with patch.object(app, 'set_timer') as mock_set_timer:
|
||||||
|
app.edit_handler.populate_edit_form_with_type_detection()
|
||||||
|
|
||||||
|
# Verify timer was set with the correct callback
|
||||||
|
mock_set_timer.assert_called_once_with(0.1, app.edit_handler._delayed_radio_setup)
|
||||||
|
|
||||||
|
# Manually call the delayed setup to test the actual logic
|
||||||
|
app.edit_handler._delayed_radio_setup()
|
||||||
|
|
||||||
app.edit_handler.populate_edit_form_with_type_detection()
|
# Verify that the DNS radio was set to True (which should be the pressed button)
|
||||||
|
assert mock_dns_radio.value
|
||||||
# Should set DNS radio button as pressed and populate DNS field
|
assert not mock_ip_radio.value
|
||||||
assert mock_radio_set.pressed_button == mock_dns_radio
|
|
||||||
assert mock_dns_input.value == "example.com"
|
assert mock_dns_input.value == "example.com"
|
||||||
app.edit_handler.handle_entry_type_change.assert_called_with("dns")
|
app.edit_handler.handle_entry_type_change.assert_called_with("dns")
|
||||||
|
|
||||||
|
@ -937,7 +946,15 @@ class TestHostsManagerApp:
|
||||||
app.manager.save_hosts_file = Mock(return_value=(True, "Success"))
|
app.manager.save_hosts_file = Mock(return_value=(True, "Success"))
|
||||||
app.table_handler.populate_entries_table = Mock()
|
app.table_handler.populate_entries_table = Mock()
|
||||||
app.details_handler.update_entry_details = Mock()
|
app.details_handler.update_entry_details = Mock()
|
||||||
app.run_worker = Mock()
|
|
||||||
|
# Create a mock that properly handles and closes coroutines
|
||||||
|
def consume_coro(coro, **kwargs):
|
||||||
|
# If it's a coroutine, close it to prevent warnings
|
||||||
|
if hasattr(coro, 'close'):
|
||||||
|
coro.close()
|
||||||
|
return None
|
||||||
|
|
||||||
|
app.run_worker = Mock(side_effect=consume_coro)
|
||||||
|
|
||||||
# Test action_refresh_dns in edit mode - should proceed
|
# Test action_refresh_dns in edit mode - should proceed
|
||||||
app.action_refresh_dns()
|
app.action_refresh_dns()
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue