(self)
| 1312 | pass |
| 1313 | |
| 1314 | def test_socket_timeout(self): |
| 1315 | no_timeout = self.client |
| 1316 | timeout_sec = 1 |
| 1317 | timeout = self.rs_or_single_client(socketTimeoutMS=1000 * timeout_sec) |
| 1318 | |
| 1319 | no_timeout.pymongo_test.drop_collection("test") |
| 1320 | no_timeout.pymongo_test.test.insert_one({"x": 1}) |
| 1321 | |
| 1322 | # A $where clause that takes a second longer than the timeout |
| 1323 | where_func = delay(timeout_sec + 1) |
| 1324 | |
| 1325 | def get_x(db): |
| 1326 | doc = next(db.test.find().where(where_func)) |
| 1327 | return doc["x"] |
| 1328 | |
| 1329 | self.assertEqual(1, get_x(no_timeout.pymongo_test)) |
| 1330 | with self.assertRaises(NetworkTimeout): |
| 1331 | get_x(timeout.pymongo_test) |
| 1332 | |
| 1333 | def test_server_selection_timeout(self): |
| 1334 | client = MongoClient(serverSelectionTimeoutMS=100, connect=False) |
nothing calls this directly
no test coverage detected