diff --git a/.codecov.yml b/.codecov.yml
index 5cbc2d3f5f..c274db5673 100644
--- a/.codecov.yml
+++ b/.codecov.yml
@@ -2,3 +2,6 @@
 # you can also set after_n_builds on comments independently
 comment:
   after_n_builds: 100
+
+ignore:
+  - "pymongo/event_loggers.py"
diff --git a/.evergreen/generated_configs/variants.yml b/.evergreen/generated_configs/variants.yml
index 4c9116d628..4dd44b5048 100644
--- a/.evergreen/generated_configs/variants.yml
+++ b/.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
diff --git a/.evergreen/scripts/generate_config.py b/.evergreen/scripts/generate_config.py
index b4760eab97..32b75a82a2 100644
--- a/.evergreen/scripts/generate_config.py
+++ b/.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()
diff --git a/doc/changelog.rst b/doc/changelog.rst
index 7ed86dfb87..ed6f23f86d 100644
--- a/doc/changelog.rst
+++ b/doc/changelog.rst
@@ -1,7 +1,7 @@
 Changelog
 =========
 
-Changes in Version 4.17.0 (2026/XX/XX)
+Changes in Version 4.17.0 (2026/04/20)
 --------------------------------------
 
 PyMongo 4.17 brings a number of changes including:
@@ -10,10 +10,9 @@ PyMongo 4.17 brings a number of changes including:
   been deprecated and will be removed in PyMongo 5.0. These methods were
   deprecated in favor of the standard dictionary containment operator ``in``
   and the ``keys()`` and ``values()`` methods, respectively.
-
 - Added the :meth:`~pymongo.asynchronous.client_session.AsyncClientSession.bind` and :meth:`~pymongo.client_session.ClientSession.bind` methods
   that allow users to bind a session to all database operations within the scope of a context manager instead of having to explicitly pass the session to each individual operation.
-  See <PLACEHOLDER> for examples and more information.
+  See the `Transactions docs <https://www.mongodb.com/docs/languages/python/pymongo-driver/current/crud/transactions/#methods>`_ for examples and more information.
 - Added support for MongoDB's Intelligent Workload Management (IWM) and ingress connection rate limiting features.
   The driver now gracefully handles write-blocking scenarios and optimizes connection establishment during high-load conditions to maintain application availability.
   See the `IWM <https://www.mongodb.com/docs/atlas/intelligent-workload-management>`_ or `Overload Errors <https://www.mongodb.com/docs/atlas/overload-errors/?interface=driver&language=python>`_ docs for more information.
diff --git a/pymongo/_version.py b/pymongo/_version.py
index 9e4363eed5..9c67ac76ef 100644
--- a/pymongo/_version.py
+++ b/pymongo/_version.py
@@ -18,7 +18,7 @@
 import re
 from typing import List, Tuple, Union
 
-__version__ = "4.17.0.dev0"
+__version__ = "4.18.0.dev0"
 
 
 def get_version_tuple(version: str) -> Tuple[Union[int, str], ...]:
diff --git a/test/test_daemon.py b/test/test_daemon.py
new file mode 100644
index 0000000000..87886d6abd
--- /dev/null
+++ b/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()
diff --git a/test/test_event_loggers.py b/test/test_event_loggers.py
new file mode 100644
index 0000000000..f609ea869b
--- /dev/null
+++ b/test/test_event_loggers.py
@@ -0,0 +1,515 @@
+# 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.
+
+"""Tests for pymongo.event_loggers."""
+from __future__ import annotations
+
+import sys
+from unittest.mock import MagicMock
+
+sys.path[0:0] = [""]
+
+from test import unittest
+
+from pymongo.event_loggers import (
+    CommandLogger,
+    ConnectionPoolLogger,
+    HeartbeatLogger,
+    ServerLogger,
+    TopologyLogger,
+)
+
+
+def _make_command_started(command_name="find", request_id=1, connection_id=("localhost", 27017)):
+    event = MagicMock()
+    event.command_name = command_name
+    event.request_id = request_id
+    event.connection_id = connection_id
+    return event
+
+
+def _make_command_succeeded(
+    command_name="find", request_id=1, connection_id=("localhost", 27017), duration_micros=100
+):
+    event = MagicMock()
+    event.command_name = command_name
+    event.request_id = request_id
+    event.connection_id = connection_id
+    event.duration_micros = duration_micros
+    return event
+
+
+def _make_command_failed(
+    command_name="find", request_id=1, connection_id=("localhost", 27017), duration_micros=200
+):
+    event = MagicMock()
+    event.command_name = command_name
+    event.request_id = request_id
+    event.connection_id = connection_id
+    event.duration_micros = duration_micros
+    return event
+
+
+def _make_server_event(server_address=("localhost", 27017), topology_id="tid1"):
+    event = MagicMock()
+    event.server_address = server_address
+    event.topology_id = topology_id
+    return event
+
+
+def _make_server_description_changed(
+    server_address=("localhost", 27017),
+    topology_id="tid1",
+    previous_type=1,
+    previous_type_name="Standalone",
+    new_type=2,
+    new_type_name="RSPrimary",
+):
+    event = MagicMock()
+    event.server_address = server_address
+    event.topology_id = topology_id
+    event.previous_description.server_type = previous_type
+    event.previous_description.server_type_name = previous_type_name
+    event.new_description.server_type = new_type
+    event.new_description.server_type_name = new_type_name
+    return event
+
+
+def _make_heartbeat_started(connection_id=("localhost", 27017)):
+    event = MagicMock()
+    event.connection_id = connection_id
+    return event
+
+
+def _make_heartbeat_succeeded(connection_id=("localhost", 27017), reply_document=None):
+    event = MagicMock()
+    event.connection_id = connection_id
+    event.reply.document = reply_document or {"ok": 1}
+    return event
+
+
+def _make_heartbeat_failed(connection_id=("localhost", 27017), reply=None):
+    event = MagicMock()
+    event.connection_id = connection_id
+    event.reply = reply or Exception("connection reset")
+    return event
+
+
+def _make_topology_event(topology_id="tid1"):
+    event = MagicMock()
+    event.topology_id = topology_id
+    return event
+
+
+def _make_topology_description_changed(
+    topology_id="tid1",
+    previous_type=0,
+    previous_type_name="Unknown",
+    new_type=1,
+    new_type_name="Single",
+    has_writable=True,
+    has_readable=True,
+):
+    event = MagicMock()
+    event.topology_id = topology_id
+    event.previous_description.topology_type = previous_type
+    event.previous_description.topology_type_name = previous_type_name
+    event.new_description.topology_type = new_type
+    event.new_description.topology_type_name = new_type_name
+    event.new_description.has_writable_server.return_value = has_writable
+    event.new_description.has_readable_server.return_value = has_readable
+    return event
+
+
+def _make_pool_event(address=("localhost", 27017)):
+    event = MagicMock()
+    event.address = address
+    return event
+
+
+def _make_connection_event(address=("localhost", 27017), connection_id=1):
+    event = MagicMock()
+    event.address = address
+    event.connection_id = connection_id
+    return event
+
+
+def _make_connection_closed_event(address=("localhost", 27017), connection_id=1, reason="stale"):
+    event = MagicMock()
+    event.address = address
+    event.connection_id = connection_id
+    event.reason = reason
+    return event
+
+
+def _make_checkout_failed_event(address=("localhost", 27017), reason="timeout"):
+    event = MagicMock()
+    event.address = address
+    event.reason = reason
+    return event
+
+
+class TestCommandLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = CommandLogger()
+
+    def test_started_logs_info(self):
+        event = _make_command_started(command_name="find", request_id=42)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.started(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("INFO", cm.output[0])
+        self.assertIn("find", cm.output[0])
+        self.assertIn("42", cm.output[0])
+        self.assertIn("started", cm.output[0])
+
+    def test_succeeded_logs_info(self):
+        event = _make_command_succeeded(command_name="insert", request_id=7, duration_micros=500)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.succeeded(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("INFO", cm.output[0])
+        self.assertIn("insert", cm.output[0])
+        self.assertIn("7", cm.output[0])
+        self.assertIn("500", cm.output[0])
+        self.assertIn("succeeded", cm.output[0])
+
+    def test_failed_logs_info(self):
+        event = _make_command_failed(command_name="delete", request_id=3, duration_micros=300)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.failed(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("INFO", cm.output[0])
+        self.assertIn("delete", cm.output[0])
+        self.assertIn("3", cm.output[0])
+        self.assertIn("300", cm.output[0])
+        self.assertIn("failed", cm.output[0])
+
+    def test_started_includes_connection_id(self):
+        event = _make_command_started(connection_id=("db.example.com", 27018))
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.started(event)
+        self.assertIn("db.example.com", cm.output[0])
+
+    def test_succeeded_includes_microseconds(self):
+        event = _make_command_succeeded(duration_micros=12345)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.succeeded(event)
+        self.assertIn("12345", cm.output[0])
+        self.assertIn("microseconds", cm.output[0])
+
+    def test_failed_includes_microseconds(self):
+        event = _make_command_failed(duration_micros=9999)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.failed(event)
+        self.assertIn("9999", cm.output[0])
+        self.assertIn("microseconds", cm.output[0])
+
+
+class TestServerLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = ServerLogger()
+
+    def test_opened_logs_info(self):
+        event = _make_server_event(server_address=("host1", 27017), topology_id="topology-abc")
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.opened(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("INFO", cm.output[0])
+        self.assertIn("host1", cm.output[0])
+        self.assertIn("topology-abc", cm.output[0])
+
+    def test_closed_logs_warning(self):
+        event = _make_server_event(server_address=("host1", 27017), topology_id="topology-abc")
+        with self.assertLogs(level="WARNING") as cm:
+            self.logger.closed(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("WARNING", cm.output[0])
+        self.assertIn("host1", cm.output[0])
+        self.assertIn("topology-abc", cm.output[0])
+
+    def test_description_changed_logs_when_type_changes(self):
+        event = _make_server_description_changed(
+            previous_type=1,
+            previous_type_name="Unknown",
+            new_type=2,
+            new_type_name="Standalone",
+        )
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.description_changed(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("Unknown", cm.output[0])
+        self.assertIn("Standalone", cm.output[0])
+
+    def test_description_changed_no_log_when_type_same(self):
+        event = _make_server_description_changed(
+            previous_type=2,
+            previous_type_name="Standalone",
+            new_type=2,
+            new_type_name="Standalone",
+        )
+        # No logs should be emitted when type is unchanged
+        import logging
+
+        root_logger = logging.getLogger()
+        original_level = root_logger.level
+        root_logger.setLevel(logging.DEBUG)
+        try:
+            with self.assertRaises(AssertionError):
+                # assertLogs raises AssertionError if no logs are emitted
+                with self.assertLogs(level="INFO"):
+                    self.logger.description_changed(event)
+        finally:
+            root_logger.setLevel(original_level)
+
+
+class TestHeartbeatLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = HeartbeatLogger()
+
+    def test_started_logs_info(self):
+        event = _make_heartbeat_started(connection_id=("mongo.host", 27017))
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.started(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("INFO", cm.output[0])
+        self.assertIn("mongo.host", cm.output[0])
+
+    def test_succeeded_logs_info(self):
+        event = _make_heartbeat_succeeded(
+            connection_id=("mongo.host", 27017), reply_document={"ok": 1, "ismaster": True}
+        )
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.succeeded(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("INFO", cm.output[0])
+        self.assertIn("mongo.host", cm.output[0])
+        self.assertIn("succeeded", cm.output[0])
+
+    def test_succeeded_includes_reply_document(self):
+        reply_doc = {"ok": 1, "maxWireVersion": 17}
+        event = _make_heartbeat_succeeded(reply_document=reply_doc)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.succeeded(event)
+        self.assertIn(str(reply_doc), cm.output[0])
+
+    def test_failed_logs_warning(self):
+        error = ConnectionRefusedError("refused")
+        event = _make_heartbeat_failed(connection_id=("mongo.host", 27017), reply=error)
+        with self.assertLogs(level="WARNING") as cm:
+            self.logger.failed(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("WARNING", cm.output[0])
+        self.assertIn("mongo.host", cm.output[0])
+        self.assertIn("failed", cm.output[0])
+
+    def test_failed_includes_error(self):
+        error = TimeoutError("timed out")
+        event = _make_heartbeat_failed(reply=error)
+        with self.assertLogs(level="WARNING") as cm:
+            self.logger.failed(event)
+        self.assertIn("timed out", cm.output[0])
+
+
+class TestTopologyLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = TopologyLogger()
+
+    def test_opened_logs_info(self):
+        event = _make_topology_event(topology_id="topo-1")
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.opened(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("INFO", cm.output[0])
+        self.assertIn("topo-1", cm.output[0])
+        self.assertIn("opened", cm.output[0])
+
+    def test_closed_logs_info(self):
+        event = _make_topology_event(topology_id="topo-1")
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.closed(event)
+        self.assertEqual(len(cm.output), 1)
+        self.assertIn("INFO", cm.output[0])
+        self.assertIn("topo-1", cm.output[0])
+        self.assertIn("closed", cm.output[0])
+
+    def test_description_changed_always_logs_update(self):
+        event = _make_topology_description_changed(
+            topology_id="topo-1",
+            previous_type=1,
+            new_type=1,
+            has_writable=True,
+            has_readable=True,
+        )
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.description_changed(event)
+        self.assertTrue(any("updated" in msg for msg in cm.output))
+        self.assertTrue(any("topo-1" in msg for msg in cm.output))
+
+    def test_description_changed_logs_type_change(self):
+        event = _make_topology_description_changed(
+            topology_id="topo-2",
+            previous_type=0,
+            previous_type_name="Unknown",
+            new_type=1,
+            new_type_name="Single",
+            has_writable=True,
+            has_readable=True,
+        )
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.description_changed(event)
+        type_change_logs = [m for m in cm.output if "Unknown" in m and "Single" in m]
+        self.assertEqual(len(type_change_logs), 1)
+
+    def test_description_changed_no_type_change_log_when_same(self):
+        event = _make_topology_description_changed(
+            previous_type=1,
+            previous_type_name="Single",
+            new_type=1,
+            new_type_name="Single",
+            has_writable=True,
+            has_readable=True,
+        )
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.description_changed(event)
+        # Only the general "updated" log, not a type-change log
+        type_change_logs = [m for m in cm.output if "changed type" in m]
+        self.assertEqual(len(type_change_logs), 0)
+
+    def test_description_changed_warns_no_writable_server(self):
+        event = _make_topology_description_changed(has_writable=False, has_readable=True)
+        with self.assertLogs(level="WARNING") as cm:
+            self.logger.description_changed(event)
+        warning_logs = [m for m in cm.output if "WARNING" in m and "writable" in m]
+        self.assertEqual(len(warning_logs), 1)
+
+    def test_description_changed_warns_no_readable_server(self):
+        event = _make_topology_description_changed(has_writable=True, has_readable=False)
+        with self.assertLogs(level="WARNING") as cm:
+            self.logger.description_changed(event)
+        warning_logs = [m for m in cm.output if "WARNING" in m and "readable" in m]
+        self.assertEqual(len(warning_logs), 1)
+
+    def test_description_changed_warns_both_unavailable(self):
+        event = _make_topology_description_changed(has_writable=False, has_readable=False)
+        with self.assertLogs(level="WARNING") as cm:
+            self.logger.description_changed(event)
+        warning_logs = [m for m in cm.output if "WARNING" in m]
+        self.assertEqual(len(warning_logs), 2)
+
+
+class TestConnectionPoolLogger(unittest.TestCase):
+    def setUp(self):
+        self.logger = ConnectionPoolLogger()
+        self.address = ("localhost", 27017)
+
+    def test_pool_created(self):
+        event = _make_pool_event(self.address)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.pool_created(event)
+        self.assertIn("pool created", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+    def test_pool_ready(self):
+        event = _make_pool_event(self.address)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.pool_ready(event)
+        self.assertIn("pool ready", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+    def test_pool_cleared(self):
+        event = _make_pool_event(self.address)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.pool_cleared(event)
+        self.assertIn("pool cleared", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+    def test_pool_closed(self):
+        event = _make_pool_event(self.address)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.pool_closed(event)
+        self.assertIn("pool closed", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+    def test_connection_created(self):
+        event = _make_connection_event(self.address, connection_id=5)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.connection_created(event)
+        self.assertIn("connection created", cm.output[0])
+        self.assertIn("5", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+    def test_connection_ready(self):
+        event = _make_connection_event(self.address, connection_id=5)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.connection_ready(event)
+        self.assertIn("connection setup succeeded", cm.output[0])
+        self.assertIn("5", cm.output[0])
+
+    def test_connection_closed(self):
+        event = _make_connection_closed_event(self.address, connection_id=5, reason="stale")
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.connection_closed(event)
+        self.assertIn("connection closed", cm.output[0])
+        self.assertIn("5", cm.output[0])
+        self.assertIn("stale", cm.output[0])
+
+    def test_connection_closed_reason_in_message(self):
+        for reason in ("stale", "idle", "error", "poolClosed"):
+            event = _make_connection_closed_event(self.address, reason=reason)
+            with self.assertLogs(level="INFO") as cm:
+                self.logger.connection_closed(event)
+            self.assertIn(reason, cm.output[0])
+
+    def test_connection_check_out_started(self):
+        event = _make_pool_event(self.address)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.connection_check_out_started(event)
+        self.assertIn("check out started", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+    def test_connection_check_out_failed(self):
+        event = _make_checkout_failed_event(self.address, reason="timeout")
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.connection_check_out_failed(event)
+        self.assertIn("check out failed", cm.output[0])
+        self.assertIn("timeout", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+    def test_connection_check_out_failed_reason_in_message(self):
+        for reason in ("timeout", "poolClosed", "connectionError"):
+            event = _make_checkout_failed_event(self.address, reason=reason)
+            with self.assertLogs(level="INFO") as cm:
+                self.logger.connection_check_out_failed(event)
+            self.assertIn(reason, cm.output[0])
+
+    def test_connection_checked_out(self):
+        event = _make_connection_event(self.address, connection_id=3)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.connection_checked_out(event)
+        self.assertIn("checked out", cm.output[0])
+        self.assertIn("3", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+    def test_connection_checked_in(self):
+        event = _make_connection_event(self.address, connection_id=3)
+        with self.assertLogs(level="INFO") as cm:
+            self.logger.connection_checked_in(event)
+        self.assertIn("checked into", cm.output[0])
+        self.assertIn("3", cm.output[0])
+        self.assertIn("localhost", cm.output[0])
+
+
+if __name__ == "__main__":
+    unittest.main()