(self)
| 254 | self.assertTrue(socket_checker.socket_closed(s)) |
| 255 | |
| 256 | def test_socket_checker(self): |
| 257 | s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 258 | s.connect((client_context.host, client_context.port)) |
| 259 | socket_checker = SocketChecker() |
| 260 | # Socket has nothing to read. |
| 261 | self.assertFalse(socket_checker.select(s, read=True)) |
| 262 | self.assertFalse(socket_checker.select(s, read=True, timeout=0)) |
| 263 | self.assertFalse(socket_checker.select(s, read=True, timeout=0.05)) |
| 264 | # Socket is writable. |
| 265 | self.assertTrue(socket_checker.select(s, write=True, timeout=None)) |
| 266 | self.assertTrue(socket_checker.select(s, write=True)) |
| 267 | self.assertTrue(socket_checker.select(s, write=True, timeout=0)) |
| 268 | self.assertTrue(socket_checker.select(s, write=True, timeout=0.05)) |
| 269 | # Make the socket readable |
| 270 | _, msg, _ = message._query( |
| 271 | 0, "admin.$cmd", 0, -1, SON([("ping", 1)]), None, DEFAULT_CODEC_OPTIONS |
| 272 | ) |
| 273 | s.sendall(msg) |
| 274 | # Block until the socket is readable. |
| 275 | self.assertTrue(socket_checker.select(s, read=True, timeout=None)) |
| 276 | self.assertTrue(socket_checker.select(s, read=True)) |
| 277 | self.assertTrue(socket_checker.select(s, read=True, timeout=0)) |
| 278 | self.assertTrue(socket_checker.select(s, read=True, timeout=0.05)) |
| 279 | # Socket is still writable. |
| 280 | self.assertTrue(socket_checker.select(s, write=True, timeout=None)) |
| 281 | self.assertTrue(socket_checker.select(s, write=True)) |
| 282 | self.assertTrue(socket_checker.select(s, write=True, timeout=0)) |
| 283 | self.assertTrue(socket_checker.select(s, write=True, timeout=0.05)) |
| 284 | s.close() |
| 285 | self.assertTrue(socket_checker.socket_closed(s)) |
| 286 | |
| 287 | def test_return_socket_after_reset(self): |
| 288 | pool = self.create_pool() |
nothing calls this directly
no test coverage detected