PYTHON-4348 Reinstate fork warning because network I/O and threads are not fork safe (#1597)

Author: ShaneHarvey

CONTRIBUTING.md

@@ -200,7 +200,7 @@ the pages will re-render and the browser will automatically refresh.
 -   Append `test/<mod_name>.py::<class_name>::<test_name>` to run
     specific tests. You can omit the `<test_name>` to test a full class
     and the `<class_name>` to test a full module. For example:
-    `tox -m test test/test_change_stream.py::TestUnifiedChangeStreamsErrors::test_change_stream_errors_on_ElectionInProgress`.
+    `tox -m test -- test/test_change_stream.py::TestUnifiedChangeStreamsErrors::test_change_stream_errors_on_ElectionInProgress`.
 -   Use the `-k` argument to select tests by pattern.
 
 ## Running Load Balancer Tests Locally

pymongo/mongo_client.py

@@ -879,8 +879,11 @@ def __init__(
             # This will be used later if we fork.
             MongoClient._clients[self._topology._topology_id] = self
 
-    def _init_background(self) -> None:
+    def _init_background(self, old_pid: Optional[int] = None) -> None:
         self._topology = Topology(self._topology_settings)
+        # Seed the topology with the old one's pid so we can detect clients
+        # that are opened before a fork and used after.
+        self._topology._pid = old_pid
 
         def target() -> bool:
             client = self_ref()
@@ -903,7 +906,7 @@ def target() -> bool:
 
     def _after_fork(self) -> None:
         """Resets topology in a child after successfully forking."""
-        self._init_background()
+        self._init_background(self._topology._pid)
 
     def _duplicate(self, **kwargs: Any) -> MongoClient:
         args = self.__init_kwargs.copy()
@@ -921,7 +924,7 @@ def _server_property(self, attr_name: str) -> Any:
         the server may change. In such cases, store a local reference to a
         ServerDescription first, then use its properties.
         """
-        server = self._topology.select_server(writable_server_selector, _Op.TEST)
+        server = self._get_topology().select_server(writable_server_selector, _Op.TEST)
 
         return getattr(server.description, attr_name)
 

pymongo/topology.py

@@ -20,9 +20,11 @@
 import os
 import queue
 import random
+import sys
 import time
 import warnings
 import weakref
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, Mapping, Optional, cast
 
 from pymongo import _csot, common, helpers, periodic_executor
@@ -70,6 +72,9 @@
     from pymongo.typings import ClusterTime, _Address
 
 
+_pymongo_dir = str(Path(__file__).parent)
+
+
 def process_events_queue(queue_ref: weakref.ReferenceType[queue.Queue]) -> bool:
     q = queue_ref()
     if not q:
@@ -187,12 +192,17 @@ def open(self) -> None:
             self._pid = pid
         elif pid != self._pid:
             self._pid = pid
-            warnings.warn(
+            if sys.version_info[:2] >= (3, 12):
+                kwargs = {"skip_file_prefixes": (_pymongo_dir,)}
+            else:
+                kwargs = {"stacklevel": 6}
+            # Ignore B028 warning for missing stacklevel.
+            warnings.warn(  # type: ignore[call-overload] # noqa: B028
                 "MongoClient opened before fork. May not be entirely fork-safe, "
                 "proceed with caution. See PyMongo's documentation for details: "
                 "https://pymongo.readthedocs.io/en/stable/faq.html#"
                 "is-pymongo-fork-safe",
-                stacklevel=2,
+                **kwargs,
             )
             with self._lock:
                 # Close servers and clear the pools.

test/test_encryption.py

@@ -26,6 +26,7 @@
 import textwrap
 import traceback
 import uuid
+import warnings
 from threading import Thread
 from typing import Any, Dict, Mapping
 
@@ -348,7 +349,9 @@ def test_fork(self):
         self.addCleanup(client.close)
 
         def target():
-            client.admin.command("ping")
+            with warnings.catch_warnings():
+                warnings.simplefilter("ignore")
+                client.admin.command("ping")
 
         with self.fork(target):
             target()

test/test_fork.py

@@ -18,6 +18,7 @@
 import os
 import sys
 import unittest
+import warnings
 from multiprocessing import Pipe
 
 sys.path[0:0] = [""]
@@ -43,7 +44,9 @@ def test_lock_client(self):
         with self.client._MongoClient__lock:
 
             def target():
-                self.client.admin.command("ping")
+                with warnings.catch_warnings():
+                    warnings.simplefilter("ignore")
+                    self.client.admin.command("ping")
 
             with self.fork(target):
                 pass
@@ -72,7 +75,11 @@ def test_topology_reset(self):
         parent_cursor_exc = self.client._kill_cursors_executor
 
         def target():
-            self.client.admin.command("ping")
+            # Catch the fork warning and send to the parent for assertion.
+            with warnings.catch_warnings(record=True) as ctx:
+                warnings.simplefilter("always")
+                self.client.admin.command("ping")
+                child_conn.send(str(ctx[0]))
             child_conn.send(self.client._topology._pid)
             child_conn.send(
                 (
@@ -83,6 +90,8 @@ def target():
 
         with self.fork(target):
             self.assertEqual(self.client._topology._pid, init_id)
+            fork_warning = parent_conn.recv()
+            self.assertIn("MongoClient opened before fork", fork_warning)
             child_id = parent_conn.recv()
             self.assertNotEqual(child_id, init_id)
             passed, msg = parent_conn.recv()