(uri: str)
| 57 | |
| 58 | |
| 59 | def main(uri: str) -> None: |
| 60 | heartbeat_logger = HeartbeatLogger() |
| 61 | client = MongoClient( |
| 62 | uri, |
| 63 | event_listeners=[heartbeat_logger], |
| 64 | heartbeatFrequencyMS=500, |
| 65 | connectTimeoutMS=500, |
| 66 | server_api=SERVER_API, |
| 67 | ) |
| 68 | client.admin.command("ping") |
| 69 | logging.info("TEST STARTED") |
| 70 | # test_sigstop_sigcont will SIGSTOP and SIGCONT this process in this loop. |
| 71 | while True: |
| 72 | try: |
| 73 | data = input('Type "q" to quit: ') |
| 74 | except EOFError: |
| 75 | break |
| 76 | if data == "q": |
| 77 | break |
| 78 | client.admin.command("ping") |
| 79 | logging.info("TEST COMPLETED") |
| 80 | heartbeat_logger.close() |
| 81 | client.close() |
| 82 | |
| 83 | |
| 84 | if __name__ == "__main__": |
no test coverage detected