PYTHON-5774 Increase daemon.py coverage to 63% (#2759)

.evergreen/generated_configs/variants.yml

@@ -615,6 +615,7 @@ buildvariants:
   - name: test-win64
     tasks:
       - name: .test-standard !.pypy
+      - name: .test-no-orchestration !.pypy
     display_name: "* Test Win64"
     run_on:
       - windows-2022-latest-small

.evergreen/scripts/generate_config.py

@@ -97,6 +97,8 @@ def create_standard_nonlinux_variants() -> list[BuildVariant]:
             tasks = [
                 f".test-standard !.pypy .server-{version}" for version in get_versions_from("6.0")
             ]
+        if host_name == "win64":
+            tasks.append(".test-no-orchestration !.pypy")
         host = HOSTS[host_name]
         tags = ["standard-non-linux"]
         expansions = dict()

test/test_daemon.py

@@ -0,0 +1,183 @@
+# Copyright 2026-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test the pymongo daemon module."""
+from __future__ import annotations
+
+import subprocess
+import sys
+import warnings
+from unittest.mock import MagicMock, patch
+
+sys.path[0:0] = [""]
+
+from test import unittest
+
+import pymongo.daemon as daemon_module
+from pymongo.daemon import _popen_wait, _silence_resource_warning, _spawn_daemon
+
+
+class TestPopenWait(unittest.TestCase):
+    def test_returns_returncode_on_success(self):
+        mock_popen = MagicMock()
+        mock_popen.wait.return_value = 0
+        self.assertEqual(0, _popen_wait(mock_popen, timeout=5))
+        mock_popen.wait.assert_called_once_with(timeout=5)
+
+    def test_returns_none_on_timeout_expired(self):
+        mock_popen = MagicMock()
+        mock_popen.wait.side_effect = subprocess.TimeoutExpired(cmd="foo", timeout=5)
+        self.assertIsNone(_popen_wait(mock_popen, timeout=5))
+
+    def test_none_timeout_passes_through(self):
+        mock_popen = MagicMock()
+        mock_popen.wait.return_value = 1
+        self.assertEqual(1, _popen_wait(mock_popen, timeout=None))
+        mock_popen.wait.assert_called_once_with(timeout=None)
+
+
+class TestSilenceResourceWarning(unittest.TestCase):
+    def test_sets_returncode_to_zero(self):
+        mock_popen = MagicMock()
+        mock_popen.returncode = None
+        _silence_resource_warning(mock_popen)
+        self.assertEqual(0, mock_popen.returncode)
+
+    def test_no_op_for_none(self):
+        # Should not raise when popen is None (mongocryptd spawn failed).
+        _silence_resource_warning(None)
+
+
+@unittest.skipIf(sys.platform == "win32", "Unix only")
+class TestSpawnUnix(unittest.TestCase):
+    def setUp(self):
+        from pymongo.daemon import _spawn
+
+        self._spawn = _spawn
+
+    def test_returns_popen_on_success(self):
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen):
+            result = self._spawn(["somecommand"])
+        self.assertIs(mock_popen, result)
+
+    def test_filenotfound_warns_and_returns_none(self):
+        with patch("subprocess.Popen", side_effect=FileNotFoundError("not found")):
+            with warnings.catch_warnings(record=True) as w:
+                warnings.simplefilter("always")
+                result = self._spawn(["nonexistent_command"])
+        self.assertIsNone(result)
+        self.assertEqual(1, len(w))
+        self.assertIs(RuntimeWarning, w[0].category)
+        self.assertIn("nonexistent_command", str(w[0].message))
+
+
+@unittest.skipIf(sys.platform == "win32", "Unix only")
+class TestSpawnDaemonDoublePopen(unittest.TestCase):
+    def setUp(self):
+        from pymongo.daemon import _spawn_daemon_double_popen
+
+        self._spawn_daemon_double_popen = _spawn_daemon_double_popen
+
+    def test_spawns_this_file_as_intermediate(self):
+        mock_popen = MagicMock()
+        mock_popen.wait.return_value = 0
+        with patch("subprocess.Popen", return_value=mock_popen) as mock_cls:
+            self._spawn_daemon_double_popen(["somecommand", "--arg"])
+        spawner_args = mock_cls.call_args[0][0]
+        self.assertEqual(sys.executable, spawner_args[0])
+        self.assertIn("daemon.py", spawner_args[1])
+        self.assertIn("somecommand", spawner_args)
+
+    def test_waits_for_intermediate_process(self):
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen):
+            self._spawn_daemon_double_popen(["somecommand"])
+        mock_popen.wait.assert_called_once_with(timeout=daemon_module._WAIT_TIMEOUT)
+
+    def test_continues_on_timeout(self):
+        # _popen_wait swallows TimeoutExpired — double Popen must not raise.
+        mock_popen = MagicMock()
+        mock_popen.wait.side_effect = subprocess.TimeoutExpired(cmd="foo", timeout=10)
+        with patch("subprocess.Popen", return_value=mock_popen):
+            self._spawn_daemon_double_popen(["somecommand"])  # must not raise
+
+
+@unittest.skipIf(sys.platform == "win32", "Unix only")
+class TestSpawnDaemonUnix(unittest.TestCase):
+    def test_uses_double_popen_when_executable_set(self):
+        with patch("pymongo.daemon._spawn_daemon_double_popen") as mock_double:
+            _spawn_daemon(["somecommand"])
+        mock_double.assert_called_once_with(["somecommand"])
+
+    def test_fallback_to_spawn_when_no_executable(self):
+        with patch("pymongo.daemon._spawn") as mock_spawn:
+            with patch.object(sys, "executable", ""):
+                _spawn_daemon(["somecommand"])
+        mock_spawn.assert_called_once_with(["somecommand"])
+
+
+@unittest.skipUnless(sys.platform == "win32", "Windows only")
+class TestSpawnDaemonWindows(unittest.TestCase):
+    def test_silences_resource_warning_on_success(self):
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen):
+            _spawn_daemon(["somecommand"])
+        self.assertEqual(0, mock_popen.returncode)
+
+    def test_filenotfound_warns(self):
+        with patch("subprocess.Popen", side_effect=FileNotFoundError("not found")):
+            with warnings.catch_warnings(record=True) as w:
+                warnings.simplefilter("always")
+                _spawn_daemon(["nonexistent_command"])
+        self.assertEqual(1, len(w))
+        self.assertIs(RuntimeWarning, w[0].category)
+        self.assertIn("nonexistent_command", str(w[0].message))
+
+    def test_uses_detached_process_flag(self):
+        # DETACHED_PROCESS must be passed so the child survives parent exit.
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen) as mock_cls:
+            _spawn_daemon(["somecommand"])
+        kwargs = mock_cls.call_args[1]
+        self.assertEqual(daemon_module._DETACHED_PROCESS, kwargs["creationflags"])
+
+    def test_uses_devnull_for_stdio(self):
+        # stdin/stdout/stderr must be redirected to devnull to fully detach.
+        mock_popen = MagicMock()
+        with patch("subprocess.Popen", return_value=mock_popen) as mock_cls:
+            _spawn_daemon(["somecommand"])
+        kwargs = mock_cls.call_args[1]
+        self.assertIsNotNone(kwargs.get("stdin"))
+        self.assertIsNotNone(kwargs.get("stdout"))
+        self.assertIsNotNone(kwargs.get("stderr"))
+
+    def test_detached_process_constant_value(self):
+        # Value must match the Windows DETACHED_PROCESS process creation flag.
+        self.assertEqual(0x00000008, daemon_module._DETACHED_PROCESS)
+
+
+@unittest.skipIf(sys.platform == "win32", "Unix only")
+class TestMainBlock(unittest.TestCase):
+    def test_exits_with_zero(self):
+        # Run daemon.py as a script with a no-op subprocess; verify it exits cleanly.
+        result = subprocess.run(
+            [sys.executable, "-m", "pymongo.daemon", sys.executable, "-c", "pass"],
+            timeout=15,
+        )
+        self.assertEqual(0, result.returncode)
+
+
+if __name__ == "__main__":
+    unittest.main()