t3code/implement-issue-five #20
7 changed files with 451 additions and 0 deletions
|
|
@ -22,6 +22,7 @@ on macOS and Linux.
|
|||
- Creates a timestamped Pre-edit Backup before Privileged Mode begins.
|
||||
- Restores the current session's Pre-edit Backup after explicit confirmation.
|
||||
- Supports undo and redo during the current Privileged Mode session.
|
||||
- Exports Host Entries as hosts, JSON, or CSV files, and imports those formats in Privileged Mode.
|
||||
|
||||
The application starts in Read-only Mode. Default Entries cannot be edited,
|
||||
deleted, activated, deactivated, or moved.
|
||||
|
|
@ -76,6 +77,8 @@ including persistence behavior and manual recovery.
|
|||
| `Space` | Activate or deactivate the selected Host Entry |
|
||||
| `Ctrl+S` | Save the current in-memory state |
|
||||
| `Ctrl+F` | Open advanced filters |
|
||||
| `Ctrl+X` | Export Host Entries (hosts, JSON, or CSV) |
|
||||
| `Ctrl+O` | Import Host Entries from hosts, JSON, or CSV (Privileged Mode) |
|
||||
| `?` | Show help |
|
||||
| `q` or `Ctrl+C` | Quit |
|
||||
|
||||
|
|
|
|||
|
|
@ -102,6 +102,17 @@ Both actions require Privileged Mode because a successful result changes
|
|||
`/etc/hosts`. A failed lookup reports the failure and keeps the previously
|
||||
stored mapping.
|
||||
|
||||
## Import and export Host Entries
|
||||
|
||||
Press `Ctrl+X` to export the current in-memory Host Entries. Choose hosts, JSON, or
|
||||
CSV and enter a destination path. The workflow validates the destination and
|
||||
requires explicit confirmation before an existing file is overwritten.
|
||||
|
||||
Press `Ctrl+O` in Privileged Mode to import a hosts, JSON, or CSV file. Choose the
|
||||
format explicitly, then select the source path. A successful import replaces
|
||||
the current Host Entries and saves the Hosts File; parse and save errors remain
|
||||
visible in the message rail and leave the current file unchanged.
|
||||
|
||||
## Verify what is on disk
|
||||
|
||||
The details shown by the application normally match the last successful save.
|
||||
|
|
@ -247,6 +258,7 @@ continually retrying it.
|
|||
| `Ctrl+R` | Reload `/etc/hosts` |
|
||||
| `Ctrl+F` | Open advanced filters |
|
||||
| `c` | Open configuration |
|
||||
| `Ctrl+X` | Export Host Entries |
|
||||
| `?` | Show help |
|
||||
| `q` or `Ctrl+C` | Quit |
|
||||
| `Ctrl+E` | Enter or leave Privileged Mode |
|
||||
|
|
@ -266,6 +278,7 @@ continually retrying it.
|
|||
| `Ctrl+Z` / `Ctrl+Y` | Undo or redo and save the result |
|
||||
| `Ctrl+S` | Save the current in-memory state |
|
||||
| `b` | Review session backup changes and confirm restoration |
|
||||
| `Ctrl+O` | Import Host Entries from hosts, JSON, or CSV |
|
||||
|
||||
Within the Entry Editor, use `Tab` and `Shift+Tab` to move between fields and
|
||||
`Escape` to leave the form. If values changed, the application asks whether to
|
||||
|
|
|
|||
|
|
@ -29,12 +29,14 @@ from ..core.manager import HostsManager, MutationState
|
|||
from ..core.restore_preview import RestorePreviewError, create_restore_preview
|
||||
from ..core.dns import DNSService
|
||||
from ..core.filters import EntryFilter, FilterOptions
|
||||
from ..core.import_export import ImportExportService, ExportFormat, ImportFormat
|
||||
from .config_modal import ConfigModal
|
||||
from .add_entry_modal import AddEntryModal
|
||||
from .backup_restore_modal import BackupRestoreModal
|
||||
from .delete_confirmation_modal import DeleteConfirmationModal
|
||||
from .filter_modal import FilterModal
|
||||
from .help_modal import HelpModal
|
||||
from .import_export_modal import ExportModal, ImportModal, ExportRequest, ImportRequest
|
||||
from .custom_footer import CustomFooter
|
||||
from .styles import HOSTS_DARK_THEME, HOSTS_MANAGER_CSS
|
||||
from .keybindings import HOSTS_MANAGER_BINDINGS
|
||||
|
|
@ -85,6 +87,7 @@ class HostsManagerApp(App):
|
|||
self.parser = HostsParser()
|
||||
self.config = Config()
|
||||
self.manager = HostsManager()
|
||||
self.import_export_service = ImportExportService()
|
||||
|
||||
# Initialize DNS service
|
||||
dns_config = self.config.get("dns_resolution", {})
|
||||
|
|
@ -904,6 +907,75 @@ class HostsManagerApp(App):
|
|||
return
|
||||
self.navigation_handler.save_hosts_file()
|
||||
|
||||
def action_export_entries(self) -> None:
|
||||
"""Export the in-memory Host Entries without requiring Privileged Mode."""
|
||||
|
||||
def handle_export(request: ExportRequest | None) -> None:
|
||||
if request is None:
|
||||
return
|
||||
exporters = {
|
||||
ExportFormat.HOSTS: self.import_export_service.export_hosts_format,
|
||||
ExportFormat.JSON: self.import_export_service.export_json_format,
|
||||
ExportFormat.CSV: self.import_export_service.export_csv_format,
|
||||
}
|
||||
result = exporters[request.format](self.hosts_file, request.path)
|
||||
if result.success:
|
||||
self.update_status(
|
||||
f"✓ Exported {result.entries_exported} Host Entries to {result.file_path}"
|
||||
)
|
||||
else:
|
||||
self.update_status(
|
||||
f"Error exporting Host Entries: {'; '.join(result.errors)}"
|
||||
)
|
||||
|
||||
self.push_screen(
|
||||
ExportModal(
|
||||
self.import_export_service.get_supported_export_formats(),
|
||||
self.import_export_service.validate_export_path,
|
||||
),
|
||||
handle_export,
|
||||
)
|
||||
|
||||
def action_import_entries(self) -> None:
|
||||
"""Import and persist replacement Host Entries in Privileged Mode."""
|
||||
if not self._allow_mutation_action():
|
||||
return
|
||||
if not self.edit_mode:
|
||||
self.update_status(
|
||||
"Cannot import Host Entries in Read-only Mode. Enter Privileged Mode first."
|
||||
)
|
||||
return
|
||||
|
||||
importers = {
|
||||
ImportFormat.HOSTS: self.import_export_service.import_hosts_format,
|
||||
ImportFormat.JSON: self.import_export_service.import_json_format,
|
||||
ImportFormat.CSV: self.import_export_service.import_csv_format,
|
||||
}
|
||||
|
||||
def import_file(format: ImportFormat, path):
|
||||
return importers[format](path)
|
||||
|
||||
def handle_import(request: ImportRequest | None) -> None:
|
||||
if request is None:
|
||||
return
|
||||
result = request.result
|
||||
snapshot = self.capture_mutation_state()
|
||||
self.hosts_file = HostsFile(entries=result.entries)
|
||||
self.selected_entry_index = 0
|
||||
if self.save_mutation(snapshot, "Import"):
|
||||
self.table_handler.populate_entries_table()
|
||||
self.details_handler.update_entry_details()
|
||||
self.update_status(
|
||||
f"✓ Imported and saved {result.successfully_imported} Host Entries from {request.path}"
|
||||
)
|
||||
|
||||
self.push_screen(
|
||||
ImportModal(
|
||||
self.import_export_service.get_supported_import_formats(), import_file
|
||||
),
|
||||
handle_import,
|
||||
)
|
||||
|
||||
def action_add_entry(self) -> None:
|
||||
"""Show the add entry modal."""
|
||||
if not self._allow_add_entry_action():
|
||||
|
|
|
|||
|
|
@ -48,6 +48,10 @@ class HelpModal(ModalScreen[None]):
|
|||
"Type in Search for immediate filtering Ctrl+F Advanced filters",
|
||||
classes="help-copy",
|
||||
)
|
||||
yield Static(
|
||||
"Ctrl+X Export Host Entries Ctrl+O Import Host Entries",
|
||||
classes="help-copy",
|
||||
)
|
||||
yield Static("Privileged Mode", classes="help-section")
|
||||
yield Static(
|
||||
"Ctrl+E Enter or leave Privileged Mode n Add Host Entry\n"
|
||||
|
|
|
|||
188
src/hosts/tui/import_export_modal.py
Normal file
188
src/hosts/tui/import_export_modal.py
Normal file
|
|
@ -0,0 +1,188 @@
|
|||
"""Keyboard-first import and export forms for Host Entries."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from textual.app import ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.containers import Horizontal, Vertical
|
||||
from textual.screen import ModalScreen
|
||||
from textual.widgets import Button, Input, Label, RadioButton, RadioSet, Static
|
||||
|
||||
from ..core.import_export import ExportFormat, ImportFormat, ImportResult
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExportRequest:
|
||||
"""A validated request to export the displayed Hosts File."""
|
||||
|
||||
path: Path
|
||||
format: ExportFormat
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ImportRequest:
|
||||
"""A validated request to replace the Hosts File from an import."""
|
||||
|
||||
path: Path
|
||||
format: ImportFormat
|
||||
result: ImportResult
|
||||
|
||||
|
||||
class _FormatModal(ModalScreen):
|
||||
"""Shared form mechanics for selecting a file and supported format."""
|
||||
|
||||
BINDINGS = [Binding("escape", "cancel", "Cancel")]
|
||||
CSS = """
|
||||
_FormatModal { align: center middle; }
|
||||
#format-container {
|
||||
width: 72; height: auto; max-height: 90%; background: $surface;
|
||||
border: thick $primary; padding: 1 2;
|
||||
}
|
||||
.format-title { text-align: center; text-style: bold; color: $primary; }
|
||||
.format-copy { margin-top: 1; color: $text-muted; }
|
||||
#path-input { margin-top: 1; }
|
||||
#workflow-error { color: $error; height: auto; margin-top: 1; }
|
||||
#workflow-warning { color: $warning; height: auto; margin-top: 1; }
|
||||
.button-row { margin-top: 1; height: 3; align: center middle; }
|
||||
"""
|
||||
|
||||
def _selected_format(self, enum_type):
|
||||
for member in enum_type:
|
||||
if self.query_one(f"#format-{member.value}", RadioButton).value:
|
||||
return member
|
||||
return next(iter(enum_type))
|
||||
|
||||
def _path_or_error(self) -> Path | None:
|
||||
value = self.query_one("#path-input", Input).value.strip()
|
||||
if value:
|
||||
self.query_one("#workflow-error", Static).update("")
|
||||
return Path(value).expanduser()
|
||||
self.query_one("#workflow-error", Static).update("Choose a file path first.")
|
||||
return None
|
||||
|
||||
def action_cancel(self) -> None:
|
||||
self.dismiss(None)
|
||||
|
||||
|
||||
class ExportModal(_FormatModal):
|
||||
"""Collect an export destination and require confirmation for overwrite."""
|
||||
|
||||
def __init__(self, formats: list[ExportFormat], validate_path):
|
||||
super().__init__()
|
||||
self._formats = formats
|
||||
self._validate_path = validate_path
|
||||
self._request: ExportRequest | None = None
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="format-container"):
|
||||
yield Static("Export Host Entries", classes="format-title")
|
||||
yield Static(
|
||||
"Choose a destination and format. Existing files require confirmation.",
|
||||
classes="format-copy",
|
||||
)
|
||||
yield Label("Export file path")
|
||||
yield Input(placeholder="/path/to/entries.hosts", id="path-input")
|
||||
with RadioSet(id="format-select"):
|
||||
for index, format in enumerate(self._formats):
|
||||
yield RadioButton(
|
||||
_format_label(format),
|
||||
id=f"format-{format.value}",
|
||||
value=index == 0,
|
||||
)
|
||||
yield Static("", id="workflow-error")
|
||||
yield Static("", id="workflow-warning")
|
||||
with Horizontal(classes="button-row"):
|
||||
yield Button("Cancel", id="cancel-button")
|
||||
yield Button("Continue", id="continue-button", variant="primary")
|
||||
yield Button("Overwrite export", id="confirm-button", variant="error")
|
||||
|
||||
def on_mount(self) -> None:
|
||||
self.query_one("#path-input", Input).focus()
|
||||
self.query_one("#confirm-button", Button).display = False
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed) -> None:
|
||||
if event.button.id == "cancel-button":
|
||||
self.action_cancel()
|
||||
elif event.button.id == "continue-button":
|
||||
self.action_continue()
|
||||
elif event.button.id == "confirm-button" and self._request is not None:
|
||||
self.dismiss(self._request)
|
||||
|
||||
def action_continue(self) -> None:
|
||||
path = self._path_or_error()
|
||||
if path is None:
|
||||
return
|
||||
request = ExportRequest(path, self._selected_format(ExportFormat))
|
||||
warnings = self._validate_path(path, request.format)
|
||||
self._request = request
|
||||
warning = self.query_one("#workflow-warning", Static)
|
||||
warning.update("\n".join(warnings))
|
||||
blocking_warnings = [item for item in warnings if "already exists" not in item]
|
||||
if blocking_warnings:
|
||||
self.query_one("#workflow-error", Static).update(
|
||||
"Correct the destination before exporting: "
|
||||
+ "; ".join(blocking_warnings)
|
||||
)
|
||||
return
|
||||
if warnings:
|
||||
self.query_one("#confirm-button", Button).display = True
|
||||
return
|
||||
self.dismiss(request)
|
||||
|
||||
|
||||
class ImportModal(_FormatModal):
|
||||
"""Collect an import file and format before replacing Host Entries."""
|
||||
|
||||
def __init__(self, formats: list[ImportFormat], import_file):
|
||||
super().__init__()
|
||||
self._formats = formats
|
||||
self._import_file = import_file
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="format-container"):
|
||||
yield Static("Import Host Entries", classes="format-title")
|
||||
yield Static(
|
||||
"Import replaces the current Host Entries and saves the Hosts File.",
|
||||
classes="format-copy",
|
||||
)
|
||||
yield Label("Import file path")
|
||||
yield Input(placeholder="/path/to/entries.hosts", id="path-input")
|
||||
with RadioSet(id="format-select"):
|
||||
for index, format in enumerate(self._formats):
|
||||
yield RadioButton(
|
||||
_format_label(format),
|
||||
id=f"format-{format.value}",
|
||||
value=index == 0,
|
||||
)
|
||||
yield Static("", id="workflow-error")
|
||||
with Horizontal(classes="button-row"):
|
||||
yield Button("Cancel", id="cancel-button")
|
||||
yield Button("Import and save", id="import-button", variant="primary")
|
||||
|
||||
def on_mount(self) -> None:
|
||||
self.query_one("#path-input", Input).focus()
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed) -> None:
|
||||
if event.button.id == "cancel-button":
|
||||
self.action_cancel()
|
||||
elif event.button.id == "import-button":
|
||||
path = self._path_or_error()
|
||||
if path is not None:
|
||||
format = self._selected_format(ImportFormat)
|
||||
result = self._import_file(format, path)
|
||||
if not result.success:
|
||||
self.query_one("#workflow-error", Static).update(
|
||||
"; ".join(result.errors)
|
||||
)
|
||||
return
|
||||
self.dismiss(ImportRequest(path, format, result))
|
||||
|
||||
|
||||
def _format_label(format: ExportFormat | ImportFormat) -> str:
|
||||
"""Give the core-supported formats concise, file-oriented labels."""
|
||||
return {
|
||||
"hosts": "Hosts file (.hosts)",
|
||||
"json": "JSON (.json)",
|
||||
"csv": "CSV (.csv)",
|
||||
}[format.value]
|
||||
|
|
@ -34,6 +34,12 @@ HOSTS_MANAGER_BINDINGS = [
|
|||
id="left:toggle_edit_mode",
|
||||
),
|
||||
Binding("c", "config", "Configuration", show=True, id="right:config"),
|
||||
Binding(
|
||||
"ctrl+x", "export_entries", "Export Host Entries", show=False, priority=True
|
||||
),
|
||||
Binding(
|
||||
"ctrl+o", "import_entries", "Import Host Entries", show=False, priority=True
|
||||
),
|
||||
Binding(
|
||||
"ctrl+f",
|
||||
"show_filters",
|
||||
|
|
|
|||
165
tests/test_import_export_workflow.py
Normal file
165
tests/test_import_export_workflow.py
Normal file
|
|
@ -0,0 +1,165 @@
|
|||
"""User-visible import and export workflows."""
|
||||
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
from textual.widgets import Input, RadioButton, Static
|
||||
|
||||
from hosts.core.import_export import (
|
||||
ExportFormat,
|
||||
ExportResult,
|
||||
ImportFormat,
|
||||
ImportResult,
|
||||
)
|
||||
from hosts.core.models import HostEntry, HostsFile
|
||||
from hosts.tui.app import HostsManagerApp
|
||||
from hosts.tui.import_export_modal import ExportModal, ImportModal
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_export_shortcut_opens_a_format_chooser_and_reports_validation():
|
||||
"""Export presents every core format and keeps invalid input visible."""
|
||||
app = HostsManagerApp()
|
||||
app.hosts_file = HostsFile(
|
||||
entries=[HostEntry(ip_address="192.0.2.10", hostnames=["export.test"])]
|
||||
)
|
||||
app.load_hosts_file = Mock()
|
||||
|
||||
async with app.run_test(size=(120, 40)) as pilot:
|
||||
await pilot.press("ctrl+x")
|
||||
await pilot.pause()
|
||||
assert isinstance(app.screen, ExportModal)
|
||||
assert [button.label.plain for button in app.screen.query(RadioButton)] == [
|
||||
"Hosts file (.hosts)",
|
||||
"JSON (.json)",
|
||||
"CSV (.csv)",
|
||||
]
|
||||
|
||||
app.screen.query_one("#path-input", Input).value = ""
|
||||
await pilot.click("#continue-button")
|
||||
assert "Choose a file path" in str(
|
||||
app.screen.query_one("#workflow-error", Static).render()
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_export_requires_overwrite_confirmation_before_calling_service(tmp_path):
|
||||
"""An existing output file is not replaced until the user confirms it."""
|
||||
target = tmp_path / "hosts.json"
|
||||
target.write_text("existing", encoding="utf-8")
|
||||
app = HostsManagerApp()
|
||||
app.hosts_file = HostsFile(
|
||||
entries=[HostEntry(ip_address="192.0.2.10", hostnames=["export.test"])]
|
||||
)
|
||||
app.import_export_service = Mock()
|
||||
app.import_export_service.get_supported_export_formats.return_value = list(
|
||||
ExportFormat
|
||||
)
|
||||
app.import_export_service.validate_export_path.return_value = [
|
||||
f"File {target} already exists and will be overwritten"
|
||||
]
|
||||
app.import_export_service.export_json_format.return_value = ExportResult(
|
||||
success=True,
|
||||
file_path=target,
|
||||
entries_exported=1,
|
||||
errors=[],
|
||||
format=ExportFormat.JSON,
|
||||
)
|
||||
|
||||
async with app.run_test(size=(120, 40)) as pilot:
|
||||
app.action_export_entries()
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#path-input", Input).value = str(target)
|
||||
app.screen.query_one("#format-json", RadioButton).value = True
|
||||
await pilot.click("#continue-button")
|
||||
assert "already exists" in str(
|
||||
app.screen.query_one("#workflow-warning", Static).render()
|
||||
)
|
||||
app.import_export_service.export_json_format.assert_not_called()
|
||||
|
||||
await pilot.click("#confirm-button")
|
||||
await pilot.pause()
|
||||
await pilot.pause()
|
||||
|
||||
app.import_export_service.export_json_format.assert_called_once_with(
|
||||
app.hosts_file, target
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_import_shortcut_requires_privileged_mode_and_surfaces_service_errors():
|
||||
"""Import cannot mutate in Read-only Mode and reports parser failures."""
|
||||
app = HostsManagerApp()
|
||||
app.hosts_file = HostsFile(
|
||||
entries=[HostEntry(ip_address="192.0.2.10", hostnames=["original.test"])]
|
||||
)
|
||||
app.load_hosts_file = Mock()
|
||||
|
||||
async with app.run_test(size=(120, 40)) as pilot:
|
||||
await pilot.press("ctrl+o")
|
||||
await pilot.pause()
|
||||
assert not isinstance(app.screen, ImportModal)
|
||||
assert "Read-only Mode" in str(app.query_one("#message-rail", Static).render())
|
||||
|
||||
app.edit_mode = True
|
||||
app.manager.edit_mode = True
|
||||
app.import_export_service = Mock()
|
||||
app.import_export_service.get_supported_import_formats.return_value = list(
|
||||
ImportFormat
|
||||
)
|
||||
app.import_export_service.import_csv_format.return_value = Mock(
|
||||
success=False, errors=["CSV column 'ip_address' is missing"]
|
||||
)
|
||||
app.action_import_entries()
|
||||
await pilot.pause()
|
||||
assert isinstance(app.screen, ImportModal)
|
||||
app.screen.query_one("#path-input", Input).value = "/tmp/broken.csv"
|
||||
csv_button = app.screen.query_one("#format-csv", RadioButton)
|
||||
csv_button.value = True
|
||||
await pilot.click("#import-button")
|
||||
await pilot.pause()
|
||||
assert "CSV column 'ip_address' is missing" in str(
|
||||
app.screen.query_one("#workflow-error", Static).render()
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_successful_import_replaces_entries_and_saves_through_the_manager():
|
||||
"""A valid import reaches the persistence seam without a real Hosts File."""
|
||||
app = HostsManagerApp()
|
||||
app.hosts_file = HostsFile(
|
||||
entries=[HostEntry(ip_address="192.0.2.10", hostnames=["original.test"])]
|
||||
)
|
||||
app.edit_mode = True
|
||||
app.manager.edit_mode = True
|
||||
app.save_mutation = Mock(return_value=True)
|
||||
app.import_export_service = Mock()
|
||||
app.import_export_service.get_supported_import_formats.return_value = list(
|
||||
ImportFormat
|
||||
)
|
||||
imported = HostEntry(ip_address="198.51.100.3", hostnames=["imported.test"])
|
||||
app.import_export_service.import_json_format.return_value = ImportResult(
|
||||
success=True,
|
||||
entries=[imported],
|
||||
errors=[],
|
||||
warnings=[],
|
||||
total_processed=1,
|
||||
successfully_imported=1,
|
||||
)
|
||||
|
||||
async with app.run_test(size=(120, 40)) as pilot:
|
||||
app.action_import_entries()
|
||||
await pilot.pause()
|
||||
app.screen.query_one("#path-input", Input).value = "/tmp/entries.json"
|
||||
app.screen.query_one("#format-json", RadioButton).value = True
|
||||
await pilot.click("#import-button")
|
||||
await pilot.pause()
|
||||
await pilot.pause()
|
||||
|
||||
assert [entry.hostnames for entry in app.hosts_file.entries] == [
|
||||
["imported.test"]
|
||||
]
|
||||
app.save_mutation.assert_called_once()
|
||||
assert "Imported and saved 1 Host Entries" in str(
|
||||
app.query_one("#message-rail", Static).render()
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue