Merge branch 'master' of github.com:mongodb/mongo-python-driver
This commit is contained in:
Steven Silvester
commit
2e6201d3c5
11
.github/workflows/dist.yml
vendored
11
.github/workflows/dist.yml
vendored
@ -35,10 +35,9 @@ jobs:
|
||||
# https://github.com/github/feedback/discussions/7835#discussioncomment-1769026
|
||||
buildplat:
|
||||
- [ubuntu-20.04, "manylinux_x86_64", "cp3*-manylinux_x86_64"]
|
||||
- [ubuntu-24.04-arm, "manylinux_aarch64", "cp3*-manylinux_aarch64"]
|
||||
# Disabled pending PYTHON-5058
|
||||
# - [ubuntu-24.04, "manylinux_ppc64le", "cp3*-manylinux_ppc64le"]
|
||||
# - [ubuntu-24.04, "manylinux_s390x", "cp3*-manylinux_s390x"]
|
||||
- [ubuntu-20.04, "manylinux_aarch64", "cp3*-manylinux_aarch64"]
|
||||
- [ubuntu-20.04, "manylinux_ppc64le", "cp3*-manylinux_ppc64le"]
|
||||
- [ubuntu-20.04, "manylinux_s390x", "cp3*-manylinux_s390x"]
|
||||
- [ubuntu-20.04, "manylinux_i686", "cp3*-manylinux_i686"]
|
||||
- [windows-2019, "win_amd6", "cp3*-win_amd64"]
|
||||
- [windows-2019, "win32", "cp3*-win32"]
|
||||
@ -63,6 +62,10 @@ jobs:
|
||||
if: runner.os == 'Linux'
|
||||
uses: docker/setup-qemu-action@v3
|
||||
with:
|
||||
# setup-qemu-action by default uses `tonistiigi/binfmt:latest` image,
|
||||
# which is out of date. This causes seg faults during build.
|
||||
# Here we manually fix the version.
|
||||
image: tonistiigi/binfmt:qemu-v8.1.5
|
||||
platforms: all
|
||||
|
||||
- name: Install cibuildwheel
|
||||
|
||||
199
test/asynchronous/test_mongos_load_balancing.py
Normal file
199
test/asynchronous/test_mongos_load_balancing.py
Normal file
@ -0,0 +1,199 @@
|
||||
# Copyright 2015-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 AsyncMongoClient's mongos load balancing using a mock."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
import threading
|
||||
from test.asynchronous.helpers import ConcurrentRunner
|
||||
|
||||
from pymongo.operations import _Op
|
||||
|
||||
sys.path[0:0] = [""]
|
||||
|
||||
from test.asynchronous import AsyncMockClientTest, async_client_context, connected, unittest
|
||||
from test.asynchronous.pymongo_mocks import AsyncMockClient
|
||||
from test.utils import async_wait_until
|
||||
|
||||
from pymongo.errors import AutoReconnect, InvalidOperation
|
||||
from pymongo.server_selectors import writable_server_selector
|
||||
from pymongo.topology_description import TOPOLOGY_TYPE
|
||||
|
||||
_IS_SYNC = False
|
||||
|
||||
|
||||
class SimpleOp(ConcurrentRunner):
|
||||
def __init__(self, client):
|
||||
super().__init__()
|
||||
self.client = client
|
||||
self.passed = False
|
||||
|
||||
async def run(self):
|
||||
await self.client.db.command("ping")
|
||||
self.passed = True # No exception raised.
|
||||
|
||||
|
||||
async def do_simple_op(client, ntasks):
|
||||
tasks = [SimpleOp(client) for _ in range(ntasks)]
|
||||
for t in tasks:
|
||||
await t.start()
|
||||
|
||||
for t in tasks:
|
||||
await t.join()
|
||||
|
||||
for t in tasks:
|
||||
assert t.passed
|
||||
|
||||
|
||||
async def writable_addresses(topology):
|
||||
return {
|
||||
server.description.address
|
||||
for server in await topology.select_servers(writable_server_selector, _Op.TEST)
|
||||
}
|
||||
|
||||
|
||||
class TestMongosLoadBalancing(AsyncMockClientTest):
|
||||
@async_client_context.require_connection
|
||||
@async_client_context.require_no_load_balancer
|
||||
async def asyncSetUp(self):
|
||||
await super().asyncSetUp()
|
||||
|
||||
def mock_client(self, **kwargs):
|
||||
mock_client = AsyncMockClient(
|
||||
standalones=[],
|
||||
members=[],
|
||||
mongoses=["a:1", "b:2", "c:3"],
|
||||
host="a:1,b:2,c:3",
|
||||
connect=False,
|
||||
**kwargs,
|
||||
)
|
||||
self.addAsyncCleanup(mock_client.aclose)
|
||||
|
||||
# Latencies in seconds.
|
||||
mock_client.mock_rtts["a:1"] = 0.020
|
||||
mock_client.mock_rtts["b:2"] = 0.025
|
||||
mock_client.mock_rtts["c:3"] = 0.045
|
||||
return mock_client
|
||||
|
||||
async def test_lazy_connect(self):
|
||||
# While connected() ensures we can trigger connection from the main
|
||||
# thread and wait for the monitors, this test triggers connection from
|
||||
# several threads at once to check for data races.
|
||||
nthreads = 10
|
||||
client = self.mock_client()
|
||||
self.assertEqual(0, len(client.nodes))
|
||||
|
||||
# Trigger initial connection.
|
||||
await do_simple_op(client, nthreads)
|
||||
await async_wait_until(lambda: len(client.nodes) == 3, "connect to all mongoses")
|
||||
|
||||
async def test_failover(self):
|
||||
ntasks = 10
|
||||
client = await connected(self.mock_client(localThresholdMS=0.001))
|
||||
await async_wait_until(lambda: len(client.nodes) == 3, "connect to all mongoses")
|
||||
|
||||
# Our chosen mongos goes down.
|
||||
client.kill_host("a:1")
|
||||
|
||||
# Trigger failover to higher-latency nodes. AutoReconnect should be
|
||||
# raised at most once in each thread.
|
||||
passed = []
|
||||
|
||||
async def f():
|
||||
try:
|
||||
await client.db.command("ping")
|
||||
except AutoReconnect:
|
||||
# Second attempt succeeds.
|
||||
await client.db.command("ping")
|
||||
|
||||
passed.append(True)
|
||||
|
||||
tasks = [ConcurrentRunner(target=f) for _ in range(ntasks)]
|
||||
for t in tasks:
|
||||
await t.start()
|
||||
|
||||
for t in tasks:
|
||||
await t.join()
|
||||
|
||||
self.assertEqual(ntasks, len(passed))
|
||||
|
||||
# Down host removed from list.
|
||||
self.assertEqual(2, len(client.nodes))
|
||||
|
||||
async def test_local_threshold(self):
|
||||
client = await connected(self.mock_client(localThresholdMS=30))
|
||||
self.assertEqual(30, client.options.local_threshold_ms)
|
||||
await async_wait_until(lambda: len(client.nodes) == 3, "connect to all mongoses")
|
||||
topology = client._topology
|
||||
|
||||
# All are within a 30-ms latency window, see self.mock_client().
|
||||
self.assertEqual({("a", 1), ("b", 2), ("c", 3)}, await writable_addresses(topology))
|
||||
|
||||
# No error
|
||||
await client.admin.command("ping")
|
||||
|
||||
client = await connected(self.mock_client(localThresholdMS=0))
|
||||
self.assertEqual(0, client.options.local_threshold_ms)
|
||||
# No error
|
||||
await client.db.command("ping")
|
||||
# Our chosen mongos goes down.
|
||||
client.kill_host("{}:{}".format(*next(iter(client.nodes))))
|
||||
try:
|
||||
await client.db.command("ping")
|
||||
except:
|
||||
pass
|
||||
|
||||
# We eventually connect to a new mongos.
|
||||
async def connect_to_new_mongos():
|
||||
try:
|
||||
return await client.db.command("ping")
|
||||
except AutoReconnect:
|
||||
pass
|
||||
|
||||
await async_wait_until(connect_to_new_mongos, "connect to a new mongos")
|
||||
|
||||
async def test_load_balancing(self):
|
||||
# Although the server selection JSON tests already prove that
|
||||
# select_servers works for sharded topologies, here we do an end-to-end
|
||||
# test of discovering servers' round trip times and configuring
|
||||
# localThresholdMS.
|
||||
client = await connected(self.mock_client())
|
||||
await async_wait_until(lambda: len(client.nodes) == 3, "connect to all mongoses")
|
||||
|
||||
# Prohibited for topology type Sharded.
|
||||
with self.assertRaises(InvalidOperation):
|
||||
await client.address
|
||||
|
||||
topology = client._topology
|
||||
self.assertEqual(TOPOLOGY_TYPE.Sharded, topology.description.topology_type)
|
||||
|
||||
# a and b are within the 15-ms latency window, see self.mock_client().
|
||||
self.assertEqual({("a", 1), ("b", 2)}, await writable_addresses(topology))
|
||||
|
||||
client.mock_rtts["a:1"] = 0.045
|
||||
|
||||
# Discover only b is within latency window.
|
||||
async def predicate():
|
||||
return {("b", 2)} == await writable_addresses(topology)
|
||||
|
||||
await async_wait_until(
|
||||
predicate,
|
||||
'discover server "a" is too far',
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
374
test/asynchronous/test_sdam_monitoring_spec.py
Normal file
374
test/asynchronous/test_sdam_monitoring_spec.py
Normal file
@ -0,0 +1,374 @@
|
||||
# Copyright 2016 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.
|
||||
|
||||
"""Run the sdam monitoring spec tests."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path[0:0] = [""]
|
||||
|
||||
from test.asynchronous import AsyncIntegrationTest, async_client_context, client_knobs, unittest
|
||||
from test.utils import (
|
||||
ServerAndTopologyEventListener,
|
||||
async_wait_until,
|
||||
server_name_to_type,
|
||||
)
|
||||
|
||||
from bson.json_util import object_hook
|
||||
from pymongo import AsyncMongoClient, monitoring
|
||||
from pymongo.asynchronous.collection import AsyncCollection
|
||||
from pymongo.asynchronous.monitor import Monitor
|
||||
from pymongo.common import clean_node
|
||||
from pymongo.errors import ConnectionFailure, NotPrimaryError
|
||||
from pymongo.hello import Hello
|
||||
from pymongo.server_description import ServerDescription
|
||||
from pymongo.topology_description import TOPOLOGY_TYPE
|
||||
|
||||
_IS_SYNC = False
|
||||
|
||||
# Location of JSON test specifications.
|
||||
if _IS_SYNC:
|
||||
TEST_PATH = os.path.join(Path(__file__).resolve().parent, "sdam_monitoring")
|
||||
else:
|
||||
TEST_PATH = os.path.join(Path(__file__).resolve().parent.parent, "sdam_monitoring")
|
||||
|
||||
|
||||
def compare_server_descriptions(expected, actual):
|
||||
if (expected["address"] != "{}:{}".format(*actual.address)) or (
|
||||
server_name_to_type(expected["type"]) != actual.server_type
|
||||
):
|
||||
return False
|
||||
expected_hosts = set(expected["arbiters"] + expected["passives"] + expected["hosts"])
|
||||
return expected_hosts == {"{}:{}".format(*s) for s in actual.all_hosts}
|
||||
|
||||
|
||||
def compare_topology_descriptions(expected, actual):
|
||||
if TOPOLOGY_TYPE.__getattribute__(expected["topologyType"]) != actual.topology_type:
|
||||
return False
|
||||
expected = expected["servers"]
|
||||
actual = actual.server_descriptions()
|
||||
if len(expected) != len(actual):
|
||||
return False
|
||||
for exp_server in expected:
|
||||
for _address, actual_server in actual.items():
|
||||
if compare_server_descriptions(exp_server, actual_server):
|
||||
break
|
||||
else:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def compare_events(expected_dict, actual):
|
||||
if not expected_dict:
|
||||
return False, "Error: Bad expected value in YAML test"
|
||||
if not actual:
|
||||
return False, "Error: Event published was None"
|
||||
|
||||
expected_type, expected = list(expected_dict.items())[0]
|
||||
|
||||
if expected_type == "server_opening_event":
|
||||
if not isinstance(actual, monitoring.ServerOpeningEvent):
|
||||
return False, "Expected ServerOpeningEvent, got %s" % (actual.__class__)
|
||||
if expected["address"] != "{}:{}".format(*actual.server_address):
|
||||
return (
|
||||
False,
|
||||
"ServerOpeningEvent published with wrong address (expected" " {}, got {}".format(
|
||||
expected["address"], actual.server_address
|
||||
),
|
||||
)
|
||||
|
||||
elif expected_type == "server_description_changed_event":
|
||||
if not isinstance(actual, monitoring.ServerDescriptionChangedEvent):
|
||||
return (False, "Expected ServerDescriptionChangedEvent, got %s" % (actual.__class__))
|
||||
if expected["address"] != "{}:{}".format(*actual.server_address):
|
||||
return (
|
||||
False,
|
||||
"ServerDescriptionChangedEvent has wrong address" " (expected {}, got {}".format(
|
||||
expected["address"], actual.server_address
|
||||
),
|
||||
)
|
||||
|
||||
if not compare_server_descriptions(expected["newDescription"], actual.new_description):
|
||||
return (False, "New ServerDescription incorrect in ServerDescriptionChangedEvent")
|
||||
if not compare_server_descriptions(
|
||||
expected["previousDescription"], actual.previous_description
|
||||
):
|
||||
return (
|
||||
False,
|
||||
"Previous ServerDescription incorrect in ServerDescriptionChangedEvent",
|
||||
)
|
||||
|
||||
elif expected_type == "server_closed_event":
|
||||
if not isinstance(actual, monitoring.ServerClosedEvent):
|
||||
return False, "Expected ServerClosedEvent, got %s" % (actual.__class__)
|
||||
if expected["address"] != "{}:{}".format(*actual.server_address):
|
||||
return (
|
||||
False,
|
||||
"ServerClosedEvent published with wrong address" " (expected {}, got {}".format(
|
||||
expected["address"], actual.server_address
|
||||
),
|
||||
)
|
||||
|
||||
elif expected_type == "topology_opening_event":
|
||||
if not isinstance(actual, monitoring.TopologyOpenedEvent):
|
||||
return False, "Expected TopologyOpenedEvent, got %s" % (actual.__class__)
|
||||
|
||||
elif expected_type == "topology_description_changed_event":
|
||||
if not isinstance(actual, monitoring.TopologyDescriptionChangedEvent):
|
||||
return (
|
||||
False,
|
||||
"Expected TopologyDescriptionChangedEvent, got %s" % (actual.__class__),
|
||||
)
|
||||
if not compare_topology_descriptions(expected["newDescription"], actual.new_description):
|
||||
return (
|
||||
False,
|
||||
"New TopologyDescription incorrect in TopologyDescriptionChangedEvent",
|
||||
)
|
||||
if not compare_topology_descriptions(
|
||||
expected["previousDescription"], actual.previous_description
|
||||
):
|
||||
return (
|
||||
False,
|
||||
"Previous TopologyDescription incorrect in TopologyDescriptionChangedEvent",
|
||||
)
|
||||
|
||||
elif expected_type == "topology_await aclosed_event":
|
||||
if not isinstance(actual, monitoring.TopologyClosedEvent):
|
||||
return False, "Expected TopologyClosedEvent, got %s" % (actual.__class__)
|
||||
|
||||
else:
|
||||
return False, f"Incorrect event: expected {expected_type}, actual {actual}"
|
||||
|
||||
return True, ""
|
||||
|
||||
|
||||
def compare_multiple_events(i, expected_results, actual_results):
|
||||
events_in_a_row = []
|
||||
j = i
|
||||
while j < len(expected_results) and isinstance(actual_results[j], actual_results[i].__class__):
|
||||
events_in_a_row.append(actual_results[j])
|
||||
j += 1
|
||||
message = ""
|
||||
for event in events_in_a_row:
|
||||
for k in range(i, j):
|
||||
passed, message = compare_events(expected_results[k], event)
|
||||
if passed:
|
||||
expected_results[k] = None
|
||||
break
|
||||
else:
|
||||
return i, False, message
|
||||
return j, True, ""
|
||||
|
||||
|
||||
class TestAllScenarios(AsyncIntegrationTest):
|
||||
async def asyncSetUp(self):
|
||||
await super().asyncSetUp()
|
||||
self.all_listener = ServerAndTopologyEventListener()
|
||||
|
||||
|
||||
def create_test(scenario_def):
|
||||
async def run_scenario(self):
|
||||
with client_knobs(events_queue_frequency=0.05, min_heartbeat_interval=0.05):
|
||||
await _run_scenario(self)
|
||||
|
||||
async def _run_scenario(self):
|
||||
class NoopMonitor(Monitor):
|
||||
"""Override the _run method to do nothing."""
|
||||
|
||||
async def _run(self):
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
m = AsyncMongoClient(
|
||||
host=scenario_def["uri"],
|
||||
port=27017,
|
||||
event_listeners=[self.all_listener],
|
||||
_monitor_class=NoopMonitor,
|
||||
)
|
||||
topology = await m._get_topology()
|
||||
|
||||
try:
|
||||
for phase in scenario_def["phases"]:
|
||||
for source, response in phase.get("responses", []):
|
||||
source_address = clean_node(source)
|
||||
await topology.on_change(
|
||||
ServerDescription(
|
||||
address=source_address, hello=Hello(response), round_trip_time=0
|
||||
)
|
||||
)
|
||||
|
||||
expected_results = phase["outcome"]["events"]
|
||||
expected_len = len(expected_results)
|
||||
await async_wait_until(
|
||||
lambda: len(self.all_listener.results) >= expected_len,
|
||||
"publish all events",
|
||||
timeout=15,
|
||||
)
|
||||
|
||||
# Wait some time to catch possible lagging extra events.
|
||||
await async_wait_until(lambda: topology._events.empty(), "publish lagging events")
|
||||
|
||||
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, monitoring.ServerClosedEvent)
|
||||
):
|
||||
i, passed, message = compare_multiple_events(
|
||||
i, expected_results, self.all_listener.results
|
||||
)
|
||||
self.assertTrue(passed, message)
|
||||
else:
|
||||
self.assertTrue(*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(f"Extra events {extra_events!r}")
|
||||
|
||||
self.all_listener.reset()
|
||||
finally:
|
||||
await m.close()
|
||||
|
||||
return run_scenario
|
||||
|
||||
|
||||
def create_tests():
|
||||
for dirpath, _, filenames in os.walk(TEST_PATH):
|
||||
for filename in filenames:
|
||||
with open(os.path.join(dirpath, filename)) as scenario_stream:
|
||||
scenario_def = json.load(scenario_stream, object_hook=object_hook)
|
||||
# Construct test from scenario.
|
||||
new_test = create_test(scenario_def)
|
||||
test_name = f"test_{os.path.splitext(filename)[0]}"
|
||||
new_test.__name__ = test_name
|
||||
setattr(TestAllScenarios, new_test.__name__, new_test)
|
||||
|
||||
|
||||
create_tests()
|
||||
|
||||
|
||||
class TestSdamMonitoring(AsyncIntegrationTest):
|
||||
knobs: client_knobs
|
||||
listener: ServerAndTopologyEventListener
|
||||
test_client: AsyncMongoClient
|
||||
coll: AsyncCollection
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
# Speed up the tests by decreasing the event publish frequency.
|
||||
cls.knobs = client_knobs(
|
||||
events_queue_frequency=0.1, heartbeat_frequency=0.1, min_heartbeat_interval=0.1
|
||||
)
|
||||
cls.knobs.enable()
|
||||
cls.listener = ServerAndTopologyEventListener()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.knobs.disable()
|
||||
|
||||
@async_client_context.require_failCommand_fail_point
|
||||
async def asyncSetUp(self):
|
||||
await super().asyncSetUp()
|
||||
|
||||
retry_writes = async_client_context.supports_transactions()
|
||||
self.test_client = await self.async_rs_or_single_client(
|
||||
event_listeners=[self.listener], retryWrites=retry_writes
|
||||
)
|
||||
self.coll = self.test_client[self.client.db.name].test
|
||||
await self.coll.insert_one({})
|
||||
self.listener.reset()
|
||||
|
||||
async def asyncTearDown(self):
|
||||
await super().asyncTearDown()
|
||||
|
||||
async def _test_app_error(self, fail_command_opts, expected_error):
|
||||
address = await self.test_client.address
|
||||
|
||||
# Test that an application error causes a ServerDescriptionChangedEvent
|
||||
# to be published.
|
||||
data = {"failCommands": ["insert"]}
|
||||
data.update(fail_command_opts)
|
||||
fail_insert = {
|
||||
"configureFailPoint": "failCommand",
|
||||
"mode": {"times": 1},
|
||||
"data": data,
|
||||
}
|
||||
async with self.fail_point(fail_insert):
|
||||
if self.test_client.options.retry_writes:
|
||||
await self.coll.insert_one({})
|
||||
else:
|
||||
with self.assertRaises(expected_error):
|
||||
await self.coll.insert_one({})
|
||||
await self.coll.insert_one({})
|
||||
|
||||
def marked_unknown(event):
|
||||
return (
|
||||
isinstance(event, monitoring.ServerDescriptionChangedEvent)
|
||||
and event.server_address == address
|
||||
and not event.new_description.is_server_type_known
|
||||
)
|
||||
|
||||
def discovered_node(event):
|
||||
return (
|
||||
isinstance(event, monitoring.ServerDescriptionChangedEvent)
|
||||
and event.server_address == address
|
||||
and not event.previous_description.is_server_type_known
|
||||
and event.new_description.is_server_type_known
|
||||
)
|
||||
|
||||
def marked_unknown_and_rediscovered():
|
||||
return (
|
||||
len(self.listener.matching(marked_unknown)) >= 1
|
||||
and len(self.listener.matching(discovered_node)) >= 1
|
||||
)
|
||||
|
||||
# Topology events are not published synchronously
|
||||
await async_wait_until(marked_unknown_and_rediscovered, "rediscover node")
|
||||
|
||||
# Expect a single ServerDescriptionChangedEvent for the network error.
|
||||
marked_unknown_events = self.listener.matching(marked_unknown)
|
||||
self.assertEqual(len(marked_unknown_events), 1, marked_unknown_events)
|
||||
self.assertIsInstance(marked_unknown_events[0].new_description.error, expected_error)
|
||||
|
||||
async def test_network_error_publishes_events(self):
|
||||
await self._test_app_error({"closeConnection": True}, ConnectionFailure)
|
||||
|
||||
# In 4.4+, not primary errors from failCommand don't cause SDAM state
|
||||
# changes because topologyVersion is not incremented.
|
||||
@async_client_context.require_version_max(4, 3)
|
||||
async def test_not_primary_error_publishes_events(self):
|
||||
await self._test_app_error(
|
||||
{"errorCode": 10107, "closeConnection": False, "errorLabels": ["RetryableWriteError"]},
|
||||
NotPrimaryError,
|
||||
)
|
||||
|
||||
async def test_shutdown_error_publishes_events(self):
|
||||
await self._test_app_error(
|
||||
{"errorCode": 91, "closeConnection": False, "errorLabels": ["RetryableWriteError"]},
|
||||
NotPrimaryError,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -15,8 +15,10 @@
|
||||
"""Test MongoClient's mongos load balancing using a mock."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
import threading
|
||||
from test.helpers import ConcurrentRunner
|
||||
|
||||
from pymongo.operations import _Op
|
||||
|
||||
@ -30,14 +32,10 @@ from pymongo.errors import AutoReconnect, InvalidOperation
|
||||
from pymongo.server_selectors import writable_server_selector
|
||||
from pymongo.topology_description import TOPOLOGY_TYPE
|
||||
|
||||
|
||||
@client_context.require_connection
|
||||
@client_context.require_no_load_balancer
|
||||
def setUpModule():
|
||||
pass
|
||||
_IS_SYNC = True
|
||||
|
||||
|
||||
class SimpleOp(threading.Thread):
|
||||
class SimpleOp(ConcurrentRunner):
|
||||
def __init__(self, client):
|
||||
super().__init__()
|
||||
self.client = client
|
||||
@ -48,15 +46,15 @@ class SimpleOp(threading.Thread):
|
||||
self.passed = True # No exception raised.
|
||||
|
||||
|
||||
def do_simple_op(client, nthreads):
|
||||
threads = [SimpleOp(client) for _ in range(nthreads)]
|
||||
for t in threads:
|
||||
def do_simple_op(client, ntasks):
|
||||
tasks = [SimpleOp(client) for _ in range(ntasks)]
|
||||
for t in tasks:
|
||||
t.start()
|
||||
|
||||
for t in threads:
|
||||
for t in tasks:
|
||||
t.join()
|
||||
|
||||
for t in threads:
|
||||
for t in tasks:
|
||||
assert t.passed
|
||||
|
||||
|
||||
@ -68,6 +66,11 @@ def writable_addresses(topology):
|
||||
|
||||
|
||||
class TestMongosLoadBalancing(MockClientTest):
|
||||
@client_context.require_connection
|
||||
@client_context.require_no_load_balancer
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
def mock_client(self, **kwargs):
|
||||
mock_client = MockClient(
|
||||
standalones=[],
|
||||
@ -98,7 +101,7 @@ class TestMongosLoadBalancing(MockClientTest):
|
||||
wait_until(lambda: len(client.nodes) == 3, "connect to all mongoses")
|
||||
|
||||
def test_failover(self):
|
||||
nthreads = 10
|
||||
ntasks = 10
|
||||
client = connected(self.mock_client(localThresholdMS=0.001))
|
||||
wait_until(lambda: len(client.nodes) == 3, "connect to all mongoses")
|
||||
|
||||
@ -118,14 +121,14 @@ class TestMongosLoadBalancing(MockClientTest):
|
||||
|
||||
passed.append(True)
|
||||
|
||||
threads = [threading.Thread(target=f) for _ in range(nthreads)]
|
||||
for t in threads:
|
||||
tasks = [ConcurrentRunner(target=f) for _ in range(ntasks)]
|
||||
for t in tasks:
|
||||
t.start()
|
||||
|
||||
for t in threads:
|
||||
for t in tasks:
|
||||
t.join()
|
||||
|
||||
self.assertEqual(nthreads, len(passed))
|
||||
self.assertEqual(ntasks, len(passed))
|
||||
|
||||
# Down host removed from list.
|
||||
self.assertEqual(2, len(client.nodes))
|
||||
@ -183,8 +186,11 @@ class TestMongosLoadBalancing(MockClientTest):
|
||||
client.mock_rtts["a:1"] = 0.045
|
||||
|
||||
# Discover only b is within latency window.
|
||||
def predicate():
|
||||
return {("b", 2)} == writable_addresses(topology)
|
||||
|
||||
wait_until(
|
||||
lambda: {("b", 2)} == writable_addresses(topology),
|
||||
predicate,
|
||||
'discover server "a" is too far',
|
||||
)
|
||||
|
||||
|
||||
@ -15,10 +15,12 @@
|
||||
"""Run the sdam monitoring spec tests."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path[0:0] = [""]
|
||||
|
||||
@ -39,8 +41,13 @@ from pymongo.synchronous.collection import Collection
|
||||
from pymongo.synchronous.monitor import Monitor
|
||||
from pymongo.topology_description import TOPOLOGY_TYPE
|
||||
|
||||
_IS_SYNC = True
|
||||
|
||||
# Location of JSON test specifications.
|
||||
_TEST_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "sdam_monitoring")
|
||||
if _IS_SYNC:
|
||||
TEST_PATH = os.path.join(Path(__file__).resolve().parent, "sdam_monitoring")
|
||||
else:
|
||||
TEST_PATH = os.path.join(Path(__file__).resolve().parent.parent, "sdam_monitoring")
|
||||
|
||||
|
||||
def compare_server_descriptions(expected, actual):
|
||||
@ -247,7 +254,7 @@ def create_test(scenario_def):
|
||||
|
||||
|
||||
def create_tests():
|
||||
for dirpath, _, filenames in os.walk(_TEST_PATH):
|
||||
for dirpath, _, filenames in os.walk(TEST_PATH):
|
||||
for filename in filenames:
|
||||
with open(os.path.join(dirpath, filename)) as scenario_stream:
|
||||
scenario_def = json.load(scenario_stream, object_hook=object_hook)
|
||||
@ -268,31 +275,33 @@ class TestSdamMonitoring(IntegrationTest):
|
||||
coll: Collection
|
||||
|
||||
@classmethod
|
||||
@client_context.require_failCommand_fail_point
|
||||
def setUpClass(cls):
|
||||
super().setUp(cls)
|
||||
# Speed up the tests by decreasing the event publish frequency.
|
||||
cls.knobs = client_knobs(
|
||||
events_queue_frequency=0.1, heartbeat_frequency=0.1, min_heartbeat_interval=0.1
|
||||
)
|
||||
cls.knobs.enable()
|
||||
cls.listener = ServerAndTopologyEventListener()
|
||||
retry_writes = client_context.supports_transactions()
|
||||
cls.test_client = cls.unmanaged_rs_or_single_client(
|
||||
event_listeners=[cls.listener], retryWrites=retry_writes
|
||||
)
|
||||
cls.coll = cls.test_client[cls.client.db.name].test
|
||||
cls.coll.insert_one({})
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.test_client.close()
|
||||
cls.knobs.disable()
|
||||
super().tearDownClass()
|
||||
|
||||
@client_context.require_failCommand_fail_point
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
|
||||
retry_writes = client_context.supports_transactions()
|
||||
self.test_client = self.rs_or_single_client(
|
||||
event_listeners=[self.listener], retryWrites=retry_writes
|
||||
)
|
||||
self.coll = self.test_client[self.client.db.name].test
|
||||
self.coll.insert_one({})
|
||||
self.listener.reset()
|
||||
|
||||
def tearDown(self):
|
||||
super().tearDown()
|
||||
|
||||
def _test_app_error(self, fail_command_opts, expected_error):
|
||||
address = self.test_client.address
|
||||
|
||||
@ -334,7 +343,7 @@ class TestSdamMonitoring(IntegrationTest):
|
||||
and len(self.listener.matching(discovered_node)) >= 1
|
||||
)
|
||||
|
||||
# Topology events are published asynchronously
|
||||
# Topology events are not published synchronously
|
||||
wait_until(marked_unknown_and_rediscovered, "rediscover node")
|
||||
|
||||
# Expect a single ServerDescriptionChangedEvent for the network error.
|
||||
|
||||
@ -221,6 +221,7 @@ converted_tests = [
|
||||
"test_logger.py",
|
||||
"test_max_staleness.py",
|
||||
"test_monitoring.py",
|
||||
"test_mongos_load_balancing.py",
|
||||
"test_on_demand_csfle.py",
|
||||
"test_raw_bson.py",
|
||||
"test_read_concern.py",
|
||||
@ -231,6 +232,7 @@ converted_tests = [
|
||||
"test_retryable_writes.py",
|
||||
"test_retryable_writes_unified.py",
|
||||
"test_run_command.py",
|
||||
"test_sdam_monitoring_spec.py",
|
||||
"test_server_selection_logging.py",
|
||||
"test_session.py",
|
||||
"test_server_selection_rtt.py",
|
||||
|
||||
Loading…
Reference in New Issue
Block a user