test: add tests for network_policy.get_policy_engine singleton (closes #2445)

backend/secuscan/parser_sandbox.py

@@ -189,10 +189,12 @@ def _unshare_net_supported() -> bool:
     _unshare_capability_checked = True
 
     if platform.system() != "Linux":
+        _unshare_available = False
         return False
 
     unshare_path = shutil.which("unshare")
     if not unshare_path:
+        _unshare_available = False
         return False
 
     try:

testing/backend/unit/test_network_policy_engine_singleton.py

@@ -0,0 +1,65 @@
+"""
+Unit tests for get_policy_engine singleton in backend/secuscan/network_policy.py.
+"""
+import pytest
+
+from backend.secuscan.network_policy import get_policy_engine
+
+
+def _reset_singleton():
+    """Reset the module-level singleton so tests get a fresh engine."""
+    import backend.secuscan.network_policy as mod
+    mod._policy_engine = None
+
+
+class TestGetPolicyEngine:
+    """Tests for the get_policy_engine singleton accessor."""
+
+    def test_first_call_creates_new_engine_instance(self):
+        """First call should return a new NetworkPolicyEngine instance."""
+        _reset_singleton()
+        engine = get_policy_engine()
+        assert engine is not None
+        assert hasattr(engine, "check_access")
+        assert hasattr(engine, "add_deny_rule")
+        assert hasattr(engine, "add_allow_rule")
+
+    def test_second_call_returns_same_instance(self):
+        """Subsequent calls must return the exact same instance (singleton)."""
+        _reset_singleton()
+        engine1 = get_policy_engine()
+        engine2 = get_policy_engine()
+        assert engine1 is engine2
+
+    def test_engine_has_expected_public_methods(self):
+        """Engine must expose the documented public API."""
+        _reset_singleton()
+        engine = get_policy_engine()
+        assert callable(engine.check_access)
+        assert callable(engine.add_deny_rule)
+        assert callable(engine.add_allow_rule)
+        assert callable(engine.export_audit_log)
+        assert callable(engine.clear_audit_entries)
+        assert callable(engine.get_audit_entries)
+
+    def test_singleton_persists_after_clear_audit_entries(self):
+        """Clearing audit entries must not replace the engine instance."""
+        _reset_singleton()
+        engine1 = get_policy_engine()
+        engine1.clear_audit_entries()
+        engine2 = get_policy_engine()
+        assert engine1 is engine2
+
+    def test_singleton_behavior_across_multiple_calls(self):
+        """Multiple get_policy_engine calls always return the same object."""
+        _reset_singleton()
+        engines = [get_policy_engine() for _ in range(5)]
+        assert all(e is engines[0] for e in engines)
+
+    def test_engine_is_initialized_with_correct_settings(self):
+        """Engine must be created and have the expected attributes."""
+        _reset_singleton()
+        engine = get_policy_engine()
+        assert engine is not None
+        assert hasattr(engine, "_max_audit_entries")
+        assert engine._max_audit_entries > 0

testing/backend/unit/test_parser_sandbox_unshare_supported.py

@@ -0,0 +1,157 @@
+"""
+Unit tests for _unshare_net_supported in backend/secuscan/parser_sandbox.py.
+
+The parser_sandbox module is importable without conftest fixtures since it only
+depends on platform, shutil, and subprocess (all stdlib).
+"""
+import sys
+from unittest.mock import patch, MagicMock
+
+# Ensure the module is re-imported fresh
+_mod_key = "backend.secuscan.parser_sandbox"
+if _mod_key in sys.modules:
+    del sys.modules[_mod_key]
+
+from backend.secuscan.parser_sandbox import _unshare_net_supported
+
+
+def _reset_module_globals():
+    """Reset module-level cache so the function re-evaluates on next call."""
+    import backend.secuscan.parser_sandbox as mod
+    mod._unshare_capability_checked = False
+    mod._unshare_available = False
+
+
+# ---------------------------------------------------------------------------
+# Non-Linux paths
+# ---------------------------------------------------------------------------
+
+class TestUnshareNetSupportedNonLinux:
+    """On non-Linux platforms, _unshare_net_supported returns False."""
+
+    def test_returns_false_on_darwin(self):
+        _reset_module_globals()
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Darwin"):
+            result = _unshare_net_supported()
+        assert result is False
+
+    def test_returns_false_on_windows(self):
+        _reset_module_globals()
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Windows"):
+            result = _unshare_net_supported()
+        assert result is False
+
+    def test_caches_result_after_non_linux_call(self):
+        """Subsequent calls must not re-evaluate — must return cached value."""
+        _reset_module_globals()
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Darwin"):
+            result1 = _unshare_net_supported()
+        assert result1 is False
+        # Subsequent calls return the cached _unshare_available value (False)
+        result2 = _unshare_net_supported()
+        assert result2 is False
+
+
+# ---------------------------------------------------------------------------
+# Linux path: shutil.which returns None
+# ---------------------------------------------------------------------------
+
+class TestUnshareNetSupportedBinaryNotFound:
+    """If the unshare binary is not in PATH, return False and cache it."""
+
+    def test_returns_false_when_which_returns_none(self):
+        _reset_module_globals()
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Linux"):
+            with patch("backend.secuscan.parser_sandbox.shutil.which", return_value=None):
+                result = _unshare_net_supported()
+        assert result is False
+
+    def test_caches_false_when_binary_not_found(self):
+        _reset_module_globals()
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Linux"):
+            with patch("backend.secuscan.parser_sandbox.shutil.which", return_value=None):
+                result1 = _unshare_net_supported()
+        assert result1 is False
+        # Second call uses cached _unshare_available from first call (False)
+        result2 = _unshare_net_supported()
+        assert result2 is False
+
+
+# ---------------------------------------------------------------------------
+# Linux path: subprocess.run probe fails
+# ---------------------------------------------------------------------------
+
+class TestUnshareNetSupportedProbeFails:
+    """If the unshare probe exits non-zero, return False."""
+
+    def test_returns_false_when_probe_exit_nonzero(self):
+        _reset_module_globals()
+        mock_proc = MagicMock()
+        mock_proc.returncode = 1
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Linux"):
+            with patch("backend.secuscan.parser_sandbox.shutil.which", return_value="/usr/bin/unshare"):
+                with patch("backend.secuscan.parser_sandbox.subprocess.run", return_value=mock_proc):
+                    result = _unshare_net_supported()
+        assert result is False
+
+    def test_caches_false_when_probe_fails(self):
+        _reset_module_globals()
+        mock_proc = MagicMock()
+        mock_proc.returncode = 1
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Linux"):
+            with patch("backend.secuscan.parser_sandbox.shutil.which", return_value="/usr/bin/unshare"):
+                with patch("backend.secuscan.parser_sandbox.subprocess.run", return_value=mock_proc):
+                    result1 = _unshare_net_supported()
+        assert result1 is False
+        # Second call must use the cached False, not re-run subprocess
+        result2 = _unshare_net_supported()
+        assert result2 is False
+
+
+# ---------------------------------------------------------------------------
+# Linux path: subprocess.run probe succeeds
+# ---------------------------------------------------------------------------
+
+class TestUnshareNetSupportedProbeSucceeds:
+    """If the unshare probe exits 0, return True."""
+
+    def test_returns_true_when_probe_exit_zero(self):
+        _reset_module_globals()
+        mock_proc = MagicMock()
+        mock_proc.returncode = 0
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Linux"):
+            with patch("backend.secuscan.parser_sandbox.shutil.which", return_value="/usr/bin/unshare"):
+                with patch("backend.secuscan.parser_sandbox.subprocess.run", return_value=mock_proc):
+                    result = _unshare_net_supported()
+        assert result is True
+
+    def test_probes_with_correct_arguments(self):
+        _reset_module_globals()
+        mock_proc = MagicMock()
+        mock_proc.returncode = 0
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Linux"):
+            with patch("backend.secuscan.parser_sandbox.shutil.which", return_value="/usr/bin/unshare"):
+                with patch("backend.secuscan.parser_sandbox.subprocess.run", return_value=mock_proc) as mock_run:
+                    _unshare_net_supported()
+        mock_run.assert_called_once()
+        args, kwargs = mock_run.call_args
+        assert "unshare" in args[0][0] if args else True  # args[0] is the cmd list
+        assert kwargs.get("capture_output") is True
+        assert kwargs.get("timeout") == 5
+
+
+# ---------------------------------------------------------------------------
+# Linux path: subprocess raises an exception
+# ---------------------------------------------------------------------------
+
+class TestUnshareNetSupportedSubprocessException:
+    """If subprocess.run raises an exception, return False."""
+
+    def test_returns_false_when_subprocess_raises(self):
+        _reset_module_globals()
+        with patch("backend.secuscan.parser_sandbox.platform.system", return_value="Linux"):
+            with patch("backend.secuscan.parser_sandbox.shutil.which", return_value="/usr/bin/unshare"):
+                with patch("backend.secuscan.parser_sandbox.subprocess.run",
+                           side_effect=OSError("exec failed")):
+                    result = _unshare_net_supported()
+        assert result is False