summaryrefslogtreecommitdiff
path: root/tests/analysis/storage
diff options
context:
space:
mode:
authorCyrille Bagard <nocbos@gmail.com>2021-07-04 17:06:28 (GMT)
committerCyrille Bagard <nocbos@gmail.com>2021-07-04 17:06:28 (GMT)
commit5958d9b25f806df73cd0634de2c9475eb6497e8c (patch)
tree7594605f9722cf5329c965cd35e11d52f2dfc4c8 /tests/analysis/storage
parent0150df2a3dafcce46bc95a2cb8642d0bb842ca8d (diff)
Store and load binary contents on demand.
Diffstat (limited to 'tests/analysis/storage')
-rw-r--r--tests/analysis/storage/storage.py81
1 files changed, 81 insertions, 0 deletions
diff --git a/tests/analysis/storage/storage.py b/tests/analysis/storage/storage.py
new file mode 100644
index 0000000..612d500
--- /dev/null
+++ b/tests/analysis/storage/storage.py
@@ -0,0 +1,81 @@
+
+from chrysacase import ChrysalideTestCase
+from pychrysalide import core
+from pychrysalide.analysis.contents import FileContent
+from pychrysalide.analysis.storage import ObjectStorage
+from pychrysalide.common import PackedBuffer
+import os
+import shutil
+import tempfile
+
+
+class TestObjectStorage(ChrysalideTestCase):
+ """TestCase for analysis.storage."""
+
+ @classmethod
+ def setUpClass(cls):
+
+ super(TestObjectStorage, cls).setUpClass()
+
+ cls._tmp_path = tempfile.mkdtemp()
+
+ config = core.get_main_configuration()
+ param = config.search(core.MainParameterKeys.TMPDIR)
+
+ cls._old_tmpdir = param.value
+ param.value = cls._tmp_path
+
+ cls.log('Using temporary directory "%s"' % cls._tmp_path)
+
+
+ @classmethod
+ def tearDownClass(cls):
+
+ super(TestObjectStorage, cls).tearDownClass()
+
+ config = core.get_main_configuration()
+ param = config.search(core.MainParameterKeys.TMPDIR)
+
+ param.value = cls._old_tmpdir
+
+ # import os
+ # os.system('ls -laihR %s' % cls._tmp_path)
+
+ cls.log('Delete directory "%s"' % cls._tmp_path)
+
+ shutil.rmtree(cls._tmp_path)
+
+
+ def testFileContentStorage(self):
+ """Store and load file binary content."""
+
+ storage = ObjectStorage('my-storage-hash')
+ self.assertIsNotNone(storage)
+
+ filename = os.path.join(self._tmp_path, 'test.bin')
+
+ with open(filename, 'wb') as fd:
+ fd.write(b'ABC')
+
+ cnt = FileContent(filename)
+ self.assertIsNotNone(cnt)
+
+ ret = storage.store_object('contents', cnt)
+ self.assertEqual(ret, 0)
+
+ pbuf = PackedBuffer()
+
+ ret = storage.store(pbuf)
+ self.assertTrue(ret)
+
+ self.assertTrue(pbuf.payload_length > 0)
+
+ pbuf.rewind()
+
+ storage2 = ObjectStorage.load(pbuf)
+ self.assertIsNotNone(storage2)
+
+ cnt2 = storage2.load_object('contents', 0)
+ self.assertIsNotNone(cnt2)
+
+ self.assertEqual(cnt.data, cnt2.data)