| 246 | handle_del_response(del_client) |
| 247 | |
| 248 | def main(test, flush): |
| 249 | clear_ops() |
| 250 | |
| 251 | if flush: |
| 252 | flush_db_sync() |
| 253 | |
| 254 | try: |
| 255 | globals()[f"init_{test}"]() |
| 256 | except KeyError: |
| 257 | print(f"Test \"{test}\" not found. Exiting...") |
| 258 | exit() |
| 259 | except ConnectionRefusedError: |
| 260 | print("Could not connect to server. Is it running?") |
| 261 | print("Exiting...") |
| 262 | exit() |
| 263 | |
| 264 | threading.Thread(target=stats_thread).start() |
| 265 | asyncore.loop() |
| 266 | print("Done.") |
| 267 | |
| 268 | parser = argparse.ArgumentParser(description="Test use cases for KeyDB.") |
| 269 | parser.add_argument('test', choices=[x[5:] for x in filter(lambda name: name.startswith("init_"), globals().keys())], help="which test to run") |