phg
0b7b52521f
Reviewed-on: #15 Co-authored-by: phg <mail@philip-henning.com> Co-committed-by: phg <mail@philip-henning.com>
371 lines
13 KiB
Python
371 lines
13 KiB
Python
"""Tests for application-level authorization flow."""
|
||
|
||
from contextlib import contextmanager
|
||
from unittest.mock import Mock, patch
|
||
|
||
import pytest
|
||
from textual.app import SuspendNotSupported
|
||
from textual.widgets import 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.filter_modal import FilterModal
|
||
from src.hosts.tui.help_modal import HelpModal
|
||
|
||
|
||
@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
|
||
|
||
|
||
@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_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()
|
||
|
||
|
||
class TestPrivilegedModeAuthorization:
|
||
"""Test the user-visible privileged-mode authorization flow."""
|
||
|
||
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")
|
||
|
||
def authorize_interactively():
|
||
assert events == ["suspended"]
|
||
events.append("authorized")
|
||
return True, "Sudo access granted"
|
||
|
||
def finish_entering_edit_mode():
|
||
assert events == ["suspended", "authorized", "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()
|
||
|
||
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", "authorized", "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_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")
|