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

Merge branch 'master' of github.com:mongodb/mongo-python-driver

Browse Source
This commit is contained in:
Steven Silvester 2025-06-30 13:56:46 -05:00
commit 354b166cb2
No known key found for this signature in database
GPG Key ID: B1BF5EC3A8B32F91
75 changed files with 1458 additions and 1918 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1478
.evergreen/generated_configs/tasks.yml
View File

File diff suppressed because it is too large Load Diff

17
.evergreen/generated_configs/variants.yml
View File

@ -146,7 +146,7 @@ buildvariants:
COMPRESSOR: zlib
- name: compression-zstd-rhel8
tasks:
- name: .test-standard !.server-4.0
- name: .test-standard !.server-4.2
display_name: Compression zstd RHEL8
run_on:
- rhel87-small
@ -522,13 +522,6 @@ buildvariants:
PYTHON_BINARY: /opt/python/3.9/bin/python3
# Server version tests
- name: mongodb-v4.0
tasks:
- name: .server-version
display_name: "* MongoDB v4.0"
run_on:
- rhel87-small
tags: [coverage_tag]
- name: mongodb-v4.2
tasks:
- name: .server-version
@ -664,11 +657,3 @@ buildvariants:
- rhel87-small
expansions:
STORAGE_ENGINE: inmemory
- name: storage-mmapv1-rhel8
tasks:
- name: .test-standard !.sharded_cluster-auth-ssl .server-4.0
display_name: Storage MMAPv1 RHEL8
run_on:
- rhel87-small
expansions:
STORAGE_ENGINE: mmapv1

3
.evergreen/resync-specs.sh
View File

@ -131,6 +131,9 @@ do
gridfs)
cpjson gridfs/tests gridfs
;;
handshake)
cpjson mongodb-handshake/tests handshake
;;
index|index-management)
cpjson index-management/tests index_management
;;

12
.evergreen/scripts/generate_config.py
View File

@ -25,7 +25,6 @@ from generate_config_utils import (
get_task_name,
get_variant_name,
get_versions_from,
get_versions_until,
handle_c_ext,
write_functions_to_file,
write_tasks_to_file,
@ -196,7 +195,7 @@ def create_compression_variants():
for compressor in "snappy", "zlib", "zstd":
expansions = dict(COMPRESSOR=compressor)
if compressor == "zstd":
tasks = [".test-standard !.server-4.0"]
tasks = [".test-standard !.server-4.2"]
else:
tasks = [".test-standard"]
display_name = get_variant_name(f"Compression {compressor}", host)
@ -249,16 +248,11 @@ def create_pyopenssl_variants():
def create_storage_engine_variants():
host = DEFAULT_HOST
engines = ["InMemory", "MMAPv1"]
engines = ["InMemory"]
variants = []
for engine in engines:
expansions = dict(STORAGE_ENGINE=engine.lower())
if engine == engines[0]:
tasks = [".test-standard .standalone-noauth-nossl"]
else:
# MongoDB 4.2 drops support for MMAPv1
versions = get_versions_until("4.0")
tasks = [f".test-standard !.sharded_cluster-auth-ssl .server-{v}" for v in versions]
tasks = [".test-standard .standalone-noauth-nossl"]
display_name = get_variant_name(f"Storage {engine}", host)
variant = create_variant(tasks, display_name, host=host, expansions=expansions)
variants.append(variant)

2
.evergreen/scripts/generate_config_utils.py
View File

@ -21,7 +21,7 @@ from shrub.v3.shrub_service import ShrubService
# Globals
##############
ALL_VERSIONS = ["4.0", "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "rapid", "latest"]
ALL_VERSIONS = ["4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "rapid", "latest"]
CPYTHONS = ["3.9", "3.10", "3.11", "3.12", "3.13"]
PYPYS = ["pypy3.10"]
ALL_PYTHONS = CPYTHONS + PYPYS

1
.github/CODEOWNERS vendored Normal file
View File

@ -0,0 +1 @@
* @mongodb/dbx-python

4
.github/workflows/codeql.yml vendored
View File

@ -46,7 +46,7 @@ jobs:
# Initializes the CodeQL tools for scanning.
- name: Initialize CodeQL
uses: github/codeql-action/init@ce28f5bb42b7a9f2c824e633a3f6ee835bab6858 # v3
uses: github/codeql-action/init@39edc492dbe16b1465b0cafca41432d857bdb31a # v3
with:
languages: ${{ matrix.language }}
build-mode: ${{ matrix.build-mode }}
@ -63,6 +63,6 @@ jobs:
pip install -e .
- name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@ce28f5bb42b7a9f2c824e633a3f6ee835bab6858 # v3
uses: github/codeql-action/analyze@39edc492dbe16b1465b0cafca41432d857bdb31a # v3
with:
category: "/language:${{matrix.language}}"

64
.github/workflows/test-python.yml vendored
View File

@ -25,7 +25,7 @@ jobs:
- name: Install just
uses: extractions/setup-just@e33e0265a09d6d736e2ee1e0eb685ef1de4669ff # v3
- name: Install uv
uses: astral-sh/setup-uv@f0ec1fc3b38f5e7cd731bb6ce540c5af426746bb # v5
uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v5
with:
enable-cache: true
python-version: "3.9"
@ -65,7 +65,7 @@ jobs:
- name: Install just
uses: extractions/setup-just@e33e0265a09d6d736e2ee1e0eb685ef1de4669ff # v3
- name: Install uv
uses: astral-sh/setup-uv@f0ec1fc3b38f5e7cd731bb6ce540c5af426746bb # v5
uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v5
with:
enable-cache: true
python-version: ${{ matrix.python-version }}
@ -88,7 +88,7 @@ jobs:
- name: Install just
uses: extractions/setup-just@e33e0265a09d6d736e2ee1e0eb685ef1de4669ff # v3
- name: Install uv
uses: astral-sh/setup-uv@f0ec1fc3b38f5e7cd731bb6ce540c5af426746bb # v5
uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v5
with:
enable-cache: true
python-version: "3.9"
@ -111,7 +111,7 @@ jobs:
with:
persist-credentials: false
- name: Install uv
uses: astral-sh/setup-uv@f0ec1fc3b38f5e7cd731bb6ce540c5af426746bb # v5
uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v5
with:
enable-cache: true
python-version: "3.9"
@ -130,7 +130,7 @@ jobs:
with:
persist-credentials: false
- name: Install uv
uses: astral-sh/setup-uv@f0ec1fc3b38f5e7cd731bb6ce540c5af426746bb # v5
uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v5
with:
enable-cache: true
python-version: "3.9"
@ -152,7 +152,7 @@ jobs:
with:
persist-credentials: false
- name: Install uv
uses: astral-sh/setup-uv@f0ec1fc3b38f5e7cd731bb6ce540c5af426746bb # v5
uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v5
with:
enable-cache: true
python-version: "${{matrix.python}}"
@ -220,3 +220,55 @@ jobs:
which python
pip install -e ".[test]"
PYMONGO_MUST_CONNECT=1 pytest -v -k client_context
test_minimum:
permissions:
contents: read
runs-on: ubuntu-latest
name: Test using minimum dependencies and supported Python
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- name: Install uv
uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v5
with:
python-version: '3.9'
- name: Start MongoDB
uses: supercharge/mongodb-github-action@90004df786821b6308fb02299e5835d0dae05d0d # 1.12.0
with:
mongodb-version: 6.0
# Async and our test_dns do not support dnspython 1.X, so we don't run async or dns tests here
- name: Run tests
shell: bash
run: |
uv venv
source .venv/bin/activate
uv pip install -e ".[test]" --resolution=lowest-direct
pytest -v test/test_srv_polling.py
test_minimum_for_async:
permissions:
contents: read
runs-on: ubuntu-latest
name: Test async's minimum dependencies and Python
steps:
- uses: actions/checkout@v4
with:
persist-credentials: false
- name: Install uv
uses: astral-sh/setup-uv@bd01e18f51369d5a26f1651c3cb451d3417e3bba # v5
with:
python-version: '3.9'
- name: Start MongoDB
uses: supercharge/mongodb-github-action@90004df786821b6308fb02299e5835d0dae05d0d # 1.12.0
with:
mongodb-version: 6.0
# The lifetime kwarg we use in srv resolution was added to the async resolver API in dnspython 2.1.0
- name: Run tests
shell: bash
run: |
uv venv
source .venv/bin/activate
uv pip install -e ".[test]" --resolution=lowest-direct dnspython==2.1.0 --force-reinstall
pytest -v test/test_srv_polling.py test/test_dns.py test/asynchronous/test_srv_polling.py test/asynchronous/test_dns.py

4
.github/workflows/zizmor.yml vendored
View File

@ -18,7 +18,7 @@ jobs:
with:
persist-credentials: false
- name: Setup Rust
uses: actions-rust-lang/setup-rust-toolchain@9d7e65c320fdb52dcd45ffaa68deb6c02c8754d9 # v1
uses: actions-rust-lang/setup-rust-toolchain@fb51252c7ba57d633bc668f941da052e410add48 # v1
- name: Get zizmor
run: cargo install zizmor
- name: Run zizmor 🌈
@ -26,7 +26,7 @@ jobs:
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
- name: Upload SARIF file
uses: github/codeql-action/upload-sarif@ce28f5bb42b7a9f2c824e633a3f6ee835bab6858 # v3
uses: github/codeql-action/upload-sarif@39edc492dbe16b1465b0cafca41432d857bdb31a # v3
with:
sarif_file: results.sarif
category: zizmor

12
CONTRIBUTING.md
View File

@ -460,3 +460,15 @@ partially-converted asynchronous version of the same name to the `test/asynchron
Use this generated file as a starting point for the completed conversion.
The script is used like so: `python tools/convert_test_to_async.py [test_file.py]`
## Generating a flame graph using py-spy
To profile a test script and generate a flame graph, follow these steps:
1. Install `py-spy` if you haven't already:
```bash
pip install py-spy
```
2. Inside your test script, perform any required setup and then loop over the code you want to profile for improved sampling.
3. Run `py-spy record -o <output.svg> -r <sample_rate=100> -- python <path/to/script>` to generate a `.svg` file containing the flame graph.
(Note: on macOS you will need to run this command using `sudo` to allow `py-spy` to attach to the Python process.)
4. If you need to include native code (for example the C extensions), profiling should be done on a Linux system, as macOS and Windows do not support the `--native` option of `py-spy`.
Creating an ubuntu Evergreen spawn host and using `scp` to copy the flamegraph `.svg` file back to your local machine is the best way to do this.

9
bson/binary.py
View File

@ -490,6 +490,11 @@ class Binary(bytes):
dtype = BinaryVectorDtype(dtype)
n_values = len(self) - position
if padding and dtype != BinaryVectorDtype.PACKED_BIT:
raise ValueError(
f"Corrupt data. Padding ({padding}) must be 0 for all but PACKED_BIT dtypes. ({dtype=})"
)
if dtype == BinaryVectorDtype.INT8:
dtype_format = "b"
format_string = f"<{n_values}{dtype_format}"
@ -510,6 +515,10 @@ class Binary(bytes):
elif dtype == BinaryVectorDtype.PACKED_BIT:
# data packed as uint8
if padding and not n_values:
raise ValueError("Corrupt data. Vector has a padding P, but no data.")
if padding > 7 or padding < 0:
raise ValueError(f"Corrupt data. Padding ({padding}) must be between 0 and 7.")
dtype_format = "B"
format_string = f"<{n_values}{dtype_format}"
unpacked_uint8s = list(struct.unpack_from(format_string, self, position))

6
doc/changelog.rst
View File

@ -7,6 +7,12 @@ PyMongo 4.14 brings a number of changes including:
- Added :attr:`bson.codec_options.TypeRegistry.codecs` and :attr:`bson.codec_options.TypeRegistry.fallback_encoder` properties
to allow users to directly access the type codecs and fallback encoder for a given :class:`bson.codec_options.TypeRegistry`.
- Added :meth:`pymongo.asynchronous.mongo_client.AsyncMongoClient.append_metadata` and
:meth:`pymongo.mongo_client.MongoClient.append_metadata` to allow instantiated MongoClients to send client metadata
on-demand
- Introduces a minor breaking change. When encoding :class:`bson.binary.BinaryVector`, a ``ValueError`` will be raised
if the 'padding' metadata field is < 0 or > 7, or non-zero for any type other than PACKED_BIT.
Changes in Version 4.13.2 (2025/06/17)
--------------------------------------

2
hatch_build.py
View File

@ -19,7 +19,7 @@ class CustomHook(BuildHookInterface):
here = Path(__file__).parent.resolve()
sys.path.insert(0, str(here))
subprocess.check_call([sys.executable, "_setup.py", "build_ext", "-i"])
subprocess.run([sys.executable, "_setup.py", "build_ext", "-i"], check=True)
# Ensure wheel is marked as binary and contains the binary files.
build_data["infer_tag"] = True

15
pymongo/asynchronous/mongo_client.py
View File

@ -70,6 +70,7 @@ from pymongo.asynchronous.command_cursor import AsyncCommandCursor
from pymongo.asynchronous.settings import TopologySettings
from pymongo.asynchronous.topology import Topology, _ErrorContext
from pymongo.client_options import ClientOptions
from pymongo.driver_info import DriverInfo
from pymongo.errors import (
AutoReconnect,
BulkWriteError,
@ -1040,6 +1041,20 @@ class AsyncMongoClient(common.BaseObject, Generic[_DocumentType]):
self._kill_cursors_executor = executor
self._opened = False
def append_metadata(self, driver_info: DriverInfo) -> None:
"""Appends the given metadata to existing driver metadata.
:param driver_info: a :class:`~pymongo.driver_info.DriverInfo`
.. versionadded:: 4.14
"""
if not isinstance(driver_info, DriverInfo):
raise TypeError(
f"driver_info must be an instance of DriverInfo, not {type(driver_info)}"
)
self._options.pool_options._update_metadata(driver_info)
def _should_pin_cursor(self, session: Optional[AsyncClientSession]) -> Optional[bool]:
return self._options.load_balanced and not (session and session.in_transaction)

4
pymongo/common.py
View File

@ -66,8 +66,8 @@ MAX_WIRE_VERSION = 0
MAX_WRITE_BATCH_SIZE = 100000
# What this version of PyMongo supports.
MIN_SUPPORTED_SERVER_VERSION = "4.0"
MIN_SUPPORTED_WIRE_VERSION = 7
MIN_SUPPORTED_SERVER_VERSION = "4.2"
MIN_SUPPORTED_WIRE_VERSION = 8
# MongoDB 8.0
MAX_SUPPORTED_WIRE_VERSION = 25

32
pymongo/pool_options.py
View File

@ -376,18 +376,7 @@ class PoolOptions:
"async",
)
if driver:
if driver.name:
self.__metadata["driver"]["name"] = "{}|{}".format(
self.__metadata["driver"]["name"],
driver.name,
)
if driver.version:
self.__metadata["driver"]["version"] = "{}|{}".format(
_METADATA["driver"]["version"],
driver.version,
)
if driver.platform:
self.__metadata["platform"] = "{}|{}".format(_METADATA["platform"], driver.platform)
self._update_metadata(driver)
env = _metadata_env()
if env:
@ -395,6 +384,25 @@ class PoolOptions:
_truncate_metadata(self.__metadata)
def _update_metadata(self, driver: DriverInfo) -> None:
"""Updates the client's metadata"""
metadata = copy.deepcopy(self.__metadata)
if driver.name:
metadata["driver"]["name"] = "{}|{}".format(
metadata["driver"]["name"],
driver.name,
)
if driver.version:
metadata["driver"]["version"] = "{}|{}".format(
metadata["driver"]["version"],
driver.version,
)
if driver.platform:
metadata["platform"] = "{}|{}".format(metadata["platform"], driver.platform)
self.__metadata = metadata
@property
def _credentials(self) -> Optional[MongoCredential]:
"""A :class:`~pymongo.auth.MongoCredentials` instance or None."""

15
pymongo/synchronous/mongo_client.py
View File

@ -62,6 +62,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry
from bson.timestamp import Timestamp
from pymongo import _csot, common, helpers_shared, periodic_executor
from pymongo.client_options import ClientOptions
from pymongo.driver_info import DriverInfo
from pymongo.errors import (
AutoReconnect,
BulkWriteError,
@ -1040,6 +1041,20 @@ class MongoClient(common.BaseObject, Generic[_DocumentType]):
self._kill_cursors_executor = executor
self._opened = False
def append_metadata(self, driver_info: DriverInfo) -> None:
"""Appends the given metadata to existing driver metadata.
:param driver_info: a :class:`~pymongo.driver_info.DriverInfo`
.. versionadded:: 4.14
"""
if not isinstance(driver_info, DriverInfo):
raise TypeError(
f"driver_info must be an instance of DriverInfo, not {type(driver_info)}"
)
self._options.pool_options._update_metadata(driver_info)
def _should_pin_cursor(self, session: Optional[ClientSession]) -> Optional[bool]:
return self._options.load_balanced and not (session and session.in_transaction)

22
test/__init__.py
View File

@ -369,7 +369,7 @@ class ClientContext:
if self._fips_enabled is not None:
return self._fips_enabled
try:
subprocess.check_call(["fips-mode-setup", "--is-enabled"])
subprocess.run(["fips-mode-setup", "--is-enabled"], check=True)
self._fips_enabled = True
except (subprocess.SubprocessError, FileNotFoundError):
self._fips_enabled = False
@ -508,19 +508,6 @@ class ClientContext:
func=func,
)
def require_no_mmap(self, func):
"""Run a test only if the server is not using the MMAPv1 storage
engine. Only works for standalone and replica sets; tests are
run regardless of storage engine on sharded clusters.
"""
def is_not_mmap():
if self.is_mongos:
return True
return self.storage_engine != "mmapv1"
return self._require(is_not_mmap, "Storage engine must not be MMAPv1", func=func)
def require_version_min(self, *ver):
"""Run a test only if the server version is at least ``version``."""
other_version = Version(*ver)
@ -651,7 +638,7 @@ class ClientContext:
def require_change_streams(self, func):
"""Run a test only if the server supports change streams."""
return self.require_no_mmap(self.require_no_standalone(func))
return self.require_no_standalone(func)
def is_topology_type(self, topologies):
unknown = set(topologies) - {
@ -754,8 +741,6 @@ class ClientContext:
return self._require(lambda: self.sessions_enabled, "Sessions not supported", func=func)
def supports_retryable_writes(self):
if self.storage_engine == "mmapv1":
return False
if not self.sessions_enabled:
return False
return self.is_mongos or self.is_rs
@ -769,9 +754,6 @@ class ClientContext:
)
def supports_transactions(self):
if self.storage_engine == "mmapv1":
return False
if self.version.at_least(4, 1, 8):
return self.is_mongos or self.is_rs

22
test/asynchronous/__init__.py
View File

@ -369,7 +369,7 @@ class AsyncClientContext:
if self._fips_enabled is not None:
return self._fips_enabled
try:
subprocess.check_call(["fips-mode-setup", "--is-enabled"])
subprocess.run(["fips-mode-setup", "--is-enabled"], check=True)
self._fips_enabled = True
except (subprocess.SubprocessError, FileNotFoundError):
self._fips_enabled = False
@ -508,19 +508,6 @@ class AsyncClientContext:
func=func,
)
def require_no_mmap(self, func):
"""Run a test only if the server is not using the MMAPv1 storage
engine. Only works for standalone and replica sets; tests are
run regardless of storage engine on sharded clusters.
"""
def is_not_mmap():
if self.is_mongos:
return True
return self.storage_engine != "mmapv1"
return self._require(is_not_mmap, "Storage engine must not be MMAPv1", func=func)
def require_version_min(self, *ver):
"""Run a test only if the server version is at least ``version``."""
other_version = Version(*ver)
@ -651,7 +638,7 @@ class AsyncClientContext:
def require_change_streams(self, func):
"""Run a test only if the server supports change streams."""
return self.require_no_mmap(self.require_no_standalone(func))
return self.require_no_standalone(func)
async def is_topology_type(self, topologies):
unknown = set(topologies) - {
@ -754,8 +741,6 @@ class AsyncClientContext:
return self._require(lambda: self.sessions_enabled, "Sessions not supported", func=func)
def supports_retryable_writes(self):
if self.storage_engine == "mmapv1":
return False
if not self.sessions_enabled:
return False
return self.is_mongos or self.is_rs
@ -769,9 +754,6 @@ class AsyncClientContext:
)
def supports_transactions(self):
if self.storage_engine == "mmapv1":
return False
if self.version.at_least(4, 1, 8):
return self.is_mongos or self.is_rs

12
test/asynchronous/test_bulk.py
View File

@ -165,7 +165,7 @@ class AsyncTestBulk(AsyncBulkTestBase):
async def test_update_many(self):
await self._test_update_many({"$set": {"foo": "bar"}})
@async_client_context.require_version_min(4, 1, 11)
@async_client_context.require_version_min(4, 2, 0)
async def test_update_many_pipeline(self):
await self._test_update_many([{"$set": {"foo": "bar"}}])
@ -206,7 +206,7 @@ class AsyncTestBulk(AsyncBulkTestBase):
async def test_update_one(self):
await self._test_update_one({"$set": {"foo": "bar"}})
@async_client_context.require_version_min(4, 1, 11)
@async_client_context.require_version_min(4, 2, 0)
async def test_update_one_pipeline(self):
await self._test_update_one([{"$set": {"foo": "bar"}}])
@ -994,7 +994,7 @@ class AsyncTestBulkWriteConcern(AsyncBulkTestBase):
# When talking to legacy servers there will be a
# write concern error for each operation.
self.assertTrue(len(details["writeConcernErrors"]) > 0)
self.assertGreater(len(details["writeConcernErrors"]), 0)
failed = details["writeConcernErrors"][0]
self.assertEqual(64, failed["code"])
@ -1035,7 +1035,7 @@ class AsyncTestBulkWriteConcern(AsyncBulkTestBase):
details,
)
self.assertTrue(len(details["writeConcernErrors"]) > 1)
self.assertGreater(len(details["writeConcernErrors"]), 1)
failed = details["writeErrors"][0]
self.assertIn("duplicate", failed["errmsg"])
@ -1073,7 +1073,7 @@ class AsyncTestBulkWriteConcern(AsyncBulkTestBase):
self.assertEqual(0, len(details["writeErrors"]))
# When talking to legacy servers there will be a
# write concern error for each operation.
self.assertTrue(len(details["writeConcernErrors"]) > 1)
self.assertGreater(len(details["writeConcernErrors"]), 1)
await self.coll.delete_many({})
await self.coll.create_index("a", unique=True)
@ -1100,7 +1100,7 @@ class AsyncTestBulkWriteConcern(AsyncBulkTestBase):
self.assertEqual(1, len(details["writeErrors"]))
# When talking to legacy servers there will be a
# write concern error for each operation.
self.assertTrue(len(details["writeConcernErrors"]) > 1)
self.assertGreater(len(details["writeConcernErrors"]), 1)
failed = details["writeErrors"][0]
self.assertEqual(2, failed["index"])

95
test/asynchronous/test_change_stream.py
View File

@ -267,7 +267,7 @@ class APITestsMixin:
# $changeStream.startAtOperationTime was added in 4.0.0.
@no_type_check
@async_client_context.require_version_min(4, 0, 0)
@async_client_context.require_version_min(4, 2, 0)
async def test_start_at_operation_time(self):
optime = await self.get_start_at_operation_time()
@ -436,7 +436,7 @@ class APITestsMixin:
await self._test_get_invalidate_event(change_stream)
@no_type_check
@async_client_context.require_version_min(4, 1, 1)
@async_client_context.require_version_min(4, 2, 0)
async def test_start_after(self):
resume_token = await self.get_resume_token(invalidate=True)
@ -452,7 +452,7 @@ class APITestsMixin:
self.assertEqual(change["fullDocument"], {"_id": 2})
@no_type_check
@async_client_context.require_version_min(4, 1, 1)
@async_client_context.require_version_min(4, 2, 0)
async def test_start_after_resume_process_with_changes(self):
resume_token = await self.get_resume_token(invalidate=True)
@ -563,27 +563,16 @@ class ProseSpecTestsMixin:
)
# Prose test no. 1
@async_client_context.require_version_min(4, 0, 7)
@async_client_context.require_version_min(4, 2, 0)
async def test_update_resume_token(self):
await self._test_update_resume_token(self._get_expected_resume_token)
# Prose test no. 1
@async_client_context.require_version_max(4, 0, 7)
async def test_update_resume_token_legacy(self):
await self._test_update_resume_token(self._get_expected_resume_token_legacy)
# Prose test no. 2
@async_client_context.require_version_min(4, 1, 8)
@async_client_context.require_version_min(4, 2, 0)
async def test_raises_error_on_missing_id_418plus(self):
# Server returns an error on 4.1.8+
await self._test_raises_error_on_missing_id(OperationFailure)
# Prose test no. 2
@async_client_context.require_version_max(4, 1, 8)
async def test_raises_error_on_missing_id_418minus(self):
# PyMongo raises an error
await self._test_raises_error_on_missing_id(InvalidOperation)
# Prose test no. 3
@no_type_check
async def test_resume_on_error(self):
@ -642,40 +631,12 @@ class ProseSpecTestsMixin:
cursor.close = raise_error
await self.insert_one_and_check(change_stream, {"_id": 2})
# Prose test no. 9
@no_type_check
@async_client_context.require_version_min(4, 0, 0)
@async_client_context.require_version_max(4, 0, 7)
async def test_start_at_operation_time_caching(self):
# Case 1: change stream not started with startAtOperationTime
client, listener = self.client_with_listener("aggregate")
async with await self.change_stream_with_client(client) as cs:
await self.kill_change_stream_cursor(cs)
await cs.try_next()
cmd = listener.started_events[-1].command
self.assertIsNotNone(cmd["pipeline"][0]["$changeStream"].get("startAtOperationTime"))
# Case 2: change stream started with startAtOperationTime
listener.reset()
optime = await self.get_start_at_operation_time()
async with await self.change_stream_with_client(
client, start_at_operation_time=optime
) as cs:
await self.kill_change_stream_cursor(cs)
await cs.try_next()
cmd = listener.started_events[-1].command
self.assertEqual(
cmd["pipeline"][0]["$changeStream"].get("startAtOperationTime"),
optime,
str([k.command for k in listener.started_events]),
)
# Prose test no. 10 - SKIPPED
# This test is identical to prose test no. 3.
# Prose test no. 11
@no_type_check
@async_client_context.require_version_min(4, 0, 7)
@async_client_context.require_version_min(4, 2, 0)
async def test_resumetoken_empty_batch(self):
client, listener = await self._client_with_listener("getMore")
async with await self.change_stream_with_client(client) as change_stream:
@ -687,7 +648,7 @@ class ProseSpecTestsMixin:
# Prose test no. 11
@no_type_check
@async_client_context.require_version_min(4, 0, 7)
@async_client_context.require_version_min(4, 2, 0)
async def test_resumetoken_exhausted_batch(self):
client, listener = await self._client_with_listener("getMore")
async with await self.change_stream_with_client(client) as change_stream:
@ -697,38 +658,6 @@ class ProseSpecTestsMixin:
response = listener.succeeded_events[-1].reply
self.assertEqual(resume_token, response["cursor"]["postBatchResumeToken"])
# Prose test no. 12
@no_type_check
@async_client_context.require_version_max(4, 0, 7)
async def test_resumetoken_empty_batch_legacy(self):
resume_point = await self.get_resume_token()
# Empty resume token when neither resumeAfter or startAfter specified.
async with await self.change_stream() as change_stream:
await change_stream.try_next()
self.assertIsNone(change_stream.resume_token)
# Resume token value is same as resumeAfter.
async with await self.change_stream(resume_after=resume_point) as change_stream:
await change_stream.try_next()
resume_token = change_stream.resume_token
self.assertEqual(resume_token, resume_point)
# Prose test no. 12
@no_type_check
@async_client_context.require_version_max(4, 0, 7)
async def test_resumetoken_exhausted_batch_legacy(self):
# Resume token is _id of last change.
async with await self.change_stream() as change_stream:
change = await self._populate_and_exhaust_change_stream(change_stream)
self.assertEqual(change_stream.resume_token, change["_id"])
resume_point = change["_id"]
# Resume token is _id of last change even if resumeAfter is specified.
async with await self.change_stream(resume_after=resume_point) as change_stream:
change = await self._populate_and_exhaust_change_stream(change_stream)
self.assertEqual(change_stream.resume_token, change["_id"])
# Prose test no. 13
@no_type_check
async def test_resumetoken_partially_iterated_batch(self):
@ -770,13 +699,13 @@ class ProseSpecTestsMixin:
# Prose test no. 14
@no_type_check
@async_client_context.require_no_mongos
@async_client_context.require_version_min(4, 1, 1)
@async_client_context.require_version_min(4, 2, 0)
async def test_resumetoken_uniterated_nonempty_batch_startafter(self):
await self._test_resumetoken_uniterated_nonempty_batch("start_after")
# Prose test no. 17
@no_type_check
@async_client_context.require_version_min(4, 1, 1)
@async_client_context.require_version_min(4, 2, 0)
async def test_startafter_resume_uses_startafter_after_empty_getMore(self):
# Resume should use startAfter after no changes have been returned.
resume_point = await self.get_resume_token()
@ -796,7 +725,7 @@ class ProseSpecTestsMixin:
# Prose test no. 18
@no_type_check
@async_client_context.require_version_min(4, 1, 1)
@async_client_context.require_version_min(4, 2, 0)
async def test_startafter_resume_uses_resumeafter_after_nonempty_getMore(self):
# Resume should use resumeAfter after some changes have been returned.
resume_point = await self.get_resume_token()
@ -843,7 +772,7 @@ class ProseSpecTestsMixin:
class TestClusterAsyncChangeStream(TestAsyncChangeStreamBase, APITestsMixin):
dbs: list
@async_client_context.require_version_min(4, 0, 0, -1)
@async_client_context.require_version_min(4, 2, 0)
@async_client_context.require_change_streams
async def asyncSetUp(self) -> None:
await super().asyncSetUp()
@ -903,7 +832,7 @@ class TestClusterAsyncChangeStream(TestAsyncChangeStreamBase, APITestsMixin):
class TestAsyncDatabaseAsyncChangeStream(TestAsyncChangeStreamBase, APITestsMixin):
@async_client_context.require_version_min(4, 0, 0, -1)
@async_client_context.require_version_min(4, 2, 0)
@async_client_context.require_change_streams
async def asyncSetUp(self) -> None:
await super().asyncSetUp()

2
test/asynchronous/test_client.py
View File

@ -1005,7 +1005,7 @@ class TestClient(AsyncIntegrationTest):
cursor = await self.client.list_databases()
self.assertIsInstance(cursor, AsyncCommandCursor)
helper_docs = await cursor.to_list()
self.assertTrue(len(helper_docs) > 0)
self.assertGreater(len(helper_docs), 0)
self.assertEqual(len(helper_docs), len(cmd_docs))
# PYTHON-3529 Some fields may change between calls, just compare names.
for helper_doc, cmd_doc in zip(helper_docs, cmd_docs):

215
test/asynchronous/test_client_metadata.py Normal file
View File

@ -0,0 +1,215 @@
# Copyright 2013-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.
from __future__ import annotations
import asyncio
import os
import pathlib
import time
import unittest
from test.asynchronous import AsyncIntegrationTest
from test.asynchronous.unified_format import generate_test_classes
from test.utils_shared import CMAPListener
from typing import Any, Optional
import pytest
from pymongo import AsyncMongoClient
from pymongo.driver_info import DriverInfo
from pymongo.monitoring import ConnectionClosedEvent
try:
from mockupdb import MockupDB, OpMsgReply
_HAVE_MOCKUPDB = True
except ImportError:
_HAVE_MOCKUPDB = False
pytestmark = pytest.mark.mockupdb
_IS_SYNC = False
# Location of JSON test specifications.
if _IS_SYNC:
_TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "handshake", "unified")
else:
_TEST_PATH = os.path.join(
pathlib.Path(__file__).resolve().parent.parent, "handshake", "unified"
)
# Generate unified tests.
globals().update(generate_test_classes(_TEST_PATH, module=__name__))
def _get_handshake_driver_info(request):
assert "client" in request
return request["client"]
class TestClientMetadataProse(AsyncIntegrationTest):
async def asyncSetUp(self):
await super().asyncSetUp()
self.server = MockupDB()
self.handshake_req = None
def respond(r):
if "ismaster" in r:
# then this is a handshake request
self.handshake_req = r
return r.reply(OpMsgReply(maxWireVersion=13))
self.server.autoresponds(respond)
self.server.run()
self.addAsyncCleanup(self.server.stop)
async def send_ping_and_get_metadata(
self, client: AsyncMongoClient, is_handshake: bool
) -> tuple[str, Optional[str], Optional[str], dict[str, Any]]:
# reset if handshake request
if is_handshake:
self.handshake_req: Optional[dict] = None
await client.admin.command("ping")
metadata = _get_handshake_driver_info(self.handshake_req)
driver_metadata = metadata["driver"]
name, version, platform = (
driver_metadata["name"],
driver_metadata["version"],
metadata["platform"],
)
return name, version, platform, metadata
async def check_metadata_added(
self,
client: AsyncMongoClient,
add_name: str,
add_version: Optional[str],
add_platform: Optional[str],
) -> None:
# send initial metadata
name, version, platform, metadata = await self.send_ping_and_get_metadata(client, True)
# wait for connection to become idle
await asyncio.sleep(0.005)
# add new metadata
client.append_metadata(DriverInfo(add_name, add_version, add_platform))
new_name, new_version, new_platform, new_metadata = await self.send_ping_and_get_metadata(
client, True
)
self.assertEqual(new_name, f"{name}|{add_name}" if add_name is not None else name)
self.assertEqual(
new_version,
f"{version}|{add_version}" if add_version is not None else version,
)
self.assertEqual(
new_platform,
f"{platform}|{add_platform}" if add_platform is not None else platform,
)
metadata.pop("driver")
metadata.pop("platform")
new_metadata.pop("driver")
new_metadata.pop("platform")
self.assertEqual(metadata, new_metadata)
async def test_append_metadata(self):
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
)
await self.check_metadata_added(client, "framework", "2.0", "Framework Platform")
async def test_append_metadata_platform_none(self):
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
)
await self.check_metadata_added(client, "framework", "2.0", None)
async def test_append_metadata_version_none(self):
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
)
await self.check_metadata_added(client, "framework", None, "Framework Platform")
async def test_append_metadata_platform_version_none(self):
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
)
await self.check_metadata_added(client, "framework", None, None)
async def test_multiple_successive_metadata_updates(self):
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string, maxIdleTimeMS=1, connect=False
)
client.append_metadata(DriverInfo("library", "1.2", "Library Platform"))
await self.check_metadata_added(client, "framework", "2.0", "Framework Platform")
async def test_multiple_successive_metadata_updates_platform_none(self):
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
)
client.append_metadata(DriverInfo("library", "1.2", "Library Platform"))
await self.check_metadata_added(client, "framework", "2.0", None)
async def test_multiple_successive_metadata_updates_version_none(self):
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
)
client.append_metadata(DriverInfo("library", "1.2", "Library Platform"))
await self.check_metadata_added(client, "framework", None, "Framework Platform")
async def test_multiple_successive_metadata_updates_platform_version_none(self):
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
)
client.append_metadata(DriverInfo("library", "1.2", "Library Platform"))
await self.check_metadata_added(client, "framework", None, None)
async def test_doesnt_update_established_connections(self):
listener = CMAPListener()
client = await self.async_rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
event_listeners=[listener],
)
# send initial metadata
name, version, platform, metadata = await self.send_ping_and_get_metadata(client, True)
self.assertIsNotNone(name)
self.assertIsNotNone(version)
self.assertIsNotNone(platform)
# add data
add_name, add_version, add_platform = "framework", "2.0", "Framework Platform"
client.append_metadata(DriverInfo(add_name, add_version, add_platform))
# check new data isn't sent
self.handshake_req: Optional[dict] = None
await client.admin.command("ping")
self.assertIsNone(self.handshake_req)
self.assertEqual(listener.event_count(ConnectionClosedEvent), 0)
if __name__ == "__main__":
unittest.main()

2
test/asynchronous/test_collection.py
View File

@ -500,7 +500,7 @@ class AsyncTestCollection(AsyncIntegrationTest):
# Sort by 'score' field.
cursor.sort([("score", {"$meta": "textScore"})])
results = await cursor.to_list()
self.assertTrue(results[0]["score"] >= results[1]["score"])
self.assertGreaterEqual(results[0]["score"], results[1]["score"])
await db.test.drop_indexes()

10
test/asynchronous/test_connections_survive_primary_stepdown_spec.py
View File

@ -122,18 +122,12 @@ class TestAsyncConnectionsSurvivePrimaryStepDown(AsyncIntegrationTest):
async def test_not_primary_keep_connection_pool(self):
await self.run_scenario(10107, True, self.verify_pool_not_cleared)
@async_client_context.require_version_min(4, 0, 0)
@async_client_context.require_version_max(4, 1, 0, -1)
@async_client_context.require_test_commands
async def test_not_primary_reset_connection_pool(self):
await self.run_scenario(10107, False, self.verify_pool_cleared)
@async_client_context.require_version_min(4, 0, 0)
@async_client_context.require_version_min(4, 2, 0)
@async_client_context.require_test_commands
async def test_shutdown_in_progress(self):
await self.run_scenario(91, False, self.verify_pool_cleared)
@async_client_context.require_version_min(4, 0, 0)
@async_client_context.require_version_min(4, 2, 0)
@async_client_context.require_test_commands
async def test_interrupted_at_shutdown(self):
await self.run_scenario(11600, False, self.verify_pool_cleared)

10
test/asynchronous/test_cursor.py
View File

@ -1190,15 +1190,6 @@ class TestCursor(AsyncIntegrationTest):
self.assertEqual(["b", "c"], distinct)
@async_client_context.require_version_max(4, 1, 0, -1)
async def test_max_scan(self):
await self.db.drop_collection("test")
await self.db.test.insert_many([{} for _ in range(100)])
self.assertEqual(100, len(await self.db.test.find().to_list()))
self.assertEqual(50, len(await self.db.test.find().max_scan(50).to_list()))
self.assertEqual(50, len(await self.db.test.find().max_scan(90).max_scan(50).to_list()))
async def test_with_statement(self):
await self.db.drop_collection("test")
await self.db.test.insert_many([{} for _ in range(100)])
@ -1600,7 +1591,6 @@ class TestRawBatchCursor(AsyncIntegrationTest):
async def test_collation(self):
await anext(self.db.test.find_raw_batches(collation=Collation("en_US")))
@async_client_context.require_no_mmap # MMAPv1 does not support read concern
async def test_read_concern(self):
await self.db.get_collection("test", write_concern=WriteConcern(w="majority")).insert_one(
{}

4
test/asynchronous/test_custom_types.py
View File

@ -953,7 +953,7 @@ class TestCollectionChangeStreamsWCustomTypes(
class TestDatabaseChangeStreamsWCustomTypes(
AsyncIntegrationTest, ChangeStreamsWCustomTypesTestMixin
):
@async_client_context.require_version_min(4, 0, 0)
@async_client_context.require_version_min(4, 2, 0)
@async_client_context.require_change_streams
async def asyncSetUp(self):
await super().asyncSetUp()
@ -973,7 +973,7 @@ class TestDatabaseChangeStreamsWCustomTypes(
class TestClusterChangeStreamsWCustomTypes(
AsyncIntegrationTest, ChangeStreamsWCustomTypesTestMixin
):
@async_client_context.require_version_min(4, 0, 0)
@async_client_context.require_version_min(4, 2, 0)
@async_client_context.require_change_streams
async def asyncSetUp(self):
await super().asyncSetUp()

2
test/asynchronous/test_database.py
View File

@ -239,7 +239,7 @@ class TestDatabase(AsyncIntegrationTest):
listener.reset()
await db.drop_collection("unique")
await db.create_collection("unique", check_exists=False)
self.assertTrue(len(listener.started_events) > 0)
self.assertGreater(len(listener.started_events), 0)
self.assertNotIn("listCollections", listener.started_command_names())
async def test_list_collections(self):

14
test/asynchronous/test_encryption.py
View File

@ -451,20 +451,6 @@ class TestClientMaxWireVersion(AsyncIntegrationTest):
async def asyncSetUp(self):
await super().asyncSetUp()
@async_client_context.require_version_max(4, 0, 99)
async def test_raise_max_wire_version_error(self):
opts = AutoEncryptionOpts(KMS_PROVIDERS, "keyvault.datakeys")
client = await self.async_rs_or_single_client(auto_encryption_opts=opts)
msg = "Auto-encryption requires a minimum MongoDB version of 4.2"
with self.assertRaisesRegex(ConfigurationError, msg):
await client.test.test.insert_one({})
with self.assertRaisesRegex(ConfigurationError, msg):
await client.admin.command("ping")
with self.assertRaisesRegex(ConfigurationError, msg):
await client.test.test.find_one({})
with self.assertRaisesRegex(ConfigurationError, msg):
await client.test.test.bulk_write([InsertOne({})])
async def test_raise_unsupported_error(self):
opts = AutoEncryptionOpts(KMS_PROVIDERS, "keyvault.datakeys")
client = await self.async_rs_or_single_client(auto_encryption_opts=opts)

1
test/asynchronous/test_examples.py
View File

@ -1162,7 +1162,6 @@ class TestTransactionExamples(AsyncIntegrationTest):
class TestCausalConsistencyExamples(AsyncIntegrationTest):
@async_client_context.require_secondaries_count(1)
@async_client_context.require_no_mmap
async def test_causal_consistency(self):
# Causal consistency examples
client = self.client

9
test/asynchronous/test_pooling.py
View File

@ -331,8 +331,9 @@ class TestPooling(_TestPoolingBase):
pass
duration = time.time() - start
self.assertTrue(
abs(wait_queue_timeout - duration) < 1,
self.assertLess(
abs(wait_queue_timeout - duration),
1,
f"Waited {duration:.2f} seconds for a socket, expected {wait_queue_timeout:f}",
)
@ -547,7 +548,7 @@ class TestPoolMaxSize(_TestPoolingBase):
await async_joinall(tasks)
self.assertEqual(ntasks, self.n_passed)
self.assertTrue(len(cx_pool.conns) > 1)
self.assertGreater(len(cx_pool.conns), 1)
self.assertEqual(0, cx_pool.requests)
async def test_max_pool_size_none(self):
@ -578,7 +579,7 @@ class TestPoolMaxSize(_TestPoolingBase):
await async_joinall(tasks)
self.assertEqual(ntasks, self.n_passed)
self.assertTrue(len(cx_pool.conns) > 1)
self.assertGreater(len(cx_pool.conns), 1)
self.assertEqual(cx_pool.max_pool_size, float("inf"))
async def test_max_pool_size_zero(self):

32
test/asynchronous/test_retryable_writes.py
View File

@ -140,40 +140,10 @@ class IgnoreDeprecationsTest(AsyncIntegrationTest):
self.deprecation_filter.stop()
class TestRetryableWritesMMAPv1(IgnoreDeprecationsTest):
knobs: client_knobs
async def asyncSetUp(self) -> None:
await super().asyncSetUp()
# Speed up the tests by decreasing the heartbeat frequency.
self.knobs = client_knobs(heartbeat_frequency=0.1, min_heartbeat_interval=0.1)
self.knobs.enable()
self.client = await self.async_rs_or_single_client(retryWrites=True)
self.db = self.client.pymongo_test
async def asyncTearDown(self) -> None:
self.knobs.disable()
@async_client_context.require_no_standalone
async def test_actionable_error_message(self):
if async_client_context.storage_engine != "mmapv1":
raise SkipTest("This cluster is not running MMAPv1")
expected_msg = (
"This MongoDB deployment does not support retryable "
"writes. Please add retryWrites=false to your "
"connection string."
)
for method, args, kwargs in retryable_single_statement_ops(self.db.retryable_write_test):
with self.assertRaisesRegex(OperationFailure, expected_msg):
await method(*args, **kwargs)
class TestRetryableWrites(IgnoreDeprecationsTest):
listener: OvertCommandListener
knobs: client_knobs
@async_client_context.require_no_mmap
async def asyncSetUp(self) -> None:
await super().asyncSetUp()
# Speed up the tests by decreasing the heartbeat frequency.
@ -425,7 +395,6 @@ class TestWriteConcernError(AsyncIntegrationTest):
fail_insert: dict
@async_client_context.require_replica_set
@async_client_context.require_no_mmap
@async_client_context.require_failCommand_fail_point
async def asyncSetUp(self) -> None:
await super().asyncSetUp()
@ -596,7 +565,6 @@ class TestPoolPausedError(AsyncIntegrationTest):
# TODO: Make this a real integration test where we stepdown the primary.
class TestRetryableWritesTxnNumber(IgnoreDeprecationsTest):
@async_client_context.require_replica_set
@async_client_context.require_no_mmap
async def test_increment_transaction_id_without_sending_command(self):
"""Test that the txnNumber field is properly incremented, even when
the first attempt fails before sending the command.

9
test/asynchronous/test_session.py
View File

@ -1045,14 +1045,6 @@ class TestCausalConsistency(AsyncUnitTest):
lambda coll, session: coll.find({}, session=session).explain()
)
@async_client_context.require_no_standalone
@async_client_context.require_version_max(4, 1, 0)
async def test_aggregate_out_does_not_include_read_concern(self):
async def alambda(coll, session):
await (await coll.aggregate([{"$out": "aggout"}], session=session)).to_list()
await self._test_no_read_concern(alambda)
@async_client_context.require_no_standalone
async def test_get_more_does_not_include_read_concern(self):
coll = self.client.pymongo_test.test
@ -1095,7 +1087,6 @@ class TestCausalConsistency(AsyncUnitTest):
self.assertIsNone(act)
@async_client_context.require_no_standalone
@async_client_context.require_no_mmap
async def test_read_concern(self):
async with self.client.start_session(causal_consistency=True) as s:
coll = self.client.pymongo_test.test

2
test/asynchronous/test_srv_polling.py
View File

@ -363,7 +363,7 @@ class TestSrvPolling(AsyncPyMongoTestCase):
# Regression test for PYTHON-4407
import dns.resolver
self.assertTrue(hasattr(dns.resolver, "resolve"))
self.assertTrue(hasattr(dns.resolver, "resolve") or hasattr(dns.resolver, "query"))
if __name__ == "__main__":

1
test/asynchronous/test_transactions_unified.py
View File

@ -27,7 +27,6 @@ from test.asynchronous.unified_format import generate_test_classes
_IS_SYNC = False
@client_context.require_no_mmap
def setUpModule():
pass

33
test/asynchronous/unified_format.py
View File

@ -75,6 +75,7 @@ from pymongo.asynchronous.command_cursor import AsyncCommandCursor
from pymongo.asynchronous.database import AsyncDatabase
from pymongo.asynchronous.encryption import AsyncClientEncryption
from pymongo.asynchronous.helpers import anext
from pymongo.driver_info import DriverInfo
from pymongo.encryption_options import _HAVE_PYMONGOCRYPT
from pymongo.errors import (
AutoReconnect,
@ -493,11 +494,6 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
raise unittest.SkipTest(f"{self.__class__.__name__} runOnRequirements not satisfied")
# add any special-casing for skipping tests here
if async_client_context.storage_engine == "mmapv1":
if "retryable-writes" in self.TEST_SPEC["description"] or "retryable_writes" in str(
self.TEST_PATH
):
raise unittest.SkipTest("MMAPv1 does not support retryWrites=True")
# Handle mongos_clients for transactions tests.
self.mongos_clients = []
@ -519,13 +515,6 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
def maybe_skip_test(self, spec):
# add any special-casing for skipping tests here
if async_client_context.storage_engine == "mmapv1":
if (
"Dirty explicit session is discarded" in spec["description"]
or "Dirty implicit session is discarded" in spec["description"]
or "Cancel server check" in spec["description"]
):
self.skipTest("MMAPv1 does not support retryWrites=True")
if "Client side error in command starting transaction" in spec["description"]:
self.skipTest("Implement PYTHON-1894")
if "timeoutMS applied to entire download" in spec["description"]:
@ -544,10 +533,6 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
if "csot" in class_name:
if "gridfs" in class_name and sys.platform == "win32":
self.skipTest("PYTHON-3522 CSOT GridFS tests are flaky on Windows")
if async_client_context.storage_engine == "mmapv1":
self.skipTest(
"MMAPv1 does not support retryable writes which is required for CSOT tests"
)
if "change" in description or "change" in class_name:
self.skipTest("CSOT not implemented for watch()")
if "cursors" in class_name:
@ -572,11 +557,6 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
self.skipTest("PyMongo does not support count()")
if name == "listIndexNames":
self.skipTest("PyMongo does not support list_index_names()")
if async_client_context.storage_engine == "mmapv1":
if name == "createChangeStream":
self.skipTest("MMAPv1 does not support change streams")
if name == "withTransaction" or name == "startTransaction":
self.skipTest("MMAPv1 does not support document-level locking")
if not async_client_context.test_commands_enabled:
if name == "failPoint" or name == "targetedFailPoint":
self.skipTest("Test commands must be enabled to use fail points")
@ -682,8 +662,6 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
self.fail(f"Operation {opname} not supported for entity of type {type(target)}")
async def __entityOperation_createChangeStream(self, target, *args, **kwargs):
if async_client_context.storage_engine == "mmapv1":
self.skipTest("MMAPv1 does not support change streams")
self.__raise_if_unsupported(
"createChangeStream", target, AsyncMongoClient, AsyncDatabase, AsyncCollection
)
@ -810,14 +788,10 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
return await (await target.list_search_indexes(name, **agg_kwargs)).to_list()
async def _sessionOperation_withTransaction(self, target, *args, **kwargs):
if async_client_context.storage_engine == "mmapv1":
self.skipTest("MMAPv1 does not support document-level locking")
self.__raise_if_unsupported("withTransaction", target, AsyncClientSession)
return await target.with_transaction(*args, **kwargs)
async def _sessionOperation_startTransaction(self, target, *args, **kwargs):
if async_client_context.storage_engine == "mmapv1":
self.skipTest("MMAPv1 does not support document-level locking")
self.__raise_if_unsupported("startTransaction", target, AsyncClientSession)
return await target.start_transaction(*args, **kwargs)
@ -840,6 +814,11 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
self.__raise_if_unsupported("close", target, NonLazyCursor, AsyncCommandCursor)
return await target.close()
async def _clientOperation_appendMetadata(self, target, *args, **kwargs):
info_opts = kwargs["driver_info_options"]
driver_info = DriverInfo(info_opts["name"], info_opts["version"], info_opts["platform"])
target.append_metadata(driver_info)
async def _clientEncryptionOperation_createDataKey(self, target, *args, **kwargs):
if "opts" in kwargs:
kwargs.update(camel_to_snake_args(kwargs.pop("opts")))

6
test/asynchronous/utils_spec_runner.py
View File

@ -648,12 +648,6 @@ class AsyncSpecRunner(AsyncIntegrationTest):
server_listener = ServerAndTopologyEventListener()
# Create a new client, to avoid interference from pooled sessions.
client_options = self.parse_client_options(test["clientOptions"])
# MMAPv1 does not support retryable writes.
if (
client_options.get("retryWrites") is True
and async_client_context.storage_engine == "mmapv1"
):
self.skipTest("MMAPv1 does not support retryWrites=True")
use_multi_mongos = test["useMultipleMongoses"]
host = None
if use_multi_mongos:

18
test/bson_binary_vector/packed_bit.json
View File

@ -20,6 +20,15 @@
"padding": 0,
"canonical_bson": "1600000005766563746F7200040000000910007F0700"
},
{
"description": "PACKED_BIT with padding",
"valid": true,
"vector": [127, 8],
"dtype_hex": "0x10",
"dtype_alias": "PACKED_BIT",
"padding": 3,
"canonical_bson": "1600000005766563746F7200040000000910037F0800"
},
{
"description": "Empty Vector PACKED_BIT",
"valid": true,
@ -29,15 +38,6 @@
"padding": 0,
"canonical_bson": "1400000005766563746F72000200000009100000"
},
{
"description": "PACKED_BIT with padding",
"valid": true,
"vector": [127, 7],
"dtype_hex": "0x10",
"dtype_alias": "PACKED_BIT",
"padding": 3,
"canonical_bson": "1600000005766563746F7200040000000910037F0700"
},
{
"description": "Overflow Vector PACKED_BIT",
"valid": false,

70
test/discovery_and_monitoring/errors/pre-42-InterruptedAtShutdown.json
View File

@ -1,70 +0,0 @@
{
"description": "Pre-4.2 InterruptedAtShutdown error",
"uri": "mongodb://a/?replicaSet=rs",
"phases": [
{
"description": "Primary A is discovered",
"responses": [
[
"a:27017",
{
"ok": 1,
"helloOk": true,
"isWritablePrimary": true,
"hosts": [
"a:27017"
],
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
}
]
],
"outcome": {
"servers": {
"a:27017": {
"type": "RSPrimary",
"setName": "rs",
"topologyVersion": null,
"pool": {
"generation": 0
}
}
},
"topologyType": "ReplicaSetWithPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
},
{
"description": "Pre-4.2 InterruptedAtShutdown error marks server Unknown and clears the pool",
"applicationErrors": [
{
"address": "a:27017",
"when": "afterHandshakeCompletes",
"maxWireVersion": 7,
"type": "command",
"response": {
"ok": 0,
"errmsg": "InterruptedAtShutdown",
"code": 11600
}
}
],
"outcome": {
"servers": {
"a:27017": {
"type": "Unknown",
"topologyVersion": null,
"pool": {
"generation": 1
}
}
},
"topologyType": "ReplicaSetNoPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
}
]
}

70
test/discovery_and_monitoring/errors/pre-42-InterruptedDueToReplStateChange.json
View File

@ -1,70 +0,0 @@
{
"description": "Pre-4.2 InterruptedDueToReplStateChange error",
"uri": "mongodb://a/?replicaSet=rs",
"phases": [
{
"description": "Primary A is discovered",
"responses": [
[
"a:27017",
{
"ok": 1,
"helloOk": true,
"isWritablePrimary": true,
"hosts": [
"a:27017"
],
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
}
]
],
"outcome": {
"servers": {
"a:27017": {
"type": "RSPrimary",
"setName": "rs",
"topologyVersion": null,
"pool": {
"generation": 0
}
}
},
"topologyType": "ReplicaSetWithPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
},
{
"description": "Pre-4.2 InterruptedDueToReplStateChange error marks server Unknown and clears the pool",
"applicationErrors": [
{
"address": "a:27017",
"when": "afterHandshakeCompletes",
"maxWireVersion": 7,
"type": "command",
"response": {
"ok": 0,
"errmsg": "InterruptedDueToReplStateChange",
"code": 11602
}
}
],
"outcome": {
"servers": {
"a:27017": {
"type": "Unknown",
"topologyVersion": null,
"pool": {
"generation": 1
}
}
},
"topologyType": "ReplicaSetNoPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
}
]
}

70
test/discovery_and_monitoring/errors/pre-42-LegacyNotPrimary.json
View File

@ -1,70 +0,0 @@
{
"description": "Pre-4.2 LegacyNotPrimary error",
"uri": "mongodb://a/?replicaSet=rs",
"phases": [
{
"description": "Primary A is discovered",
"responses": [
[
"a:27017",
{
"ok": 1,
"helloOk": true,
"isWritablePrimary": true,
"hosts": [
"a:27017"
],
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
}
]
],
"outcome": {
"servers": {
"a:27017": {
"type": "RSPrimary",
"setName": "rs",
"topologyVersion": null,
"pool": {
"generation": 0
}
}
},
"topologyType": "ReplicaSetWithPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
},
{
"description": "Pre-4.2 LegacyNotPrimary error marks server Unknown and clears the pool",
"applicationErrors": [
{
"address": "a:27017",
"when": "afterHandshakeCompletes",
"maxWireVersion": 7,
"type": "command",
"response": {
"ok": 0,
"errmsg": "LegacyNotPrimary",
"code": 10058
}
}
],
"outcome": {
"servers": {
"a:27017": {
"type": "Unknown",
"topologyVersion": null,
"pool": {
"generation": 1
}
}
},
"topologyType": "ReplicaSetNoPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
}
]
}

70
test/discovery_and_monitoring/errors/pre-42-NotPrimaryNoSecondaryOk.json
View File

@ -1,70 +0,0 @@
{
"description": "Pre-4.2 NotPrimaryNoSecondaryOk error",
"uri": "mongodb://a/?replicaSet=rs",
"phases": [
{
"description": "Primary A is discovered",
"responses": [
[
"a:27017",
{
"ok": 1,
"helloOk": true,
"isWritablePrimary": true,
"hosts": [
"a:27017"
],
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
}
]
],
"outcome": {
"servers": {
"a:27017": {
"type": "RSPrimary",
"setName": "rs",
"topologyVersion": null,
"pool": {
"generation": 0
}
}
},
"topologyType": "ReplicaSetWithPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
},
{
"description": "Pre-4.2 NotPrimaryNoSecondaryOk error marks server Unknown and clears the pool",
"applicationErrors": [
{
"address": "a:27017",
"when": "afterHandshakeCompletes",
"maxWireVersion": 7,
"type": "command",
"response": {
"ok": 0,
"errmsg": "NotPrimaryNoSecondaryOk",
"code": 13435
}
}
],
"outcome": {
"servers": {
"a:27017": {
"type": "Unknown",
"topologyVersion": null,
"pool": {
"generation": 1
}
}
},
"topologyType": "ReplicaSetNoPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
}
]
}

70
test/discovery_and_monitoring/errors/pre-42-NotPrimaryOrSecondary.json
View File

@ -1,70 +0,0 @@
{
"description": "Pre-4.2 NotPrimaryOrSecondary error",
"uri": "mongodb://a/?replicaSet=rs",
"phases": [
{
"description": "Primary A is discovered",
"responses": [
[
"a:27017",
{
"ok": 1,
"helloOk": true,
"isWritablePrimary": true,
"hosts": [
"a:27017"
],
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
}
]
],
"outcome": {
"servers": {
"a:27017": {
"type": "RSPrimary",
"setName": "rs",
"topologyVersion": null,
"pool": {
"generation": 0
}
}
},
"topologyType": "ReplicaSetWithPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
},
{
"description": "Pre-4.2 NotPrimaryOrSecondary error marks server Unknown and clears the pool",
"applicationErrors": [
{
"address": "a:27017",
"when": "afterHandshakeCompletes",
"maxWireVersion": 7,
"type": "command",
"response": {
"ok": 0,
"errmsg": "NotPrimaryOrSecondary",
"code": 13436
}
}
],
"outcome": {
"servers": {
"a:27017": {
"type": "Unknown",
"topologyVersion": null,
"pool": {
"generation": 1
}
}
},
"topologyType": "ReplicaSetNoPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
}
]
}

70
test/discovery_and_monitoring/errors/pre-42-NotWritablePrimary.json
View File

@ -1,70 +0,0 @@
{
"description": "Pre-4.2 NotWritablePrimary error",
"uri": "mongodb://a/?replicaSet=rs",
"phases": [
{
"description": "Primary A is discovered",
"responses": [
[
"a:27017",
{
"ok": 1,
"helloOk": true,
"isWritablePrimary": true,
"hosts": [
"a:27017"
],
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
}
]
],
"outcome": {
"servers": {
"a:27017": {
"type": "RSPrimary",
"setName": "rs",
"topologyVersion": null,
"pool": {
"generation": 0
}
}
},
"topologyType": "ReplicaSetWithPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
},
{
"description": "Pre-4.2 NotWritablePrimary error marks server Unknown and clears the pool",
"applicationErrors": [
{
"address": "a:27017",
"when": "afterHandshakeCompletes",
"maxWireVersion": 7,
"type": "command",
"response": {
"ok": 0,
"errmsg": "NotWritablePrimary",
"code": 10107
}
}
],
"outcome": {
"servers": {
"a:27017": {
"type": "Unknown",
"topologyVersion": null,
"pool": {
"generation": 1
}
}
},
"topologyType": "ReplicaSetNoPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
}
]
}

70
test/discovery_and_monitoring/errors/pre-42-PrimarySteppedDown.json
View File

@ -1,70 +0,0 @@
{
"description": "Pre-4.2 PrimarySteppedDown error",
"uri": "mongodb://a/?replicaSet=rs",
"phases": [
{
"description": "Primary A is discovered",
"responses": [
[
"a:27017",
{
"ok": 1,
"helloOk": true,
"isWritablePrimary": true,
"hosts": [
"a:27017"
],
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
}
]
],
"outcome": {
"servers": {
"a:27017": {
"type": "RSPrimary",
"setName": "rs",
"topologyVersion": null,
"pool": {
"generation": 0
}
}
},
"topologyType": "ReplicaSetWithPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
},
{
"description": "Pre-4.2 PrimarySteppedDown error marks server Unknown and clears the pool",
"applicationErrors": [
{
"address": "a:27017",
"when": "afterHandshakeCompletes",
"maxWireVersion": 7,
"type": "command",
"response": {
"ok": 0,
"errmsg": "PrimarySteppedDown",
"code": 189
}
}
],
"outcome": {
"servers": {
"a:27017": {
"type": "Unknown",
"topologyVersion": null,
"pool": {
"generation": 1
}
}
},
"topologyType": "ReplicaSetNoPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
}
]
}

70
test/discovery_and_monitoring/errors/pre-42-ShutdownInProgress.json
View File

@ -1,70 +0,0 @@
{
"description": "Pre-4.2 ShutdownInProgress error",
"uri": "mongodb://a/?replicaSet=rs",
"phases": [
{
"description": "Primary A is discovered",
"responses": [
[
"a:27017",
{
"ok": 1,
"helloOk": true,
"isWritablePrimary": true,
"hosts": [
"a:27017"
],
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
}
]
],
"outcome": {
"servers": {
"a:27017": {
"type": "RSPrimary",
"setName": "rs",
"topologyVersion": null,
"pool": {
"generation": 0
}
}
},
"topologyType": "ReplicaSetWithPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
},
{
"description": "Pre-4.2 ShutdownInProgress error marks server Unknown and clears the pool",
"applicationErrors": [
{
"address": "a:27017",
"when": "afterHandshakeCompletes",
"maxWireVersion": 7,
"type": "command",
"response": {
"ok": 0,
"errmsg": "ShutdownInProgress",
"code": 91
}
}
],
"outcome": {
"servers": {
"a:27017": {
"type": "Unknown",
"topologyVersion": null,
"pool": {
"generation": 1
}
}
},
"topologyType": "ReplicaSetNoPrimary",
"logicalSessionTimeoutMinutes": null,
"setName": "rs"
}
}
]
}

8
test/discovery_and_monitoring/rs/null_election_id-pre-6.0.json
View File

@ -18,7 +18,7 @@
"setVersion": 1,
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],
@ -66,7 +66,7 @@
"$oid": "000000000000000000000002"
},
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],
@ -116,7 +116,7 @@
"setVersion": 1,
"setName": "rs",
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],
@ -167,7 +167,7 @@
"$oid": "000000000000000000000001"
},
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],

4
test/discovery_and_monitoring/rs/primary_mismatched_me_not_removed.json
View File

@ -18,7 +18,7 @@
"primary": "localhost:27017",
"me": "a:27017",
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 25
}
]
],
@ -55,7 +55,7 @@
"primary": "localhost:27017",
"me": "localhost:27018",
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 25
}
]
],

4
test/discovery_and_monitoring/rs/setversion_without_electionid-pre-6.0.json
View File

@ -17,7 +17,7 @@
"setName": "rs",
"setVersion": 2,
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],
@ -56,7 +56,7 @@
"setName": "rs",
"setVersion": 1,
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],

6
test/discovery_and_monitoring/rs/use_setversion_without_electionid-pre-6.0.json
View File

@ -20,7 +20,7 @@
"$oid": "000000000000000000000001"
},
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],
@ -64,7 +64,7 @@
"setName": "rs",
"setVersion": 2,
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],
@ -109,7 +109,7 @@
"$oid": "000000000000000000000002"
},
"minWireVersion": 0,
"maxWireVersion": 7
"maxWireVersion": 16
}
]
],

100
test/handshake/unified/metadata-not-propagated.json Normal file
View File

@ -0,0 +1,100 @@
{
"description": "client metadata is not propagated to the server",
"schemaVersion": "1.9",
"runOnRequirements": [
{
"minServerVersion": "6.0"
}
],
"createEntities": [
{
"client": {
"id": "client",
"observeEvents": [
"commandSucceededEvent",
"commandFailedEvent",
"connectionClosedEvent",
"connectionCreatedEvent"
]
}
},
{
"database": {
"id": "database",
"client": "client",
"databaseName": "test"
}
}
],
"tests": [
{
"description": "metadata append does not create new connections or close existing ones and no hello command is sent",
"operations": [
{
"name": "runCommand",
"object": "database",
"arguments": {
"commandName": "ping",
"command": {
"ping": 1
}
},
"expectResult": {
"ok": 1
}
},
{
"name": "appendMetadata",
"object": "client",
"arguments": {
"driverInfoOptions": {
"name": "framework",
"version": "2.0",
"platform": "Framework Platform"
}
}
},
{
"name": "runCommand",
"object": "database",
"arguments": {
"commandName": "ping",
"command": {
"ping": 1
}
},
"expectResult": {
"ok": 1
}
}
],
"expectEvents": [
{
"client": "client",
"eventType": "cmap",
"events": [
{
"connectionCreatedEvent": {}
}
]
},
{
"client": "client",
"eventType": "command",
"events": [
{
"commandSucceededEvent": {
"commandName": "ping"
}
},
{
"commandSucceededEvent": {
"commandName": "ping"
}
}
]
}
]
}
]
}

2
test/mockupdb/test_cursor_namespace.py
View File

@ -40,7 +40,7 @@ class TestCursorNamespace(PyMongoTestCase):
@classmethod
def setUpClass(cls):
cls.server = MockupDB(auto_ismaster={"maxWireVersion": 7})
cls.server = MockupDB(auto_ismaster={"maxWireVersion": 8})
cls.server.run()
cls.client = cls.unmanaged_simple_client(cls.server.uri)

2
test/test_bson.py
View File

@ -1045,6 +1045,8 @@ class TestBSON(unittest.TestCase):
def test_minkey_maxkey_comparison(self):
# MinKey's <, <=, >, >=, !=, and ==.
# These tests should be kept as assertTrue as opposed to using unittest's built-in comparison assertions because
# MinKey and MaxKey define their own __ge__, __le__, and other comparison attributes, and we want to explicitly test that.
self.assertTrue(MinKey() < None)
self.assertTrue(MinKey() < 1)
self.assertTrue(MinKey() <= 1)

30
test/test_bson_binary_vector.py
View File

@ -48,7 +48,7 @@ def create_test(case_spec):
def run_test(self):
for test_case in case_spec.get("tests", []):
description = test_case["description"]
vector_exp = test_case.get("vector", [])
vector_exp = test_case.get("vector")
dtype_hex_exp = test_case["dtype_hex"]
dtype_alias_exp = test_case.get("dtype_alias")
padding_exp = test_case.get("padding", 0)
@ -85,14 +85,26 @@ def create_test(case_spec):
self.assertEqual(cB_obs, canonical_bson_exp, description)
else:
with self.assertRaises((struct.error, ValueError), msg=description):
# Tests Binary.from_vector
Binary.from_vector(vector_exp, dtype_exp, padding_exp)
# Tests Binary.as_vector
cB_exp = binascii.unhexlify(canonical_bson_exp.encode("utf8"))
decoded_doc = decode(cB_exp)
binary_obs = decoded_doc[test_key]
binary_obs.as_vector()
"""
#### To prove correct in an invalid case (`valid:false`), one MUST
- (encoding case) if the vector field is present, raise an exception
when attempting to encode a document from the numeric values,dtype, and padding.
- (decoding case) if the canonical_bson field is present, raise an exception
when attempting to deserialize it into the corresponding
numeric values, as the field contains corrupted data.
"""
# Tests Binary.from_vector()
if vector_exp is not None:
with self.assertRaises((struct.error, ValueError), msg=description):
Binary.from_vector(vector_exp, dtype_exp, padding_exp)
# Tests Binary.as_vector()
if canonical_bson_exp is not None:
with self.assertRaises((struct.error, ValueError), msg=description):
cB_exp = binascii.unhexlify(canonical_bson_exp.encode("utf8"))
decoded_doc = decode(cB_exp)
binary_obs = decoded_doc[test_key]
binary_obs.as_vector()
return run_test

12
test/test_bulk.py
View File

@ -165,7 +165,7 @@ class TestBulk(BulkTestBase):
def test_update_many(self):
self._test_update_many({"$set": {"foo": "bar"}})
@client_context.require_version_min(4, 1, 11)
@client_context.require_version_min(4, 2, 0)
def test_update_many_pipeline(self):
self._test_update_many([{"$set": {"foo": "bar"}}])
@ -206,7 +206,7 @@ class TestBulk(BulkTestBase):
def test_update_one(self):
self._test_update_one({"$set": {"foo": "bar"}})
@client_context.require_version_min(4, 1, 11)
@client_context.require_version_min(4, 2, 0)
def test_update_one_pipeline(self):
self._test_update_one([{"$set": {"foo": "bar"}}])
@ -992,7 +992,7 @@ class TestBulkWriteConcern(BulkTestBase):
# When talking to legacy servers there will be a
# write concern error for each operation.
self.assertTrue(len(details["writeConcernErrors"]) > 0)
self.assertGreater(len(details["writeConcernErrors"]), 0)
failed = details["writeConcernErrors"][0]
self.assertEqual(64, failed["code"])
@ -1033,7 +1033,7 @@ class TestBulkWriteConcern(BulkTestBase):
details,
)
self.assertTrue(len(details["writeConcernErrors"]) > 1)
self.assertGreater(len(details["writeConcernErrors"]), 1)
failed = details["writeErrors"][0]
self.assertIn("duplicate", failed["errmsg"])
@ -1069,7 +1069,7 @@ class TestBulkWriteConcern(BulkTestBase):
self.assertEqual(0, len(details["writeErrors"]))
# When talking to legacy servers there will be a
# write concern error for each operation.
self.assertTrue(len(details["writeConcernErrors"]) > 1)
self.assertGreater(len(details["writeConcernErrors"]), 1)
self.coll.delete_many({})
self.coll.create_index("a", unique=True)
@ -1096,7 +1096,7 @@ class TestBulkWriteConcern(BulkTestBase):
self.assertEqual(1, len(details["writeErrors"]))
# When talking to legacy servers there will be a
# write concern error for each operation.
self.assertTrue(len(details["writeConcernErrors"]) > 1)
self.assertGreater(len(details["writeConcernErrors"]), 1)
failed = details["writeErrors"][0]
self.assertEqual(2, failed["index"])

93
test/test_change_stream.py
View File

@ -263,7 +263,7 @@ class APITestsMixin:
# $changeStream.startAtOperationTime was added in 4.0.0.
@no_type_check
@client_context.require_version_min(4, 0, 0)
@client_context.require_version_min(4, 2, 0)
def test_start_at_operation_time(self):
optime = self.get_start_at_operation_time()
@ -432,7 +432,7 @@ class APITestsMixin:
self._test_get_invalidate_event(change_stream)
@no_type_check
@client_context.require_version_min(4, 1, 1)
@client_context.require_version_min(4, 2, 0)
def test_start_after(self):
resume_token = self.get_resume_token(invalidate=True)
@ -448,7 +448,7 @@ class APITestsMixin:
self.assertEqual(change["fullDocument"], {"_id": 2})
@no_type_check
@client_context.require_version_min(4, 1, 1)
@client_context.require_version_min(4, 2, 0)
def test_start_after_resume_process_with_changes(self):
resume_token = self.get_resume_token(invalidate=True)
@ -553,27 +553,16 @@ class ProseSpecTestsMixin:
)
# Prose test no. 1
@client_context.require_version_min(4, 0, 7)
@client_context.require_version_min(4, 2, 0)
def test_update_resume_token(self):
self._test_update_resume_token(self._get_expected_resume_token)
# Prose test no. 1
@client_context.require_version_max(4, 0, 7)
def test_update_resume_token_legacy(self):
self._test_update_resume_token(self._get_expected_resume_token_legacy)
# Prose test no. 2
@client_context.require_version_min(4, 1, 8)
@client_context.require_version_min(4, 2, 0)
def test_raises_error_on_missing_id_418plus(self):
# Server returns an error on 4.1.8+
self._test_raises_error_on_missing_id(OperationFailure)
# Prose test no. 2
@client_context.require_version_max(4, 1, 8)
def test_raises_error_on_missing_id_418minus(self):
# PyMongo raises an error
self._test_raises_error_on_missing_id(InvalidOperation)
# Prose test no. 3
@no_type_check
def test_resume_on_error(self):
@ -632,38 +621,12 @@ class ProseSpecTestsMixin:
cursor.close = raise_error
self.insert_one_and_check(change_stream, {"_id": 2})
# Prose test no. 9
@no_type_check
@client_context.require_version_min(4, 0, 0)
@client_context.require_version_max(4, 0, 7)
def test_start_at_operation_time_caching(self):
# Case 1: change stream not started with startAtOperationTime
client, listener = self.client_with_listener("aggregate")
with self.change_stream_with_client(client) as cs:
self.kill_change_stream_cursor(cs)
cs.try_next()
cmd = listener.started_events[-1].command
self.assertIsNotNone(cmd["pipeline"][0]["$changeStream"].get("startAtOperationTime"))
# Case 2: change stream started with startAtOperationTime
listener.reset()
optime = self.get_start_at_operation_time()
with self.change_stream_with_client(client, start_at_operation_time=optime) as cs:
self.kill_change_stream_cursor(cs)
cs.try_next()
cmd = listener.started_events[-1].command
self.assertEqual(
cmd["pipeline"][0]["$changeStream"].get("startAtOperationTime"),
optime,
str([k.command for k in listener.started_events]),
)
# Prose test no. 10 - SKIPPED
# This test is identical to prose test no. 3.
# Prose test no. 11
@no_type_check
@client_context.require_version_min(4, 0, 7)
@client_context.require_version_min(4, 2, 0)
def test_resumetoken_empty_batch(self):
client, listener = self._client_with_listener("getMore")
with self.change_stream_with_client(client) as change_stream:
@ -675,7 +638,7 @@ class ProseSpecTestsMixin:
# Prose test no. 11
@no_type_check
@client_context.require_version_min(4, 0, 7)
@client_context.require_version_min(4, 2, 0)
def test_resumetoken_exhausted_batch(self):
client, listener = self._client_with_listener("getMore")
with self.change_stream_with_client(client) as change_stream:
@ -685,38 +648,6 @@ class ProseSpecTestsMixin:
response = listener.succeeded_events[-1].reply
self.assertEqual(resume_token, response["cursor"]["postBatchResumeToken"])
# Prose test no. 12
@no_type_check
@client_context.require_version_max(4, 0, 7)
def test_resumetoken_empty_batch_legacy(self):
resume_point = self.get_resume_token()
# Empty resume token when neither resumeAfter or startAfter specified.
with self.change_stream() as change_stream:
change_stream.try_next()
self.assertIsNone(change_stream.resume_token)
# Resume token value is same as resumeAfter.
with self.change_stream(resume_after=resume_point) as change_stream:
change_stream.try_next()
resume_token = change_stream.resume_token
self.assertEqual(resume_token, resume_point)
# Prose test no. 12
@no_type_check
@client_context.require_version_max(4, 0, 7)
def test_resumetoken_exhausted_batch_legacy(self):
# Resume token is _id of last change.
with self.change_stream() as change_stream:
change = self._populate_and_exhaust_change_stream(change_stream)
self.assertEqual(change_stream.resume_token, change["_id"])
resume_point = change["_id"]
# Resume token is _id of last change even if resumeAfter is specified.
with self.change_stream(resume_after=resume_point) as change_stream:
change = self._populate_and_exhaust_change_stream(change_stream)
self.assertEqual(change_stream.resume_token, change["_id"])
# Prose test no. 13
@no_type_check
def test_resumetoken_partially_iterated_batch(self):
@ -758,13 +689,13 @@ class ProseSpecTestsMixin:
# Prose test no. 14
@no_type_check
@client_context.require_no_mongos
@client_context.require_version_min(4, 1, 1)
@client_context.require_version_min(4, 2, 0)
def test_resumetoken_uniterated_nonempty_batch_startafter(self):
self._test_resumetoken_uniterated_nonempty_batch("start_after")
# Prose test no. 17
@no_type_check
@client_context.require_version_min(4, 1, 1)
@client_context.require_version_min(4, 2, 0)
def test_startafter_resume_uses_startafter_after_empty_getMore(self):
# Resume should use startAfter after no changes have been returned.
resume_point = self.get_resume_token()
@ -782,7 +713,7 @@ class ProseSpecTestsMixin:
# Prose test no. 18
@no_type_check
@client_context.require_version_min(4, 1, 1)
@client_context.require_version_min(4, 2, 0)
def test_startafter_resume_uses_resumeafter_after_nonempty_getMore(self):
# Resume should use resumeAfter after some changes have been returned.
resume_point = self.get_resume_token()
@ -827,7 +758,7 @@ class ProseSpecTestsMixin:
class TestClusterChangeStream(TestChangeStreamBase, APITestsMixin):
dbs: list
@client_context.require_version_min(4, 0, 0, -1)
@client_context.require_version_min(4, 2, 0)
@client_context.require_change_streams
def setUp(self) -> None:
super().setUp()
@ -887,7 +818,7 @@ class TestClusterChangeStream(TestChangeStreamBase, APITestsMixin):
class TestDatabaseChangeStream(TestChangeStreamBase, APITestsMixin):
@client_context.require_version_min(4, 0, 0, -1)
@client_context.require_version_min(4, 2, 0)
@client_context.require_change_streams
def setUp(self) -> None:
super().setUp()

2
test/test_client.py
View File

@ -978,7 +978,7 @@ class TestClient(IntegrationTest):
cursor = self.client.list_databases()
self.assertIsInstance(cursor, CommandCursor)
helper_docs = cursor.to_list()
self.assertTrue(len(helper_docs) > 0)
self.assertGreater(len(helper_docs), 0)
self.assertEqual(len(helper_docs), len(cmd_docs))
# PYTHON-3529 Some fields may change between calls, just compare names.
for helper_doc, cmd_doc in zip(helper_docs, cmd_docs):

215
test/test_client_metadata.py Normal file
View File

@ -0,0 +1,215 @@
# Copyright 2013-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.
from __future__ import annotations
import asyncio
import os
import pathlib
import time
import unittest
from test import IntegrationTest
from test.unified_format import generate_test_classes
from test.utils_shared import CMAPListener
from typing import Any, Optional
import pytest
from pymongo import MongoClient
from pymongo.driver_info import DriverInfo
from pymongo.monitoring import ConnectionClosedEvent
try:
from mockupdb import MockupDB, OpMsgReply
_HAVE_MOCKUPDB = True
except ImportError:
_HAVE_MOCKUPDB = False
pytestmark = pytest.mark.mockupdb
_IS_SYNC = True
# Location of JSON test specifications.
if _IS_SYNC:
_TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "handshake", "unified")
else:
_TEST_PATH = os.path.join(
pathlib.Path(__file__).resolve().parent.parent, "handshake", "unified"
)
# Generate unified tests.
globals().update(generate_test_classes(_TEST_PATH, module=__name__))
def _get_handshake_driver_info(request):
assert "client" in request
return request["client"]
class TestClientMetadataProse(IntegrationTest):
def setUp(self):
super().setUp()
self.server = MockupDB()
self.handshake_req = None
def respond(r):
if "ismaster" in r:
# then this is a handshake request
self.handshake_req = r
return r.reply(OpMsgReply(maxWireVersion=13))
self.server.autoresponds(respond)
self.server.run()
self.addCleanup(self.server.stop)
def send_ping_and_get_metadata(
self, client: MongoClient, is_handshake: bool
) -> tuple[str, Optional[str], Optional[str], dict[str, Any]]:
# reset if handshake request
if is_handshake:
self.handshake_req: Optional[dict] = None
client.admin.command("ping")
metadata = _get_handshake_driver_info(self.handshake_req)
driver_metadata = metadata["driver"]
name, version, platform = (
driver_metadata["name"],
driver_metadata["version"],
metadata["platform"],
)
return name, version, platform, metadata
def check_metadata_added(
self,
client: MongoClient,
add_name: str,
add_version: Optional[str],
add_platform: Optional[str],
) -> None:
# send initial metadata
name, version, platform, metadata = self.send_ping_and_get_metadata(client, True)
# wait for connection to become idle
time.sleep(0.005)
# add new metadata
client.append_metadata(DriverInfo(add_name, add_version, add_platform))
new_name, new_version, new_platform, new_metadata = self.send_ping_and_get_metadata(
client, True
)
self.assertEqual(new_name, f"{name}|{add_name}" if add_name is not None else name)
self.assertEqual(
new_version,
f"{version}|{add_version}" if add_version is not None else version,
)
self.assertEqual(
new_platform,
f"{platform}|{add_platform}" if add_platform is not None else platform,
)
metadata.pop("driver")
metadata.pop("platform")
new_metadata.pop("driver")
new_metadata.pop("platform")
self.assertEqual(metadata, new_metadata)
def test_append_metadata(self):
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
)
self.check_metadata_added(client, "framework", "2.0", "Framework Platform")
def test_append_metadata_platform_none(self):
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
)
self.check_metadata_added(client, "framework", "2.0", None)
def test_append_metadata_version_none(self):
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
)
self.check_metadata_added(client, "framework", None, "Framework Platform")
def test_append_metadata_platform_version_none(self):
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
)
self.check_metadata_added(client, "framework", None, None)
def test_multiple_successive_metadata_updates(self):
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string, maxIdleTimeMS=1, connect=False
)
client.append_metadata(DriverInfo("library", "1.2", "Library Platform"))
self.check_metadata_added(client, "framework", "2.0", "Framework Platform")
def test_multiple_successive_metadata_updates_platform_none(self):
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
)
client.append_metadata(DriverInfo("library", "1.2", "Library Platform"))
self.check_metadata_added(client, "framework", "2.0", None)
def test_multiple_successive_metadata_updates_version_none(self):
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
)
client.append_metadata(DriverInfo("library", "1.2", "Library Platform"))
self.check_metadata_added(client, "framework", None, "Framework Platform")
def test_multiple_successive_metadata_updates_platform_version_none(self):
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
)
client.append_metadata(DriverInfo("library", "1.2", "Library Platform"))
self.check_metadata_added(client, "framework", None, None)
def test_doesnt_update_established_connections(self):
listener = CMAPListener()
client = self.rs_or_single_client(
"mongodb://" + self.server.address_string,
maxIdleTimeMS=1,
driver=DriverInfo("library", "1.2", "Library Platform"),
event_listeners=[listener],
)
# send initial metadata
name, version, platform, metadata = self.send_ping_and_get_metadata(client, True)
self.assertIsNotNone(name)
self.assertIsNotNone(version)
self.assertIsNotNone(platform)
# add data
add_name, add_version, add_platform = "framework", "2.0", "Framework Platform"
client.append_metadata(DriverInfo(add_name, add_version, add_platform))
# check new data isn't sent
self.handshake_req: Optional[dict] = None
client.admin.command("ping")
self.assertIsNone(self.handshake_req)
self.assertEqual(listener.event_count(ConnectionClosedEvent), 0)
if __name__ == "__main__":
unittest.main()

2
test/test_collection.py
View File

@ -490,7 +490,7 @@ class TestCollection(IntegrationTest):
# Sort by 'score' field.
cursor.sort([("score", {"$meta": "textScore"})])
results = cursor.to_list()
self.assertTrue(results[0]["score"] >= results[1]["score"])
self.assertGreaterEqual(results[0]["score"], results[1]["score"])
db.test.drop_indexes()

10
test/test_connections_survive_primary_stepdown_spec.py
View File

@ -122,18 +122,12 @@ class TestConnectionsSurvivePrimaryStepDown(IntegrationTest):
def test_not_primary_keep_connection_pool(self):
self.run_scenario(10107, True, self.verify_pool_not_cleared)
@client_context.require_version_min(4, 0, 0)
@client_context.require_version_max(4, 1, 0, -1)
@client_context.require_test_commands
def test_not_primary_reset_connection_pool(self):
self.run_scenario(10107, False, self.verify_pool_cleared)
@client_context.require_version_min(4, 0, 0)
@client_context.require_version_min(4, 2, 0)
@client_context.require_test_commands
def test_shutdown_in_progress(self):
self.run_scenario(91, False, self.verify_pool_cleared)
@client_context.require_version_min(4, 0, 0)
@client_context.require_version_min(4, 2, 0)
@client_context.require_test_commands
def test_interrupted_at_shutdown(self):
self.run_scenario(11600, False, self.verify_pool_cleared)

10
test/test_cursor.py
View File

@ -1181,15 +1181,6 @@ class TestCursor(IntegrationTest):
self.assertEqual(["b", "c"], distinct)
@client_context.require_version_max(4, 1, 0, -1)
def test_max_scan(self):
self.db.drop_collection("test")
self.db.test.insert_many([{} for _ in range(100)])
self.assertEqual(100, len(self.db.test.find().to_list()))
self.assertEqual(50, len(self.db.test.find().max_scan(50).to_list()))
self.assertEqual(50, len(self.db.test.find().max_scan(90).max_scan(50).to_list()))
def test_with_statement(self):
self.db.drop_collection("test")
self.db.test.insert_many([{} for _ in range(100)])
@ -1591,7 +1582,6 @@ class TestRawBatchCursor(IntegrationTest):
def test_collation(self):
next(self.db.test.find_raw_batches(collation=Collation("en_US")))
@client_context.require_no_mmap # MMAPv1 does not support read concern
def test_read_concern(self):
self.db.get_collection("test", write_concern=WriteConcern(w="majority")).insert_one({})
c = self.db.get_collection("test", read_concern=ReadConcern("majority"))

4
test/test_custom_types.py
View File

@ -949,7 +949,7 @@ class TestCollectionChangeStreamsWCustomTypes(IntegrationTest, ChangeStreamsWCus
class TestDatabaseChangeStreamsWCustomTypes(IntegrationTest, ChangeStreamsWCustomTypesTestMixin):
@client_context.require_version_min(4, 0, 0)
@client_context.require_version_min(4, 2, 0)
@client_context.require_change_streams
def setUp(self):
super().setUp()
@ -967,7 +967,7 @@ class TestDatabaseChangeStreamsWCustomTypes(IntegrationTest, ChangeStreamsWCusto
class TestClusterChangeStreamsWCustomTypes(IntegrationTest, ChangeStreamsWCustomTypesTestMixin):
@client_context.require_version_min(4, 0, 0)
@client_context.require_version_min(4, 2, 0)
@client_context.require_change_streams
def setUp(self):
super().setUp()

2
test/test_database.py
View File

@ -238,7 +238,7 @@ class TestDatabase(IntegrationTest):
listener.reset()
db.drop_collection("unique")
db.create_collection("unique", check_exists=False)
self.assertTrue(len(listener.started_events) > 0)
self.assertGreater(len(listener.started_events), 0)
self.assertNotIn("listCollections", listener.started_command_names())
def test_list_collections(self):

14
test/test_encryption.py
View File

@ -451,20 +451,6 @@ class TestClientMaxWireVersion(IntegrationTest):
def setUp(self):
super().setUp()
@client_context.require_version_max(4, 0, 99)
def test_raise_max_wire_version_error(self):
opts = AutoEncryptionOpts(KMS_PROVIDERS, "keyvault.datakeys")
client = self.rs_or_single_client(auto_encryption_opts=opts)
msg = "Auto-encryption requires a minimum MongoDB version of 4.2"
with self.assertRaisesRegex(ConfigurationError, msg):
client.test.test.insert_one({})
with self.assertRaisesRegex(ConfigurationError, msg):
client.admin.command("ping")
with self.assertRaisesRegex(ConfigurationError, msg):
client.test.test.find_one({})
with self.assertRaisesRegex(ConfigurationError, msg):
client.test.test.bulk_write([InsertOne({})])
def test_raise_unsupported_error(self):
opts = AutoEncryptionOpts(KMS_PROVIDERS, "keyvault.datakeys")
client = self.rs_or_single_client(auto_encryption_opts=opts)

1
test/test_examples.py
View File

@ -1160,7 +1160,6 @@ class TestTransactionExamples(IntegrationTest):
class TestCausalConsistencyExamples(IntegrationTest):
@client_context.require_secondaries_count(1)
@client_context.require_no_mmap
def test_causal_consistency(self):
# Causal consistency examples
client = self.client

2
test/test_objectid.py
View File

@ -92,7 +92,7 @@ class TestObjectId(unittest.TestCase):
self.assertEqual(utc, d2.tzinfo)
d2 = d2.replace(tzinfo=None)
self.assertTrue(d2 - d1 < datetime.timedelta(seconds=2))
self.assertLess(d2 - d1, datetime.timedelta(seconds=2))
def test_from_datetime(self):
d = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)

9
test/test_pooling.py
View File

@ -331,8 +331,9 @@ class TestPooling(_TestPoolingBase):
pass
duration = time.time() - start
self.assertTrue(
abs(wait_queue_timeout - duration) < 1,
self.assertLess(
abs(wait_queue_timeout - duration),
1,
f"Waited {duration:.2f} seconds for a socket, expected {wait_queue_timeout:f}",
)
@ -545,7 +546,7 @@ class TestPoolMaxSize(_TestPoolingBase):
joinall(tasks)
self.assertEqual(ntasks, self.n_passed)
self.assertTrue(len(cx_pool.conns) > 1)
self.assertGreater(len(cx_pool.conns), 1)
self.assertEqual(0, cx_pool.requests)
def test_max_pool_size_none(self):
@ -576,7 +577,7 @@ class TestPoolMaxSize(_TestPoolingBase):
joinall(tasks)
self.assertEqual(ntasks, self.n_passed)
self.assertTrue(len(cx_pool.conns) > 1)
self.assertGreater(len(cx_pool.conns), 1)
self.assertEqual(cx_pool.max_pool_size, float("inf"))
def test_max_pool_size_zero(self):

32
test/test_retryable_writes.py
View File

@ -140,40 +140,10 @@ class IgnoreDeprecationsTest(IntegrationTest):
self.deprecation_filter.stop()
class TestRetryableWritesMMAPv1(IgnoreDeprecationsTest):
knobs: client_knobs
def setUp(self) -> None:
super().setUp()
# Speed up the tests by decreasing the heartbeat frequency.
self.knobs = client_knobs(heartbeat_frequency=0.1, min_heartbeat_interval=0.1)
self.knobs.enable()
self.client = self.rs_or_single_client(retryWrites=True)
self.db = self.client.pymongo_test
def tearDown(self) -> None:
self.knobs.disable()
@client_context.require_no_standalone
def test_actionable_error_message(self):
if client_context.storage_engine != "mmapv1":
raise SkipTest("This cluster is not running MMAPv1")
expected_msg = (
"This MongoDB deployment does not support retryable "
"writes. Please add retryWrites=false to your "
"connection string."
)
for method, args, kwargs in retryable_single_statement_ops(self.db.retryable_write_test):
with self.assertRaisesRegex(OperationFailure, expected_msg):
method(*args, **kwargs)
class TestRetryableWrites(IgnoreDeprecationsTest):
listener: OvertCommandListener
knobs: client_knobs
@client_context.require_no_mmap
def setUp(self) -> None:
super().setUp()
# Speed up the tests by decreasing the heartbeat frequency.
@ -423,7 +393,6 @@ class TestWriteConcernError(IntegrationTest):
fail_insert: dict
@client_context.require_replica_set
@client_context.require_no_mmap
@client_context.require_failCommand_fail_point
def setUp(self) -> None:
super().setUp()
@ -592,7 +561,6 @@ class TestPoolPausedError(IntegrationTest):
# TODO: Make this a real integration test where we stepdown the primary.
class TestRetryableWritesTxnNumber(IgnoreDeprecationsTest):
@client_context.require_replica_set
@client_context.require_no_mmap
def test_increment_transaction_id_without_sending_command(self):
"""Test that the txnNumber field is properly incremented, even when
the first attempt fails before sending the command.

9
test/test_session.py
View File

@ -1031,14 +1031,6 @@ class TestCausalConsistency(UnitTest):
# Not a write, but explain also doesn't support readConcern.
self._test_no_read_concern(lambda coll, session: coll.find({}, session=session).explain())
@client_context.require_no_standalone
@client_context.require_version_max(4, 1, 0)
def test_aggregate_out_does_not_include_read_concern(self):
def alambda(coll, session):
(coll.aggregate([{"$out": "aggout"}], session=session)).to_list()
self._test_no_read_concern(alambda)
@client_context.require_no_standalone
def test_get_more_does_not_include_read_concern(self):
coll = self.client.pymongo_test.test
@ -1081,7 +1073,6 @@ class TestCausalConsistency(UnitTest):
self.assertIsNone(act)
@client_context.require_no_standalone
@client_context.require_no_mmap
def test_read_concern(self):
with self.client.start_session(causal_consistency=True) as s:
coll = self.client.pymongo_test.test

2
test/test_srv_polling.py
View File

@ -363,7 +363,7 @@ class TestSrvPolling(PyMongoTestCase):
# Regression test for PYTHON-4407
import dns.resolver
self.assertTrue(hasattr(dns.resolver, "resolve"))
self.assertTrue(hasattr(dns.resolver, "resolve") or hasattr(dns.resolver, "query"))
if __name__ == "__main__":

2
test/test_topology.py
View File

@ -559,7 +559,7 @@ class TestMultiServerTopology(TopologyTest):
)
self.assertEqual(server.description.min_wire_version, 1)
self.assertEqual(server.description.max_wire_version, 7)
self.assertEqual(server.description.max_wire_version, 8)
t.select_servers(any_server_selector, _Op.TEST)
# Incompatible.

1
test/test_transactions_unified.py
View File

@ -27,7 +27,6 @@ from test.unified_format import generate_test_classes
_IS_SYNC = True
@client_context.require_no_mmap
def setUpModule():
pass

33
test/unified_format.py
View File

@ -67,6 +67,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS
from bson.objectid import ObjectId
from gridfs import GridFSBucket, GridOut, NoFile
from pymongo import ASCENDING, CursorType, MongoClient, _csot
from pymongo.driver_info import DriverInfo
from pymongo.encryption_options import _HAVE_PYMONGOCRYPT
from pymongo.errors import (
AutoReconnect,
@ -492,11 +493,6 @@ class UnifiedSpecTestMixinV1(IntegrationTest):
raise unittest.SkipTest(f"{self.__class__.__name__} runOnRequirements not satisfied")
# add any special-casing for skipping tests here
if client_context.storage_engine == "mmapv1":
if "retryable-writes" in self.TEST_SPEC["description"] or "retryable_writes" in str(
self.TEST_PATH
):
raise unittest.SkipTest("MMAPv1 does not support retryWrites=True")
# Handle mongos_clients for transactions tests.
self.mongos_clients = []
@ -518,13 +514,6 @@ class UnifiedSpecTestMixinV1(IntegrationTest):
def maybe_skip_test(self, spec):
# add any special-casing for skipping tests here
if client_context.storage_engine == "mmapv1":
if (
"Dirty explicit session is discarded" in spec["description"]
or "Dirty implicit session is discarded" in spec["description"]
or "Cancel server check" in spec["description"]
):
self.skipTest("MMAPv1 does not support retryWrites=True")
if "Client side error in command starting transaction" in spec["description"]:
self.skipTest("Implement PYTHON-1894")
if "timeoutMS applied to entire download" in spec["description"]:
@ -543,10 +532,6 @@ class UnifiedSpecTestMixinV1(IntegrationTest):
if "csot" in class_name:
if "gridfs" in class_name and sys.platform == "win32":
self.skipTest("PYTHON-3522 CSOT GridFS tests are flaky on Windows")
if client_context.storage_engine == "mmapv1":
self.skipTest(
"MMAPv1 does not support retryable writes which is required for CSOT tests"
)
if "change" in description or "change" in class_name:
self.skipTest("CSOT not implemented for watch()")
if "cursors" in class_name:
@ -571,11 +556,6 @@ class UnifiedSpecTestMixinV1(IntegrationTest):
self.skipTest("PyMongo does not support count()")
if name == "listIndexNames":
self.skipTest("PyMongo does not support list_index_names()")
if client_context.storage_engine == "mmapv1":
if name == "createChangeStream":
self.skipTest("MMAPv1 does not support change streams")
if name == "withTransaction" or name == "startTransaction":
self.skipTest("MMAPv1 does not support document-level locking")
if not client_context.test_commands_enabled:
if name == "failPoint" or name == "targetedFailPoint":
self.skipTest("Test commands must be enabled to use fail points")
@ -681,8 +661,6 @@ class UnifiedSpecTestMixinV1(IntegrationTest):
self.fail(f"Operation {opname} not supported for entity of type {type(target)}")
def __entityOperation_createChangeStream(self, target, *args, **kwargs):
if client_context.storage_engine == "mmapv1":
self.skipTest("MMAPv1 does not support change streams")
self.__raise_if_unsupported("createChangeStream", target, MongoClient, Database, Collection)
stream = target.watch(*args, **kwargs)
self.addCleanup(stream.close)
@ -807,14 +785,10 @@ class UnifiedSpecTestMixinV1(IntegrationTest):
return (target.list_search_indexes(name, **agg_kwargs)).to_list()
def _sessionOperation_withTransaction(self, target, *args, **kwargs):
if client_context.storage_engine == "mmapv1":
self.skipTest("MMAPv1 does not support document-level locking")
self.__raise_if_unsupported("withTransaction", target, ClientSession)
return target.with_transaction(*args, **kwargs)
def _sessionOperation_startTransaction(self, target, *args, **kwargs):
if client_context.storage_engine == "mmapv1":
self.skipTest("MMAPv1 does not support document-level locking")
self.__raise_if_unsupported("startTransaction", target, ClientSession)
return target.start_transaction(*args, **kwargs)
@ -837,6 +811,11 @@ class UnifiedSpecTestMixinV1(IntegrationTest):
self.__raise_if_unsupported("close", target, NonLazyCursor, CommandCursor)
return target.close()
def _clientOperation_appendMetadata(self, target, *args, **kwargs):
info_opts = kwargs["driver_info_options"]
driver_info = DriverInfo(info_opts["name"], info_opts["version"], info_opts["platform"])
target.append_metadata(driver_info)
def _clientEncryptionOperation_createDataKey(self, target, *args, **kwargs):
if "opts" in kwargs:
kwargs.update(camel_to_snake_args(kwargs.pop("opts")))

3
test/utils_spec_runner.py
View File

@ -648,9 +648,6 @@ class SpecRunner(IntegrationTest):
server_listener = ServerAndTopologyEventListener()
# Create a new client, to avoid interference from pooled sessions.
client_options = self.parse_client_options(test["clientOptions"])
# MMAPv1 does not support retryable writes.
if client_options.get("retryWrites") is True and client_context.storage_engine == "mmapv1":
self.skipTest("MMAPv1 does not support retryWrites=True")
use_multi_mongos = test["useMultipleMongoses"]
host = None
if use_multi_mongos:

1
tools/synchro.py
View File

@ -212,6 +212,7 @@ converted_tests = [
"test_client.py",
"test_client_bulk_write.py",
"test_client_context.py",
"test_client_metadata.py",
"test_collation.py",
"test_collection.py",
"test_collection_management.py",