(self)
| 1058 | |
| 1059 | @client_context.require_no_standalone |
| 1060 | def test_writes(self): |
| 1061 | self._test_writes( |
| 1062 | lambda coll, session: coll.bulk_write([InsertOne[dict]({})], session=session) |
| 1063 | ) |
| 1064 | self._test_writes(lambda coll, session: coll.insert_one({}, session=session)) |
| 1065 | self._test_writes(lambda coll, session: coll.insert_many([{}], session=session)) |
| 1066 | self._test_writes( |
| 1067 | lambda coll, session: coll.replace_one({"_id": 1}, {"x": 1}, session=session) |
| 1068 | ) |
| 1069 | self._test_writes( |
| 1070 | lambda coll, session: coll.update_one({}, {"$set": {"X": 1}}, session=session) |
| 1071 | ) |
| 1072 | self._test_writes( |
| 1073 | lambda coll, session: coll.update_many({}, {"$set": {"x": 1}}, session=session) |
| 1074 | ) |
| 1075 | self._test_writes(lambda coll, session: coll.delete_one({}, session=session)) |
| 1076 | self._test_writes(lambda coll, session: coll.delete_many({}, session=session)) |
| 1077 | self._test_writes( |
| 1078 | lambda coll, session: coll.find_one_and_replace({"x": 1}, {"y": 1}, session=session) |
| 1079 | ) |
| 1080 | self._test_writes( |
| 1081 | lambda coll, session: coll.find_one_and_update( |
| 1082 | {"y": 1}, {"$set": {"x": 1}}, session=session |
| 1083 | ) |
| 1084 | ) |
| 1085 | self._test_writes(lambda coll, session: coll.find_one_and_delete({"x": 1}, session=session)) |
| 1086 | self._test_writes(lambda coll, session: coll.create_index("foo", session=session)) |
| 1087 | self._test_writes( |
| 1088 | lambda coll, session: coll.create_indexes( |
| 1089 | [IndexModel([("bar", ASCENDING)])], session=session |
| 1090 | ) |
| 1091 | ) |
| 1092 | self._test_writes(lambda coll, session: coll.drop_index("foo_1", session=session)) |
| 1093 | self._test_writes(lambda coll, session: coll.drop_indexes(session=session)) |
| 1094 | |
| 1095 | def _test_no_read_concern(self, op): |
| 1096 | coll = self.client.pymongo_test.test |
nothing calls this directly
no test coverage detected