# Copyright 2020-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 database module.""" from __future__ import annotations import sys import time sys.path[0:0] = [""] from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest from test.utils_shared import ( HeartbeatEventListener, ServerEventListener, async_wait_until, ) from pymongo import monitoring from pymongo.hello import HelloCompat _IS_SYNC = False class TestStreamingProtocol(AsyncIntegrationTest): @async_client_context.require_failCommand_appName async def test_failCommand_streaming(self): listener = ServerEventListener() hb_listener = HeartbeatEventListener() client = await self.async_rs_or_single_client( event_listeners=[listener, hb_listener], heartbeatFrequencyMS=500, appName="failingHeartbeatTest", ) # Force a connection. await client.admin.command("ping") address = await client.address listener.reset() fail_hello = { "configureFailPoint": "failCommand", "mode": {"times": 4}, "data": { "failCommands": [HelloCompat.LEGACY_CMD, "hello"], "closeConnection": False, "errorCode": 10107, "appName": "failingHeartbeatTest", }, } async with self.fail_point(fail_hello): def _marked_unknown(event): return ( event.server_address == address and not event.new_description.is_server_type_known ) def _discovered_node(event): return ( event.server_address == address and not event.previous_description.is_server_type_known and event.new_description.is_server_type_known ) def marked_unknown(): return len(listener.matching(_marked_unknown)) >= 1 def rediscovered(): return len(listener.matching(_discovered_node)) >= 1 # Topology events are not published synchronously await async_wait_until(marked_unknown, "mark node unknown") await async_wait_until(rediscovered, "rediscover node") # Server should be selectable. await client.admin.command("ping") @async_client_context.require_failCommand_appName async def test_streaming_rtt(self): listener = ServerEventListener() hb_listener = HeartbeatEventListener() # On Windows, RTT can actually be 0.0 because time.time() only has # 1-15 millisecond resolution. We need to delay the initial hello # to ensure that RTT is never zero. name = "streamingRttTest" delay_hello: dict = { "configureFailPoint": "failCommand", "mode": {"times": 1000}, "data": { "failCommands": [HelloCompat.LEGACY_CMD, "hello"], "blockConnection": True, "blockTimeMS": 20, # This can be uncommented after SERVER-49220 is fixed. # 'appName': name, }, } async with self.fail_point(delay_hello): client = await self.async_rs_or_single_client( event_listeners=[listener, hb_listener], heartbeatFrequencyMS=500, appName=name ) # Force a connection. await client.admin.command("ping") address = await client.address delay_hello["data"]["blockTimeMS"] = 500 delay_hello["data"]["appName"] = name async with self.fail_point(delay_hello): def rtt_exceeds_250_ms(): # XXX: Add a public TopologyDescription getter to MongoClient? topology = client._topology sd = topology.description.server_descriptions()[address] assert sd.round_trip_time is not None return sd.round_trip_time > 0.250 await async_wait_until(rtt_exceeds_250_ms, "exceed 250ms RTT") # Server should be selectable. await client.admin.command("ping") def changed_event(event): return event.server_address == address and isinstance( event, monitoring.ServerDescriptionChangedEvent ) # There should only be one event published, for the initial discovery. events = listener.matching(changed_event) self.assertEqual(1, len(events)) self.assertGreater(events[0].new_description.round_trip_time, 0) @async_client_context.require_failCommand_appName async def test_monitor_waits_after_server_check_error(self): # This test implements: # https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring-tests.md#monitors-sleep-at-least-minheartbeatfreqencyms-between-checks fail_hello = { "mode": {"times": 5}, "data": { "failCommands": [HelloCompat.LEGACY_CMD, "hello"], "errorCode": 1234, "appName": "SDAMMinHeartbeatFrequencyTest", }, } async with self.fail_point(fail_hello): start = time.time() client = await self.async_single_client( appName="SDAMMinHeartbeatFrequencyTest", serverSelectionTimeoutMS=5000 ) # Force a connection. await client.admin.command("ping") duration = time.time() - start # Explanation of the expected events: # 0ms: run configureFailPoint # 1ms: create MongoClient # 2ms: failed monitor handshake, 1 # 502ms: failed monitor handshake, 2 # 1002ms: failed monitor handshake, 3 # 1502ms: failed monitor handshake, 4 # 2002ms: failed monitor handshake, 5 # 2502ms: monitor handshake succeeds # 2503ms: run awaitable hello # 2504ms: application handshake succeeds # 2505ms: ping command succeeds self.assertGreaterEqual(duration, 2) self.assertLessEqual(duration, 4.0) @async_client_context.require_failCommand_appName async def test_heartbeat_awaited_flag(self): hb_listener = HeartbeatEventListener() client = await self.async_single_client( event_listeners=[hb_listener], heartbeatFrequencyMS=500, appName="heartbeatEventAwaitedFlag", ) # Force a connection. await client.admin.command("ping") def hb_succeeded(event): return isinstance(event, monitoring.ServerHeartbeatSucceededEvent) def hb_failed(event): return isinstance(event, monitoring.ServerHeartbeatFailedEvent) fail_heartbeat = { "mode": {"times": 2}, "data": { "failCommands": [HelloCompat.LEGACY_CMD, "hello"], "closeConnection": True, "appName": "heartbeatEventAwaitedFlag", }, } async with self.fail_point(fail_heartbeat): await async_wait_until( lambda: hb_listener.matching(hb_failed), "published failed event" ) # Reconnect. await client.admin.command("ping") hb_succeeded_events = hb_listener.matching(hb_succeeded) hb_failed_events = hb_listener.matching(hb_failed) self.assertFalse(hb_succeeded_events[0].awaited) self.assertTrue(hb_failed_events[0].awaited) # Depending on thread scheduling, the failed heartbeat could occur on # the second or third check. events = [type(e) for e in hb_listener.events[:4]] if events == [ monitoring.ServerHeartbeatStartedEvent, monitoring.ServerHeartbeatSucceededEvent, monitoring.ServerHeartbeatStartedEvent, monitoring.ServerHeartbeatFailedEvent, ]: self.assertFalse(hb_succeeded_events[1].awaited) else: self.assertTrue(hb_succeeded_events[1].awaited) if __name__ == "__main__": unittest.main()