summaryrefslogtreecommitdiff
path: root/tests/analysis/storage/storage.py
blob: 612d50053ae6556b398ad884ef0218bac9d88273 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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)