(self)
| 938 | |
| 939 | if checkpointer_cls is Checkpointer: |
| 940 | self.assertIsNone(ckpt._gc_thread) |
| 941 | |
| 942 | def test_stop_on_exception(self): |
| 943 | # Ensure that checkpointer gc thread terminates if there's an exception. |
| 944 | ckpt = _checkpointer_config().instantiate(parent=None) |
| 945 | |
| 946 | def run(): |
| 947 | ckpt._start_gc_thread() |
| 948 | raise ValueError("expected error") |
| 949 | |
| 950 | # By default, an exception in the main thread does not terminate child threads. |
| 951 | run_thread = threading.Thread(target=run) |
| 952 | run_thread.start() |
| 953 | run_thread.join() |
| 954 | self.assertFalse(ckpt._gc_stopping.is_set()) |
| 955 | ckpt.stop() # Stop it explicitly, otherwise test will run forever. |
| 956 | |
| 957 | def run_in_context(): |
| 958 | with ckpt: |
| 959 | raise ValueError("expected error") |
| 960 | |
| 961 | # With a context manager, we stop when context is exited. |
| 962 | run_thread = threading.Thread(target=run_in_context) |
| 963 | run_thread.start() |
| 964 | run_thread.join() |
| 965 | self.assertTrue(ckpt._gc_stopping.is_set()) |
| 966 |
nothing calls this directly
no test coverage detected