(self)
| 766 | self.assertNotIn(conn, server._pool.conns) |
| 767 | |
| 768 | def test_max_idle_time_checkout(self): |
| 769 | # Use high frequency to test _get_socket_no_auth. |
| 770 | with client_knobs(kill_cursor_frequency=99999999): |
| 771 | client = self.rs_or_single_client(maxIdleTimeMS=500) |
| 772 | server = (client._get_topology()).select_server(readable_server_selector, _Op.TEST) |
| 773 | with server._pool.checkout() as conn: |
| 774 | pass |
| 775 | self.assertEqual(1, len(server._pool.conns)) |
| 776 | time.sleep(1) # Sleep so that the socket becomes stale. |
| 777 | |
| 778 | with server._pool.checkout() as new_con: |
| 779 | self.assertNotEqual(conn, new_con) |
| 780 | self.assertEqual(1, len(server._pool.conns)) |
| 781 | self.assertNotIn(conn, server._pool.conns) |
| 782 | self.assertIn(new_con, server._pool.conns) |
| 783 | |
| 784 | # Test that connections are reused if maxIdleTimeMS is not set. |
| 785 | client = self.rs_or_single_client() |
| 786 | server = (client._get_topology()).select_server(readable_server_selector, _Op.TEST) |
| 787 | with server._pool.checkout() as conn: |
| 788 | pass |
| 789 | self.assertEqual(1, len(server._pool.conns)) |
| 790 | time.sleep(1) |
| 791 | with server._pool.checkout() as new_con: |
| 792 | self.assertEqual(conn, new_con) |
| 793 | self.assertEqual(1, len(server._pool.conns)) |
| 794 | |
| 795 | def test_constants(self): |
| 796 | """This test uses MongoClient explicitly to make sure that host and |
nothing calls this directly
no test coverage detected