mongodb/mongo-python-driver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

PYTHON-5421 Make parse_uri() return "options" as a dict rather than _CaseInsensitiveDictionary (#2413)

Browse Source
This commit is contained in:
Iris 2025-07-02 09:51:50 -07:00 committed by GitHub
parent 2eb18f18b2
commit 947fbe33ee
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 91 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pymongo/asynchronous/uri_parser.py
View File

@ -29,6 +29,7 @@ from pymongo.uri_parser_shared import (
SCHEME_LEN,
SRV_SCHEME_LEN,
_check_options,
_make_options_case_sensitive,
_validate_uri,
split_hosts,
split_options,
@ -113,6 +114,7 @@ async def parse_uri(
srv_max_hosts,
)
)
result["options"] = _make_options_case_sensitive(result["options"])
return result

2
pymongo/synchronous/uri_parser.py
View File

@ -29,6 +29,7 @@ from pymongo.uri_parser_shared import (
SCHEME_LEN,
SRV_SCHEME_LEN,
_check_options,
_make_options_case_sensitive,
_validate_uri,
split_hosts,
split_options,
@ -113,6 +114,7 @@ def parse_uri(
srv_max_hosts,
)
)
result["options"] = _make_options_case_sensitive(result["options"])
return result

62
pymongo/uri_parser_shared.py
View File

@ -54,6 +54,57 @@ SRV_SCHEME = "mongodb+srv://"
SRV_SCHEME_LEN = len(SRV_SCHEME)
DEFAULT_PORT = 27017
URI_OPTIONS = frozenset(
[
"appname",
"authMechanism",
"authMechanismProperties",
"authSource",
"compressors",
"connectTimeoutMS",
"directConnection",
"heartbeatFrequencyMS",
"journal",
"loadBalanced",
"localThresholdMS",
"maxIdleTimeMS",
"maxPoolSize",
"maxConnecting",
"maxStalenessSeconds",
"minPoolSize",
"proxyHost",
"proxyPort",
"proxyUsername",
"proxyPassword",
"readConcernLevel",
"readPreference",
"readPreferenceTags",
"replicaSet",
"retryReads",
"retryWrites",
"serverMonitoringMode",
"serverSelectionTimeoutMS",
"serverSelectionTryOnce",
"socketTimeoutMS",
"srvMaxHosts",
"srvServiceName",
"ssl",
"tls",
"tlsAllowInvalidCertificates",
"tlsAllowInvalidHostnames",
"tlsCAFile",
"tlsCertificateKeyFile",
"tlsCertificateKeyFilePassword",
"tlsDisableCertificateRevocationCheck",
"tlsDisableOCSPEndpointCheck",
"tlsInsecure",
"w",
"waitQueueTimeoutMS",
"wTimeoutMS",
"zlibCompressionLevel",
]
)
def _unquoted_percent(s: str) -> bool:
"""Check for unescaped percent signs.
@ -550,3 +601,14 @@ def _validate_uri(
"options": options,
"fqdn": fqdn,
}
def _make_options_case_sensitive(options: _CaseInsensitiveDictionary) -> dict[str, Any]:
case_sensitive = {}
for option in URI_OPTIONS:
if option.lower() in options:
case_sensitive[option] = options[option]
options.pop(option)
for k, v in options.items():
case_sensitive[k] = v
return case_sensitive

4
test/asynchronous/test_discovery_and_monitoring.py
View File

@ -91,8 +91,8 @@ async def create_mock_topology(uri, monitor_class=DummyMonitor):
replica_set_name = None
direct_connection = None
load_balanced = None
if "replicaset" in parsed_uri["options"]:
replica_set_name = parsed_uri["options"]["replicaset"]
if "replicaSet" in parsed_uri["options"]:
replica_set_name = parsed_uri["options"]["replicaSet"]
if "directConnection" in parsed_uri["options"]:
direct_connection = parsed_uri["options"]["directConnection"]
if "loadBalanced" in parsed_uri["options"]:

4
test/test_discovery_and_monitoring.py
View File

@ -91,8 +91,8 @@ def create_mock_topology(uri, monitor_class=DummyMonitor):
replica_set_name = None
direct_connection = None
load_balanced = None
if "replicaset" in parsed_uri["options"]:
replica_set_name = parsed_uri["options"]["replicaset"]
if "replicaSet" in parsed_uri["options"]:
replica_set_name = parsed_uri["options"]["replicaSet"]
if "directConnection" in parsed_uri["options"]:
direct_connection = parsed_uri["options"]["directConnection"]
if "loadBalanced" in parsed_uri["options"]:

35
test/test_uri_parser.py
View File

@ -142,9 +142,9 @@ class TestURI(unittest.TestCase):
self.assertTrue(split_options("wtimeoutms=500"))
self.assertEqual({"fsync": True}, split_options("fsync=true"))
self.assertEqual({"fsync": False}, split_options("fsync=false"))
self.assertEqual({"authmechanism": "GSSAPI"}, split_options("authMechanism=GSSAPI"))
self.assertEqual({"authMechanism": "GSSAPI"}, split_options("authMechanism=GSSAPI"))
self.assertEqual(
{"authmechanism": "SCRAM-SHA-1"}, split_options("authMechanism=SCRAM-SHA-1")
{"authMechanism": "SCRAM-SHA-1"}, split_options("authMechanism=SCRAM-SHA-1")
)
self.assertEqual({"authsource": "foobar"}, split_options("authSource=foobar"))
self.assertEqual({"maxpoolsize": 50}, split_options("maxpoolsize=50"))
@ -290,12 +290,12 @@ class TestURI(unittest.TestCase):
self.assertEqual(res, parse_uri('mongodb://localhost/test.name/with "delimiters'))
res = copy.deepcopy(orig)
res["options"] = {"readpreference": ReadPreference.SECONDARY.mongos_mode}
res["options"] = {"readPreference": ReadPreference.SECONDARY.mongos_mode}
self.assertEqual(res, parse_uri("mongodb://localhost/?readPreference=secondary"))
# Various authentication tests
res = copy.deepcopy(orig)
res["options"] = {"authmechanism": "SCRAM-SHA-256"}
res["options"] = {"authMechanism": "SCRAM-SHA-256"}
res["username"] = "user"
res["password"] = "password"
self.assertEqual(
@ -303,7 +303,7 @@ class TestURI(unittest.TestCase):
)
res = copy.deepcopy(orig)
res["options"] = {"authmechanism": "SCRAM-SHA-256", "authsource": "bar"}
res["options"] = {"authMechanism": "SCRAM-SHA-256", "authSource": "bar"}
res["username"] = "user"
res["password"] = "password"
res["database"] = "foo"
@ -315,7 +315,7 @@ class TestURI(unittest.TestCase):
)
res = copy.deepcopy(orig)
res["options"] = {"authmechanism": "SCRAM-SHA-256"}
res["options"] = {"authMechanism": "SCRAM-SHA-256"}
res["username"] = "user"
res["password"] = ""
self.assertEqual(res, parse_uri("mongodb://user:@localhost/?authMechanism=SCRAM-SHA-256"))
@ -327,7 +327,7 @@ class TestURI(unittest.TestCase):
self.assertEqual(res, parse_uri("mongodb://user%40domain.com:password@localhost/foo"))
res = copy.deepcopy(orig)
res["options"] = {"authmechanism": "GSSAPI"}
res["options"] = {"authMechanism": "GSSAPI"}
res["username"] = "user@domain.com"
res["password"] = "password"
res["database"] = "foo"
@ -337,7 +337,7 @@ class TestURI(unittest.TestCase):
)
res = copy.deepcopy(orig)
res["options"] = {"authmechanism": "GSSAPI"}
res["options"] = {"authMechanism": "GSSAPI"}
res["username"] = "user@domain.com"
res["password"] = ""
res["database"] = "foo"
@ -347,8 +347,8 @@ class TestURI(unittest.TestCase):
res = copy.deepcopy(orig)
res["options"] = {
"readpreference": ReadPreference.SECONDARY.mongos_mode,
"readpreferencetags": [
"readPreference": ReadPreference.SECONDARY.mongos_mode,
"readPreferenceTags": [
{"dc": "west", "use": "website"},
{"dc": "east", "use": "website"},
],
@ -368,8 +368,8 @@ class TestURI(unittest.TestCase):
res = copy.deepcopy(orig)
res["options"] = {
"readpreference": ReadPreference.SECONDARY.mongos_mode,
"readpreferencetags": [
"readPreference": ReadPreference.SECONDARY.mongos_mode,
"readPreferenceTags": [
{"dc": "west", "use": "website"},
{"dc": "east", "use": "website"},
{},
@ -462,6 +462,7 @@ class TestURI(unittest.TestCase):
"tlsInsecure": True,
"tlsDisableOCSPEndpointCheck": True,
}
print(parse_uri(uri)["options"])
self.assertEqual(res, parse_uri(uri)["options"])
def test_normalize_options(self):
@ -479,8 +480,8 @@ class TestURI(unittest.TestCase):
)
res = parse_uri(uri)
options: dict[str, Any] = {
"authmechanism": "MONGODB-AWS",
"authmechanismproperties": {"AWS_SESSION_TOKEN": unquoted_val},
"authMechanism": "MONGODB-AWS",
"authMechanismProperties": {"AWS_SESSION_TOKEN": unquoted_val},
}
self.assertEqual(options, res["options"])
@ -491,8 +492,8 @@ class TestURI(unittest.TestCase):
)
res = parse_uri(uri)
options = {
"readpreference": ReadPreference.SECONDARY.mongos_mode,
"readpreferencetags": [
"readPreference": ReadPreference.SECONDARY.mongos_mode,
"readPreferenceTags": [
{"dc": "west", unquoted_val: unquoted_val},
{"dc": "east", "use": unquoted_val},
],
@ -519,7 +520,7 @@ class TestURI(unittest.TestCase):
)
res = parse_uri(uri)
options = {
"authmechanism": "MONGODB-AWS",
"authMechanism": "MONGODB-AWS",
"authMechanismProperties": {"AWS_SESSION_TOKEN": token},
}
self.assertEqual(options, res["options"])

5
test/test_uri_spec.py
View File

@ -27,7 +27,7 @@ sys.path[0:0] = [""]
from test import unittest
from test.helpers import clear_warning_registry
from pymongo.common import INTERNAL_URI_OPTION_NAME_MAP, validate
from pymongo.common import INTERNAL_URI_OPTION_NAME_MAP, _CaseInsensitiveDictionary, validate
from pymongo.compression_support import _have_snappy
from pymongo.synchronous.uri_parser import parse_uri
@ -169,7 +169,8 @@ def create_test(test, test_workdir):
# Compare URI options.
err_msg = "For option %s expected %s but got %s"
if test["options"]:
opts = options["options"]
opts = _CaseInsensitiveDictionary()
opts.update(options["options"])
for opt in test["options"]:
lopt = opt.lower()
optname = INTERNAL_URI_OPTION_NAME_MAP.get(lopt, lopt)