(self)
| 1273 | |
| 1274 | @async_client_context.require_no_tls |
| 1275 | async def test_unix_socket(self): |
| 1276 | if not hasattr(socket, "AF_UNIX"): |
| 1277 | raise SkipTest("UNIX-sockets are not supported on this system") |
| 1278 | |
| 1279 | mongodb_socket = "/tmp/mongodb-%d.sock" % (await async_client_context.port,) |
| 1280 | encoded_socket = "%2Ftmp%2F" + "mongodb-%d.sock" % (await async_client_context.port,) |
| 1281 | if not os.access(mongodb_socket, os.R_OK): |
| 1282 | raise SkipTest("Socket file is not accessible") |
| 1283 | |
| 1284 | uri = "mongodb://%s" % encoded_socket |
| 1285 | # Confirm we can do operations via the socket. |
| 1286 | client = await self.async_rs_or_single_client(uri) |
| 1287 | await client.pymongo_test.test.insert_one({"dummy": "object"}) |
| 1288 | dbs = await client.list_database_names() |
| 1289 | self.assertIn("pymongo_test", dbs) |
| 1290 | |
| 1291 | self.assertIn(mongodb_socket, repr(client)) |
| 1292 | |
| 1293 | # Confirm it fails with a missing socket. |
| 1294 | with self.assertRaises(ConnectionFailure): |
| 1295 | c = self.simple_client( |
| 1296 | "mongodb://%2Ftmp%2Fnon-existent.sock", serverSelectionTimeoutMS=100 |
| 1297 | ) |
| 1298 | await connected(c) |
| 1299 | |
| 1300 | async def test_document_class(self): |
| 1301 | c = self.client |
nothing calls this directly
no test coverage detected