PYTHON-2024 Skip publishing SDAM events for "equivalent" ServerDescriptions

Author: ShaneHarvey

pymongo/server_description.py

@@ -207,5 +207,26 @@ def retryable_reads_supported(self):
         """Checks if this server supports retryable writes."""
         return self._max_wire_version >= 6
 
+    def __eq__(self, other):
+        if isinstance(other, ServerDescription):
+            return ((self._address == other.address) and
+                    (self._server_type == other.server_type) and
+                    (self._min_wire_version == other.min_wire_version) and
+                    (self._max_wire_version == other.max_wire_version) and
+                    (self._me == other.me) and
+                    (self._all_hosts == other.all_hosts) and
+                    (self._tags == other.tags) and
+                    (self._replica_set_name == other.replica_set_name) and
+                    (self._set_version == other.set_version) and
+                    (self._election_id == other.election_id) and
+                    (self._primary == other.primary) and
+                    (self._ls_timeout_minutes ==
+                     other.logical_session_timeout_minutes))
+
+        return NotImplemented
+
+    def __ne__(self, other):
+        return not self == other
+
     # For unittesting only. Use under no circumstances!
     _host_to_round_trip_time = {}

pymongo/topology.py

@@ -264,9 +264,11 @@ def _process_change(self, server_description):
         Hold the lock when calling this.
         """
         td_old = self._description
-        if self._publish_server:
-            old_server_description = td_old._server_descriptions[
-                server_description.address]
+        old_server_description = td_old._server_descriptions[
+            server_description.address]
+        suppress_event = ((self._publish_server or self._publish_tp)
+                          and old_server_description == server_description)
+        if self._publish_server and not suppress_event:
             self._events.put((
                 self._listeners.publish_server_description_changed,
                 (old_server_description, server_description,
@@ -278,7 +280,7 @@ def _process_change(self, server_description):
         self._update_servers()
         self._receive_cluster_time_no_lock(server_description.cluster_time)
 
-        if self._publish_tp:
+        if self._publish_tp and not suppress_event:
             self._events.put((
                 self._listeners.publish_topology_description_changed,
                 (td_old, self._description, self._topology_id)))

test/test_sdam_monitoring_spec.py

@@ -207,15 +207,16 @@ def _run(self):
                     lambda: len(self.all_listener.results) >= expected_len,
                     "publish all events", timeout=15)
 
+                # Wait some time to catch possible lagging extra events.
+                time.sleep(0.5)
+
                 i = 0
                 while i < expected_len:
                     result = self.all_listener.results[i] if len(
                         self.all_listener.results) > i else None
                     # The order of ServerOpening/ClosedEvents doesn't matter
-                    if (isinstance(result,
-                                   monitoring.ServerOpeningEvent) or
-                            isinstance(result,
-                                       monitoring.ServerClosedEvent)):
+                    if isinstance(result, (monitoring.ServerOpeningEvent,
+                                           monitoring.ServerClosedEvent)):
                         i, passed, message = compare_multiple_events(
                             i, expected_results, self.all_listener.results)
                         self.assertTrue(passed, message)
@@ -224,6 +225,11 @@ def _run(self):
                             *compare_events(expected_results[i], result))
                         i += 1
 
+                # Assert no extra events.
+                extra_events = self.all_listener.results[expected_len:]
+                if extra_events:
+                    self.fail('Extra events %r' % (extra_events,))
+
                 self.all_listener.reset()
         finally:
             m.close()