diff --git a/README.md b/README.md index 17c8cdc..a40dcec 100644 --- a/README.md +++ b/README.md @@ -75,8 +75,6 @@ sorting, DNS refresh, movement, undo, and redo. ## Safety and known limitations - Reloading discards unsaved in-memory state. -- A failed save can leave the interface changed while `/etc/hosts` remains - unchanged. - Pre-edit Backups are not listed or restored by the TUI and have no retention management. Manual recovery is documented in the user guide. - Leaving Privileged Mode clears the application's session state but does not diff --git a/docs/user-guide.md b/docs/user-guide.md index e98a92b..da6bc7c 100644 --- a/docs/user-guide.md +++ b/docs/user-guide.md @@ -75,8 +75,9 @@ Default Entries represent protected baseline localhost or broadcasthost mappings. They cannot be changed, deleted, activated, deactivated, or moved. Successful mutations are saved to `/etc/hosts` immediately. Read the -status message after every action. A message that reports a save failure means -the interface may have changed while the system file did not. +status message after every action. If saving fails, the application restores +the Host Entries, visible selection, and undo/redo history from before the +action. ## Use a DNS Entry @@ -120,13 +121,9 @@ Adding, editing, deleting, moving, activating, deactivating, undoing, redoing, and successfully refreshing DNS Entries save immediately. `Ctrl+S` explicitly saves the entire current in-memory Hosts File. -If saving an undo or redo result fails, the application restores the prior -in-memory state and undo/redo history. Reload with `Ctrl+R` to verify the -on-disk state before attempting another change. - -Other save failures can leave the interface changed while the file on disk is -unchanged. Reload with `Ctrl+R` to return the interface to the on-disk state -before attempting another change. +If an immediate save fails, the application restores the complete pre-action +in-memory state, visible selection, and undo/redo history. A later mutation +therefore cannot accidentally persist the failed change. ### Locate a Pre-edit Backup diff --git a/src/hosts/core/manager.py b/src/hosts/core/manager.py index 1057604..f244599 100644 --- a/src/hosts/core/manager.py +++ b/src/hosts/core/manager.py @@ -8,6 +8,8 @@ and safe file modifications with backup and validation. import os import subprocess import tempfile +from copy import deepcopy +from dataclasses import dataclass from pathlib import Path from typing import Optional, Tuple from .models import HostEntry, HostsFile @@ -23,6 +25,14 @@ from .commands import ( ) +@dataclass +class MutationState: + """Hosts File and history state from before a persisted mutation.""" + + hosts_file: HostsFile + undo_redo_history: UndoRedoHistory + + class PermissionManager: """ Manages sudo permissions for hosts file editing. @@ -591,6 +601,22 @@ class HostsManager: """Get description of the operation that would be redone.""" return self.undo_redo_history.get_redo_description() + def capture_mutation_state(self, hosts_file: HostsFile) -> MutationState: + """Capture the Hosts File and history before a persisted mutation.""" + saved_hosts_file, saved_history = deepcopy((hosts_file, self.undo_redo_history)) + return MutationState(saved_hosts_file, saved_history) + + def save_mutation( + self, hosts_file: HostsFile, state: MutationState + ) -> Tuple[bool, str, HostsFile]: + """Save a mutation or restore its pre-action model and history.""" + success, message = self.save_hosts_file(hosts_file) + if success: + return True, message, hosts_file + + self.undo_redo_history = state.undo_redo_history + return False, message, state.hosts_file + def save_hosts_file(self, hosts_file: HostsFile) -> Tuple[bool, str]: """ Save the hosts file to disk with sudo permissions. diff --git a/src/hosts/tui/app.py b/src/hosts/tui/app.py index 599c6e5..ed47f84 100644 --- a/src/hosts/tui/app.py +++ b/src/hosts/tui/app.py @@ -5,7 +5,7 @@ This module contains the main application class that orchestrates all the handlers and provides the primary user interface. """ -from collections.abc import Callable +from dataclasses import dataclass from textual.app import App, ComposeResult, SuspendNotSupported from textual.containers import Horizontal, Vertical @@ -21,10 +21,9 @@ from textual.widgets import ( from textual.reactive import reactive from ..core.parser import HostsParser -from ..core.models import HostsFile -from ..core.commands import OperationResult +from ..core.models import HostEntry, HostsFile from ..core.config import Config -from ..core.manager import HostsManager +from ..core.manager import HostsManager, MutationState from ..core.dns import DNSService from ..core.filters import EntryFilter, FilterOptions from .config_modal import ConfigModal @@ -40,6 +39,14 @@ from .edit_handler import EditHandler from .navigation_handler import NavigationHandler +@dataclass +class MutationSnapshot: + """TUI and core state that must survive a failed persistence attempt.""" + + manager_state: MutationState + selected_entry_index: int + + class HostsManagerApp(App): """ Main application class for the hosts TUI manager. @@ -353,6 +360,44 @@ class HostsManagerApp(App): except Exception: pass + def capture_mutation_state(self) -> MutationSnapshot: + """Capture the model, selection, and history before a mutation.""" + return MutationSnapshot( + manager_state=self.manager.capture_mutation_state(self.hosts_file), + selected_entry_index=self.selected_entry_index, + ) + + def save_mutation(self, snapshot: MutationSnapshot, action: str) -> bool: + """Persist a mutation, restoring its complete pre-action state on failure.""" + save_success, save_message, hosts_file = self.manager.save_mutation( + self.hosts_file, snapshot.manager_state + ) + if save_success: + return True + + self.hosts_file = hosts_file + self.selected_entry_index = snapshot.selected_entry_index + + self.table_handler.populate_entries_table() + self.table_handler.move_cursor_to_entry_index(self.selected_entry_index) + self.details_handler.update_entry_details() + self.update_status( + f"❌ {action} save failed; previous state restored: {save_message}" + ) + return False + + def get_pending_dns_entry( + self, entry_index: int, dns_name: str, hostnames: list[str] + ) -> HostEntry | None: + """Return the original DNS Entry if it is unchanged after an await.""" + if not 0 <= entry_index < len(self.hosts_file.entries): + return None + + entry = self.hosts_file.entries[entry_index] + if entry.dns_name != dns_name or entry.hostnames != hostnames: + return None + return entry + # Event handlers def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None: """Handle row highlighting (cursor movement) in the DataTable.""" @@ -624,14 +669,13 @@ class HostsManagerApp(App): self.update_status("Entry creation cancelled") return + snapshot = self.capture_mutation_state() + # 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: + if self.save_mutation(snapshot, "Add"): # Refresh the table self.table_handler.populate_entries_table() # Move cursor to the newly added entry (last entry) @@ -649,8 +693,6 @@ class HostsManagerApp(App): 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}") @@ -682,16 +724,15 @@ class HostsManagerApp(App): self.update_status("Entry deletion cancelled") return + snapshot = self.capture_mutation_state() + # 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: + if self.save_mutation(snapshot, "Delete"): # Adjust selected index if needed if self.selected_entry_index >= len(self.hosts_file.entries): self.selected_entry_index = max( @@ -704,8 +745,6 @@ class HostsManagerApp(App): 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}") @@ -715,25 +754,6 @@ class HostsManagerApp(App): """Quit the application.""" self.navigation_handler.quit_application() - def _save_history_operation( - self, action: str, rollback: Callable[[HostsFile], OperationResult] - ) -> bool: - """Save a history change or restore the previous state after a failure.""" - save_success, save_message = self.manager.save_hosts_file(self.hosts_file) - if save_success: - return True - - rollback_result = rollback(self.hosts_file) - if rollback_result.success: - self.update_status( - f"❌ {action} save failed; previous state restored: {save_message}" - ) - else: - self.update_status( - f"❌ {action} save failed and could not restore the previous state: {save_message}; {rollback_result.message}. Reload the Hosts File." - ) - return False - def action_undo(self) -> None: """Undo the last operation.""" if not self.edit_mode: @@ -746,13 +766,12 @@ class HostsManagerApp(App): # Get description before undoing description = self.manager.get_undo_description() + snapshot = self.capture_mutation_state() # Perform undo result = self.manager.undo_last_operation(self.hosts_file) if result.success: - if not self._save_history_operation( - "Undo", self.manager.redo_last_operation - ): + if not self.save_mutation(snapshot, "Undo"): return # Refresh the table and update UI @@ -774,13 +793,12 @@ class HostsManagerApp(App): # Get description before redoing description = self.manager.get_redo_description() + snapshot = self.capture_mutation_state() # Perform redo result = self.manager.redo_last_operation(self.hosts_file) if result.success: - if not self._save_history_operation( - "Redo", self.manager.undo_last_operation - ): + if not self.save_mutation(snapshot, "Redo"): return # Refresh the table and update UI @@ -808,13 +826,6 @@ class HostsManagerApp(App): 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 @@ -827,18 +838,31 @@ class HostsManagerApp(App): 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) + resolutions = [ + (dns_name, await self.dns_service.resolve_entry_async(dns_name)) + for dns_name in dns_names + ] + snapshot = self.capture_mutation_state() + current_entry = ( + self.hosts_file.entries[self.selected_entry_index] + if 0 <= self.selected_entry_index < len(self.hosts_file.entries) + else None + ) + current_dns_entries = self.hosts_file.get_dns_entries() + # Apply each DNS result back to the current entries. + for dns_name, resolution in resolutions: # Find the corresponding entry and update it - for entry in dns_entries: + for entry in current_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(): + if ( + resolution.is_success() + and resolution.resolved_ip is not None + ): # Update both resolved_ip and ip_address for the hosts file entry.ip_address = resolution.resolved_ip entry.resolved_ip = resolution.resolved_ip @@ -849,13 +873,7 @@ class HostsManagerApp(App): # 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}" - ) + if not self.save_mutation(snapshot, "DNS refresh"): return # Update the UI and restore cursor position @@ -907,33 +925,36 @@ class HostsManagerApp(App): self.update_status("❌ Selected entry has no DNS name to resolve") return - # Remember the currently selected entry before DNS update - current_entry = entry + dns_name = entry.dns_name + entry_hostnames = entry.hostnames.copy() + entry_index = self.selected_entry_index 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) + current_entry = self.get_pending_dns_entry( + entry_index, dns_name, entry_hostnames + ) + if current_entry is None: + self.update_status( + f"❌ DNS entry changed before resolution completed: {dns_name}" + ) + return + + snapshot = self.capture_mutation_state() # Apply resolution results to entry fields - entry.last_resolved = resolution.resolved_at - entry.dns_resolution_status = resolution.status.value + current_entry.last_resolved = resolution.resolved_at + current_entry.dns_resolution_status = resolution.status.value - if resolution.is_success(): + if resolution.is_success() and resolution.resolved_ip is not None: # Update both resolved_ip and ip_address for the hosts file - entry.ip_address = resolution.resolved_ip - entry.resolved_ip = resolution.resolved_ip + current_entry.ip_address = resolution.resolved_ip + current_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}" - ) + if not self.save_mutation(snapshot, "DNS refresh"): return # Update the UI and restore cursor position @@ -946,14 +967,13 @@ class HostsManagerApp(App): ) 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() + if not self.save_mutation(snapshot, "DNS refresh"): + return + + # 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( @@ -1029,62 +1049,62 @@ class HostsManagerApp(App): if not hasattr(entry, "dns_name") or not entry.dns_name: return + dns_name = entry.dns_name + entry_hostnames = entry.hostnames.copy() + entry_index = next( + ( + index + for index, candidate in enumerate(self.hosts_file.entries) + if candidate is entry + ), + -1, + ) + async def resolve_and_activate(): try: # Resolve the DNS name - resolution = await self.dns_service.resolve_entry_async(entry.dns_name) + resolution = await self.dns_service.resolve_entry_async(dns_name) + current_entry = self.get_pending_dns_entry( + entry_index, dns_name, entry_hostnames + ) + if current_entry is None: + self.update_status( + f"❌ DNS entry changed before resolution completed: {dns_name}" + ) + return - 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 + snapshot = self.capture_mutation_state() - # 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 + if resolution.is_success() and resolution.resolved_ip is not None: + # Update the entry with resolved IP + current_entry.ip_address = resolution.resolved_ip + current_entry.resolved_ip = resolution.resolved_ip + current_entry.last_resolved = resolution.resolved_at + current_entry.dns_resolution_status = resolution.status.value + current_entry.is_active = True # Activate the entry + + # Save the updated hosts file + if self.save_mutation(snapshot, "DNS activation"): + # 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: {dns_name} → {resolution.resolved_ip} (entry activated)" + ) 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 + current_entry.dns_resolution_status = resolution.status.value + current_entry.last_resolved = resolution.resolved_at + + if not self.save_mutation(snapshot, "DNS refresh"): + return self.update_status( - f"❌ DNS resolution failed for {entry.dns_name}: {resolution.error_message or 'Unknown error'}" + f"❌ DNS resolution failed for {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)}" - ) + self.update_status(f"❌ DNS resolution error for {dns_name}: {str(e)}") # Start the resolution in background self.run_worker(resolve_and_activate(), exclusive=False) diff --git a/src/hosts/tui/edit_handler.py b/src/hosts/tui/edit_handler.py index 3723059..ff8b505 100644 --- a/src/hosts/tui/edit_handler.py +++ b/src/hosts/tui/edit_handler.py @@ -355,6 +355,7 @@ class EditHandler: hostnames = [h.strip() for h in hostname_input.value.split(",") if h.strip()] comment = comment_input.value.strip() or None is_active = active_checkbox.value + snapshot = self.app.capture_mutation_state() # Update entry based on type if entry_type == "ip": @@ -390,8 +391,7 @@ class EditHandler: entry.is_active = is_active # Save to file - success, message = self.app.manager.save_hosts_file(self.app.hosts_file) - if success: + if self.app.save_mutation(snapshot, "Edit"): # Update the table display self.app.table_handler.populate_entries_table() # Restore cursor position @@ -410,9 +410,7 @@ class EditHandler: else: self.app.update_status("Entry saved successfully") return True - else: - self.app.update_status(f"❌ Error saving entry: {message}") - return False + return False def navigate_to_next_field(self) -> None: """Move to the next field in edit mode.""" diff --git a/src/hosts/tui/navigation_handler.py b/src/hosts/tui/navigation_handler.py index 4322dbc..8a5d3ba 100644 --- a/src/hosts/tui/navigation_handler.py +++ b/src/hosts/tui/navigation_handler.py @@ -29,6 +29,7 @@ class NavigationHandler: # Remember current entry for cursor position restoration current_entry = self.app.hosts_file.entries[self.app.selected_entry_index] + snapshot = self.app.capture_mutation_state() # Use command-based method for undo/redo support result = self.app.manager.execute_toggle_command( @@ -36,10 +37,7 @@ class NavigationHandler: ) if result.success: # Auto-save the changes immediately - save_success, save_message = self.app.manager.save_hosts_file( - self.app.hosts_file - ) - if save_success: + if self.app.save_mutation(snapshot, "Toggle"): self.app.table_handler.populate_entries_table() # Restore cursor position to the same entry self.app.set_timer( @@ -52,8 +50,6 @@ class NavigationHandler: self.app.update_status( f"{result.message} - Changes saved automatically" ) - else: - self.app.update_status(f"Entry toggled but save failed: {save_message}") else: self.app.update_status(f"Error toggling entry: {result.message}") @@ -69,16 +65,15 @@ class NavigationHandler: self.app.update_status("No entries to move") return + snapshot = self.app.capture_mutation_state() + # Use command-based method for undo/redo support result = self.app.manager.execute_move_command( self.app.hosts_file, self.app.selected_entry_index, "up" ) if result.success: # Auto-save the changes immediately - save_success, save_message = self.app.manager.save_hosts_file( - self.app.hosts_file - ) - if save_success: + if self.app.save_mutation(snapshot, "Move"): # Update the selection index to follow the moved entry if self.app.selected_entry_index > 0: self.app.selected_entry_index -= 1 @@ -94,8 +89,6 @@ class NavigationHandler: self.app.update_status( f"{result.message} - Changes saved automatically" ) - else: - self.app.update_status(f"Entry moved but save failed: {save_message}") else: self.app.update_status(f"Error moving entry: {result.message}") @@ -111,16 +104,15 @@ class NavigationHandler: self.app.update_status("No entries to move") return + snapshot = self.app.capture_mutation_state() + # Use command-based method for undo/redo support result = self.app.manager.execute_move_command( self.app.hosts_file, self.app.selected_entry_index, "down" ) if result.success: # Auto-save the changes immediately - save_success, save_message = self.app.manager.save_hosts_file( - self.app.hosts_file - ) - if save_success: + if self.app.save_mutation(snapshot, "Move"): # Update the selection index to follow the moved entry if self.app.selected_entry_index < len(self.app.hosts_file.entries) - 1: self.app.selected_entry_index += 1 @@ -136,8 +128,6 @@ class NavigationHandler: self.app.update_status( f"{result.message} - Changes saved automatically" ) - else: - self.app.update_status(f"Entry moved but save failed: {save_message}") else: self.app.update_status(f"Error moving entry: {result.message}") diff --git a/src/hosts/tui/table_handler.py b/src/hosts/tui/table_handler.py index 0384257..769fe79 100644 --- a/src/hosts/tui/table_handler.py +++ b/src/hosts/tui/table_handler.py @@ -123,7 +123,7 @@ class TableHandler: def actual_index_to_display_index(self, actual_index: int) -> int: """Convert an actual hosts file entry index to a display table index.""" - if actual_index >= len(self.app.hosts_file.entries): + if actual_index < 0 or actual_index >= len(self.app.hosts_file.entries): return 0 target_entry = self.app.hosts_file.entries[actual_index] @@ -136,6 +136,14 @@ class TableHandler: return 0 + def move_cursor_to_entry_index(self, actual_index: int) -> None: + """Move the table cursor to an exact Hosts File entry index.""" + table = self.app.query_one("#entries-table", DataTable) + display_index = self.actual_index_to_display_index(actual_index) + if table.row_count > 0 and display_index < table.row_count: + table.move_cursor(row=display_index) + table.focus() + def populate_entries_table(self) -> None: """Populate the left pane with hosts entries using DataTable.""" table = self.app.query_one("#entries-table", DataTable) diff --git a/tests/test_main.py b/tests/test_main.py index 520fa02..bda5b49 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -5,19 +5,31 @@ This module contains unit tests for the HostsManagerApp class, validating application behavior, navigation, and user interactions. """ -from unittest.mock import Mock, patch +from datetime import datetime +from unittest.mock import AsyncMock, Mock, patch + +import pytest from hosts.tui.app import HostsManagerApp from hosts.core.models import HostEntry, HostsFile from hosts.core.parser import HostsParser from hosts.core.config import Config -from hosts.core.manager import HostsManager +from hosts.core.dns import DNSResolution, DNSResolutionStatus class TestHostsManagerApp: """Test cases for the HostsManagerApp class.""" + @staticmethod + def fail_saves(app: HostsManagerApp) -> None: + """Configure a TUI app with an isolated failed-persistence boundary.""" + app.manager.save_hosts_file = Mock(return_value=(False, "Permission denied")) + app.table_handler.populate_entries_table = Mock() + app.table_handler.move_cursor_to_entry_index = Mock() + app.details_handler.update_entry_details = Mock() + app.update_status = Mock() + def test_app_initialization(self): """Test application initialization.""" with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): @@ -537,8 +549,7 @@ class TestHostsManagerApp: app.update_status = Mock() serialized_files = [] - app.manager = Mock(spec=HostsManager) - app.manager.enter_edit_mode.return_value = (True, "Edit mode enabled") + app.manager.enter_edit_mode = Mock(return_value=(True, "Edit mode enabled")) def toggle_entry(hosts_file, index): hosts_file.toggle_entry(index) @@ -548,8 +559,8 @@ class TestHostsManagerApp: serialized_files.append(HostsParser().serialize(hosts_file)) return True, "Hosts file saved successfully" - app.manager.execute_toggle_command.side_effect = toggle_entry - app.manager.save_hosts_file.side_effect = save_hosts_file + app.manager.execute_toggle_command = Mock(side_effect=toggle_entry) + app.manager.save_hosts_file = Mock(side_effect=save_hosts_file) app.action_sort_by_ip() assert app.edit_mode is False @@ -1138,20 +1149,16 @@ class TestHostsManagerApp: entry = HostEntry(ip_address="192.168.1.1", hostnames=["test.local"]) app.hosts_file.add_entry(entry) app.manager.execute_toggle_command(app.hosts_file, 0) - app.manager.save_hosts_file = Mock( - return_value=(False, "Permission denied") - ) - app.table_handler.populate_entries_table = Mock() - app.details_handler.update_entry_details = Mock() - app.update_status = Mock() + self.fail_saves(app) app.action_undo() - assert entry.is_active is False + assert app.hosts_file.entries[0].is_active is False assert app.manager.can_undo() assert not app.manager.can_redo() - app.table_handler.populate_entries_table.assert_not_called() - app.details_handler.update_entry_details.assert_not_called() + app.table_handler.populate_entries_table.assert_called_once() + app.table_handler.move_cursor_to_entry_index.assert_called_once_with(0) + app.details_handler.update_entry_details.assert_called_once() app.update_status.assert_called_once_with( "❌ Undo save failed; previous state restored: Permission denied" ) @@ -1190,22 +1197,449 @@ class TestHostsManagerApp: app.hosts_file.add_entry(entry) app.manager.execute_toggle_command(app.hosts_file, 0) app.manager.undo_last_operation(app.hosts_file) + self.fail_saves(app) + + app.action_redo() + + assert app.hosts_file.entries[0].is_active is True + assert not app.manager.can_undo() + assert app.manager.can_redo() + app.table_handler.populate_entries_table.assert_called_once() + app.table_handler.move_cursor_to_entry_index.assert_called_once_with(0) + app.details_handler.update_entry_details.assert_called_once() + app.update_status.assert_called_once_with( + "❌ Redo save failed; previous state restored: Permission denied" + ) + + def test_toggle_save_failure_restores_model_selection_and_history(self): + """A failed toggle restores all user-visible and undoable state.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + app.hosts_file = HostsFile( + entries=[ + HostEntry(ip_address="192.0.2.1", hostnames=["one.test"]), + HostEntry(ip_address="192.0.2.2", hostnames=["two.test"]), + ] + ) + app.selected_entry_index = 0 + app.manager.execute_toggle_command(app.hosts_file, 1) + app.manager.undo_last_operation(app.hosts_file) + self.fail_saves(app) + + app.action_toggle_entry() + + assert app.hosts_file.entries[0].is_active is True + assert app.selected_entry_index == 0 + assert not app.manager.can_undo() + assert app.manager.can_redo() + app.update_status.assert_called_once_with( + "❌ Toggle save failed; previous state restored: Permission denied" + ) + + def test_move_save_failure_restores_entry_order_and_selection(self): + """A failed move leaves the selected Host Entry in its original position.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + app.hosts_file = HostsFile( + entries=[ + HostEntry(ip_address="192.0.2.1", hostnames=["one.test"]), + HostEntry(ip_address="192.0.2.2", hostnames=["two.test"]), + ] + ) + app.selected_entry_index = 1 + self.fail_saves(app) + + app.action_move_entry_up() + + assert [entry.hostnames[0] for entry in app.hosts_file.entries] == [ + "one.test", + "two.test", + ] + assert app.selected_entry_index == 1 + assert not app.manager.can_undo() + app.update_status.assert_called_once_with( + "❌ Move save failed; previous state restored: Permission denied" + ) + + def test_add_save_failure_removes_entry_and_restores_history(self): + """A failed add restores the prior entries and history.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + existing = HostEntry(ip_address="192.0.2.1", hostnames=["one.test"]) + added = HostEntry(ip_address="192.0.2.2", hostnames=["two.test"]) + app.hosts_file = HostsFile(entries=[existing]) + app.selected_entry_index = 0 + self.fail_saves(app) + app.push_screen = Mock( + side_effect=lambda _screen, callback: callback(added) + ) + + app.action_add_entry() + + assert [entry.hostnames[0] for entry in app.hosts_file.entries] == [ + "one.test" + ] + assert app.selected_entry_index == 0 + assert not app.manager.can_undo() + app.update_status.assert_called_once_with( + "❌ Add save failed; previous state restored: Permission denied" + ) + + def test_later_mutation_does_not_persist_failed_add(self): + """A later successful save excludes a previously failed addition.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + existing = HostEntry(ip_address="192.0.2.1", hostnames=["one.test"]) + added = HostEntry(ip_address="192.0.2.2", hostnames=["two.test"]) + app.hosts_file = HostsFile(entries=[existing]) app.manager.save_hosts_file = Mock( return_value=(False, "Permission denied") ) app.table_handler.populate_entries_table = Mock() + app.table_handler.move_cursor_to_entry_index = Mock() + app.table_handler.restore_cursor_position = Mock() + app.details_handler.update_entry_details = Mock() + app.update_status = Mock() + app.set_timer = Mock() + app.push_screen = Mock( + side_effect=lambda _screen, callback: callback(added) + ) + app.action_add_entry() + + persisted = [] + + def save_hosts_file(hosts_file): + persisted.append(HostsParser().serialize(hosts_file)) + return True, "Hosts file saved" + + app.manager.save_hosts_file = Mock(side_effect=save_hosts_file) + app.action_toggle_entry() + + assert len(persisted) == 1 + assert "one.test" in persisted[0] + assert "two.test" not in persisted[0] + + def test_save_failure_restores_exact_duplicate_selection(self): + """Rollback keeps the second of two identical Host Entries selected.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + app.hosts_file = HostsFile( + entries=[ + HostEntry(ip_address="192.0.2.1", hostnames=["same.test"]), + HostEntry(ip_address="192.0.2.1", hostnames=["same.test"]), + ] + ) + app.selected_entry_index = 1 + app.manager.save_hosts_file = Mock( + return_value=(False, "Permission denied") + ) + table = Mock(row_count=2) + app.query_one = Mock(return_value=table) + app.table_handler.populate_entries_table = Mock() app.details_handler.update_entry_details = Mock() app.update_status = Mock() - app.action_redo() + app.action_toggle_entry() - assert entry.is_active is True + assert app.selected_entry_index == 1 + table.move_cursor.assert_called_once_with(row=1) + + def test_delete_save_failure_restores_entry_and_selection(self): + """A failed delete restores the removed Host Entry and selection.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + app.hosts_file = HostsFile( + entries=[ + HostEntry(ip_address="192.0.2.1", hostnames=["one.test"]), + HostEntry(ip_address="192.0.2.2", hostnames=["two.test"]), + ] + ) + app.selected_entry_index = 1 + self.fail_saves(app) + app.push_screen = Mock(side_effect=lambda _screen, callback: callback(True)) + + app.action_delete_entry() + + assert [entry.hostnames[0] for entry in app.hosts_file.entries] == [ + "one.test", + "two.test", + ] + assert app.selected_entry_index == 1 assert not app.manager.can_undo() - assert app.manager.can_redo() - app.table_handler.populate_entries_table.assert_not_called() - app.details_handler.update_entry_details.assert_not_called() app.update_status.assert_called_once_with( - "❌ Redo save failed; previous state restored: Permission denied" + "❌ Delete save failed; previous state restored: Permission denied" + ) + + def test_entry_editor_save_failure_restores_all_entry_fields(self): + """A failed editor save restores ordinary fields and DNS metadata.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + resolved_at = datetime(2026, 9, 4, 10, 30) + app.hosts_file = HostsFile( + entries=[ + HostEntry( + ip_address="192.0.2.1", + hostnames=["one.test", "alias.test"], + comment="original", + is_active=False, + dns_name="source.test", + resolved_ip="192.0.2.1", + last_resolved=resolved_at, + dns_resolution_status="resolved", + ) + ] + ) + app.selected_entry_index = 0 + app.entry_edit_mode = True + app.manager.save_hosts_file = Mock( + return_value=(False, "Permission denied") + ) + app.table_handler.populate_entries_table = Mock() + app.table_handler.move_cursor_to_entry_index = Mock() + app.update_status = Mock() + + ip_input = Mock(value="198.51.100.8") + hostname_input = Mock(value="changed.test") + comment_input = Mock(value="changed") + active_checkbox = Mock(value=True) + dns_input = Mock(value="") + ip_radio = Mock(id="edit-ip-entry-radio", value=True) + dns_radio = Mock(id="edit-dns-entry-radio", value=False) + radio_set = Mock(pressed_button=ip_radio) + widgets = { + "#entry-details-display": Mock(), + "#entry-edit-form": Mock(), + "#ip-input": ip_input, + "#hostname-input": hostname_input, + "#comment-input": comment_input, + "#active-checkbox": active_checkbox, + "#dns-name-input": dns_input, + "#edit-entry-type-radio": radio_set, + "#edit-ip-entry-radio": ip_radio, + "#edit-dns-entry-radio": dns_radio, + "#edit-ip-section": Mock(), + "#edit-dns-section": Mock(), + } + app.query_one = Mock(side_effect=lambda selector, *_args: widgets[selector]) + app.set_timer = Mock(side_effect=lambda _delay, callback: callback()) + + saved = app.edit_handler.validate_and_save_entry_changes() + + entry = app.hosts_file.entries[0] + assert saved is False + assert entry.ip_address == "192.0.2.1" + assert entry.hostnames == ["one.test", "alias.test"] + assert entry.comment == "original" + assert entry.is_active is False + assert entry.dns_name == "source.test" + assert entry.resolved_ip == "192.0.2.1" + assert entry.last_resolved == resolved_at + assert entry.dns_resolution_status == "resolved" + assert ip_input.value == "192.0.2.1" + assert hostname_input.value == "one.test, alias.test" + assert comment_input.value == "original" + assert active_checkbox.value is False + assert dns_input.value == "source.test" + app.update_status.assert_called_once_with( + "❌ Edit save failed; previous state restored: Permission denied" + ) + + @pytest.mark.asyncio + async def test_batch_dns_save_failure_restores_mapping_and_metadata(self): + """A failed batch DNS refresh restores every changed DNS field.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + original_time = datetime(2026, 9, 3, 8, 0) + resolved_time = datetime(2026, 9, 4, 11, 0) + app.hosts_file = HostsFile( + entries=[ + HostEntry( + ip_address="192.0.2.1", + hostnames=["one.test"], + dns_name="source.test", + resolved_ip="192.0.2.1", + last_resolved=original_time, + dns_resolution_status="match", + ) + ] + ) + app.selected_entry_index = 0 + app.dns_service.resolve_entry_async = AsyncMock( + return_value=DNSResolution( + hostname="source.test", + resolved_ip="198.51.100.9", + status=DNSResolutionStatus.RESOLVED, + resolved_at=resolved_time, + ) + ) + self.fail_saves(app) + workers = [] + app.run_worker = Mock( + side_effect=lambda worker, **_kwargs: workers.append(worker) + ) + + app.action_refresh_dns() + await workers[0] + + entry = app.hosts_file.entries[0] + assert entry.ip_address == "192.0.2.1" + assert entry.resolved_ip == "192.0.2.1" + assert entry.last_resolved == original_time + assert entry.dns_resolution_status == "match" + assert app.selected_entry_index == 0 + app.update_status.assert_any_call( + "❌ DNS refresh save failed; previous state restored: Permission denied" + ) + + @pytest.mark.asyncio + async def test_single_dns_save_failure_restores_mapping_and_metadata(self): + """A failed selected-entry DNS refresh restores every changed field.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + original_time = datetime(2026, 9, 3, 8, 0) + app.hosts_file = HostsFile( + entries=[ + HostEntry( + ip_address="192.0.2.1", + hostnames=["one.test"], + dns_name="source.test", + resolved_ip="192.0.2.1", + last_resolved=original_time, + dns_resolution_status="match", + ) + ] + ) + app.dns_service.resolve_entry_async = AsyncMock( + return_value=DNSResolution( + hostname="source.test", + resolved_ip="198.51.100.9", + status=DNSResolutionStatus.RESOLVED, + resolved_at=datetime(2026, 9, 4, 11, 0), + ) + ) + self.fail_saves(app) + workers = [] + app.run_worker = Mock( + side_effect=lambda worker, **_kwargs: workers.append(worker) + ) + + app.action_update_single_dns() + await workers[0] + + entry = app.hosts_file.entries[0] + assert entry.ip_address == "192.0.2.1" + assert entry.resolved_ip == "192.0.2.1" + assert entry.last_resolved == original_time + assert entry.dns_resolution_status == "match" + app.update_status.assert_any_call( + "❌ DNS refresh save failed; previous state restored: Permission denied" + ) + + @pytest.mark.asyncio + async def test_single_dns_refresh_updates_selected_duplicate(self): + """DNS refresh retains the selected entry identity across its await.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + app.hosts_file = HostsFile( + entries=[ + HostEntry( + ip_address="192.0.2.1", + hostnames=["same.test"], + dns_name="source.test", + ), + HostEntry( + ip_address="192.0.2.2", + hostnames=["same.test"], + dns_name="source.test", + ), + ] + ) + app.selected_entry_index = 1 + app.dns_service.resolve_entry_async = AsyncMock( + return_value=DNSResolution( + hostname="source.test", + resolved_ip="198.51.100.9", + status=DNSResolutionStatus.RESOLVED, + resolved_at=datetime(2026, 9, 4, 11, 0), + ) + ) + app.manager.save_hosts_file = Mock(return_value=(True, "Saved")) + app.table_handler.populate_entries_table = Mock() + app.table_handler.restore_cursor_position = Mock() + app.details_handler.update_entry_details = Mock() + app.update_status = Mock() + workers = [] + app.run_worker = Mock( + side_effect=lambda worker, **_kwargs: workers.append(worker) + ) + + app.action_update_single_dns() + await workers[0] + + assert app.hosts_file.entries[0].ip_address == "192.0.2.1" + assert app.hosts_file.entries[1].ip_address == "198.51.100.9" + + @pytest.mark.asyncio + async def test_new_entry_dns_save_failure_keeps_saved_placeholder(self): + """Failed post-add DNS persistence keeps the already-saved DNS placeholder.""" + with patch("hosts.tui.app.HostsParser"), patch("hosts.tui.app.Config"): + app = HostsManagerApp() + app.edit_mode = True + app.manager.edit_mode = True + app.hosts_file = HostsFile() + entry = HostEntry( + ip_address="0.0.0.0", + hostnames=["one.test"], + is_active=False, + dns_name="source.test", + ) + app.manager.execute_add_command(app.hosts_file, entry) + app.dns_service.resolve_entry_async = AsyncMock( + return_value=DNSResolution( + hostname="source.test", + resolved_ip="198.51.100.9", + status=DNSResolutionStatus.RESOLVED, + resolved_at=datetime(2026, 9, 4, 11, 0), + ) + ) + self.fail_saves(app) + workers = [] + app.run_worker = Mock( + side_effect=lambda worker, **_kwargs: workers.append(worker) + ) + + app._resolve_new_dns_entry(entry) + await workers[0] + + restored = app.hosts_file.entries[0] + assert restored.ip_address == "0.0.0.0" + assert restored.resolved_ip is None + assert restored.last_resolved is None + assert restored.dns_resolution_status is None + assert restored.is_active is False + assert app.manager.can_undo() + app.update_status.assert_any_call( + "❌ DNS activation save failed; previous state restored: Permission denied" ) def test_main_function(self): diff --git a/tests/test_manager.py b/tests/test_manager.py index 2f35464..9a1c91c 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -556,6 +556,28 @@ class TestHostsManager: assert not success assert "No sudo permissions" in message + def test_save_mutation_failure_restores_hosts_file_and_history(self): + """The manager owns rollback of model and undo/redo state.""" + manager = HostsManager() + manager.edit_mode = True + hosts_file = HostsFile( + entries=[HostEntry("192.0.2.1", ["one.test"], is_active=True)] + ) + manager.execute_toggle_command(hosts_file, 0) + manager.undo_last_operation(hosts_file) + state = manager.capture_mutation_state(hosts_file) + + manager.execute_toggle_command(hosts_file, 0) + manager.save_hosts_file = Mock(return_value=(False, "Permission denied")) + + success, message, restored_hosts_file = manager.save_mutation(hosts_file, state) + + assert success is False + assert message == "Permission denied" + assert restored_hosts_file.entries[0].is_active is True + assert not manager.can_undo() + assert manager.can_redo() + @patch("subprocess.run") def test_restore_backup_success(self, mock_run): """Test restoring backup successfully."""