Compare commits
2 commits
d7ca9cc87f
...
4761f48885
Author | SHA1 | Date | |
---|---|---|---|
4761f48885 | |||
8d99cfe53c |
3 changed files with 52 additions and 21 deletions
|
@ -115,7 +115,10 @@ class TestResolveHostname:
|
|||
@pytest.mark.asyncio
|
||||
async def test_timeout_resolution(self):
|
||||
"""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)
|
||||
|
||||
assert resolution.hostname == "slow.example"
|
||||
|
@ -139,11 +142,14 @@ class TestResolveHostname:
|
|||
@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_loop.return_value = mock_event_loop
|
||||
|
||||
with patch("asyncio.wait_for", return_value=[]):
|
||||
with patch("asyncio.wait_for", side_effect=mock_wait_for):
|
||||
resolution = await resolve_hostname("empty.example")
|
||||
|
||||
assert resolution.hostname == "empty.example"
|
||||
|
@ -161,7 +167,7 @@ class TestResolveHostnamesBatch:
|
|||
"""Test successful batch hostname resolution."""
|
||||
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_resolve.side_effect = [
|
||||
DNSResolution(
|
||||
|
@ -191,23 +197,25 @@ class TestResolveHostnamesBatch:
|
|||
"""Test batch resolution with mixed success/failure."""
|
||||
hostnames = ["example.com", "nonexistent.example"]
|
||||
|
||||
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
||||
# Mock mixed results
|
||||
mock_resolve.side_effect = [
|
||||
DNSResolution(
|
||||
# Create a direct async function replacement instead of using AsyncMock
|
||||
async def mock_resolve_hostname(hostname, timeout=5.0):
|
||||
if hostname == "example.com":
|
||||
return DNSResolution(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.1",
|
||||
status=DNSResolutionStatus.RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
),
|
||||
DNSResolution(
|
||||
)
|
||||
else:
|
||||
return DNSResolution(
|
||||
hostname="nonexistent.example",
|
||||
resolved_ip=None,
|
||||
status=DNSResolutionStatus.RESOLUTION_FAILED,
|
||||
resolved_at=datetime.now(),
|
||||
error_message="Name not found",
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
with patch("src.hosts.core.dns.resolve_hostname", mock_resolve_hostname):
|
||||
|
||||
resolutions = await resolve_hostnames_batch(hostnames)
|
||||
|
||||
|
@ -270,14 +278,17 @@ class TestDNSService:
|
|||
"""Test async resolution when service is enabled."""
|
||||
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(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.1",
|
||||
status=DNSResolutionStatus.RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
)
|
||||
mock_resolve.return_value = mock_resolution
|
||||
# Use proper async setup
|
||||
async def mock_side_effect(hostname, timeout=5.0):
|
||||
return mock_resolution
|
||||
mock_resolve.side_effect = mock_side_effect
|
||||
|
||||
resolution = await service.resolve_entry_async("example.com")
|
||||
|
||||
|
@ -301,14 +312,17 @@ class TestDNSService:
|
|||
"""Test manual entry refresh."""
|
||||
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(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.1",
|
||||
status=DNSResolutionStatus.RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
)
|
||||
mock_resolve.return_value = mock_resolution
|
||||
# Use proper async setup
|
||||
async def mock_side_effect(hostname, timeout=5.0):
|
||||
return mock_resolution
|
||||
mock_resolve.side_effect = mock_side_effect
|
||||
|
||||
result = await service.refresh_entry("example.com")
|
||||
|
||||
|
|
|
@ -421,7 +421,7 @@ class TestEntryFilter:
|
|||
|
||||
# Apply filters and check preset name is preserved
|
||||
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
|
||||
assert preset_options.preset_name == "Active Only"
|
||||
|
|
|
@ -833,11 +833,20 @@ class TestHostsManagerApp:
|
|||
|
||||
app.query_one = mock_query_one
|
||||
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()
|
||||
|
||||
# Should set DNS radio button as pressed and populate DNS field
|
||||
assert mock_radio_set.pressed_button == mock_dns_radio
|
||||
# Verify that the DNS radio was set to True (which should be the pressed button)
|
||||
assert mock_dns_radio.value
|
||||
assert not mock_ip_radio.value
|
||||
assert mock_dns_input.value == "example.com"
|
||||
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.table_handler.populate_entries_table = 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
|
||||
app.action_refresh_dns()
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue