280 lines
9.6 KiB
Python
280 lines
9.6 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 HelpPanel, 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
|
|
|
|
|
|
@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 Filter entries" in str(footer_text)
|
|
|
|
app.action_help()
|
|
await pilot.pause()
|
|
assert isinstance(app.query_one(HelpPanel), HelpPanel)
|
|
assert any(
|
|
binding.action == "show_filters" and binding.description == "Filter entries"
|
|
for _, binding, _, _ in app.screen.active_bindings.values()
|
|
)
|
|
app.action_help()
|
|
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_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",
|
|
"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",
|
|
]
|
|
|
|
|
|
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")
|