db_test.py 1.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546
  1. from caffe2.python import workspace
  2. import os
  3. import tempfile
  4. import unittest
  5. class TestDB(unittest.TestCase):
  6. def setUp(self):
  7. handle, self.file_name = tempfile.mkstemp()
  8. os.close(handle)
  9. self.data = [
  10. (
  11. "key{}".format(i).encode("ascii"),
  12. "value{}".format(i).encode("ascii")
  13. )
  14. for i in range(1, 10)
  15. ]
  16. def testSimple(self):
  17. db = workspace.C.create_db(
  18. "minidb", self.file_name, workspace.C.Mode.write)
  19. for key, value in self.data:
  20. transaction = db.new_transaction()
  21. transaction.put(key, value)
  22. del transaction
  23. del db # should close DB
  24. db = workspace.C.create_db(
  25. "minidb", self.file_name, workspace.C.Mode.read)
  26. cursor = db.new_cursor()
  27. data = []
  28. while cursor.valid():
  29. data.append((cursor.key(), cursor.value()))
  30. cursor.next() # noqa: B305
  31. del cursor
  32. db.close() # test explicit db closer
  33. self.assertEqual(data, self.data)