Compare commits
No commits in common. "4761f488852e8dcf922cff6a76381b20114d5b4f" and "d7ca9cc87f9d40c4d3f6f7909983f2d2baf8b345" have entirely different histories.
4761f48885
...
d7ca9cc87f
3 changed files with 21 additions and 52 deletions
|
@ -115,10 +115,7 @@ 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."""
|
||||||
async def mock_wait_for(*args, **kwargs):
|
with patch("asyncio.wait_for", side_effect=asyncio.TimeoutError()):
|
||||||
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"
|
||||||
|
@ -142,14 +139,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 = AsyncMock()
|
||||||
mock_loop.return_value = mock_event_loop
|
mock_loop.return_value = mock_event_loop
|
||||||
|
|
||||||
with patch("asyncio.wait_for", side_effect=mock_wait_for):
|
with patch("asyncio.wait_for", return_value=[]):
|
||||||
resolution = await resolve_hostname("empty.example")
|
resolution = await resolve_hostname("empty.example")
|
||||||
|
|
||||||
assert resolution.hostname == "empty.example"
|
assert resolution.hostname == "empty.example"
|
||||||
|
@ -167,7 +161,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", new_callable=AsyncMock) as mock_resolve:
|
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
||||||
# Mock successful resolutions
|
# Mock successful resolutions
|
||||||
mock_resolve.side_effect = [
|
mock_resolve.side_effect = [
|
||||||
DNSResolution(
|
DNSResolution(
|
||||||
|
@ -197,25 +191,23 @@ 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"]
|
||||||
|
|
||||||
# Create a direct async function replacement instead of using AsyncMock
|
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
||||||
async def mock_resolve_hostname(hostname, timeout=5.0):
|
# Mock mixed results
|
||||||
if hostname == "example.com":
|
mock_resolve.side_effect = [
|
||||||
return DNSResolution(
|
DNSResolution(
|
||||||
hostname="example.com",
|
hostname="example.com",
|
||||||
resolved_ip="192.0.2.1",
|
resolved_ip="192.0.2.1",
|
||||||
status=DNSResolutionStatus.RESOLVED,
|
status=DNSResolutionStatus.RESOLVED,
|
||||||
resolved_at=datetime.now(),
|
resolved_at=datetime.now(),
|
||||||
)
|
),
|
||||||
else:
|
DNSResolution(
|
||||||
return DNSResolution(
|
|
||||||
hostname="nonexistent.example",
|
hostname="nonexistent.example",
|
||||||
resolved_ip=None,
|
resolved_ip=None,
|
||||||
status=DNSResolutionStatus.RESOLUTION_FAILED,
|
status=DNSResolutionStatus.RESOLUTION_FAILED,
|
||||||
resolved_at=datetime.now(),
|
resolved_at=datetime.now(),
|
||||||
error_message="Name not found",
|
error_message="Name not found",
|
||||||
)
|
),
|
||||||
|
]
|
||||||
with patch("src.hosts.core.dns.resolve_hostname", mock_resolve_hostname):
|
|
||||||
|
|
||||||
resolutions = await resolve_hostnames_batch(hostnames)
|
resolutions = await resolve_hostnames_batch(hostnames)
|
||||||
|
|
||||||
|
@ -278,17 +270,14 @@ 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", new_callable=AsyncMock) as mock_resolve:
|
with patch("src.hosts.core.dns.resolve_hostname") 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",
|
||||||
status=DNSResolutionStatus.RESOLVED,
|
status=DNSResolutionStatus.RESOLVED,
|
||||||
resolved_at=datetime.now(),
|
resolved_at=datetime.now(),
|
||||||
)
|
)
|
||||||
# Use proper async setup
|
mock_resolve.return_value = mock_resolution
|
||||||
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")
|
resolution = await service.resolve_entry_async("example.com")
|
||||||
|
|
||||||
|
@ -312,17 +301,14 @@ 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", new_callable=AsyncMock) as mock_resolve:
|
with patch("src.hosts.core.dns.resolve_hostname") 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",
|
||||||
status=DNSResolutionStatus.RESOLVED,
|
status=DNSResolutionStatus.RESOLVED,
|
||||||
resolved_at=datetime.now(),
|
resolved_at=datetime.now(),
|
||||||
)
|
)
|
||||||
# Use proper async setup
|
mock_resolve.return_value = mock_resolution
|
||||||
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")
|
result = await service.refresh_entry("example.com")
|
||||||
|
|
||||||
|
|
|
@ -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)
|
||||||
entry_filter.apply_filters([sample_entry], preset_options)
|
result = 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"
|
||||||
|
|
|
@ -834,19 +834,10 @@ 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()
|
app.edit_handler.populate_edit_form_with_type_detection()
|
||||||
|
|
||||||
# Verify timer was set with the correct callback
|
# Should set DNS radio button as pressed and populate DNS field
|
||||||
mock_set_timer.assert_called_once_with(0.1, app.edit_handler._delayed_radio_setup)
|
assert mock_radio_set.pressed_button == mock_dns_radio
|
||||||
|
|
||||||
# Manually call the delayed setup to test the actual logic
|
|
||||||
app.edit_handler._delayed_radio_setup()
|
|
||||||
|
|
||||||
# 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"
|
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")
|
||||||
|
|
||||||
|
@ -946,15 +937,7 @@ 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