(self)
| 1206 | |
| 1207 | @test_util.run_deprecated_v1 |
| 1208 | def testNonSharded(self): |
| 1209 | save_dir = self._get_test_dir("max_to_keep_non_sharded") |
| 1210 | |
| 1211 | with self.cached_session() as sess: |
| 1212 | v = variables.VariableV1(10.0, name="v") |
| 1213 | save = saver_module.Saver({"v": v}, max_to_keep=2) |
| 1214 | self.evaluate(variables.global_variables_initializer()) |
| 1215 | self.assertEqual([], save.last_checkpoints) |
| 1216 | |
| 1217 | s1 = save.save(sess, os.path.join(save_dir, "s1")) |
| 1218 | self.assertEqual([s1], save.last_checkpoints) |
| 1219 | self.assertTrue(checkpoint_management.checkpoint_exists(s1)) |
| 1220 | self.assertCheckpointState( |
| 1221 | model_checkpoint_path=s1, |
| 1222 | all_model_checkpoint_paths=[s1], |
| 1223 | save_dir=save_dir) |
| 1224 | |
| 1225 | s2 = save.save(sess, os.path.join(save_dir, "s2")) |
| 1226 | self.assertEqual([s1, s2], save.last_checkpoints) |
| 1227 | self.assertTrue(checkpoint_management.checkpoint_exists(s1)) |
| 1228 | self.assertTrue(checkpoint_management.checkpoint_exists(s2)) |
| 1229 | self.assertCheckpointState( |
| 1230 | model_checkpoint_path=s2, |
| 1231 | all_model_checkpoint_paths=[s1, s2], |
| 1232 | save_dir=save_dir) |
| 1233 | |
| 1234 | s3 = save.save(sess, os.path.join(save_dir, "s3")) |
| 1235 | self.assertEqual([s2, s3], save.last_checkpoints) |
| 1236 | self.assertFalse(checkpoint_management.checkpoint_exists(s1)) |
| 1237 | self.assertTrue(checkpoint_management.checkpoint_exists(s2)) |
| 1238 | self.assertTrue(checkpoint_management.checkpoint_exists(s3)) |
| 1239 | self.assertCheckpointState( |
| 1240 | model_checkpoint_path=s3, |
| 1241 | all_model_checkpoint_paths=[s2, s3], |
| 1242 | save_dir=save_dir) |
| 1243 | |
| 1244 | # Create a second helper, identical to the first. |
| 1245 | save2 = saver_module.Saver(saver_def=save.as_saver_def()) |
| 1246 | save2.set_last_checkpoints(save.last_checkpoints) |
| 1247 | |
| 1248 | # Create a third helper, with the same configuration but no knowledge of |
| 1249 | # previous checkpoints. |
| 1250 | save3 = saver_module.Saver(saver_def=save.as_saver_def()) |
| 1251 | |
| 1252 | # Exercise the first helper. |
| 1253 | |
| 1254 | # Adding s2 again (old s2 is removed first, then new s2 appended) |
| 1255 | s2 = save.save(sess, os.path.join(save_dir, "s2")) |
| 1256 | self.assertEqual([s3, s2], save.last_checkpoints) |
| 1257 | self.assertFalse(checkpoint_management.checkpoint_exists(s1)) |
| 1258 | self.assertFalse( |
| 1259 | checkpoint_management.checkpoint_exists( |
| 1260 | checkpoint_management.meta_graph_filename(s1))) |
| 1261 | self.assertTrue(checkpoint_management.checkpoint_exists(s3)) |
| 1262 | self.assertTrue( |
| 1263 | checkpoint_management.checkpoint_exists( |
| 1264 | checkpoint_management.meta_graph_filename(s3))) |
| 1265 | self.assertTrue(checkpoint_management.checkpoint_exists(s2)) |
nothing calls this directly
no test coverage detected