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 2024-10-15 19:22:56 -05:00
commit 01733cf050
No known key found for this signature in database
GPG Key ID: B1BF5EC3A8B32F91
10 changed files with 280 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doc/changelog.rst
View File

@ -17,6 +17,8 @@ PyMongo 4.11 brings a number of changes including:
- :attr:`~pymongo.asynchronous.mongo_client.AsyncMongoClient.address` and
:attr:`~pymongo.mongo_client.MongoClient.address` now correctly block when called on unconnected clients
until either connection succeeds or a server selection timeout error is raised.
- Added :func:`repr` support to :class:`pymongo.operations.IndexModel`.
- Added :func:`repr` support to :class:`pymongo.operations.SearchIndexModel`.
Issues Resolved
...............

13
pymongo/operations.py
View File

@ -773,6 +773,13 @@ class IndexModel:
"""
return self.__document
def __repr__(self) -> str:
return "{}({}{})".format(
self.__class__.__name__,
self.document["key"],
"".join([f", {key}={value!r}" for key, value in self.document.items() if key != "key"]),
)
class SearchIndexModel:
"""Represents a search index to create."""
@ -812,3 +819,9 @@ class SearchIndexModel:
def document(self) -> Mapping[str, Any]:
"""The document for this index."""
return self.__document
def __repr__(self) -> str:
return "{}({})".format(
self.__class__.__name__,
", ".join([f"{key}={value!r}" for key, value in self.document.items()]),
)

41
test/asynchronous/test_collection_management.py Normal file
View File

@ -0,0 +1,41 @@
# Copyright 2021-present MongoDB, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test the collection management unified spec tests."""
from __future__ import annotations
import os
import pathlib
import sys
sys.path[0:0] = [""]
from test import unittest
from test.asynchronous.unified_format import generate_test_classes
_IS_SYNC = False
# Location of JSON test specifications.
if _IS_SYNC:
_TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "collection_management")
else:
_TEST_PATH = os.path.join(
pathlib.Path(__file__).resolve().parent.parent, "collection_management"
)
# Generate unified tests.
globals().update(generate_test_classes(_TEST_PATH, module=__name__))
if __name__ == "__main__":
unittest.main()

128
test/asynchronous/test_create_entities.py Normal file
View File

@ -0,0 +1,128 @@
# Copyright 2021-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 sys
import unittest
sys.path[0:0] = [""]
from test.asynchronous import AsyncIntegrationTest
from test.asynchronous.unified_format import UnifiedSpecTestMixinV1
_IS_SYNC = False
class TestCreateEntities(AsyncIntegrationTest):
async def test_store_events_as_entities(self):
self.scenario_runner = UnifiedSpecTestMixinV1()
spec = {
"description": "blank",
"schemaVersion": "1.2",
"createEntities": [
{
"client": {
"id": "client0",
"storeEventsAsEntities": [
{
"id": "events1",
"events": [
"PoolCreatedEvent",
],
}
],
}
},
],
"tests": [{"description": "foo", "operations": []}],
}
self.scenario_runner.TEST_SPEC = spec
await self.scenario_runner.asyncSetUp()
await self.scenario_runner.run_scenario(spec["tests"][0])
await self.scenario_runner.entity_map["client0"].close()
final_entity_map = self.scenario_runner.entity_map
self.assertIn("events1", final_entity_map)
self.assertGreater(len(final_entity_map["events1"]), 0)
for event in final_entity_map["events1"]:
self.assertIn("PoolCreatedEvent", event["name"])
async def test_store_all_others_as_entities(self):
self.scenario_runner = UnifiedSpecTestMixinV1()
spec = {
"description": "Find",
"schemaVersion": "1.2",
"createEntities": [
{
"client": {
"id": "client0",
"uriOptions": {"retryReads": True},
}
},
{"database": {"id": "database0", "client": "client0", "databaseName": "dat"}},
{
"collection": {
"id": "collection0",
"database": "database0",
"collectionName": "dat",
}
},
],
"tests": [
{
"description": "test loops",
"operations": [
{
"name": "loop",
"object": "testRunner",
"arguments": {
"storeIterationsAsEntity": "iterations",
"storeSuccessesAsEntity": "successes",
"storeFailuresAsEntity": "failures",
"storeErrorsAsEntity": "errors",
"numIterations": 5,
"operations": [
{
"name": "insertOne",
"object": "collection0",
"arguments": {"document": {"_id": 1, "x": 44}},
},
{
"name": "insertOne",
"object": "collection0",
"arguments": {"document": {"_id": 2, "x": 44}},
},
],
},
}
],
}
],
}
await self.client.dat.dat.delete_many({})
self.scenario_runner.TEST_SPEC = spec
await self.scenario_runner.asyncSetUp()
await self.scenario_runner.run_scenario(spec["tests"][0])
await self.scenario_runner.entity_map["client0"].close()
entity_map = self.scenario_runner.entity_map
self.assertEqual(len(entity_map["errors"]), 4)
for error in entity_map["errors"]:
self.assertEqual(error["type"], "DuplicateKeyError")
self.assertEqual(entity_map["failures"], [])
self.assertEqual(entity_map["successes"], 2)
self.assertEqual(entity_map["iterations"], 5)
if __name__ == "__main__":
unittest.main()

2
test/asynchronous/unified_format.py
View File

@ -773,7 +773,7 @@ class UnifiedSpecTestMixinV1(AsyncIntegrationTest):
if "batch_size" in kwargs:
kwargs["cursor"] = {"batchSize": kwargs.pop("batch_size")}
cursor = await target.list_collections(*args, **kwargs)
return list(cursor)
return await cursor.to_list()
async def _databaseOperation_createCollection(self, target, *args, **kwargs):
# PYTHON-1936 Ignore the listCollections event from create_collection.

12
test/test_collection_management.py
View File

@ -16,6 +16,7 @@
from __future__ import annotations
import os
import pathlib
import sys
sys.path[0:0] = [""]
@ -23,11 +24,18 @@ sys.path[0:0] = [""]
from test import unittest
from test.unified_format import generate_test_classes
_IS_SYNC = True
# Location of JSON test specifications.
TEST_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "collection_management")
if _IS_SYNC:
_TEST_PATH = os.path.join(pathlib.Path(__file__).resolve().parent, "collection_management")
else:
_TEST_PATH = os.path.join(
pathlib.Path(__file__).resolve().parent.parent, "collection_management"
)
# Generate unified tests.
globals().update(generate_test_classes(TEST_PATH, module=__name__))
globals().update(generate_test_classes(_TEST_PATH, module=__name__))
if __name__ == "__main__":
unittest.main()

2
test/test_create_entities.py
View File

@ -21,6 +21,8 @@ sys.path[0:0] = [""]
from test import IntegrationTest
from test.unified_format import UnifiedSpecTestMixinV1
_IS_SYNC = True
class TestCreateEntities(IntegrationTest):
def test_store_events_as_entities(self):

80
test/test_operations.py Normal file
View File

@ -0,0 +1,80 @@
# Copyright 2024-present MongoDB, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Test the operations module."""
from __future__ import annotations
from test import UnitTest, unittest
from pymongo import ASCENDING, DESCENDING
from pymongo.collation import Collation
from pymongo.errors import OperationFailure
from pymongo.operations import IndexModel, SearchIndexModel
class TestOperationsBase(UnitTest):
"""Base class for testing operations module."""
def assertRepr(self, obj):
new_obj = eval(repr(obj))
self.assertEqual(type(new_obj), type(obj))
self.assertEqual(repr(new_obj), repr(obj))
class TestIndexModel(TestOperationsBase):
"""Test IndexModel features."""
def test_repr(self):
# Based on examples in test_collection.py
self.assertRepr(IndexModel("hello"))
self.assertRepr(IndexModel([("hello", DESCENDING), ("world", ASCENDING)]))
self.assertRepr(
IndexModel([("hello", DESCENDING), ("world", ASCENDING)], name="hello_world")
)
# Test all the kwargs
self.assertRepr(IndexModel("name", name="name"))
self.assertRepr(IndexModel("unique", unique=False))
self.assertRepr(IndexModel("background", background=True))
self.assertRepr(IndexModel("sparse", sparse=True))
self.assertRepr(IndexModel("bucketSize", bucketSize=1))
self.assertRepr(IndexModel("min", min=1))
self.assertRepr(IndexModel("max", max=1))
self.assertRepr(IndexModel("expireAfterSeconds", expireAfterSeconds=1))
self.assertRepr(
IndexModel("partialFilterExpression", partialFilterExpression={"hello": "world"})
)
self.assertRepr(IndexModel("collation", collation=Collation(locale="en_US")))
self.assertRepr(IndexModel("wildcardProjection", wildcardProjection={"$**": 1}))
self.assertRepr(IndexModel("hidden", hidden=False))
# Test string literal
self.assertEqual(repr(IndexModel("hello")), "IndexModel({'hello': 1}, name='hello_1')")
self.assertEqual(
repr(IndexModel({"hello": 1, "world": -1})),
"IndexModel({'hello': 1, 'world': -1}, name='hello_1_world_-1')",
)
class TestSearchIndexModel(TestOperationsBase):
"""Test SearchIndexModel features."""
def test_repr(self):
self.assertRepr(SearchIndexModel({"hello": "hello"}, key=1))
self.assertEqual(
repr(SearchIndexModel({"hello": "hello"}, key=1)),
"SearchIndexModel(definition={'hello': 'hello'}, key=1)",
)
if __name__ == "__main__":
unittest.main()

2
test/unified_format.py
View File

@ -769,7 +769,7 @@ class UnifiedSpecTestMixinV1(IntegrationTest):
if "batch_size" in kwargs:
kwargs["cursor"] = {"batchSize": kwargs.pop("batch_size")}
cursor = target.list_collections(*args, **kwargs)
return list(cursor)
return cursor.to_list()
def _databaseOperation_createCollection(self, target, *args, **kwargs):
# PYTHON-1936 Ignore the listCollections event from create_collection.

2
tools/synchro.py
View File

@ -192,6 +192,7 @@ converted_tests = [
"test_client_context.py",
"test_collation.py",
"test_collection.py",
"test_collection_management.py",
"test_command_logging.py",
"test_command_logging.py",
"test_command_monitoring.py",
@ -199,6 +200,7 @@ converted_tests = [
"test_common.py",
"test_connection_logging.py",
"test_connections_survive_primary_stepdown_spec.py",
"test_create_entities.py",
"test_crud_unified.py",
"test_cursor.py",
"test_database.py",