(self, enable_async_checkpoint_saving)
| 301 | ("_async_checkpoint_saving", True) |
| 302 | ) |
| 303 | def test_has_checkpoint_no_summaries(self, enable_async_checkpoint_saving): |
| 304 | test_runner = TestRunner() |
| 305 | # Has checkpoint, but no summary directories. |
| 306 | checkpoint = tf.train.Checkpoint(model=test_runner.model) |
| 307 | checkpoint_manager = tf.train.CheckpointManager( |
| 308 | checkpoint, |
| 309 | self.model_dir, |
| 310 | max_to_keep=None, |
| 311 | step_counter=test_runner.global_step) |
| 312 | test_controller = controller.Controller( |
| 313 | trainer=test_runner, |
| 314 | evaluator=test_runner, |
| 315 | global_step=test_runner.global_step, |
| 316 | checkpoint_manager=checkpoint_manager, |
| 317 | enable_async_checkpointing=enable_async_checkpoint_saving, |
| 318 | steps_per_loop=2) |
| 319 | test_controller.train_and_evaluate( |
| 320 | train_steps=10, eval_steps=2, eval_interval=6) |
| 321 | self.assertEqual(test_runner.global_step, 10) |
| 322 | self.assertTrue(controller._orbit_api_gauge.get_cell().value()) |
| 323 | |
| 324 | # No summaries are saved. |
| 325 | self.assertEmpty(tf.io.gfile.glob( |
| 326 | os.path.join(checkpoint_manager.directory, "events.*"))) |
| 327 | |
| 328 | @parameterized.named_parameters( |
| 329 | ("_sync_checkpoint_saving", False), |
nothing calls this directly
no test coverage detected