(self)
| 1236 | |
| 1237 | @client_context.require_no_tls |
| 1238 | def test_unix_socket(self): |
| 1239 | if not hasattr(socket, "AF_UNIX"): |
| 1240 | raise SkipTest("UNIX-sockets are not supported on this system") |
| 1241 | |
| 1242 | mongodb_socket = "/tmp/mongodb-%d.sock" % (client_context.port,) |
| 1243 | encoded_socket = "%2Ftmp%2F" + "mongodb-%d.sock" % (client_context.port,) |
| 1244 | if not os.access(mongodb_socket, os.R_OK): |
| 1245 | raise SkipTest("Socket file is not accessible") |
| 1246 | |
| 1247 | uri = "mongodb://%s" % encoded_socket |
| 1248 | # Confirm we can do operations via the socket. |
| 1249 | client = self.rs_or_single_client(uri) |
| 1250 | client.pymongo_test.test.insert_one({"dummy": "object"}) |
| 1251 | dbs = client.list_database_names() |
| 1252 | self.assertIn("pymongo_test", dbs) |
| 1253 | |
| 1254 | self.assertIn(mongodb_socket, repr(client)) |
| 1255 | |
| 1256 | # Confirm it fails with a missing socket. |
| 1257 | with self.assertRaises(ConnectionFailure): |
| 1258 | c = self.simple_client( |
| 1259 | "mongodb://%2Ftmp%2Fnon-existent.sock", serverSelectionTimeoutMS=100 |
| 1260 | ) |
| 1261 | connected(c) |
| 1262 | |
| 1263 | def test_document_class(self): |
| 1264 | c = self.client |
nothing calls this directly
no test coverage detected