834 lines
30 KiB
Python
834 lines
30 KiB
Python
"""Tests for application-level authorization flow."""
|
||
|
||
from contextlib import contextmanager
|
||
import asyncio
|
||
from io import StringIO
|
||
import os
|
||
import signal
|
||
from unittest.mock import Mock, patch
|
||
|
||
import pytest
|
||
from rich.cells import cell_len
|
||
from textual.app import SuspendNotSupported
|
||
from textual.events import Key
|
||
from textual.widgets import Button, Input, RadioButton, Static
|
||
|
||
from src.hosts.core.filters import FilterOptions
|
||
from src.hosts.core.models import HostEntry, HostsFile
|
||
from src.hosts.tui.app import HostsManagerApp
|
||
from src.hosts.tui.custom_footer import CustomFooter
|
||
from src.hosts.tui.backup_restore_modal import BackupRestoreModal
|
||
from src.hosts.tui.filter_modal import FilterModal
|
||
from src.hosts.tui.help_modal import HelpModal
|
||
from src.hosts.tui.add_entry_modal import AddEntryModal
|
||
from src.hosts.tui.keyboard_protocol import HostsXTermParser
|
||
from src.hosts.tui.keyboard_driver import HostsKeyboardDriver
|
||
from src.hosts.tui.privilege_prompt import (
|
||
render_sudo_authentication_notice,
|
||
sudo_authentication_screen,
|
||
)
|
||
|
||
|
||
@contextmanager
|
||
def suspended_tui():
|
||
"""Stand in for Textual's terminal suspension in action tests."""
|
||
yield
|
||
|
||
|
||
def app_with_filterable_entries() -> HostsManagerApp:
|
||
"""Create an app whose filter workflow cannot access the system Hosts File."""
|
||
app = HostsManagerApp()
|
||
app.hosts_file = HostsFile(
|
||
entries=[
|
||
HostEntry(ip_address="192.0.2.1", hostnames=["active.test"]),
|
||
HostEntry(
|
||
ip_address="192.0.2.2",
|
||
hostnames=["inactive.test"],
|
||
is_active=False,
|
||
),
|
||
]
|
||
)
|
||
app.load_hosts_file = Mock()
|
||
return app
|
||
|
||
|
||
def test_app_uses_the_modifier_preserving_keyboard_driver():
|
||
"""The live terminal app receives enhanced keyboard events through the adapter."""
|
||
assert HostsManagerApp().driver_class is HostsKeyboardDriver
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_filter_shortcut_opens_modal_and_applies_the_selected_filter():
|
||
"""The filter shortcut opens a modal whose Apply action changes the view state."""
|
||
app = app_with_filterable_entries()
|
||
|
||
async with app.run_test() as pilot:
|
||
footer = app.query_one("#custom-footer", CustomFooter)
|
||
footer_text = footer.query_one("#footer-right", Static).render()
|
||
assert "ctrl+f Filters" in str(footer_text)
|
||
|
||
app.action_help()
|
||
await pilot.pause()
|
||
assert isinstance(app.screen, HelpModal)
|
||
assert app.screen.query_one("#help-close") is not None
|
||
await pilot.press("escape")
|
||
await pilot.pause()
|
||
|
||
await pilot.press("ctrl+f")
|
||
|
||
assert isinstance(app.screen, FilterModal)
|
||
app.screen.query_one("#status-active", RadioButton).value = True
|
||
await pilot.pause()
|
||
await pilot.click("#apply")
|
||
await pilot.pause()
|
||
|
||
assert app.current_filter_options.active_only
|
||
assert not app.current_filter_options.show_inactive
|
||
assert [entry.hostnames[0] for entry in app.get_visible_entries()] == [
|
||
"active.test"
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_filter_shortcut_does_not_stack_filter_modals():
|
||
"""Repeated filter shortcuts keep the existing filter modal in focus."""
|
||
app = app_with_filterable_entries()
|
||
|
||
async with app.run_test() as pilot:
|
||
await pilot.press("ctrl+f")
|
||
await pilot.pause()
|
||
|
||
await pilot.press("ctrl+f")
|
||
await pilot.press("ctrl+f")
|
||
await pilot.pause()
|
||
|
||
assert isinstance(app.screen, FilterModal)
|
||
assert sum(isinstance(screen, FilterModal) for screen in app.screen_stack) == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_new_from_selected_prefills_the_highlighted_visible_entry():
|
||
"""Shift+N snapshots the highlighted visible Host Entry into the Add form."""
|
||
app = app_with_filterable_entries()
|
||
source = app.hosts_file.entries[1]
|
||
source.hostnames = ["inactive.test", "alias.test"]
|
||
source.comment = "Copied comment"
|
||
source.is_active = False
|
||
app.edit_mode = True
|
||
app.sort_column = "hostname"
|
||
|
||
async with app.run_test(size=(120, 40)) as pilot:
|
||
app.table_handler.populate_entries_table()
|
||
app.table_handler.move_cursor_to_entry_index(1)
|
||
await pilot.pause()
|
||
|
||
await pilot.press("shift+n")
|
||
await pilot.pause()
|
||
|
||
assert isinstance(app.screen, AddEntryModal)
|
||
assert app.screen.query_one("#hostnames-input", Input).value == (
|
||
"inactive.test, alias.test"
|
||
)
|
||
assert app.screen.query_one("#ip-address-input", Input).value == "192.0.2.2"
|
||
assert app.screen.query_one("#comment-input", Input).value == "Copied comment"
|
||
assert not app.screen.query_one("#active-checkbox").value
|
||
assert "Based on: inactive.test" in str(
|
||
app.screen.query_one("#entry-source", Static).render()
|
||
)
|
||
assert app.focused is app.screen.query_one("#hostnames-input", Input)
|
||
|
||
assert source.hostnames == ["inactive.test", "alias.test"]
|
||
assert source.comment == "Copied comment"
|
||
assert not source.is_active
|
||
|
||
|
||
def test_enhanced_shift_n_preserves_the_shift_modifier():
|
||
"""Kitty Shift+N remains distinct from an uppercase text character."""
|
||
event = list(HostsXTermParser().feed("\x1b[110;2;78u"))[0]
|
||
|
||
assert isinstance(event, Key)
|
||
assert event.key == "shift+n"
|
||
assert event.character == "N"
|
||
|
||
|
||
def test_enhanced_caps_lock_n_does_not_become_shift_n():
|
||
"""Caps Lock does not invoke the New from selected binding."""
|
||
event = list(HostsXTermParser().feed("\x1b[110;65;78u"))[0]
|
||
|
||
assert isinstance(event, Key)
|
||
assert event.key == "caps_lock+n"
|
||
assert event.character == "N"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_enhanced_shift_n_opens_new_from_selected_but_caps_lock_n_does_not():
|
||
"""The real enhanced-key events distinguish Shift from Caps Lock at the app."""
|
||
app = app_with_filterable_entries()
|
||
app.edit_mode = True
|
||
|
||
async with app.run_test(size=(120, 40)) as pilot:
|
||
app.table_handler.populate_entries_table()
|
||
app.query_one("#entries-table").focus()
|
||
|
||
shift_n = list(HostsXTermParser().feed("\x1b[110;2;78u"))[0]
|
||
app.post_message(shift_n)
|
||
await pilot.pause()
|
||
assert isinstance(app.screen, AddEntryModal)
|
||
|
||
await pilot.press("escape")
|
||
await pilot.pause()
|
||
caps_lock_n = list(HostsXTermParser().feed("\x1b[110;65;78u"))[0]
|
||
app.post_message(caps_lock_n)
|
||
await pilot.pause()
|
||
assert not isinstance(app.screen, AddEntryModal)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_uppercase_n_is_a_fallback_when_the_terminal_lacks_enhanced_keys():
|
||
"""Legacy terminals use uppercase N for Shift+N and Caps Lock+N alike."""
|
||
app = app_with_filterable_entries()
|
||
app.edit_mode = True
|
||
|
||
async with app.run_test(size=(120, 40)) as pilot:
|
||
app.table_handler.populate_entries_table()
|
||
app.query_one("#entries-table").focus()
|
||
await pilot.press("N")
|
||
await pilot.pause()
|
||
|
||
assert isinstance(app.screen, AddEntryModal)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_new_from_selected_requires_a_visible_host_entry():
|
||
"""Shift+N does not reuse a stale selection when filters hide every entry."""
|
||
app = app_with_filterable_entries()
|
||
app.edit_mode = True
|
||
app.current_filter_options.search_term = "not-a-match"
|
||
|
||
async with app.run_test(size=(120, 40)) as pilot:
|
||
app.table_handler.populate_entries_table()
|
||
await pilot.press("shift+n")
|
||
await pilot.pause()
|
||
|
||
assert not isinstance(app.screen, AddEntryModal)
|
||
assert "No Host Entry selected; use n to add a blank Host Entry." in str(
|
||
app.query_one("#message-rail", Static).render()
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_new_from_selected_is_available_at_the_minimum_viewport():
|
||
"""Shift+N keeps the selected-entry add workflow reachable at 100×30."""
|
||
app = app_with_filterable_entries()
|
||
app.edit_mode = True
|
||
|
||
async with app.run_test(size=(100, 30)) as pilot:
|
||
app.table_handler.populate_entries_table()
|
||
await pilot.press("shift+n")
|
||
await pilot.pause()
|
||
|
||
assert isinstance(app.screen, AddEntryModal)
|
||
assert app.focused is app.screen.query_one("#hostnames-input", Input)
|
||
await pilot.press("escape")
|
||
await pilot.pause()
|
||
assert not isinstance(app.screen, AddEntryModal)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_help_overlay_closes_to_the_control_that_opened_it():
|
||
"""Help is an overlay and returns keyboard focus to its opener."""
|
||
app = app_with_filterable_entries()
|
||
|
||
async with app.run_test(size=(120, 40)) as pilot:
|
||
search = app.query_one("#search-input", Input)
|
||
search.focus()
|
||
app.action_help()
|
||
await pilot.pause()
|
||
|
||
assert isinstance(app.screen, HelpModal)
|
||
await pilot.press("escape")
|
||
await pilot.pause()
|
||
|
||
assert app.focused is search
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_filter_modal_tabs_through_controls_in_visual_order():
|
||
"""The modal starts at Presets and tabs through the visible controls in order."""
|
||
app = app_with_filterable_entries()
|
||
|
||
async with app.run_test() as pilot:
|
||
await pilot.press("ctrl+f")
|
||
await pilot.pause()
|
||
|
||
assert app.focused is app.screen.query_one("#preset-select")
|
||
for expected_id in (
|
||
"load-preset",
|
||
"save-preset",
|
||
"delete-preset",
|
||
"status-filter-type",
|
||
"type-filter-type",
|
||
"resolution-filter-type",
|
||
"search-term",
|
||
"search-hostnames",
|
||
"search-comments",
|
||
"search-ips",
|
||
"search-case-sensitive",
|
||
"cancel",
|
||
"reset",
|
||
"apply",
|
||
):
|
||
await pilot.press("tab")
|
||
assert app.focused is app.screen.query_one(f"#{expected_id}")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_filter_reset_clears_filters_and_cancel_preserves_applied_filters():
|
||
"""Reset clears the form on Apply, while Cancel leaves the applied filter alone."""
|
||
app = app_with_filterable_entries()
|
||
app.current_filter_options = FilterOptions(
|
||
active_only=True,
|
||
show_inactive=False,
|
||
)
|
||
|
||
async with app.run_test() as pilot:
|
||
await pilot.press("ctrl+f")
|
||
await pilot.click("#reset")
|
||
await pilot.click("#apply")
|
||
await pilot.pause()
|
||
|
||
assert app.current_filter_options.is_empty()
|
||
|
||
await pilot.press("ctrl+f")
|
||
assert isinstance(app.screen, FilterModal)
|
||
app.screen.query_one("#status-active", RadioButton).value = True
|
||
await pilot.pause()
|
||
await pilot.press("escape")
|
||
await pilot.pause()
|
||
|
||
assert app.current_filter_options.is_empty()
|
||
assert [entry.hostnames[0] for entry in app.get_visible_entries()] == [
|
||
"active.test",
|
||
"inactive.test",
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_workspace_uses_read_only_detail_rows_and_hides_irrelevant_dns_fields():
|
||
"""An IP Host Entry has compact text details instead of disabled controls."""
|
||
app = app_with_filterable_entries()
|
||
|
||
async with app.run_test(size=(120, 40)) as pilot:
|
||
app.table_handler.populate_entries_table()
|
||
app.details_handler.update_entry_details()
|
||
await pilot.pause()
|
||
|
||
details = app.query_one("#entry-details-display")
|
||
assert not list(details.query(Input))
|
||
assert "192.0.2.1" in str(app.query_one("#details-ip-input", Static).render())
|
||
assert "Active" in str(
|
||
app.query_one("#details-active-checkbox", Static).render()
|
||
)
|
||
assert app.query_one("#details-dns-rows").has_class("hidden")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_workspace_exposes_dns_and_protection_details_and_durable_state():
|
||
"""DNS and Default Entry states remain explicit independently of colour."""
|
||
app = app_with_filterable_entries()
|
||
app.hosts_file.entries[0].dns_name = "active.example"
|
||
app.hosts_file.entries[0].dns_resolution_status = "resolved"
|
||
app.hosts_file.entries[0].resolved_ip = "192.0.2.1"
|
||
app.current_filter_options.active_only = True
|
||
app.current_filter_options.show_inactive = False
|
||
|
||
async with app.run_test(size=(100, 30)) as pilot:
|
||
app.table_handler.populate_entries_table()
|
||
app.details_handler.update_entry_details()
|
||
app._update_footer_status()
|
||
await pilot.pause()
|
||
|
||
assert not app.query_one("#details-dns-rows").has_class("hidden")
|
||
assert "Resolved" in str(
|
||
app.query_one("#details-dns-status-input", Static).render()
|
||
)
|
||
footer = app.query_one("#custom-footer", CustomFooter)
|
||
assert "Filters: 1" in footer._status_text
|
||
assert "READ-ONLY" in footer._status_text
|
||
assert app.has_class("compact-layout")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_small_viewport_replaces_workspace_with_an_explicit_safe_message():
|
||
"""A viewport below the contract cannot expose clipped mutating controls."""
|
||
app = app_with_filterable_entries()
|
||
|
||
async with app.run_test(size=(80, 24)) as pilot:
|
||
await pilot.pause()
|
||
|
||
assert not app.query_one("#workspace").display
|
||
minimum_size = app.query_one("#minimum-size", Static)
|
||
assert "requires at least 100 columns × 30 rows" in str(minimum_size.render())
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_small_viewport_blocks_hidden_mutation_shortcuts():
|
||
"""A minimum-size presentation cannot open a hidden editor or privilege flow."""
|
||
app = app_with_filterable_entries()
|
||
app.push_screen = Mock()
|
||
app.manager = Mock()
|
||
|
||
async with app.run_test(size=(80, 24)) as pilot:
|
||
await pilot.press("n", "ctrl+e")
|
||
await pilot.pause()
|
||
|
||
app.push_screen.assert_not_called()
|
||
app.manager.enter_edit_mode.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_restore_opens_a_side_by_side_preview_with_exact_paths(tmp_path):
|
||
"""The recovery workflow makes its before-and-after files reviewable."""
|
||
app = app_with_filterable_entries()
|
||
backup_path = tmp_path / "hosts.backup"
|
||
hosts_file_path = tmp_path / "hosts"
|
||
backup_path.write_text("127.0.0.1 restored.test\n")
|
||
hosts_file_path.write_text("127.0.0.1 current.test\n")
|
||
app.edit_mode = True
|
||
app.manager = Mock()
|
||
app.manager.backup_path = backup_path
|
||
app.manager.parser.file_path = hosts_file_path
|
||
|
||
async with app.run_test(size=(120, 40)) as pilot:
|
||
app.action_restore_backup()
|
||
await pilot.pause()
|
||
await pilot.pause()
|
||
|
||
assert isinstance(app.screen, BackupRestoreModal)
|
||
assert str(backup_path) in str(
|
||
app.screen.query_one("#backup-restored-path", Static).render()
|
||
)
|
||
assert str(hosts_file_path) in str(
|
||
app.screen.query_one("#backup-current-path", Static).render()
|
||
)
|
||
diff = str(app.screen.query_one("#backup-diff", Static).render())
|
||
assert "current.test" in diff
|
||
assert "restored.test" in diff
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_restore_side_by_side_divider_uses_terminal_cell_widths(tmp_path):
|
||
"""Tabs in Hosts File lines cannot move the right-hand file between rows."""
|
||
app = app_with_filterable_entries()
|
||
backup_path = tmp_path / "hosts.backup"
|
||
hosts_file_path = tmp_path / "hosts"
|
||
hosts_file_path.write_text(
|
||
"127.0.0.1\tlocalhost\n"
|
||
"255.255.255.255\tbroadcasthost\n"
|
||
"::1\tlocalhost\n"
|
||
"192.0.2.10\tcurrent.test\n"
|
||
)
|
||
backup_path.write_text(
|
||
"127.0.0.1\tlocalhost\n"
|
||
"255.255.255.255\tbroadcasthost\n"
|
||
"::1\tlocalhost\n"
|
||
"192.0.2.10\trestored.test\n"
|
||
)
|
||
app.edit_mode = True
|
||
app.manager = Mock()
|
||
app.manager.backup_path = backup_path
|
||
app.manager.parser.file_path = hosts_file_path
|
||
|
||
async with app.run_test(size=(160, 40)) as pilot:
|
||
app.action_restore_backup()
|
||
await pilot.pause()
|
||
await pilot.pause()
|
||
|
||
modal = app.screen
|
||
assert isinstance(modal, BackupRestoreModal)
|
||
lines = [
|
||
line for line in modal._render_diff().plain.splitlines() if " │ " in line
|
||
]
|
||
divider_columns = {
|
||
cell_len(line[: line.index("│")].expandtabs(8)) for line in lines
|
||
}
|
||
|
||
assert "\t" not in "\n".join(lines)
|
||
assert len(divider_columns) == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_restore_preview_disables_a_no_change_operation(tmp_path):
|
||
"""A no-op restore cannot perform an unnecessary privileged write."""
|
||
app = app_with_filterable_entries()
|
||
backup_path = tmp_path / "hosts.backup"
|
||
hosts_file_path = tmp_path / "hosts"
|
||
contents = "127.0.0.1 localhost\n"
|
||
backup_path.write_text(contents)
|
||
hosts_file_path.write_text(contents)
|
||
app.edit_mode = True
|
||
app.manager = Mock()
|
||
app.manager.backup_path = backup_path
|
||
app.manager.parser.file_path = hosts_file_path
|
||
|
||
async with app.run_test(size=(120, 40)) as pilot:
|
||
app.action_restore_backup()
|
||
await pilot.pause()
|
||
await pilot.pause()
|
||
|
||
modal = app.screen
|
||
assert isinstance(modal, BackupRestoreModal)
|
||
assert modal.query_one("#restore-button", Button).disabled
|
||
assert "would make no changes" in str(
|
||
modal.query_one("#backup-restore-status", Static).render()
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_restore_preview_fills_width_and_routes_navigation_from_cancel(tmp_path):
|
||
"""The safe initial focus does not prevent full mode or two-axis scrolling."""
|
||
app = app_with_filterable_entries()
|
||
backup_path = tmp_path / "hosts.backup"
|
||
hosts_file_path = tmp_path / "hosts"
|
||
current_lines = [
|
||
f"192.0.2.{index} host-{index}.test " + "x" * 120 for index in range(1, 81)
|
||
]
|
||
restored_lines = current_lines.copy()
|
||
restored_lines[39] = "192.0.2.40 restored.test " + "y" * 120
|
||
hosts_file_path.write_text("\n".join(current_lines) + "\n")
|
||
backup_path.write_text("\n".join(restored_lines) + "\n")
|
||
app.edit_mode = True
|
||
app.manager = Mock()
|
||
app.manager.backup_path = backup_path
|
||
app.manager.parser.file_path = hosts_file_path
|
||
|
||
async with app.run_test(size=(160, 40)) as pilot:
|
||
app.action_restore_backup()
|
||
await pilot.pause()
|
||
await pilot.pause()
|
||
|
||
modal = app.screen
|
||
assert isinstance(modal, BackupRestoreModal)
|
||
container = modal.query_one("#backup-restore-container")
|
||
assert container.outer_size.width == 156
|
||
|
||
changed_diff = str(modal.query_one("#backup-diff", Static).render())
|
||
await pilot.press("f")
|
||
await pilot.pause()
|
||
full_diff = str(modal.query_one("#backup-diff", Static).render())
|
||
assert full_diff != changed_diff
|
||
assert "host-1.test" in full_diff
|
||
|
||
scroller = modal.query_one("#backup-diff-scroll")
|
||
assert app.focused is modal.query_one("#cancel-button")
|
||
await pilot.press("down", "right")
|
||
await pilot.pause()
|
||
assert scroller.scroll_y > 0
|
||
assert scroller.scroll_x > 0
|
||
|
||
|
||
class TestPrivilegedModeAuthorization:
|
||
"""Test the user-visible privileged-mode authorization flow."""
|
||
|
||
def test_sudo_notice_has_a_framed_and_a_narrow_terminal_variant(self):
|
||
"""The notice keeps its safety language when there is no room for a frame."""
|
||
framed = render_sudo_authentication_notice(width=80, color=False)
|
||
narrow = render_sudo_authentication_notice(width=40, color=False)
|
||
|
||
for notice in (framed, narrow):
|
||
assert "PRIVILEGED MODE REQUESTED" in notice
|
||
assert "sudo authentication required" in notice
|
||
assert "does not collect or store your credentials" in notice
|
||
assert "Ctrl+C to cancel and remain in Read-only Mode" in notice
|
||
|
||
assert "+--------------------------------------------------+" in framed
|
||
assert "+--------------------------------------------------+" not in narrow
|
||
|
||
def test_sudo_notice_uses_the_alternate_screen_for_the_entire_conversation(self):
|
||
"""The shell viewport is restored after sudo/PAM is finished."""
|
||
output = StringIO()
|
||
|
||
with sudo_authentication_screen(stream=output, width=80):
|
||
output.write("sudo output\n")
|
||
|
||
rendered = output.getvalue()
|
||
assert rendered.startswith("\x1b[?1049h")
|
||
assert "PRIVILEGED MODE REQUESTED" in rendered
|
||
assert "sudo output" in rendered
|
||
assert rendered.endswith("\x1b[0m\x1b[?1049l")
|
||
|
||
def test_sudo_screen_turns_sigint_into_a_catchable_keyboard_interrupt(self):
|
||
"""Asyncio's SIGINT handler must not cancel the Textual main task."""
|
||
output = StringIO()
|
||
previous_handler = signal.getsignal(signal.SIGINT)
|
||
runner_handler_called = False
|
||
|
||
def runner_handler(_signum, _frame):
|
||
nonlocal runner_handler_called
|
||
runner_handler_called = True
|
||
|
||
signal.signal(signal.SIGINT, runner_handler)
|
||
try:
|
||
with pytest.raises(KeyboardInterrupt):
|
||
with sudo_authentication_screen(stream=output, width=80):
|
||
os.kill(os.getpid(), signal.SIGINT)
|
||
finally:
|
||
signal.signal(signal.SIGINT, previous_handler)
|
||
|
||
assert not runner_handler_called
|
||
|
||
def test_interactive_authorization_suspends_and_restores_the_tui(self):
|
||
"""Test foreground PAM authorization runs while Textual is suspended."""
|
||
events = []
|
||
|
||
@contextmanager
|
||
def suspended_tui():
|
||
events.append("suspended")
|
||
yield
|
||
events.append("restored")
|
||
|
||
@contextmanager
|
||
def authentication_screen():
|
||
assert events == ["suspended"]
|
||
events.append("notice shown")
|
||
yield
|
||
events.append("notice dismissed")
|
||
|
||
def authorize_interactively():
|
||
assert events == ["suspended", "notice shown"]
|
||
events.append("authorized")
|
||
return True, "Sudo access granted"
|
||
|
||
def finish_entering_edit_mode():
|
||
assert events == [
|
||
"suspended",
|
||
"notice shown",
|
||
"authorized",
|
||
"notice dismissed",
|
||
"restored",
|
||
]
|
||
events.append("edit mode enabled")
|
||
return True, "Edit mode enabled"
|
||
|
||
app = HostsManagerApp()
|
||
app.manager = Mock()
|
||
app.manager.enter_edit_mode.return_value = (
|
||
False,
|
||
"Interactive authorization required",
|
||
)
|
||
app.manager.authorize_interactively.side_effect = authorize_interactively
|
||
app.manager.finish_entering_edit_mode.side_effect = finish_entering_edit_mode
|
||
app.suspend = Mock(return_value=suspended_tui())
|
||
app.update_status = Mock()
|
||
|
||
with patch(
|
||
"src.hosts.tui.app.sudo_authentication_screen", authentication_screen
|
||
):
|
||
app.action_toggle_edit_mode()
|
||
|
||
assert app.edit_mode
|
||
app.suspend.assert_called_once_with()
|
||
app.manager.enter_edit_mode.assert_called_once_with()
|
||
app.manager.authorize_interactively.assert_called_once_with()
|
||
app.manager.finish_entering_edit_mode.assert_called_once_with()
|
||
assert events == [
|
||
"suspended",
|
||
"notice shown",
|
||
"authorized",
|
||
"notice dismissed",
|
||
"restored",
|
||
"edit mode enabled",
|
||
]
|
||
app.update_status.assert_called_once_with("Edit mode enabled")
|
||
|
||
def test_rejected_authorization_keeps_the_app_read_only(self):
|
||
"""Test a rejected PAM conversation reports the required message."""
|
||
app = HostsManagerApp()
|
||
app.manager = Mock()
|
||
app.manager.enter_edit_mode.return_value = (
|
||
False,
|
||
"Interactive authorization required",
|
||
)
|
||
app.manager.authorize_interactively.return_value = (
|
||
False,
|
||
"Authorization was not granted",
|
||
)
|
||
app.suspend = Mock(return_value=suspended_tui())
|
||
app.update_status = Mock()
|
||
|
||
app.action_toggle_edit_mode()
|
||
|
||
assert not app.edit_mode
|
||
app.update_status.assert_called_once_with(
|
||
"Authorization was not granted; remaining in Read-only Mode."
|
||
)
|
||
|
||
def test_interrupted_authorization_keeps_the_app_read_only(self):
|
||
"""Test interruption during PAM authorization restores read-only mode."""
|
||
events = []
|
||
|
||
@contextmanager
|
||
def suspended_tui():
|
||
events.append("suspended")
|
||
yield
|
||
events.append("restored")
|
||
|
||
app = HostsManagerApp()
|
||
app.manager = Mock()
|
||
app.manager.enter_edit_mode.return_value = (
|
||
False,
|
||
"Interactive authorization required",
|
||
)
|
||
app.manager.authorize_interactively.side_effect = KeyboardInterrupt()
|
||
app.suspend = Mock(return_value=suspended_tui())
|
||
app.update_status = Mock()
|
||
|
||
app.action_toggle_edit_mode()
|
||
|
||
assert not app.edit_mode
|
||
assert events == ["suspended", "restored"]
|
||
app.update_status.assert_called_once_with(
|
||
"Authorization was not granted; remaining in Read-only Mode."
|
||
)
|
||
|
||
def test_interrupt_while_dismissing_the_sudo_notice_keeps_the_app_running(self):
|
||
"""An interrupt during terminal cleanup must not escape and quit the app."""
|
||
app = HostsManagerApp()
|
||
app.manager = Mock()
|
||
app.manager.enter_edit_mode.return_value = (
|
||
False,
|
||
"Interactive authorization required",
|
||
)
|
||
app.manager.authorize_interactively.return_value = (
|
||
False,
|
||
"Authorization was not granted",
|
||
)
|
||
app.suspend = Mock(return_value=suspended_tui())
|
||
app.update_status = Mock()
|
||
|
||
@contextmanager
|
||
def interrupted_notice():
|
||
yield
|
||
raise KeyboardInterrupt
|
||
|
||
with patch("src.hosts.tui.app.sudo_authentication_screen", interrupted_notice):
|
||
app.action_toggle_edit_mode()
|
||
|
||
assert not app.edit_mode
|
||
app.update_status.assert_called_once_with(
|
||
"Authorization was not granted; remaining in Read-only Mode."
|
||
)
|
||
|
||
def test_unsupported_suspension_keeps_the_app_read_only(self):
|
||
"""Test the user receives an actionable suspension failure message."""
|
||
app = HostsManagerApp()
|
||
app.manager = Mock()
|
||
app.manager.enter_edit_mode.return_value = (
|
||
False,
|
||
"Interactive authorization required",
|
||
)
|
||
app.suspend = Mock(side_effect=SuspendNotSupported())
|
||
app.update_status = Mock()
|
||
|
||
app.action_toggle_edit_mode()
|
||
|
||
assert not app.edit_mode
|
||
app.update_status.assert_called_once_with(
|
||
"Interactive authorization requires terminal suspension; remaining in Read-only Mode."
|
||
)
|
||
|
||
@patch("src.hosts.core.manager.subprocess.run")
|
||
def test_leaving_privileged_mode_preserves_the_sudo_timestamp(self, mock_run):
|
||
"""Test Ctrl+E clears app state without invalidating sudo globally."""
|
||
app = HostsManagerApp()
|
||
app.edit_mode = True
|
||
app.manager.edit_mode = True
|
||
app.manager.permission_manager.has_sudo = True
|
||
app.manager.permission_manager._sudo_validated = True
|
||
app.update_status = Mock()
|
||
|
||
app.action_toggle_edit_mode()
|
||
|
||
assert not app.edit_mode
|
||
assert not app.manager.edit_mode
|
||
assert not app.manager.permission_manager.has_sudo
|
||
assert not app.manager.permission_manager._sudo_validated
|
||
mock_run.assert_not_called()
|
||
app.update_status.assert_called_once_with("Edit mode disabled")
|
||
|
||
|
||
class TestPreEditBackupRestoration:
|
||
"""Test the user-reachable Pre-edit Backup recovery flow."""
|
||
|
||
def test_restore_action_confirms_paths_then_reloads_after_success(self, tmp_path):
|
||
app = HostsManagerApp()
|
||
app.edit_mode = True
|
||
app.manager = Mock()
|
||
backup_path = tmp_path / "hosts.backup.123"
|
||
hosts_file_path = tmp_path / "hosts"
|
||
backup_path.write_text("127.0.0.1 restored.test\n")
|
||
hosts_file_path.write_text("127.0.0.1 current.test\n")
|
||
app.manager.backup_path = backup_path
|
||
app.manager.parser.file_path = hosts_file_path
|
||
app.manager.restore_backup.return_value = (True, "Backup restored successfully")
|
||
app.push_screen = Mock()
|
||
app.load_hosts_file = Mock()
|
||
app.update_status = Mock()
|
||
|
||
app.run_worker = Mock(side_effect=lambda coroutine, **_: asyncio.run(coroutine))
|
||
app.action_restore_backup()
|
||
|
||
modal, callback = app.push_screen.call_args.args
|
||
assert modal.preview.backup_path == backup_path
|
||
assert modal.preview.hosts_file_path == hosts_file_path
|
||
|
||
callback(True)
|
||
|
||
app.manager.restore_backup.assert_called_once_with(
|
||
modal.preview.backup_digest, modal.preview.hosts_file_digest
|
||
)
|
||
app.load_hosts_file.assert_called_once_with()
|
||
assert app.update_status.call_args.args == (
|
||
f"Pre-edit Backup restored from {backup_path} to {hosts_file_path}",
|
||
)
|
||
|
||
def test_failed_restore_keeps_interface_state_and_reports_paths(self, tmp_path):
|
||
app = HostsManagerApp()
|
||
app.edit_mode = True
|
||
app.manager = Mock()
|
||
backup_path = tmp_path / "hosts.backup.123"
|
||
hosts_file_path = tmp_path / "hosts"
|
||
backup_path.write_text("127.0.0.1 restored.test\n")
|
||
hosts_file_path.write_text("127.0.0.1 current.test\n")
|
||
app.manager.backup_path = backup_path
|
||
app.manager.parser.file_path = hosts_file_path
|
||
app.manager.restore_backup.return_value = (False, "Permission denied")
|
||
app.push_screen = Mock()
|
||
app.load_hosts_file = Mock()
|
||
app.update_status = Mock()
|
||
|
||
app.run_worker = Mock(side_effect=lambda coroutine, **_: asyncio.run(coroutine))
|
||
app.action_restore_backup()
|
||
_, callback = app.push_screen.call_args.args
|
||
callback(True)
|
||
|
||
app.load_hosts_file.assert_not_called()
|
||
assert app.update_status.call_args.args == (
|
||
f"Failed to restore Pre-edit Backup from {backup_path} to {hosts_file_path}: Permission denied. "
|
||
"The Hosts File and interface are unchanged. Check write access and try again.",
|
||
)
|
||
|
||
def test_restore_is_unavailable_while_the_entry_editor_has_unsaved_state(self):
|
||
app = HostsManagerApp()
|
||
app.entry_edit_mode = True
|
||
app.manager = Mock()
|
||
app.push_screen = Mock()
|
||
app.update_status = Mock()
|
||
|
||
app.action_restore_backup()
|
||
|
||
app.push_screen.assert_not_called()
|
||
app.manager.restore_backup.assert_not_called()
|
||
app.update_status.assert_called_once_with(
|
||
"Finish or cancel the Entry Editor before restoring the Pre-edit Backup."
|
||
)
|