Refactor DNS resolution: remove background service components, simplify manual resolution, and update configuration and tests accordingly.
This commit is contained in:
parent
b2d48be045
commit
f8b235ab24
6 changed files with 177 additions and 348 deletions
110
implementation_plan.md
Normal file
110
implementation_plan.md
Normal file
|
|
@ -0,0 +1,110 @@
|
|||
# Implementation Plan
|
||||
|
||||
## [Overview]
|
||||
Remove the DNS Resolution Service background functionality while preserving manual DNS resolution capabilities.
|
||||
|
||||
The current DNS system includes both background automatic resolution that runs periodically and manual user-triggered resolution. This implementation will remove all background resolution components including the DNS service lifecycle, caching system, automatic refresh intervals, and service toggle functionality. Manual DNS resolution will be preserved and simplified to work without caching, making DNS resolution purely on-demand when users explicitly trigger it via the refresh DNS action (Ctrl+R).
|
||||
|
||||
## [Types]
|
||||
Simplify DNS-related data structures by removing background service state management.
|
||||
|
||||
**DNSService class modifications:**
|
||||
- Remove `_background_task: Optional[asyncio.Task]` field
|
||||
- Remove `_stop_event: asyncio.Event` field
|
||||
- Remove `_resolution_cache: Dict[str, DNSResolution]` field
|
||||
- Remove `_update_callback: Optional[Callable]` field
|
||||
- Remove `update_interval: int` parameter and field
|
||||
- Keep `enabled: bool` and `timeout: float` for manual resolution control
|
||||
- Remove cache-related methods: `get_cached_resolution()`, `clear_cache()`, `get_cache_stats()`
|
||||
|
||||
**Configuration type changes:**
|
||||
- Remove `dns_resolution.interval` setting
|
||||
- Remove `dns_resolution.cache_ttl` setting
|
||||
- Keep `dns_resolution.enabled` and `dns_resolution.timeout` for manual resolution
|
||||
|
||||
## [Files]
|
||||
Remove background DNS service components across multiple files.
|
||||
|
||||
**Files to be modified:**
|
||||
- `src/hosts/core/dns.py` - Remove background service, caching, and lifecycle management
|
||||
- `src/hosts/tui/app.py` - Remove background service initialization and toggle functionality
|
||||
- `src/hosts/core/config.py` - Remove background resolution configuration options
|
||||
- `src/hosts/tui/keybindings.py` - Remove DNS service toggle keybinding
|
||||
- `tests/test_dns.py` - Update tests to reflect manual-only functionality
|
||||
- `tests/test_config.py` - Remove tests for background DNS configuration
|
||||
|
||||
**No new files to be created**
|
||||
**No files to be deleted**
|
||||
|
||||
## [Functions]
|
||||
Remove background service functions and simplify DNS resolution interface.
|
||||
|
||||
**Functions to be removed from DNSService:**
|
||||
- `start_background_resolution()` - Background service startup
|
||||
- `stop_background_resolution()` - Background service shutdown
|
||||
- `_background_worker()` - Background resolution worker loop
|
||||
- `_resolve_and_cache()` - Background resolution with caching
|
||||
- `set_update_callback()` - Callback registration for background updates
|
||||
- `get_cached_resolution()` - Cache retrieval method
|
||||
- `clear_cache()` - Cache clearing method
|
||||
- `get_cache_stats()` - Cache statistics method
|
||||
|
||||
**Functions to be modified:**
|
||||
- `__init__()` - Remove update_interval, cache, background task initialization
|
||||
- `resolve_entry_async()` - Remove cache logic, perform direct resolution
|
||||
- `resolve_entry()` - Remove cache logic and background task scheduling
|
||||
- `refresh_entry()` - Simplify to direct resolution without cache management
|
||||
- `refresh_all_entries()` - Simplify to direct batch resolution without caching
|
||||
|
||||
**Functions to be removed from HostsManagerApp:**
|
||||
- `action_toggle_dns_service()` - DNS service toggle functionality
|
||||
|
||||
**Configuration functions to be removed:**
|
||||
- `get_dns_resolution_interval()` - Background interval setting
|
||||
- `set_dns_resolution_interval()` - Background interval configuration
|
||||
- `get_dns_cache_ttl()` - Cache TTL setting
|
||||
|
||||
## [Classes]
|
||||
Simplify DNSService class by removing background service capabilities.
|
||||
|
||||
**Modified classes:**
|
||||
- **DNSService** (src/hosts/core/dns.py):
|
||||
- Remove background service state management
|
||||
- Remove caching infrastructure
|
||||
- Remove callback system for background updates
|
||||
- Simplify constructor to only accept `enabled` and `timeout` parameters
|
||||
- Keep manual resolution methods: `resolve_entry_async()`, `refresh_entry()`, `refresh_all_entries()`
|
||||
|
||||
**No new classes to be created**
|
||||
**No classes to be removed**
|
||||
|
||||
## [Dependencies]
|
||||
No changes to external dependencies required.
|
||||
|
||||
All required dependencies for manual DNS resolution (asyncio, socket) remain unchanged. Background service removal eliminates some asyncio usage but doesn't require dependency modifications.
|
||||
|
||||
## [Testing]
|
||||
Update existing DNS tests to focus on manual resolution functionality.
|
||||
|
||||
**Test modifications required:**
|
||||
- Remove background service lifecycle tests from `TestDNSService`
|
||||
- Remove cache-related tests: cache hit/miss, cache operations, cache stats
|
||||
- Remove background worker and callback tests
|
||||
- Keep and update manual resolution tests
|
||||
- Update configuration tests to remove background DNS settings
|
||||
- Simplify DNSService initialization tests to reflect new constructor
|
||||
|
||||
**Test files to modify:**
|
||||
- `tests/test_dns.py` - Remove ~15 background service and cache tests
|
||||
- `tests/test_config.py` - Remove background DNS configuration tests
|
||||
|
||||
## [Implementation Order]
|
||||
Sequential implementation to minimize conflicts and ensure system stability.
|
||||
|
||||
1. **Update DNSService class** - Remove background service infrastructure and caching
|
||||
2. **Update configuration system** - Remove background DNS settings and related methods
|
||||
3. **Remove keybinding** - Remove DNS service toggle from keybindings
|
||||
4. **Update main application** - Remove background service initialization and toggle action
|
||||
5. **Update tests** - Remove background service tests and update remaining tests
|
||||
6. **Verify manual DNS resolution** - Test that manual refresh functionality still works
|
||||
7. **Test complete removal** - Ensure no background DNS activity occurs
|
||||
|
|
@ -37,9 +37,7 @@ class Config:
|
|||
},
|
||||
"dns_resolution": {
|
||||
"enabled": True,
|
||||
"interval": 300, # 5 minutes in seconds
|
||||
"timeout": 5.0, # 5 seconds timeout
|
||||
"cache_ttl": 300, # 5 minutes cache time-to-live
|
||||
},
|
||||
"filter_settings": {
|
||||
"remember_filter_state": True,
|
||||
|
|
@ -112,18 +110,10 @@ class Config:
|
|||
"""Check if DNS resolution is enabled."""
|
||||
return self.get("dns_resolution", {}).get("enabled", True)
|
||||
|
||||
def get_dns_resolution_interval(self) -> int:
|
||||
"""Get DNS resolution update interval in seconds."""
|
||||
return self.get("dns_resolution", {}).get("interval", 300)
|
||||
|
||||
def get_dns_timeout(self) -> float:
|
||||
"""Get DNS resolution timeout in seconds."""
|
||||
return self.get("dns_resolution", {}).get("timeout", 5.0)
|
||||
|
||||
def get_dns_cache_ttl(self) -> int:
|
||||
"""Get DNS cache time-to-live in seconds."""
|
||||
return self.get("dns_resolution", {}).get("cache_ttl", 300)
|
||||
|
||||
def set_dns_resolution_enabled(self, enabled: bool) -> None:
|
||||
"""Enable or disable DNS resolution."""
|
||||
dns_settings = self.get("dns_resolution", {})
|
||||
|
|
@ -131,13 +121,6 @@ class Config:
|
|||
self.set("dns_resolution", dns_settings)
|
||||
self.save()
|
||||
|
||||
def set_dns_resolution_interval(self, interval: int) -> None:
|
||||
"""Set DNS resolution update interval in seconds."""
|
||||
dns_settings = self.get("dns_resolution", {})
|
||||
dns_settings["interval"] = interval
|
||||
self.set("dns_resolution", dns_settings)
|
||||
self.save()
|
||||
|
||||
def set_dns_timeout(self, timeout: float) -> None:
|
||||
"""Set DNS resolution timeout in seconds."""
|
||||
dns_settings = self.get("dns_resolution", {})
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
"""DNS resolution service for hosts manager.
|
||||
|
||||
Provides background DNS resolution capabilities with timeout handling,
|
||||
Provides manual DNS resolution capabilities with timeout handling,
|
||||
batch processing, and status tracking for hostname to IP address resolution.
|
||||
"""
|
||||
|
||||
|
|
@ -134,78 +134,21 @@ async def resolve_hostnames_batch(hostnames: List[str], timeout: float = 5.0) ->
|
|||
|
||||
|
||||
class DNSService:
|
||||
"""Background DNS resolution service for hosts entries."""
|
||||
"""DNS resolution service for hosts entries."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
update_interval: int = 300, # 5 minutes
|
||||
enabled: bool = True,
|
||||
timeout: float = 5.0
|
||||
):
|
||||
"""Initialize DNS service.
|
||||
|
||||
Args:
|
||||
update_interval: Seconds between background updates
|
||||
enabled: Whether DNS resolution is enabled
|
||||
timeout: Timeout for individual DNS queries
|
||||
"""
|
||||
self.update_interval = update_interval
|
||||
self.enabled = enabled
|
||||
self.timeout = timeout
|
||||
self._background_task: Optional[asyncio.Task] = None
|
||||
self._stop_event = asyncio.Event()
|
||||
self._resolution_cache: Dict[str, DNSResolution] = {}
|
||||
self._update_callback: Optional[Callable] = None
|
||||
|
||||
def set_update_callback(self, callback: Callable) -> None:
|
||||
"""Set callback function for resolution updates.
|
||||
|
||||
Args:
|
||||
callback: Function to call when resolutions are updated
|
||||
"""
|
||||
self._update_callback = callback
|
||||
|
||||
async def start_background_resolution(self) -> None:
|
||||
"""Start background DNS resolution service."""
|
||||
if not self.enabled or self._background_task is not None:
|
||||
return
|
||||
|
||||
self._stop_event.clear()
|
||||
self._background_task = asyncio.create_task(self._background_worker())
|
||||
logger.info("DNS background resolution service started")
|
||||
|
||||
async def stop_background_resolution(self) -> None:
|
||||
"""Stop background DNS resolution service gracefully."""
|
||||
if self._background_task is None:
|
||||
return
|
||||
|
||||
self._stop_event.set()
|
||||
try:
|
||||
await asyncio.wait_for(self._background_task, timeout=10.0)
|
||||
except asyncio.TimeoutError:
|
||||
self._background_task.cancel()
|
||||
|
||||
self._background_task = None
|
||||
logger.info("DNS background resolution service stopped")
|
||||
|
||||
async def _background_worker(self) -> None:
|
||||
"""Background worker for periodic DNS resolution."""
|
||||
while not self._stop_event.is_set():
|
||||
try:
|
||||
# Wait for either stop event or update interval
|
||||
await asyncio.wait_for(
|
||||
self._stop_event.wait(),
|
||||
timeout=self.update_interval
|
||||
)
|
||||
# If we get here, stop was requested
|
||||
break
|
||||
except asyncio.TimeoutError:
|
||||
# Time for periodic update
|
||||
if self.enabled and self._update_callback:
|
||||
try:
|
||||
await self._update_callback()
|
||||
except Exception as e:
|
||||
logger.error(f"Error in DNS update callback: {e}")
|
||||
|
||||
async def resolve_entry_async(self, hostname: str) -> DNSResolution:
|
||||
"""Resolve DNS for a hostname asynchronously.
|
||||
|
|
@ -216,59 +159,16 @@ class DNSService:
|
|||
Returns:
|
||||
DNSResolution result
|
||||
"""
|
||||
# Check cache first
|
||||
if hostname in self._resolution_cache:
|
||||
cached = self._resolution_cache[hostname]
|
||||
# Use cached result if less than 5 minutes old
|
||||
if cached.get_age_seconds() < 300:
|
||||
return cached
|
||||
|
||||
# Perform new resolution
|
||||
resolution = await resolve_hostname(hostname, self.timeout)
|
||||
self._resolution_cache[hostname] = resolution
|
||||
return resolution
|
||||
|
||||
def resolve_entry(self, hostname: str) -> DNSResolution:
|
||||
"""Resolve DNS for a hostname synchronously.
|
||||
|
||||
Args:
|
||||
hostname: Hostname to resolve
|
||||
|
||||
Returns:
|
||||
DNSResolution result (may be cached or indicate resolution in progress)
|
||||
"""
|
||||
# Check cache first
|
||||
if hostname in self._resolution_cache:
|
||||
cached = self._resolution_cache[hostname]
|
||||
# Use cached result if less than 5 minutes old
|
||||
if cached.get_age_seconds() < 300:
|
||||
return cached
|
||||
|
||||
# Return "resolving" status and trigger async resolution
|
||||
resolving_result = DNSResolution(
|
||||
if not self.enabled:
|
||||
return DNSResolution(
|
||||
hostname=hostname,
|
||||
resolved_ip=None,
|
||||
status=DNSResolutionStatus.RESOLVING,
|
||||
resolved_at=datetime.now()
|
||||
status=DNSResolutionStatus.NOT_RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
error_message="DNS resolution is disabled"
|
||||
)
|
||||
|
||||
# Schedule async resolution
|
||||
if self.enabled:
|
||||
asyncio.create_task(self._resolve_and_cache(hostname))
|
||||
|
||||
return resolving_result
|
||||
|
||||
async def _resolve_and_cache(self, hostname: str) -> None:
|
||||
"""Resolve hostname and update cache."""
|
||||
try:
|
||||
resolution = await resolve_hostname(hostname, self.timeout)
|
||||
self._resolution_cache[hostname] = resolution
|
||||
|
||||
# Notify callback if available
|
||||
if self._update_callback:
|
||||
await self._update_callback()
|
||||
except Exception as e:
|
||||
logger.error(f"Error resolving {hostname}: {e}")
|
||||
return await resolve_hostname(hostname, self.timeout)
|
||||
|
||||
async def refresh_entry(self, hostname: str) -> DNSResolution:
|
||||
"""Manually refresh DNS resolution for hostname.
|
||||
|
|
@ -279,13 +179,7 @@ class DNSService:
|
|||
Returns:
|
||||
Fresh DNSResolution result
|
||||
"""
|
||||
# Remove from cache to force fresh resolution
|
||||
self._resolution_cache.pop(hostname, None)
|
||||
|
||||
# Perform fresh resolution
|
||||
resolution = await resolve_hostname(hostname, self.timeout)
|
||||
self._resolution_cache[hostname] = resolution
|
||||
return resolution
|
||||
return await self.resolve_entry_async(hostname)
|
||||
|
||||
async def refresh_all_entries(self, hostnames: List[str]) -> List[DNSResolution]:
|
||||
"""Manually refresh DNS resolution for multiple hostnames.
|
||||
|
|
@ -296,49 +190,19 @@ class DNSService:
|
|||
Returns:
|
||||
List of fresh DNSResolution results
|
||||
"""
|
||||
# Clear cache for all hostnames
|
||||
for hostname in hostnames:
|
||||
self._resolution_cache.pop(hostname, None)
|
||||
if not self.enabled:
|
||||
return [
|
||||
DNSResolution(
|
||||
hostname=hostname,
|
||||
resolved_ip=None,
|
||||
status=DNSResolutionStatus.NOT_RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
error_message="DNS resolution is disabled"
|
||||
)
|
||||
for hostname in hostnames
|
||||
]
|
||||
|
||||
# Perform batch resolution
|
||||
resolutions = await resolve_hostnames_batch(hostnames, self.timeout)
|
||||
|
||||
# Update cache
|
||||
for resolution in resolutions:
|
||||
self._resolution_cache[resolution.hostname] = resolution
|
||||
|
||||
return resolutions
|
||||
|
||||
def get_cached_resolution(self, hostname: str) -> Optional[DNSResolution]:
|
||||
"""Get cached DNS resolution for hostname.
|
||||
|
||||
Args:
|
||||
hostname: Hostname to look up
|
||||
|
||||
Returns:
|
||||
Cached DNSResolution if available
|
||||
"""
|
||||
return self._resolution_cache.get(hostname)
|
||||
|
||||
def clear_cache(self) -> None:
|
||||
"""Clear DNS resolution cache."""
|
||||
self._resolution_cache.clear()
|
||||
|
||||
def get_cache_stats(self) -> Dict[str, int]:
|
||||
"""Get cache statistics.
|
||||
|
||||
Returns:
|
||||
Dictionary with cache statistics
|
||||
"""
|
||||
total = len(self._resolution_cache)
|
||||
successful = sum(1 for r in self._resolution_cache.values() if r.is_success())
|
||||
failed = total - successful
|
||||
|
||||
return {
|
||||
"total_entries": total,
|
||||
"successful": successful,
|
||||
"failed": failed
|
||||
}
|
||||
return await resolve_hostnames_batch(hostnames, self.timeout)
|
||||
|
||||
|
||||
def compare_ips(stored_ip: str, resolved_ip: str) -> DNSResolutionStatus:
|
||||
|
|
|
|||
|
|
@ -65,7 +65,6 @@ class HostsManagerApp(App):
|
|||
# Initialize DNS service
|
||||
dns_config = self.config.get("dns_resolution", {})
|
||||
self.dns_service = DNSService(
|
||||
update_interval=dns_config.get("interval", 300),
|
||||
enabled=dns_config.get("enabled", True),
|
||||
timeout=dns_config.get("timeout", 5.0)
|
||||
)
|
||||
|
|
@ -198,10 +197,6 @@ class HostsManagerApp(App):
|
|||
self.load_hosts_file()
|
||||
self._setup_footer()
|
||||
|
||||
# Start DNS service if enabled
|
||||
if self.dns_service.enabled:
|
||||
self.run_worker(self.dns_service.start_background_resolution(), exclusive=False)
|
||||
|
||||
def load_hosts_file(self) -> None:
|
||||
"""Load the hosts file and populate the table."""
|
||||
try:
|
||||
|
|
@ -743,18 +738,6 @@ class HostsManagerApp(App):
|
|||
self.run_worker(refresh_dns(), exclusive=False)
|
||||
self.update_status("🔄 Starting DNS resolution...")
|
||||
|
||||
def action_toggle_dns_service(self) -> None:
|
||||
"""Toggle DNS resolution service on/off."""
|
||||
if self.dns_service.enabled:
|
||||
# Stop the background resolution service
|
||||
self.run_worker(self.dns_service.stop_background_resolution(), exclusive=False)
|
||||
self.dns_service.enabled = False
|
||||
self.update_status("DNS resolution service stopped")
|
||||
else:
|
||||
# Enable and start the background resolution service
|
||||
self.dns_service.enabled = True
|
||||
self.run_worker(self.dns_service.start_background_resolution(), exclusive=False)
|
||||
self.update_status("DNS resolution service started")
|
||||
|
||||
def action_show_filters(self) -> None:
|
||||
"""Show advanced filtering modal."""
|
||||
|
|
@ -859,8 +842,8 @@ class HostsManagerApp(App):
|
|||
|
||||
async def on_shutdown(self) -> None:
|
||||
"""Clean up resources when the app is shutting down."""
|
||||
if hasattr(self, 'dns_service') and self.dns_service:
|
||||
await self.dns_service.stop_background_resolution()
|
||||
# No DNS service cleanup needed for manual-only resolution
|
||||
pass
|
||||
|
||||
# Delegated methods for backward compatibility with tests
|
||||
def has_entry_changes(self) -> bool:
|
||||
|
|
|
|||
|
|
@ -45,7 +45,6 @@ HOSTS_MANAGER_BINDINGS = [
|
|||
Binding("ctrl+z", "undo", "Undo", show=False, id="left:undo"),
|
||||
Binding("ctrl+y", "redo", "Redo", show=False, id="left:redo"),
|
||||
Binding("ctrl+r", "refresh_dns", "Refresh DNS", show=False, id="left:refresh_dns"),
|
||||
Binding("ctrl+t", "toggle_dns_service", "Toggle DNS service", show=False),
|
||||
Binding("escape", "exit_edit_entry", "Exit edit mode", show=False),
|
||||
Binding("tab", "next_field", "Next field", show=False),
|
||||
Binding("shift+tab", "prev_field", "Previous field", show=False),
|
||||
|
|
|
|||
|
|
@ -253,66 +253,23 @@ class TestDNSService:
|
|||
|
||||
def test_initialization(self):
|
||||
"""Test DNS service initialization."""
|
||||
service = DNSService(update_interval=600, enabled=True, timeout=10.0)
|
||||
service = DNSService(enabled=True, timeout=10.0)
|
||||
|
||||
assert service.update_interval == 600
|
||||
assert service.enabled is True
|
||||
assert service.timeout == 10.0
|
||||
assert service._background_task is None
|
||||
assert service._resolution_cache == {}
|
||||
|
||||
def test_update_callback_setting(self):
|
||||
"""Test setting update callback."""
|
||||
def test_initialization_defaults(self):
|
||||
"""Test DNS service initialization with defaults."""
|
||||
service = DNSService()
|
||||
callback = MagicMock()
|
||||
|
||||
service.set_update_callback(callback)
|
||||
assert service._update_callback is callback
|
||||
assert service.enabled is True
|
||||
assert service.timeout == 5.0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_background_service_lifecycle(self):
|
||||
"""Test starting and stopping background service."""
|
||||
async def test_resolve_entry_async_enabled(self):
|
||||
"""Test async resolution when service is enabled."""
|
||||
service = DNSService(enabled=True)
|
||||
|
||||
# Start service
|
||||
await service.start_background_resolution()
|
||||
assert service._background_task is not None
|
||||
assert not service._stop_event.is_set()
|
||||
|
||||
# Stop service
|
||||
await service.stop_background_resolution()
|
||||
assert service._background_task is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_background_service_disabled(self):
|
||||
"""Test background service when disabled."""
|
||||
service = DNSService(enabled=False)
|
||||
|
||||
await service.start_background_resolution()
|
||||
assert service._background_task is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_entry_async_cache_hit(self):
|
||||
"""Test async resolution with cache hit."""
|
||||
service = DNSService()
|
||||
|
||||
# Add entry to cache
|
||||
cached_resolution = DNSResolution(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.1",
|
||||
status=DNSResolutionStatus.RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
)
|
||||
service._resolution_cache["example.com"] = cached_resolution
|
||||
|
||||
resolution = await service.resolve_entry_async("example.com")
|
||||
assert resolution is cached_resolution
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_entry_async_cache_miss(self):
|
||||
"""Test async resolution with cache miss."""
|
||||
service = DNSService()
|
||||
|
||||
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
||||
mock_resolution = DNSResolution(
|
||||
hostname="example.com",
|
||||
|
|
@ -325,84 +282,47 @@ class TestDNSService:
|
|||
resolution = await service.resolve_entry_async("example.com")
|
||||
|
||||
assert resolution is mock_resolution
|
||||
assert service._resolution_cache["example.com"] is mock_resolution
|
||||
mock_resolve.assert_called_once_with("example.com", 5.0)
|
||||
|
||||
def test_resolve_entry_sync_cache_hit(self):
|
||||
"""Test synchronous resolution with cache hit."""
|
||||
service = DNSService()
|
||||
|
||||
# Add entry to cache
|
||||
cached_resolution = DNSResolution(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.1",
|
||||
status=DNSResolutionStatus.RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
)
|
||||
service._resolution_cache["example.com"] = cached_resolution
|
||||
|
||||
resolution = service.resolve_entry("example.com")
|
||||
assert resolution is cached_resolution
|
||||
|
||||
def test_resolve_entry_sync_cache_miss(self):
|
||||
"""Test synchronous resolution with cache miss."""
|
||||
service = DNSService(enabled=True)
|
||||
|
||||
with patch("asyncio.create_task") as mock_create_task:
|
||||
resolution = service.resolve_entry("example.com")
|
||||
|
||||
assert resolution.hostname == "example.com"
|
||||
assert resolution.status == DNSResolutionStatus.RESOLVING
|
||||
assert resolution.resolved_ip is None
|
||||
mock_create_task.assert_called_once()
|
||||
|
||||
def test_resolve_entry_sync_disabled(self):
|
||||
"""Test synchronous resolution when service is disabled."""
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_entry_async_disabled(self):
|
||||
"""Test async resolution when service is disabled."""
|
||||
service = DNSService(enabled=False)
|
||||
|
||||
with patch("asyncio.create_task") as mock_create_task:
|
||||
resolution = service.resolve_entry("example.com")
|
||||
resolution = await service.resolve_entry_async("example.com")
|
||||
|
||||
assert resolution.hostname == "example.com"
|
||||
assert resolution.status == DNSResolutionStatus.RESOLVING
|
||||
mock_create_task.assert_not_called()
|
||||
assert resolution.resolved_ip is None
|
||||
assert resolution.status == DNSResolutionStatus.NOT_RESOLVED
|
||||
assert resolution.error_message == "DNS resolution is disabled"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_entry(self):
|
||||
"""Test manual entry refresh."""
|
||||
service = DNSService()
|
||||
service = DNSService(enabled=True)
|
||||
|
||||
# Add stale entry to cache
|
||||
stale_resolution = DNSResolution(
|
||||
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
||||
mock_resolution = DNSResolution(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.1",
|
||||
status=DNSResolutionStatus.RESOLVED,
|
||||
resolved_at=datetime.now() - timedelta(hours=1),
|
||||
)
|
||||
service._resolution_cache["example.com"] = stale_resolution
|
||||
|
||||
with patch("src.hosts.core.dns.resolve_hostname") as mock_resolve:
|
||||
fresh_resolution = DNSResolution(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.2",
|
||||
status=DNSResolutionStatus.RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
)
|
||||
mock_resolve.return_value = fresh_resolution
|
||||
mock_resolve.return_value = mock_resolution
|
||||
|
||||
result = await service.refresh_entry("example.com")
|
||||
|
||||
assert result is fresh_resolution
|
||||
assert service._resolution_cache["example.com"] is fresh_resolution
|
||||
assert "example.com" not in service._resolution_cache or service._resolution_cache["example.com"].resolved_ip == "192.0.2.2"
|
||||
assert result is mock_resolution
|
||||
mock_resolve.assert_called_once_with("example.com", 5.0)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_all_entries(self):
|
||||
"""Test manual refresh of all entries."""
|
||||
service = DNSService()
|
||||
async def test_refresh_all_entries_enabled(self):
|
||||
"""Test manual refresh of all entries when enabled."""
|
||||
service = DNSService(enabled=True)
|
||||
hostnames = ["example.com", "test.example"]
|
||||
|
||||
with patch("src.hosts.core.dns.resolve_hostnames_batch") as mock_batch:
|
||||
fresh_resolutions = [
|
||||
mock_resolutions = [
|
||||
DNSResolution(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.1",
|
||||
|
|
@ -416,57 +336,27 @@ class TestDNSService:
|
|||
resolved_at=datetime.now(),
|
||||
),
|
||||
]
|
||||
mock_batch.return_value = fresh_resolutions
|
||||
mock_batch.return_value = mock_resolutions
|
||||
|
||||
results = await service.refresh_all_entries(hostnames)
|
||||
|
||||
assert results == fresh_resolutions
|
||||
assert len(service._resolution_cache) == 2
|
||||
assert service._resolution_cache["example.com"].resolved_ip == "192.0.2.1"
|
||||
assert service._resolution_cache["test.example"].resolved_ip == "192.0.2.2"
|
||||
assert results == mock_resolutions
|
||||
mock_batch.assert_called_once_with(hostnames, 5.0)
|
||||
|
||||
def test_cache_operations(self):
|
||||
"""Test cache operations."""
|
||||
service = DNSService()
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_all_entries_disabled(self):
|
||||
"""Test manual refresh of all entries when disabled."""
|
||||
service = DNSService(enabled=False)
|
||||
hostnames = ["example.com", "test.example"]
|
||||
|
||||
# Test empty cache
|
||||
assert service.get_cached_resolution("example.com") is None
|
||||
results = await service.refresh_all_entries(hostnames)
|
||||
|
||||
# Add to cache
|
||||
resolution = DNSResolution(
|
||||
hostname="example.com",
|
||||
resolved_ip="192.0.2.1",
|
||||
status=DNSResolutionStatus.RESOLVED,
|
||||
resolved_at=datetime.now(),
|
||||
)
|
||||
service._resolution_cache["example.com"] = resolution
|
||||
|
||||
# Test cache retrieval
|
||||
assert service.get_cached_resolution("example.com") is resolution
|
||||
|
||||
# Test cache stats
|
||||
stats = service.get_cache_stats()
|
||||
assert stats["total_entries"] == 1
|
||||
assert stats["successful"] == 1
|
||||
assert stats["failed"] == 0
|
||||
|
||||
# Add failed resolution
|
||||
failed_resolution = DNSResolution(
|
||||
hostname="failed.example",
|
||||
resolved_ip=None,
|
||||
status=DNSResolutionStatus.RESOLUTION_FAILED,
|
||||
resolved_at=datetime.now(),
|
||||
)
|
||||
service._resolution_cache["failed.example"] = failed_resolution
|
||||
|
||||
stats = service.get_cache_stats()
|
||||
assert stats["total_entries"] == 2
|
||||
assert stats["successful"] == 1
|
||||
assert stats["failed"] == 1
|
||||
|
||||
# Clear cache
|
||||
service.clear_cache()
|
||||
assert len(service._resolution_cache) == 0
|
||||
assert len(results) == 2
|
||||
for i, result in enumerate(results):
|
||||
assert result.hostname == hostnames[i]
|
||||
assert result.resolved_ip is None
|
||||
assert result.status == DNSResolutionStatus.NOT_RESOLVED
|
||||
assert result.error_message == "DNS resolution is disabled"
|
||||
|
||||
|
||||
class TestHostEntryDNSIntegration:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue