diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 5c33375..920782a 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -43,10 +43,11 @@ jobs: run: | python -m pytest tests/ --cov=dnx --cov-report=term - name: Upload coverage to Codecov - uses: codecov/codecov-action@v4 + uses: codecov/codecov-action@v5 with: - fail_ci_if_error: true token: ${{ secrets.CODECOV_TOKEN }} + fail_ci_if_error: true + slug: openscilab/dnx if: matrix.python-version == env.TEST_PYTHON_VERSION && matrix.os == env.TEST_OS - name: Vulture, Bandit and Pydocstyle tests run: | diff --git a/dnx/dns.py b/dnx/dns.py index beff670..5d4bf32 100644 --- a/dnx/dns.py +++ b/dnx/dns.py @@ -49,7 +49,7 @@ def require_admin() -> None: Raises: AdminRequiredError: If not running with sufficient privileges. """ - if os.name != "nt": + if get_platform() != Platform.WINDOWS: if os.geteuid() != 0: raise AdminRequiredError("Please run as root (sudo)") else: diff --git a/dnx/ping.py b/dnx/ping.py index bbf6c73..5d45ef2 100644 --- a/dnx/ping.py +++ b/dnx/ping.py @@ -13,6 +13,7 @@ from .dns import Platform, get_platform from .exceptions import CommandFailedError +from .params import DEFAULT_PING_COUNT, DEFAULT_PING_TIMEOUT @dataclass @@ -167,7 +168,7 @@ def _parse_ping_output_windows(output: str) -> dict: return result -def ping_server(ip: str, count: int = 3, timeout: int = 5) -> PingResult: +def ping_server(ip: str, count: int = DEFAULT_PING_COUNT, timeout: int = DEFAULT_PING_TIMEOUT) -> PingResult: """ Ping a DNS server and return latency statistics. @@ -232,7 +233,7 @@ def ping_server(ip: str, count: int = 3, timeout: int = 5) -> PingResult: ) -def ping_servers(servers: List[str], count: int = 3) -> List[PingResult]: +def ping_servers(servers: List[str], count: int = DEFAULT_PING_COUNT) -> List[PingResult]: """ Ping multiple DNS servers. diff --git a/tests/conftest.py b/tests/conftest.py index 26e8e81..7327bd9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,9 +1,28 @@ # -*- coding: utf-8 -*- """Pytest configuration and fixtures.""" -import platform +import os +import subprocess + import pytest +from dnx.dns import Platform, get_platform + + +def _has_admin() -> bool: + """Return True if the current process has admin/root privileges.""" + if get_platform() == Platform.WINDOWS: + try: + subprocess.check_call( + ["net", "session"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + return True + except (subprocess.CalledProcessError, FileNotFoundError): + return False + return os.geteuid() == 0 + def pytest_configure(config): """Register custom markers.""" @@ -16,24 +35,23 @@ def pytest_configure(config): def pytest_collection_modifyitems(config, items): - """Skip tests based on platform markers.""" - current_platform = platform.system().lower() - - platform_map = { - "linux": "linux", - "darwin": "macos", - "windows": "windows", - } - current = platform_map.get(current_platform) + """Skip tests based on platform and privilege markers.""" + current = get_platform() + admin = _has_admin() for item in items: - if "linux" in item.keywords and current != "linux": + if "linux" in item.keywords and current != Platform.LINUX: item.add_marker(pytest.mark.skip(reason="Test requires Linux")) - elif "windows" in item.keywords and current != "windows": + elif "windows" in item.keywords and current != Platform.WINDOWS: item.add_marker(pytest.mark.skip(reason="Test requires Windows")) - elif "macos" in item.keywords and current != "macos": + elif "macos" in item.keywords and current != Platform.MACOS: item.add_marker(pytest.mark.skip(reason="Test requires macOS")) + if "requires_admin" in item.keywords and not admin: + item.add_marker( + pytest.mark.skip(reason="Test requires admin/root privileges") + ) + @pytest.fixture def sample_resolv_conf(tmp_path): diff --git a/tests/test_cli.py b/tests/test_cli.py index e1ed27b..a277eb9 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -6,7 +6,9 @@ from io import StringIO from unittest.mock import patch, MagicMock from dnx.cli import main +from dnx.exceptions import DNXError from dnx.params import DNS_PRESETS, DNX_VERSION +from dnx.ping import PingResult class TestCLIVersion: @@ -178,6 +180,81 @@ def test_iface_passed_to_backend(self): mock_get_backend.assert_called_once_with("eth0") +class TestCLIListLatency: + """Tests for 'dnx list --latency' command.""" + + def test_list_latency_shows_results(self): + """Verify --latency pings every preset and prints results.""" + fake_result = PingResult( + ip="8.8.8.8", reachable=True, avg_ms=10.0, loss_percent=0.0 + ) + with patch.object(sys, "argv", ["dnx", "list", "--latency"]): + with patch("dnx.cli.ping_servers", return_value=[fake_result]): + with patch("sys.stdout", new_callable=StringIO) as mock_stdout: + main() + output = mock_stdout.getvalue() + + assert "Checking latency" in output + for name in DNS_PRESETS: + assert name in output + + +class TestCLICurrentLatency: + """Tests for 'dnx current --latency' command.""" + + def test_current_latency_shows_results(self): + """Verify --latency pings current servers and prints latency.""" + mock_backend = MagicMock() + mock_backend.get_dns.return_value = ["8.8.8.8"] + fake_result = PingResult( + ip="8.8.8.8", reachable=True, avg_ms=12.0, loss_percent=0.0 + ) + + with patch.object(sys, "argv", ["dnx", "current", "--latency"]): + with patch("dnx.cli.get_backend", return_value=mock_backend): + with patch("dnx.cli.ping_servers", return_value=[fake_result]): + with patch("sys.stdout", new_callable=StringIO) as mock_stdout: + main() + output = mock_stdout.getvalue() + + assert "Latency" in output + assert "8.8.8.8" in output + + +class TestCLIKeyboardInterrupt: + """Tests for CLI KeyboardInterrupt handling.""" + + def test_keyboard_interrupt_exits_130(self): + """Verify KeyboardInterrupt during execution exits with code 130.""" + with patch.object(sys, "argv", ["dnx", "current"]): + with patch("dnx.cli.get_backend", side_effect=KeyboardInterrupt): + with patch("sys.stdout", new_callable=StringIO): + with pytest.raises(SystemExit) as exc_info: + main() + assert exc_info.value.code == 130 + + def test_dnx_error_exits_1(self): + """Verify DNXError during execution exits with code 1.""" + with patch.object(sys, "argv", ["dnx", "current"]): + with patch("dnx.cli.get_backend", side_effect=DNXError("boom")): + with pytest.raises(SystemExit) as exc_info: + main() + assert exc_info.value.code == 1 + + +class TestCLICurrentEndToEnd: + """End-to-end tests for 'dnx current' (no admin required).""" + + def test_current_end_to_end(self): + """Run 'dnx current' with real backend -- reads DNS without admin.""" + with patch.object(sys, "argv", ["dnx", "current"]): + with patch("sys.stdout", new_callable=StringIO) as mock_stdout: + main() + output = mock_stdout.getvalue() + + assert "Current DNS servers" in output or "No DNS servers" in output + + class TestCLIErrorHandling: """Tests for CLI error handling.""" diff --git a/tests/test_linux.py b/tests/test_linux.py index 33697c5..bda8102 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -6,9 +6,11 @@ from unittest.mock import patch, MagicMock from dnx.dns import ( + DNSBackend, ResolvConfDNS, SystemdResolvedDNS, NetworkManagerDNS, + get_backend, get_linux_backend, _is_systemd_resolved_active, _is_networkmanager_active, @@ -110,6 +112,23 @@ def test_get_dns_parses_resolvectl_output(self): assert "8.8.8.8" in servers assert "8.8.4.4" in servers + def test_get_dns_real(self): + """Read DNS via resolvectl on a system where systemd-resolved is active.""" + if not _is_systemd_resolved_active(): + pytest.skip("systemd-resolved not active") + backend = SystemdResolvedDNS() + servers = backend.get_dns() + assert isinstance(servers, list) + + def test_get_dns_command_failure_returns_empty(self): + """Verify CommandFailedError from resolvectl returns empty list.""" + from dnx.exceptions import CommandFailedError + + with patch("dnx.dns.run_command", side_effect=CommandFailedError("fail")): + backend = SystemdResolvedDNS(iface="eth0") + servers = backend.get_dns() + assert servers == [] + @pytest.mark.linux class TestNetworkManagerDNS: @@ -133,6 +152,26 @@ def test_get_dns_parses_nmcli_output(self): assert servers == ["8.8.8.8", "8.8.4.4"] + def test_get_dns_real(self): + """Read DNS via nmcli on a system where NetworkManager is active.""" + if not _is_networkmanager_active(): + pytest.skip("NetworkManager not active") + backend = NetworkManagerDNS() + servers = backend.get_dns() + assert isinstance(servers, list) + + def test_get_connection_name_real(self): + """Read connection name via nmcli when NetworkManager is active.""" + if not _is_networkmanager_active(): + pytest.skip("NetworkManager not active") + backend = NetworkManagerDNS() + try: + conn_name = backend._get_connection_name() + assert isinstance(conn_name, str) + assert len(conn_name) > 0 + except InterfaceNotFoundError: + pytest.skip("No active NM connection found") + @pytest.mark.linux class TestLinuxBackendFactory: @@ -166,6 +205,24 @@ def test_passes_interface(self): assert backend.iface == "eth1" +@pytest.mark.linux +class TestLinuxGetBackend: + """Tests for get_backend() on Linux.""" + + def test_get_backend_returns_dns_backend(self): + """Verify get_backend() returns a DNSBackend subclass on Linux.""" + backend = get_backend() + assert isinstance(backend, DNSBackend) + assert isinstance( + backend, (ResolvConfDNS, SystemdResolvedDNS, NetworkManagerDNS) + ) + + def test_get_backend_passes_iface(self): + """Verify get_backend() forwards the iface argument.""" + backend = get_backend(iface="eth0") + assert backend.iface == "eth0" + + @pytest.mark.linux class TestSystemDetection: """Tests for system detection functions.""" diff --git a/tests/test_macos.py b/tests/test_macos.py index 7bf357f..fe3080e 100644 --- a/tests/test_macos.py +++ b/tests/test_macos.py @@ -4,7 +4,7 @@ import pytest from unittest.mock import patch, MagicMock -from dnx.dns import MacOSDNS +from dnx.dns import MacOSDNS, DNSBackend, get_backend from dnx.exceptions import InterfaceNotFoundError, ServiceNotFoundError @@ -149,6 +149,22 @@ def test_service_caching(self): assert service == "Cached-Service" +@pytest.mark.macos +class TestMacOSGetBackend: + """Tests for get_backend() on macOS.""" + + def test_get_backend_returns_macos_dns(self): + """Verify get_backend() returns a MacOSDNS instance on macOS.""" + backend = get_backend() + assert isinstance(backend, MacOSDNS) + assert isinstance(backend, DNSBackend) + + def test_get_backend_passes_iface(self): + """Verify get_backend() forwards the iface argument.""" + backend = get_backend(iface="en0") + assert backend.iface == "en0" + + @pytest.mark.macos class TestMacOSDNSIntegration: """Integration tests that actually call macOS APIs.""" diff --git a/tests/test_ping.py b/tests/test_ping.py index 00898f5..52be952 100644 --- a/tests/test_ping.py +++ b/tests/test_ping.py @@ -1,7 +1,11 @@ # -*- coding: utf-8 -*- """Tests for ping functionality.""" +import subprocess + import pytest +from unittest.mock import patch + from dnx.ping import ( PingResult, ping_server, @@ -129,6 +133,45 @@ def test_verify_unreachable(self): assert all_ok is False +class TestPingServerEdgeCases: + """Tests for ping_server error branches.""" + + def test_timeout_returns_unreachable(self): + """Verify TimeoutExpired results in unreachable PingResult.""" + with patch( + "dnx.ping.subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd="ping", timeout=5), + ): + result = ping_server("8.8.8.8", count=1, timeout=1) + + assert result.reachable is False + assert result.error == "Timeout" + assert result.packets_sent == 1 + assert result.loss_percent == 100.0 + + def test_ping_not_found_returns_error(self): + """Verify FileNotFoundError results in error PingResult.""" + with patch( + "dnx.ping.subprocess.run", + side_effect=FileNotFoundError("ping not found"), + ): + result = ping_server("8.8.8.8", count=1) + + assert result.reachable is False + assert "not found" in result.error.lower() + + def test_generic_exception_returns_error(self): + """Verify unexpected exception is caught gracefully.""" + with patch( + "dnx.ping.subprocess.run", + side_effect=OSError("unexpected"), + ): + result = ping_server("8.8.8.8", count=1) + + assert result.reachable is False + assert "unexpected" in result.error + + class TestFormatPingResult: """Tests for formatting ping results.""" diff --git a/tests/test_validation.py b/tests/test_validation.py index bf37fc4..2c2527b 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -1,9 +1,28 @@ # -*- coding: utf-8 -*- """Tests for validation functions.""" +import os +import subprocess + import pytest -from dnx.dns import validate_ips, get_platform, Platform -from dnx.exceptions import InvalidIPError, UnsupportedPlatformError +from unittest.mock import patch, MagicMock + +from dnx.dns import ( + Platform, + validate_ips, + get_platform, + run_command, + require_admin, + DNSBackend, + ResolvConfDNS, +) +from dnx.exceptions import ( + InvalidIPError, + UnsupportedPlatformError, + CommandFailedError, + AdminRequiredError, + DNSOperationError, +) class TestValidateIPs: @@ -86,3 +105,122 @@ def test_platform_has_value(self): """Test that Platform enum has string value.""" result = get_platform() assert result.value in ["linux", "macos", "windows"] + + def test_unsupported_platform_raises(self): + """Test that unsupported OS raises UnsupportedPlatformError.""" + with patch("dnx.dns.platform.system", return_value="FreeBSD"): + with pytest.raises(UnsupportedPlatformError) as exc_info: + get_platform() + assert "freebsd" in str(exc_info.value).lower() + + +class TestRunCommand: + """Tests for run_command error handling.""" + + def test_command_not_found_raises(self): + """Test that a nonexistent command raises CommandFailedError.""" + with pytest.raises(CommandFailedError) as exc_info: + run_command(["__nonexistent_command_dnx_test__"]) + assert "not found" in str(exc_info.value).lower() + + def test_command_failure_raises(self): + """Test that a failing command raises CommandFailedError.""" + with pytest.raises(CommandFailedError): + run_command(["python", "-c", "import sys; sys.exit(1)"]) + + def test_successful_command(self): + """Test that a successful command returns CompletedProcess.""" + result = run_command(["python", "-c", "print('hello')"]) + assert result.returncode == 0 + assert "hello" in result.stdout + + +class TestRequireAdmin: + """Tests for require_admin function.""" + + def test_non_admin_raises_on_unix(self): + """Test that non-root on Unix raises AdminRequiredError.""" + with patch("dnx.dns.get_platform", return_value=Platform.LINUX): + with patch("dnx.dns.os.geteuid", create=True, return_value=1000): + with pytest.raises(AdminRequiredError): + require_admin() + + def test_admin_passes_on_unix(self): + """Test that root on Unix passes without error.""" + with patch("dnx.dns.get_platform", return_value=Platform.LINUX): + with patch("dnx.dns.os.geteuid", create=True, return_value=0): + require_admin() + + def test_non_admin_raises_on_windows(self): + """Test that non-admin on Windows raises AdminRequiredError.""" + with patch("dnx.dns.get_platform", return_value=Platform.WINDOWS): + with patch( + "dnx.dns.subprocess.check_call", + side_effect=subprocess.CalledProcessError(1, "net"), + ): + with pytest.raises(AdminRequiredError): + require_admin() + + def test_admin_passes_on_windows(self): + """Test that admin on Windows passes without error.""" + with patch("dnx.dns.get_platform", return_value=Platform.WINDOWS): + with patch("dnx.dns.subprocess.check_call"): + require_admin() + + +class TestDNSBackendAbstract: + """Tests for DNSBackend base class abstract methods.""" + + def test_get_active_interface_raises(self): + """Test that base class raises NotImplementedError.""" + backend = DNSBackend() + with pytest.raises(NotImplementedError): + backend.get_active_interface() + + def test_get_dns_raises(self): + """Test that base class raises NotImplementedError.""" + backend = DNSBackend() + with pytest.raises(NotImplementedError): + backend.get_dns() + + def test_set_dns_raises(self): + """Test that base class raises NotImplementedError.""" + backend = DNSBackend() + with pytest.raises(NotImplementedError): + backend.set_dns(["8.8.8.8"]) + + def test_reset_dns_raises(self): + """Test that base class raises NotImplementedError.""" + backend = DNSBackend() + with pytest.raises(NotImplementedError): + backend.reset_dns() + + def test_get_interface_returns_iface_when_set(self): + """Test that get_interface returns user-specified iface.""" + backend = DNSBackend(iface="eth0") + assert backend.get_interface() == "eth0" + + def test_get_interface_falls_back_to_active(self): + """Test that get_interface calls get_active_interface when iface is None.""" + backend = DNSBackend() + with patch.object(backend, "get_active_interface", return_value="wlan0"): + assert backend.get_interface() == "wlan0" + + +class TestResolvConfGetDNSPermission: + """Tests for ResolvConfDNS.get_dns permission error handling.""" + + @pytest.mark.linux + def test_get_dns_permission_error(self, tmp_path): + """Test that PermissionError on get_dns raises DNSOperationError.""" + resolv_file = tmp_path / "resolv.conf" + resolv_file.write_text("nameserver 8.8.8.8\n") + os.chmod(resolv_file, 0o000) + + try: + with patch("dnx.dns.LINUX_RESOLV_CONF", str(resolv_file)): + backend = ResolvConfDNS() + with pytest.raises(DNSOperationError): + backend.get_dns() + finally: + os.chmod(resolv_file, 0o644) diff --git a/tests/test_windows.py b/tests/test_windows.py index bc74ad3..c81d221 100644 --- a/tests/test_windows.py +++ b/tests/test_windows.py @@ -4,7 +4,7 @@ import pytest from unittest.mock import patch, MagicMock -from dnx.dns import WindowsDNS +from dnx.dns import WindowsDNS, DNSBackend, get_backend from dnx.exceptions import InterfaceNotFoundError @@ -91,6 +91,22 @@ def test_interface_with_spaces(self): assert "Ethernet 2" in call_args[-1] +@pytest.mark.windows +class TestWindowsGetBackend: + """Tests for get_backend() on Windows.""" + + def test_get_backend_returns_windows_dns(self): + """Verify get_backend() returns a WindowsDNS instance on Windows.""" + backend = get_backend() + assert isinstance(backend, WindowsDNS) + assert isinstance(backend, DNSBackend) + + def test_get_backend_passes_iface(self): + """Verify get_backend() forwards the iface argument.""" + backend = get_backend(iface="Wi-Fi") + assert backend.iface == "Wi-Fi" + + @pytest.mark.windows class TestWindowsDNSIntegration: """Integration tests that actually call Windows APIs."""