Enhance :: Test suite

.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: |

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:

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.
 

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):

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."""
 

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."""

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."""

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."""