hosts/tests/test_manager.py

707 lines
25 KiB
Python

"""
Tests for the hosts manager module.
This module tests permission management, edit mode operations,
and safe file modifications with backup and validation.
"""
import pytest
import tempfile
import subprocess
from pathlib import Path
from unittest.mock import Mock, patch
from src.hosts.core.manager import PermissionManager, HostsManager
from src.hosts.core.models import HostEntry, HostsFile
from src.hosts.core.commands import ToggleEntryCommand
class TestPermissionManager:
"""Test the PermissionManager class."""
def test_init(self):
"""Test PermissionManager initialization."""
pm = PermissionManager()
assert not pm.has_sudo
assert not pm._sudo_validated
@patch("subprocess.run")
def test_request_sudo_uses_cached_authorization(self, mock_run):
"""Test cached sudo authorization enters without an interaction."""
mock_run.return_value = Mock(returncode=0)
pm = PermissionManager()
success, message = pm.request_sudo()
assert success
assert "already available" in message
assert pm.has_sudo
assert pm._sudo_validated
mock_run.assert_called_once_with(
["sudo", "-n", "-v"], capture_output=True, text=True, timeout=5
)
@patch("subprocess.run")
def test_request_sudo_interactively_uses_foreground_pam(self, mock_run):
"""Test an uncached authorization delegates to foreground sudo."""
mock_run.return_value = Mock(returncode=0)
pm = PermissionManager()
success, message = pm.request_sudo(interactive=True)
assert success
assert "access granted" in message
assert pm.has_sudo
assert pm._sudo_validated
mock_run.assert_called_once_with(["sudo", "-v"])
@patch("subprocess.run")
def test_request_sudo_requires_interaction_when_cache_unavailable(self, mock_run):
"""Test an uncached authorization does not start PAM from the TUI."""
mock_run.return_value = Mock(returncode=1)
pm = PermissionManager()
success, message = pm.request_sudo()
assert not success
assert message == "Interactive authorization required"
assert not pm.has_sudo
assert not pm._sudo_validated
mock_run.assert_called_once_with(
["sudo", "-n", "-v"], capture_output=True, text=True, timeout=5
)
@patch("subprocess.run")
def test_request_sudo_interactively_denied(self, mock_run):
"""Test a rejected PAM conversation keeps sudo unavailable."""
mock_run.return_value = Mock(returncode=1)
pm = PermissionManager()
success, message = pm.request_sudo(interactive=True)
assert not success
assert message == "Authorization was not granted"
assert not pm.has_sudo
assert not pm._sudo_validated
@patch("subprocess.run")
def test_request_sudo_interactively_cancelled(self, mock_run):
"""Test cancellation during PAM authorization keeps sudo unavailable."""
mock_run.return_value = Mock(returncode=130)
pm = PermissionManager()
success, message = pm.request_sudo(interactive=True)
assert not success
assert message == "Authorization was not granted"
assert not pm.has_sudo
assert not pm._sudo_validated
@patch("subprocess.run")
def test_request_sudo_timeout(self, mock_run):
"""Test requesting sudo with timeout."""
mock_run.side_effect = subprocess.TimeoutExpired(["sudo", "-n", "true"], 5)
pm = PermissionManager()
success, message = pm.request_sudo()
assert not success
assert "timed out" in message
assert not pm.has_sudo
@patch("subprocess.run")
def test_request_sudo_exception(self, mock_run):
"""Test requesting sudo with exception."""
mock_run.side_effect = Exception("Test error")
pm = PermissionManager()
success, message = pm.request_sudo()
assert not success
assert "Test error" in message
assert not pm.has_sudo
@patch("subprocess.run")
def test_request_sudo_reports_when_sudo_is_unavailable(self, mock_run):
"""Test a missing sudo executable produces an actionable result."""
mock_run.side_effect = FileNotFoundError()
pm = PermissionManager()
success, message = pm.request_sudo()
assert not success
assert message == "sudo is unavailable; remaining in Read-only Mode"
assert not pm.has_sudo
@patch("subprocess.run")
def test_validate_permissions_success(self, mock_run):
"""Test validating permissions successfully."""
mock_run.return_value = Mock(returncode=0)
pm = PermissionManager()
pm.has_sudo = True
result = pm.validate_permissions("/etc/hosts")
assert result
mock_run.assert_called_once_with(
["sudo", "-n", "test", "-w", "/etc/hosts"], capture_output=True, timeout=5
)
@patch("subprocess.run")
def test_validate_permissions_no_sudo(self, mock_run):
"""Test validating permissions without sudo."""
pm = PermissionManager()
pm.has_sudo = False
result = pm.validate_permissions("/etc/hosts")
assert not result
mock_run.assert_not_called()
@patch("subprocess.run")
def test_validate_permissions_failure(self, mock_run):
"""Test validating permissions failure."""
mock_run.return_value = Mock(returncode=1)
pm = PermissionManager()
pm.has_sudo = True
result = pm.validate_permissions("/etc/hosts")
assert not result
@patch("subprocess.run")
def test_validate_permissions_exception(self, mock_run):
"""Test validating permissions with exception."""
mock_run.side_effect = Exception("Test error")
pm = PermissionManager()
pm.has_sudo = True
result = pm.validate_permissions("/etc/hosts")
assert not result
@patch("subprocess.run")
def test_clear_permission_state_preserves_global_sudo_timestamp(self, mock_run):
"""Test exiting Privileged Mode preserves the global sudo timestamp."""
pm = PermissionManager()
pm.has_sudo = True
pm._sudo_validated = True
pm.clear_permission_state()
assert not pm.has_sudo
assert not pm._sudo_validated
mock_run.assert_not_called()
class TestHostsManager:
"""Test the HostsManager class."""
def test_init(self):
"""Test HostsManager initialization."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
assert not manager.edit_mode
assert manager._backup_path is None
assert manager.parser.file_path == Path(temp_file.name)
@patch("src.hosts.core.manager.HostsManager._create_backup")
def test_enter_edit_mode_success(self, mock_backup):
"""Test entering edit mode successfully."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
# Mock permission manager
manager.permission_manager.request_sudo = Mock(
return_value=(True, "Success")
)
manager.permission_manager.validate_permissions = Mock(return_value=True)
success, message = manager.enter_edit_mode()
assert success
assert "enabled" in message
assert manager.edit_mode
mock_backup.assert_called_once()
def test_enter_edit_mode_already_in_edit(self):
"""Test entering edit mode when already in edit mode."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
success, message = manager.enter_edit_mode()
assert success
assert "Already in edit mode" in message
def test_enter_edit_mode_sudo_failure(self):
"""Test entering edit mode with sudo failure."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
# Mock permission manager failure
manager.permission_manager.request_sudo = Mock(
return_value=(False, "Denied")
)
success, message = manager.enter_edit_mode()
assert not success
assert message == "Denied"
assert not manager.edit_mode
def test_enter_edit_mode_permission_validation_failure(self):
"""Test entering edit mode with permission validation failure."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
# Mock permission manager
manager.permission_manager.request_sudo = Mock(
return_value=(True, "Success")
)
manager.permission_manager.validate_permissions = Mock(return_value=False)
success, message = manager.enter_edit_mode()
assert not success
assert "Cannot write to hosts file" in message
assert not manager.edit_mode
@patch("src.hosts.core.manager.HostsManager._create_backup")
def test_enter_edit_mode_backup_failure(self, mock_backup):
"""Test entering edit mode with backup failure."""
mock_backup.side_effect = Exception("Backup failed")
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
# Mock permission manager
manager.permission_manager.request_sudo = Mock(
return_value=(True, "Success")
)
manager.permission_manager.validate_permissions = Mock(return_value=True)
success, message = manager.enter_edit_mode()
assert not success
assert "Failed to create backup" in message
assert not manager.edit_mode
def test_exit_edit_mode_success(self):
"""Test exit clears the Privileged Mode session state."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
manager._backup_path = Path("/tmp/backup")
hosts_file = HostsFile(
entries=[
HostEntry("192.0.2.1", ["example.test"]),
HostEntry("192.0.2.2", ["example-two.test"]),
]
)
manager.undo_redo_history.execute_command(ToggleEntryCommand(0), hosts_file)
manager.undo_redo_history.execute_command(ToggleEntryCommand(1), hosts_file)
manager.undo_redo_history.undo(hosts_file)
assert manager.can_undo()
assert manager.can_redo()
# Mock permission manager
manager.permission_manager.clear_permission_state = Mock()
success, message = manager.exit_edit_mode()
assert success
assert "disabled" in message
assert not manager.edit_mode
assert manager._backup_path is None
assert not manager.can_undo()
assert not manager.can_redo()
manager.permission_manager.clear_permission_state.assert_called_once()
def test_exit_edit_mode_not_in_edit(self):
"""Test exiting edit mode when not in edit mode."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = False
success, message = manager.exit_edit_mode()
assert success
assert "Already in read-only mode" in message
def test_exit_edit_mode_exception(self):
"""Test exiting edit mode with exception."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
# Mock permission manager to raise exception
manager.permission_manager.clear_permission_state = Mock(
side_effect=Exception("Test error")
)
success, message = manager.exit_edit_mode()
assert not success
assert "Test error" in message
def test_toggle_entry_success(self):
"""Test toggling entry successfully."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
hosts_file = HostsFile()
entry = HostEntry(
"192.168.1.1", ["router"], is_active=True
) # Non-default entry
hosts_file.entries.append(entry)
success, message = manager.toggle_entry(hosts_file, 0)
assert success
assert "active to inactive" in message
assert not hosts_file.entries[0].is_active
def test_toggle_entry_not_in_edit_mode(self):
"""Test toggling entry when not in edit mode."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = False
hosts_file = HostsFile()
success, message = manager.toggle_entry(hosts_file, 0)
assert not success
assert "Not in edit mode" in message
def test_toggle_entry_invalid_index(self):
"""Test toggling entry with invalid index."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
hosts_file = HostsFile()
success, message = manager.toggle_entry(hosts_file, 0)
assert not success
assert "Invalid entry index" in message
def test_move_entry_up_success(self):
"""Test moving entry up successfully."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
hosts_file = HostsFile()
entry1 = HostEntry("10.0.0.1", ["test1"]) # Non-default entries
entry2 = HostEntry("192.168.1.1", ["router"])
hosts_file.entries.extend([entry1, entry2])
success, message = manager.move_entry_up(hosts_file, 1)
assert success
assert "moved up" in message
assert hosts_file.entries[0].hostnames[0] == "router"
assert hosts_file.entries[1].hostnames[0] == "test1"
def test_move_entry_up_invalid_index(self):
"""Test moving entry up with invalid index."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
hosts_file = HostsFile()
entry = HostEntry("127.0.0.1", ["localhost"])
hosts_file.entries.append(entry)
success, message = manager.move_entry_up(hosts_file, 0)
assert not success
assert "Cannot move entry up" in message
def test_move_entry_down_success(self):
"""Test moving entry down successfully."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
hosts_file = HostsFile()
entry1 = HostEntry("10.0.0.1", ["test1"]) # Non-default entries
entry2 = HostEntry("192.168.1.1", ["router"])
hosts_file.entries.extend([entry1, entry2])
success, message = manager.move_entry_down(hosts_file, 0)
assert success
assert "moved down" in message
assert hosts_file.entries[0].hostnames[0] == "router"
assert hosts_file.entries[1].hostnames[0] == "test1"
def test_move_entry_down_invalid_index(self):
"""Test moving entry down with invalid index."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
hosts_file = HostsFile()
entry = HostEntry("127.0.0.1", ["localhost"])
hosts_file.entries.append(entry)
success, message = manager.move_entry_down(hosts_file, 0)
assert not success
assert "Cannot move entry down" in message
def test_update_entry_success(self):
"""Test updating entry successfully."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
hosts_file = HostsFile()
entry = HostEntry("10.0.0.1", ["test"]) # Non-default entry
hosts_file.entries.append(entry)
success, message = manager.update_entry(
hosts_file, 0, "192.168.1.1", ["newhost"], "New comment"
)
assert success
assert "updated successfully" in message
assert hosts_file.entries[0].ip_address == "192.168.1.1"
assert hosts_file.entries[0].hostnames == ["newhost"]
assert hosts_file.entries[0].comment == "New comment"
def test_update_entry_invalid_data(self):
"""Test updating entry with invalid data."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
hosts_file = HostsFile()
entry = HostEntry(
"127.0.0.1", ["localhost"]
) # Default entry - cannot be modified
hosts_file.entries.append(entry)
success, message = manager.update_entry(
hosts_file, 0, "invalid-ip", ["newhost"]
)
assert not success
assert "Cannot modify default system entries" in message
@patch("tempfile.NamedTemporaryFile")
@patch("subprocess.run")
@patch("os.unlink")
def test_save_hosts_file_success(self, mock_unlink, mock_run, mock_temp):
"""Test saving hosts file successfully."""
# Mock temporary file
mock_temp_file = Mock()
mock_temp_file.name = "/tmp/test.hosts"
mock_temp_file.__enter__ = Mock(return_value=mock_temp_file)
mock_temp_file.__exit__ = Mock(return_value=None)
mock_temp.return_value = mock_temp_file
# Mock subprocess success
mock_run.return_value = Mock(returncode=0)
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
manager.permission_manager.has_sudo = True
hosts_file = HostsFile()
entry = HostEntry("127.0.0.1", ["localhost"])
hosts_file.entries.append(entry)
success, message = manager.save_hosts_file(hosts_file)
assert success
assert "saved successfully" in message
mock_run.assert_called_once()
mock_unlink.assert_called_once_with("/tmp/test.hosts")
def test_save_hosts_file_not_in_edit_mode(self):
"""Test saving hosts file when not in edit mode."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = False
hosts_file = HostsFile()
success, message = manager.save_hosts_file(hosts_file)
assert not success
assert "Not in edit mode" in message
def test_save_hosts_file_no_sudo(self):
"""Test saving hosts file without sudo."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
manager.permission_manager.has_sudo = False
hosts_file = HostsFile()
success, message = manager.save_hosts_file(hosts_file)
assert not success
assert "No sudo permissions" in message
@patch("subprocess.run")
def test_restore_backup_success(self, mock_run):
"""Test restoring backup successfully."""
mock_run.return_value = Mock(returncode=0)
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
# Create a mock backup file
with tempfile.NamedTemporaryFile(delete=False) as backup_file:
manager._backup_path = Path(backup_file.name)
try:
success, message = manager.restore_backup()
assert success
assert "restored successfully" in message
mock_run.assert_called_once()
finally:
# Clean up
manager._backup_path.unlink()
def test_restore_backup_not_in_edit_mode(self):
"""Test restoring backup when not in edit mode."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = False
success, message = manager.restore_backup()
assert not success
assert "Not in edit mode" in message
def test_restore_backup_no_backup(self):
"""Test restoring backup when no backup exists."""
with tempfile.NamedTemporaryFile() as temp_file:
manager = HostsManager(temp_file.name)
manager.edit_mode = True
manager._backup_path = None
success, message = manager.restore_backup()
assert not success
assert "No backup available" in message
@patch("subprocess.run")
@patch("tempfile.gettempdir")
@patch("time.time")
def test_create_backup_success(self, mock_time, mock_tempdir, mock_run):
"""Test creating backup successfully."""
mock_time.return_value = 1234567890
mock_tempdir.return_value = "/tmp"
mock_run.side_effect = [
Mock(returncode=0), # cp command
Mock(returncode=0), # chmod command
]
# Create a real temporary file for testing
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_file.write(b"test content")
temp_path = temp_file.name
try:
manager = HostsManager(temp_path)
manager._create_backup()
expected_backup = Path("/tmp/hosts-manager-backups/hosts.backup.1234567890")
assert manager._backup_path == expected_backup
assert mock_run.call_count == 2
finally:
# Clean up
Path(temp_path).unlink()
@patch("subprocess.run")
def test_create_backup_failure(self, mock_run):
"""Test creating backup with failure."""
mock_run.return_value = Mock(returncode=1, stderr="Permission denied")
# Create a real temporary file for testing
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_file.write(b"test content")
temp_path = temp_file.name
try:
manager = HostsManager(temp_path)
with pytest.raises(Exception) as exc_info:
manager._create_backup()
assert "Failed to create backup" in str(exc_info.value)
finally:
# Clean up
Path(temp_path).unlink()
def test_add_entry_success(self):
"""Test successfully adding an entry with immediate save."""
manager = HostsManager()
manager.edit_mode = True
# Mock the save operation to succeed
with patch.object(manager, "save_hosts_file") as mock_save:
mock_save.return_value = (True, "File saved successfully")
hosts_file = HostsFile()
test_entry = HostEntry(ip_address="192.168.1.100", hostnames=["testhost"])
success, message = manager.add_entry(hosts_file, test_entry)
assert success
assert "Entry added: testhost" in message
assert len(hosts_file.entries) == 1
assert hosts_file.entries[0] == test_entry
mock_save.assert_called_once_with(hosts_file)
def test_add_entry_save_failure(self):
"""Test adding an entry when save fails."""
manager = HostsManager()
manager.edit_mode = True
# Mock the save operation to fail
with patch.object(manager, "save_hosts_file") as mock_save:
mock_save.return_value = (False, "Permission denied")
hosts_file = HostsFile()
test_entry = HostEntry(ip_address="192.168.1.100", hostnames=["testhost"])
success, message = manager.add_entry(hosts_file, test_entry)
assert not success
assert "Failed to save after adding entry" in message
assert (
len(hosts_file.entries) == 0
) # Entry should be removed on save failure
mock_save.assert_called_once_with(hosts_file)
def test_add_entry_not_in_edit_mode(self):
"""Test adding an entry when not in edit mode."""
manager = HostsManager()
# edit_mode defaults to False
hosts_file = HostsFile()
test_entry = HostEntry(ip_address="192.168.1.100", hostnames=["testhost"])
success, message = manager.add_entry(hosts_file, test_entry)
assert not success
assert "Not in edit mode" in message
assert len(hosts_file.entries) == 0