hosts/src/hosts/tui/app.py

1116 lines
46 KiB
Python

"""
Main application class for the hosts TUI application.
This module contains the main application class that orchestrates
all the handlers and provides the primary user interface.
"""
from textual.app import App, ComposeResult, SuspendNotSupported
from textual.containers import Horizontal, Vertical
from textual.widgets import (
Header,
Static,
DataTable,
Input,
Checkbox,
RadioSet,
RadioButton,
)
from textual.reactive import reactive
from ..core.parser import HostsParser
from ..core.models import HostsFile
from ..core.config import Config
from ..core.manager import HostsManager
from ..core.dns import DNSService
from ..core.filters import EntryFilter, FilterOptions
from .config_modal import ConfigModal
from .add_entry_modal import AddEntryModal
from .delete_confirmation_modal import DeleteConfirmationModal
from .filter_modal import FilterModal
from .custom_footer import CustomFooter
from .styles import HOSTS_MANAGER_CSS
from .keybindings import HOSTS_MANAGER_BINDINGS
from .table_handler import TableHandler
from .details_handler import DetailsHandler
from .edit_handler import EditHandler
from .navigation_handler import NavigationHandler
class HostsManagerApp(App):
"""
Main application class for the hosts TUI manager.
Provides a two-pane interface for managing hosts file entries
with read-only mode by default and explicit edit mode.
"""
ENABLE_COMMAND_PALETTE = False
CSS = HOSTS_MANAGER_CSS
BINDINGS = HOSTS_MANAGER_BINDINGS
help_visible = False
# Reactive attributes
hosts_file: reactive[HostsFile] = reactive(HostsFile())
selected_entry_index: reactive[int] = reactive(0)
edit_mode: reactive[bool] = reactive(False)
entry_edit_mode: reactive[bool] = reactive(False)
sort_column: reactive[str] = reactive("") # "ip" or "hostname"
sort_ascending: reactive[bool] = reactive(True)
search_term: reactive[str] = reactive("")
def __init__(self):
super().__init__()
self.title = "/etc/hosts Manager"
# Initialize core components
self.parser = HostsParser()
self.config = Config()
self.manager = HostsManager()
# Initialize DNS service
dns_config = self.config.get("dns_resolution", {})
self.dns_service = DNSService(
enabled=dns_config.get("enabled", True),
timeout=dns_config.get("timeout", 5.0),
)
# Initialize filtering system
self.entry_filter = EntryFilter()
self.current_filter_options = FilterOptions()
# Initialize handlers
self.table_handler = TableHandler(self)
self.details_handler = DetailsHandler(self)
self.edit_handler = EditHandler(self)
self.navigation_handler = NavigationHandler(self)
# State for edit mode
self.original_entry_values = None
def compose(self) -> ComposeResult:
"""Create child widgets for the app."""
yield Header()
yield CustomFooter(id="custom-footer")
# Search bar above the panes
with Horizontal(classes="search-container") as search_container:
search_container.border_title = "Search"
yield Input(
placeholder="Filter by hostname, IP address, or comment...",
id="search-input",
classes="search-input",
)
with Horizontal(classes="hosts-container"):
# Left pane - entries table
with Vertical(classes="common-pane left-pane") as left_pane:
left_pane.border_title = "Host Entries"
yield DataTable(id="entries-table")
# Right pane - entry details or edit form
with Vertical(classes="common-pane right-pane") as right_pane:
right_pane.border_title = "Entry Details"
# Details display form (disabled inputs)
with Vertical(id="entry-details-display", classes="entry-form"):
with Vertical(
classes="default-section section-no-top-margin"
) as ip_address:
ip_address.border_title = "IP Address"
yield Input(
placeholder="No entry selected",
id="details-ip-input",
disabled=True,
classes="default-input",
)
with Vertical(classes="default-section") as hostnames:
hostnames.border_title = "Hostnames (comma-separated)"
yield Input(
placeholder="No entry selected",
id="details-hostname-input",
disabled=True,
classes="default-input",
)
with Vertical(classes="default-section") as dns_name:
dns_name.border_title = "DNS Name"
yield Input(
placeholder="No DNS name",
id="details-dns-name-input",
disabled=True,
classes="default-input",
)
with Vertical(classes="default-section") as dns_status:
dns_status.border_title = "DNS Status"
yield Input(
placeholder="No DNS status",
id="details-dns-status-input",
disabled=True,
classes="default-input",
)
with Vertical(classes="default-section") as dns_resolved:
dns_resolved.border_title = "Last Resolved"
yield Input(
placeholder="Not resolved yet",
id="details-dns-resolved-input",
disabled=True,
classes="default-input",
)
with Vertical(classes="default-section") as comment:
comment.border_title = "Comment:"
yield Input(
placeholder="No entry selected",
id="details-comment-input",
disabled=True,
classes="default-input",
)
with Vertical(classes="default-section") as active:
active.border_title = "Active"
yield Checkbox(
"Active",
id="details-active-checkbox",
disabled=True,
classes="default-checkbox",
)
# Edit form (initially hidden)
with Vertical(id="entry-edit-form", classes="entry-form hidden"):
# Entry Type Selection
with Vertical(
classes="default-flex-section section-no-top-margin"
) as entry_type:
entry_type.border_title = "Entry Type"
with RadioSet(
id="edit-entry-type-radio", classes="default-radio-set"
):
yield RadioButton(
"IP Address Entry", value=True, id="edit-ip-entry-radio"
)
yield RadioButton(
"DNS Name Entry", id="edit-dns-entry-radio"
)
# IP Address Section
with Vertical(
classes="default-section", id="edit-ip-section"
) as ip_address:
ip_address.border_title = "IP Address"
yield Input(
placeholder="Enter IP address",
id="ip-input",
classes="default-input",
)
# DNS Name Section (initially hidden)
with Vertical(
classes="default-section hidden", id="edit-dns-section"
) as dns_name:
dns_name.border_title = "DNS Name (to resolve)"
yield Input(
placeholder="e.g., example.com",
id="dns-name-input",
classes="default-input",
)
with Vertical(classes="default-section") as hostnames:
hostnames.border_title = "Hostnames (comma-separated)"
yield Input(
placeholder="Enter hostnames",
id="hostname-input",
classes="default-input",
)
with Vertical(classes="default-section") as comment:
comment.border_title = "Comment:"
yield Input(
placeholder="Enter comment (optional)",
id="comment-input",
classes="default-input",
)
with Vertical(classes="default-section") as active:
active.border_title = "Active"
yield Checkbox(
"Active", id="active-checkbox", classes="default-checkbox"
)
# Status bar for error/temporary messages (overlay, doesn't affect layout)
yield Static("", id="status-bar", classes="status-bar hidden")
def on_ready(self) -> None:
"""Called when the app is ready."""
self.load_hosts_file()
self._setup_footer()
def load_hosts_file(self) -> None:
"""Load the hosts file and populate the table."""
try:
# Remember the currently selected entry before reload
previous_entry = None
if self.hosts_file.entries and self.selected_entry_index < len(
self.hosts_file.entries
):
previous_entry = self.hosts_file.entries[self.selected_entry_index]
# Load the hosts file
self.hosts_file = self.parser.parse()
self.table_handler.populate_entries_table()
self.table_handler.restore_cursor_position(previous_entry)
self.update_status()
except Exception as e:
self.update_status(f"❌ Error loading hosts file: {e}")
def _setup_footer(self) -> None:
"""Setup the footer with initial content based on keybindings."""
try:
footer = self.query_one("#custom-footer", CustomFooter)
# Clear existing items
footer.clear_left_items()
footer.clear_right_items()
# Process keybindings and add to appropriate sections
for binding in self.BINDINGS:
# Skip tuple-style bindings and only process Binding objects
if not hasattr(binding, "show"):
continue
# Only show bindings marked with show=True
if binding.show:
# Get the display key
key_display = getattr(binding, "key_display", None) or binding.key
# Get the description
description = binding.description or binding.action
# Determine positioning from id attribute
binding_id = getattr(binding, "id", None)
if binding_id and binding_id.startswith("left:"):
footer.add_left_item(key_display, description)
elif binding_id and binding_id.startswith("right:"):
footer.add_right_item(key_display, description)
else:
# Default to right if no specific positioning
footer.add_right_item(key_display, description)
# Status section will be updated by update_status
self._update_footer_status()
except Exception:
pass # Footer not ready yet
def _update_footer_status(self) -> None:
"""Update the footer status section."""
try:
footer = self.query_one("#custom-footer", CustomFooter)
mode = "Edit" if self.edit_mode else "Read-only"
entry_count = len(self.hosts_file.entries)
active_count = len(self.hosts_file.get_active_entries())
status = f"{entry_count} entries ({active_count} active) | {mode}"
footer.set_status(status)
except Exception:
pass # Footer not ready yet
def update_status(self, message: str = "") -> None:
"""Update the header subtitle and status bar with status information."""
if message:
# Show temporary message in the status bar
try:
status_bar = self.query_one("#status-bar", Static)
status_bar.update(message)
status_bar.remove_class("hidden")
if message.startswith(""):
# Auto-clear error message after 5 seconds
self.set_timer(5.0, lambda: self._clear_status_message())
else:
# Auto-clear regular message after 3 seconds
self.set_timer(3.0, lambda: self._clear_status_message())
except Exception:
# Fallback if status bar not found (during initialization)
pass
# Always update the header subtitle with current status
# Update the footer status
self._update_footer_status()
def _clear_status_message(self) -> None:
"""Clear the temporary status message."""
try:
status_bar = self.query_one("#status-bar", Static)
status_bar.update("")
status_bar.add_class("hidden")
except Exception:
pass
# Event handlers
def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
"""Handle row highlighting (cursor movement) in the DataTable."""
if event.data_table.id == "entries-table":
# Convert display index to actual index
self.selected_entry_index = (
self.table_handler.display_index_to_actual_index(event.cursor_row)
)
self.details_handler.update_entry_details()
def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
"""Handle row selection in the DataTable."""
if event.data_table.id == "entries-table":
# Convert display index to actual index
self.selected_entry_index = (
self.table_handler.display_index_to_actual_index(event.cursor_row)
)
self.details_handler.update_entry_details()
def on_data_table_header_selected(self, event: DataTable.HeaderSelected) -> None:
"""Handle column header clicks for sorting."""
if event.data_table.id == "entries-table":
# Check if the column key contains "IP Address" (handles sort indicators)
if "IP Address" in str(event.column_key):
self.action_sort_by_ip()
elif "Canonical Hostname" in str(event.column_key):
self.action_sort_by_hostname()
def on_key(self, event) -> None:
"""Handle key events to override default tab behavior in edit mode."""
# Handle tab navigation for search bar and data table
if event.key == "tab" and not self.entry_edit_mode:
search_input = self.query_one("#search-input", Input)
entries_table = self.query_one("#entries-table", DataTable)
# Check which widget currently has focus
if self.focused == search_input:
# Focus on entries table
entries_table.focus()
event.prevent_default()
return
elif self.focused == entries_table:
# Focus on search input
search_input.focus()
event.prevent_default()
return
# Delegate to edit handler for edit mode navigation
if self.edit_handler.handle_entry_edit_key_event(event):
return # Event was handled by edit handler
def on_input_changed(self, event: Input.Changed) -> None:
"""Handle input field changes (no auto-save - changes saved on exit)."""
if event.input.id == "search-input":
# Update search term and filter entries
self.search_term = event.value.strip()
# Also update the current filter options to keep them synchronized
self.current_filter_options.search_term = (
self.search_term if self.search_term else None
)
self.table_handler.populate_entries_table()
self.details_handler.update_entry_details()
else:
# Edit form input changes are tracked but not automatically saved
# Changes will be validated and saved when exiting edit mode
pass
def on_checkbox_changed(self, event: Checkbox.Changed) -> None:
"""Handle checkbox changes (no auto-save - changes saved on exit)."""
# Checkbox changes are tracked but not automatically saved
# Changes will be validated and saved when exiting edit mode
pass
def on_radio_set_changed(self, event) -> None:
"""Handle entry type radio button changes in edit mode."""
if (
hasattr(event, "radio_set")
and event.radio_set.id == "edit-entry-type-radio"
):
pressed_radio = event.pressed
if pressed_radio and pressed_radio.id == "edit-ip-entry-radio":
# Handle switch to IP entry type
self.edit_handler.handle_entry_type_change("ip")
elif pressed_radio and pressed_radio.id == "edit-dns-entry-radio":
# Handle switch to DNS entry type
self.edit_handler.handle_entry_type_change("dns")
# Action handlers
def action_reload(self) -> None:
"""Reload the hosts file."""
# Reset sort state on reload
self.sort_column = ""
self.sort_ascending = True
self.load_hosts_file()
self.update_status("Hosts file reloaded")
def action_help(self) -> None:
"""Show help panel."""
if self.help_visible:
self.action_hide_help_panel()
self.help_visible = False
else:
self.action_show_help_panel()
self.help_visible = True
def action_config(self) -> None:
"""Show configuration modal."""
def handle_config_result(config_changed: bool) -> None:
if config_changed:
# Reload the table to apply new filtering
self.table_handler.populate_entries_table()
self.update_status("Configuration saved")
self.push_screen(ConfigModal(self.config), handle_config_result)
def action_sort_by_ip(self) -> None:
"""Sort entries by IP address, toggle ascending/descending."""
self.table_handler.sort_entries_by_ip()
direction = "ascending" if self.sort_ascending else "descending"
self.update_status(f"Sorted by IP address ({direction})")
def action_sort_by_hostname(self) -> None:
"""Sort entries by canonical hostname, toggle ascending/descending."""
self.table_handler.sort_entries_by_hostname()
direction = "ascending" if self.sort_ascending else "descending"
self.update_status(f"Sorted by hostname ({direction})")
def action_toggle_edit_mode(self) -> None:
"""Toggle between read-only and edit mode."""
if self.edit_mode:
# Exit edit mode
success, message = self.manager.exit_edit_mode()
if success:
self.edit_mode = False
self.update_status(message)
else:
self.update_status(f"Error exiting edit mode: {message}")
else:
# First check whether the current sudo authorization is cached.
success, message = self.manager.enter_edit_mode()
if success:
self.edit_mode = True
self.update_status(message)
elif message == "Interactive authorization required":
self._enter_edit_mode_interactively()
else:
self.update_status(f"Error entering edit mode: {message}")
def _enter_edit_mode_interactively(self) -> None:
"""Run one foreground sudo/PAM conversation outside the TUI."""
interrupted = False
try:
with self.suspend():
try:
success, message = self.manager.authorize_interactively()
except KeyboardInterrupt:
interrupted = True
except SuspendNotSupported:
self.update_status(
"Interactive authorization requires terminal suspension; remaining in Read-only Mode."
)
return
if interrupted:
self.update_status(
"Authorization was not granted; remaining in Read-only Mode."
)
return
if not success:
if message == "Authorization was not granted":
self.update_status(
"Authorization was not granted; remaining in Read-only Mode."
)
else:
self.update_status(f"Error entering edit mode: {message}")
return
success, message = self.manager.finish_entering_edit_mode()
if success:
self.edit_mode = True
self.update_status(message)
elif message == "Authorization was not granted":
self.update_status(
"Authorization was not granted; remaining in Read-only Mode."
)
else:
self.update_status(f"Error entering edit mode: {message}")
def action_edit_entry(self) -> None:
"""Enter edit mode for the selected entry."""
if not self.edit_mode:
self.update_status(
"❌ Cannot edit entry: Application is in read-only mode. Press 'Ctrl+E' to enable edit mode."
)
return
if not self.hosts_file.entries:
self.update_status("No entries to edit")
return
if self.selected_entry_index >= len(self.hosts_file.entries):
self.update_status("Invalid entry selected")
return
entry = self.hosts_file.entries[self.selected_entry_index]
if entry.is_default_entry():
self.update_status("❌ Cannot edit system default entry")
return
# Store original values for change detection
self.original_entry_values = {
"ip_address": entry.ip_address,
"hostnames": entry.hostnames.copy(),
"comment": entry.comment,
"is_active": entry.is_active,
"dns_name": getattr(entry, "dns_name", None),
}
self.entry_edit_mode = True
self.details_handler.update_entry_details()
# Focus on the IP address input field
ip_input = self.query_one("#edit-entry-type-radio", RadioSet)
ip_input.focus()
self.update_status("Editing entry - Use Tab/Shift+Tab to navigate, ESC to exit")
def action_exit_edit_entry(self) -> None:
"""Exit entry edit mode and return focus to the entries table."""
self.edit_handler.exit_edit_entry_with_confirmation()
def action_next_field(self) -> None:
"""Move to the next field in edit mode."""
self.edit_handler.navigate_to_next_field()
def action_prev_field(self) -> None:
"""Move to the previous field in edit mode."""
self.edit_handler.navigate_to_prev_field()
def action_toggle_entry(self) -> None:
"""Toggle the active state of the selected entry."""
self.navigation_handler.toggle_entry()
def action_move_entry_up(self) -> None:
"""Move the selected entry up in the list."""
self.navigation_handler.move_entry_up()
def action_move_entry_down(self) -> None:
"""Move the selected entry down in the list."""
self.navigation_handler.move_entry_down()
def action_save_file(self) -> None:
"""Save the hosts file to disk."""
self.navigation_handler.save_hosts_file()
def action_add_entry(self) -> None:
"""Show the add entry modal."""
if not self.edit_mode:
self.update_status(
"❌ Cannot add entry: Application is in read-only mode. Press 'Ctrl+E' to enable edit mode."
)
return
def handle_add_entry_result(new_entry) -> None:
if new_entry is None:
# User cancelled
self.update_status("Entry creation cancelled")
return
# Add the entry using the command-based manager method
result = self.manager.execute_add_command(self.hosts_file, new_entry)
if result.success:
# Save the changes
save_success, save_message = self.manager.save_hosts_file(
self.hosts_file
)
if save_success:
# Refresh the table
self.table_handler.populate_entries_table()
# Move cursor to the newly added entry (last entry)
self.selected_entry_index = len(self.hosts_file.entries) - 1
self.table_handler.restore_cursor_position(new_entry)
# For DNS entries, trigger resolution and provide feedback
if hasattr(new_entry, "dns_name") and new_entry.dns_name:
self.update_status(
f"{result.message} - Starting DNS resolution for {new_entry.dns_name}"
)
# Trigger DNS resolution in background
self._resolve_new_dns_entry(new_entry)
else:
self.update_status(
f"{result.message} - Changes saved automatically"
)
else:
self.update_status(f"Entry added but save failed: {save_message}")
else:
self.update_status(f"{result.message}")
self.push_screen(AddEntryModal(), handle_add_entry_result)
def action_delete_entry(self) -> None:
"""Show the delete confirmation modal for the selected entry."""
if not self.edit_mode:
self.update_status(
"❌ Cannot delete entry: Application is in read-only mode. Press 'Ctrl+E' to enable edit mode."
)
return
if not self.hosts_file.entries:
self.update_status("No entries to delete")
return
if self.selected_entry_index >= len(self.hosts_file.entries):
self.update_status("Invalid entry selected")
return
entry = self.hosts_file.entries[self.selected_entry_index]
if entry.is_default_entry():
self.update_status("❌ Cannot delete system default entry")
return
def handle_delete_confirmation(confirmed: bool) -> None:
if not confirmed:
self.update_status("Entry deletion cancelled")
return
# Delete the entry using the command-based manager method
result = self.manager.execute_delete_command(
self.hosts_file, self.selected_entry_index
)
if result.success:
# Save the changes
save_success, save_message = self.manager.save_hosts_file(
self.hosts_file
)
if save_success:
# Adjust selected index if needed
if self.selected_entry_index >= len(self.hosts_file.entries):
self.selected_entry_index = max(
0, len(self.hosts_file.entries) - 1
)
# Refresh the table
self.table_handler.populate_entries_table()
self.details_handler.update_entry_details()
self.update_status(
f"{result.message} - Changes saved automatically"
)
else:
self.update_status(f"Entry deleted but save failed: {save_message}")
else:
self.update_status(f"{result.message}")
self.push_screen(DeleteConfirmationModal(entry), handle_delete_confirmation)
def action_quit(self) -> None:
"""Quit the application."""
self.navigation_handler.quit_application()
def action_undo(self) -> None:
"""Undo the last operation."""
if not self.edit_mode:
self.update_status("❌ Cannot undo: Application is in read-only mode")
return
if not self.manager.can_undo():
self.update_status("Nothing to undo")
return
# Get description before undoing
description = self.manager.get_undo_description()
# Perform undo
result = self.manager.undo_last_operation(self.hosts_file)
if result.success:
# Refresh the table and update UI
self.table_handler.populate_entries_table()
self.details_handler.update_entry_details()
self.update_status(f"✅ Undone: {description or 'operation'}")
else:
self.update_status(f"❌ Undo failed: {result.message}")
def action_redo(self) -> None:
"""Redo the last undone operation."""
if not self.edit_mode:
self.update_status("❌ Cannot redo: Application is in read-only mode")
return
if not self.manager.can_redo():
self.update_status("Nothing to redo")
return
# Get description before redoing
description = self.manager.get_redo_description()
# Perform redo
result = self.manager.redo_last_operation(self.hosts_file)
if result.success:
# Refresh the table and update UI
self.table_handler.populate_entries_table()
self.details_handler.update_entry_details()
self.update_status(f"✅ Redone: {description or 'operation'}")
else:
self.update_status(f"❌ Redo failed: {result.message}")
def action_refresh_dns(self) -> None:
"""Manually refresh DNS resolution for all entries."""
if not self.edit_mode:
self.update_status(
"❌ Cannot resolve DNS names: Application is in read-only mode. Press 'Ctrl+E' to enable edit mode."
)
return
if not self.hosts_file.entries:
self.update_status("No entries to resolve")
return
# Get entries that need DNS resolution
dns_entries = self.hosts_file.get_dns_entries()
if not dns_entries:
self.update_status("No entries with DNS names found")
return
# Remember the currently selected entry before DNS update
current_entry = None
if self.hosts_file.entries and self.selected_entry_index < len(
self.hosts_file.entries
):
current_entry = self.hosts_file.entries[self.selected_entry_index]
async def refresh_dns():
try:
# Extract DNS names (not hostnames!) from entries
dns_names = [entry.dns_name for entry in dns_entries if entry.dns_name]
if not dns_names:
self.update_status("No valid DNS names found to resolve")
return
resolved_count = 0
failed_count = 0
# Resolve each DNS name and apply results back to entries
for dns_name in dns_names:
resolution = await self.dns_service.resolve_entry_async(dns_name)
# Find the corresponding entry and update it
for entry in dns_entries:
if entry.dns_name == dns_name:
# Apply resolution results to entry fields
entry.last_resolved = resolution.resolved_at
entry.dns_resolution_status = resolution.status.value
if resolution.is_success():
# Update both resolved_ip and ip_address for the hosts file
entry.ip_address = resolution.resolved_ip
entry.resolved_ip = resolution.resolved_ip
resolved_count += 1
else:
failed_count += 1
break
# Save hosts file with updated DNS information
if resolved_count > 0 or failed_count > 0:
save_success, save_message = self.manager.save_hosts_file(
self.hosts_file
)
if not save_success:
self.update_status(
f"❌ DNS resolution completed but save failed: {save_message}"
)
return
# Update the UI and restore cursor position
self.table_handler.populate_entries_table()
self.table_handler.restore_cursor_position(current_entry)
self.details_handler.update_entry_details()
# Provide detailed status message
if failed_count == 0:
self.update_status(
f"✅ DNS resolution completed for {resolved_count} entries"
)
elif resolved_count == 0:
self.update_status(
f"❌ DNS resolution failed for all {failed_count} entries"
)
else:
self.update_status(
f"⚠️ DNS resolution: {resolved_count} succeeded, {failed_count} failed"
)
except Exception as e:
self.update_status(f"❌ DNS resolution failed: {e}")
# Run DNS resolution in background
self.run_worker(refresh_dns(), exclusive=False)
self.update_status("🔄 Starting DNS resolution...")
def action_update_single_dns(self) -> None:
"""Manually refresh DNS resolution for the currently selected entry."""
if not self.edit_mode:
self.update_status(
"❌ Cannot resolve DNS names: Application is in read-only mode. Press 'Ctrl+E' to enable edit mode."
)
return
if not self.hosts_file.entries:
self.update_status("No entries available")
return
if self.selected_entry_index >= len(self.hosts_file.entries):
self.update_status("Invalid entry selected")
return
entry = self.hosts_file.entries[self.selected_entry_index]
# Check if the entry has a DNS name to resolve
if not hasattr(entry, "dns_name") or not entry.dns_name:
self.update_status("❌ Selected entry has no DNS name to resolve")
return
# Remember the currently selected entry before DNS update
current_entry = entry
async def update_single_dns():
try:
dns_name = entry.dns_name
# Resolve the DNS name
resolution = await self.dns_service.resolve_entry_async(dns_name)
# Apply resolution results to entry fields
entry.last_resolved = resolution.resolved_at
entry.dns_resolution_status = resolution.status.value
if resolution.is_success():
# Update both resolved_ip and ip_address for the hosts file
entry.ip_address = resolution.resolved_ip
entry.resolved_ip = resolution.resolved_ip
# Save hosts file with updated DNS information
save_success, save_message = self.manager.save_hosts_file(
self.hosts_file
)
if not save_success:
self.update_status(
f"❌ DNS resolution completed but save failed: {save_message}"
)
return
# Update the UI and restore cursor position
self.table_handler.populate_entries_table()
self.table_handler.restore_cursor_position(current_entry)
self.details_handler.update_entry_details()
self.update_status(
f"✅ DNS updated: {dns_name}{resolution.resolved_ip}"
)
else:
# Resolution failed, save the status update
save_success, save_message = self.manager.save_hosts_file(
self.hosts_file
)
if save_success:
# Update the UI to show failed status and restore cursor position
self.table_handler.populate_entries_table()
self.table_handler.restore_cursor_position(current_entry)
self.details_handler.update_entry_details()
error_msg = resolution.error_message or "Unknown error"
self.update_status(
f"❌ DNS resolution failed for {dns_name}: {error_msg}"
)
except Exception as e:
self.update_status(f"❌ DNS resolution error: {e}")
# Run DNS resolution in background
self.run_worker(update_single_dns(), exclusive=False)
self.update_status(f"🔄 Resolving DNS for {entry.dns_name}...")
def action_show_filters(self) -> None:
"""Show advanced filtering modal."""
def handle_filter_result(filter_options: FilterOptions) -> None:
if filter_options is None:
# User cancelled
self.update_status("Filtering cancelled")
return
# Apply the new filter options
self.current_filter_options = filter_options
# Update the search term from filter if it has one
if filter_options.search_term:
self.search_term = filter_options.search_term
# Update the search input to reflect the filter search term
try:
search_input = self.query_one("#search-input", Input)
search_input.value = filter_options.search_term
except Exception:
pass # Search input not ready
else:
# Clear search term if no search in filter
self.search_term = ""
try:
search_input = self.query_one("#search-input", Input)
search_input.value = ""
except Exception:
pass
# Refresh the table with new filtering
self.table_handler.populate_entries_table()
self.details_handler.update_entry_details()
# Get filter statistics for status message
counts = self.entry_filter.count_filtered_entries(
self.hosts_file.entries, filter_options
)
preset_info = (
f" (preset: {filter_options.preset_name})"
if filter_options.preset_name
else ""
)
self.update_status(
f"✅ Filter applied: showing {counts['filtered']} of {counts['total']} entries{preset_info}"
)
# Show the filter modal with current options and entries for preview
self.push_screen(
FilterModal(
initial_options=self.current_filter_options,
entries=self.hosts_file.entries,
entry_filter=self.entry_filter,
),
handle_filter_result,
)
def _resolve_new_dns_entry(self, entry) -> None:
"""Trigger DNS resolution for a newly added DNS entry."""
if not hasattr(entry, "dns_name") or not entry.dns_name:
return
async def resolve_and_activate():
try:
# Resolve the DNS name
resolution = await self.dns_service.resolve_entry_async(entry.dns_name)
if resolution.is_success():
# Find the entry in the hosts file and update it
for hosts_entry in self.hosts_file.entries:
if (
hasattr(hosts_entry, "dns_name")
and hosts_entry.dns_name == entry.dns_name
and hosts_entry.hostnames == entry.hostnames
):
# Update the entry with resolved IP
hosts_entry.ip_address = resolution.resolved_ip
hosts_entry.resolved_ip = resolution.resolved_ip
hosts_entry.last_resolved = resolution.resolved_at
hosts_entry.dns_resolution_status = resolution.status.value
hosts_entry.is_active = True # Activate the entry
# Save the updated hosts file
save_success, save_message = self.manager.save_hosts_file(
self.hosts_file
)
if save_success:
# Update UI - use direct calls since we're in the same async context
self.table_handler.populate_entries_table()
self.details_handler.update_entry_details()
self.update_status(
f"✅ DNS resolved: {entry.dns_name}{resolution.resolved_ip} (entry activated)"
)
else:
self.update_status(
f"❌ DNS resolved but save failed: {save_message}"
)
break
else:
# Resolution failed, update status but keep entry inactive
for hosts_entry in self.hosts_file.entries:
if (
hasattr(hosts_entry, "dns_name")
and hosts_entry.dns_name == entry.dns_name
and hosts_entry.hostnames == entry.hostnames
):
hosts_entry.dns_resolution_status = resolution.status.value
hosts_entry.last_resolved = resolution.resolved_at
break
self.update_status(
f"❌ DNS resolution failed for {entry.dns_name}: {resolution.error_message or 'Unknown error'}"
)
except Exception as e:
self.update_status(
f"❌ DNS resolution error for {entry.dns_name}: {str(e)}"
)
# Start the resolution in background
self.run_worker(resolve_and_activate(), exclusive=False)
async def on_shutdown(self) -> None:
"""Clean up resources when the app is shutting down."""
# No DNS service cleanup needed for manual-only resolution
pass
# Delegated methods for backward compatibility with tests
def has_entry_changes(self) -> bool:
"""Check if the current entry has been modified from its original values."""
return self.edit_handler.has_entry_changes()
def exit_edit_entry_mode(self) -> None:
"""Helper method to exit entry edit mode and clean up."""
self.edit_handler.exit_edit_entry_mode()
def populate_entries_table(self) -> None:
"""Populate the left pane with hosts entries using DataTable."""
self.table_handler.populate_entries_table()
def restore_cursor_position(self, previous_entry) -> None:
"""Restore cursor position after reload, maintaining selection if possible."""
self.table_handler.restore_cursor_position(previous_entry)
def get_visible_entries(self) -> list:
"""Get the list of entries that are visible in the table (after filtering)."""
return self.table_handler.get_visible_entries()
def display_index_to_actual_index(self, display_index: int) -> int:
"""Convert a display table index to the actual hosts file entry index."""
return self.table_handler.display_index_to_actual_index(display_index)
def actual_index_to_display_index(self, actual_index: int) -> int:
"""Convert an actual hosts file entry index to a display table index."""
return self.table_handler.actual_index_to_display_index(actual_index)
def update_entry_details(self) -> None:
"""Update the right pane with selected entry details."""
self.details_handler.update_entry_details()
def update_details_display(self) -> None:
"""Update the static details display."""
self.details_handler.update_details_display()
def update_edit_form(self) -> None:
"""Update the edit form with current entry values."""
self.details_handler.update_edit_form()
def watch_entry_edit_mode(self, entry_edit_mode: bool) -> None:
"""Update the right pane border title when entry edit mode changes."""
try:
right_pane = self.query_one(".right-pane")
if entry_edit_mode:
right_pane.border_title = "Edit Entry"
else:
right_pane.border_title = "Entry Details"
except Exception:
# App not fully initialized yet, ignore
pass