pytorch/caffe2/python/operator_test/checkpoint_test.py
Nikita Shulga 2f61aca17b Skip DataIO tests relying on LevelDB if compiled without it (#42169)
Summary:
Found while trying to get RocM Caffe2 job green

Pull Request resolved: https://github.com/pytorch/pytorch/pull/42169

Reviewed By: seemethere

Differential Revision: D22791896

Pulled By: malfet

fbshipit-source-id: 9df6233876aec5ead056365499bab970aa7e8bdc
2020-07-28 10:18:26 -07:00

45 lines
1.6 KiB
Python

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
from caffe2.python import core, workspace, test_util
import os
import shutil
import tempfile
import unittest
class CheckpointTest(test_util.TestCase):
"""A simple test case to make sure that the checkpoint behavior is correct.
"""
@unittest.skipIf("LevelDB" not in core.C.registered_dbs(), "Need LevelDB")
def testCheckpoint(self):
temp_root = tempfile.mkdtemp()
net = core.Net("test_checkpoint")
# Note(jiayq): I am being a bit lazy here and am using the old iter
# convention that does not have an input. Optionally change it to the
# new style if needed.
net.Iter([], "iter")
net.ConstantFill([], "value", shape=[1, 2, 3])
net.Checkpoint(["iter", "value"], [],
db=os.path.join(temp_root, "test_checkpoint_at_%05d"),
db_type="leveldb", every=10, absolute_path=True)
self.assertTrue(workspace.CreateNet(net))
for i in range(100):
self.assertTrue(workspace.RunNet("test_checkpoint"))
for i in range(1, 10):
# Print statements are only for debugging purposes.
# print("Asserting %d" % i)
# print(os.path.join(temp_root, "test_checkpoint_at_%05d" % (i * 10)))
self.assertTrue(os.path.exists(
os.path.join(temp_root, "test_checkpoint_at_%05d" % (i * 10))))
# Finally, clean up.
shutil.rmtree(temp_root)
if __name__ == "__main__":
unittest.main()