← Back to branch summary

~libbls/libbls/libbls-package

« back to all changes in this revision

Viewing changes to test/data_object_file_tests.py

  • Committer: Alexandros Frantzis
  • Date: 2011-10-01 10:10:46 UTC
  • Revision ID: alf82@freemail.gr-20111001101046-etz52jurwzwc007x
Tags: upstream-0.3.2
Import upstream version 0.3.2

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/data_object_file_tests.py
 
1
import unittest
 
2
import os
 
3
import hashlib
 
4
import errno
 
5
import tempfile
 
6
import shutil
 
7
from libbls import *
 
8
 
 
9
def get_file_fd(name):
 
10
        """Get the fd of a file.
 
11
           We need this so that the tests can be executed from within
 
12
           the test directory as well as from the source tree root"""
 
13
        try:
 
14
                fd = os.open("test/%s" % name, os.O_RDONLY)
 
15
                return fd
 
16
        except:
 
17
                pass
 
18
        
 
19
        fd = os.open("%s" % name, os.O_RDONLY)
 
20
        return fd
 
21
        
 
22
def get_tmp_copy_file_fd(name, flags = os.O_RDONLY):
 
23
        """Get the fd of a temporary file which is a copy of the specified
 
24
        file."""
 
25
 
 
26
        (tmp_fd, tmp_path) = tempfile.mkstemp();
 
27
        os.close(tmp_fd)
 
28
 
 
29
        success = False
 
30
        try:
 
31
                shutil.copyfile("test/%s" % name, tmp_path)
 
32
                success = True
 
33
        except:
 
34
                pass
 
35
        
 
36
        if not success:
 
37
                shutil.copyfile(name, tmp_path)
 
38
 
 
39
        fd = os.open(tmp_path, flags)
 
40
 
 
41
        return (fd, tmp_path)
 
42
 
 
43
class DataObjectFileTests(unittest.TestCase):
 
44
 
 
45
        def setUp(self):
 
46
                fd = get_file_fd("data_object_test_file.bin")
 
47
 
 
48
                (err, self.obj) = data_object_file_new(fd)
 
49
 
 
50
        def tearDown(self):
 
51
                if self.obj is not None:
 
52
                        data_object_free(self.obj)
 
53
 
 
54
        def testNew(self):
 
55
                "Create a file data object"
 
56
 
 
57
                self.assert_(self.obj is not None)
 
58
 
 
59
                self.assertEqual(data_object_get_size(self.obj)[1], 10)
 
60
 
 
61
        def testGetData(self):
 
62
                "Get data from a file data object"
 
63
 
 
64
                (err, buf) = data_object_get_data(self.obj, 0, 10, DATA_OBJECT_READ)
 
65
                self.assertEqual(err, 0)
 
66
                self.assertEqual(len(buf), 10)
 
67
 
 
68
                expected_data = "1234567890"
 
69
 
 
70
                for i in range(len(buf)):
 
71
                        self.assertEqual(buf[i], expected_data[i])
 
72
                
 
73
                (err, buf) = data_object_get_data(self.obj, 5, 5, DATA_OBJECT_READ)
 
74
                self.assertEqual(err, 0)
 
75
                self.assertEqual(len(buf), 5)
 
76
 
 
77
                for i in range(len(buf)):
 
78
                        self.assertEqual(buf[i], expected_data[5 + i])
 
79
 
 
80
                (err, buf) = data_object_get_data(self.obj, 0, 0, DATA_OBJECT_READ)
 
81
                self.assertEqual(err, 0)
 
82
                self.assertEqual(len(buf), 0)
 
83
        
 
84
        def testGetDataInvalid(self):
 
85
                "Try to get data from invalid offsets"
 
86
 
 
87
                (err, buf) = data_object_get_data(self.obj, -1, 10, DATA_OBJECT_READ)
 
88
                self.assertNotEqual(err, 0)
 
89
 
 
90
                (err, buf) = data_object_get_data(self.obj, 5, 10, DATA_OBJECT_READ)
 
91
                self.assertNotEqual(err, 0)
 
92
 
 
93
                (err, buf) = data_object_get_data(self.obj, 10, 0, DATA_OBJECT_READ)
 
94
                self.assertNotEqual(err, 0)
 
95
 
 
96
        def testGetDataLarge(self):
 
97
                "Get data from a large file data object"
 
98
 
 
99
                # Create the file data object
 
100
                try:
 
101
                        fd = get_file_fd("data_object_test_file_large.bin")
 
102
                except:
 
103
                        print "Warning: Not running data_object_file_tests.testGetDataLarge "\
 
104
                                "because the file data_object_test_file_large.bin was not found."
 
105
                        return
 
106
 
 
107
                # Get the expected digest
 
108
                try:
 
109
                        f = os.fdopen(get_file_fd("data_object_test_file_large.md5"))
 
110
                        expected_digest = f.readline().strip()
 
111
                except:
 
112
                        print "Warning: Not running data_object_file_tests.testGetDataLarge "\
 
113
                                "because the file data_object_test_file_large.md5 was not found."
 
114
                        return
 
115
 
 
116
 
 
117
                (err, dobj) = data_object_file_new(fd)
 
118
                self.assertEqual(err, 0)
 
119
 
 
120
                (err, size) = data_object_get_size(dobj)
 
121
                self.assertEqual(err, 0)
 
122
 
 
123
                hash = hashlib.md5()
 
124
                bytes_left = size
 
125
 
 
126
                # Read the file and update the digest
 
127
                while bytes_left > 0:
 
128
                        (err, buf) = data_object_get_data(dobj, size - bytes_left, 
 
129
                                        bytes_left, DATA_OBJECT_READ)
 
130
                        
 
131
                        self.assertEqual(err, 0)
 
132
 
 
133
                        hash.update(buf)
 
134
 
 
135
                        bytes_left = bytes_left - len(buf)
 
136
                        
 
137
                digest = hash.hexdigest()
 
138
 
 
139
                # Compare our hash digest with the precomputed one
 
140
                self.assertEqual(digest, expected_digest)
 
141
 
 
142
                data_object_free(dobj)
 
143
 
 
144
        def testGetDataOverflow(self):
 
145
                "Test boundary cases for get_data overflow"
 
146
 
 
147
                # This one fails because of invalid range, not overflow
 
148
                (err, buf) = data_object_get_data(self.obj, get_max_off_t(), 0, 
 
149
                                DATA_OBJECT_READ)
 
150
                self.assertEqual(err, errno.EINVAL)
 
151
 
 
152
                # But this one fails because of overflow (overflow
 
153
                # is checked before range)
 
154
                (err, buf) = data_object_get_data(self.obj, get_max_off_t(), 2,
 
155
                                DATA_OBJECT_READ)
 
156
                self.assertEqual(err, errno.EOVERFLOW)
 
157
 
 
158
        def testCompare(self):
 
159
                "Compare two data objects"
 
160
 
 
161
                fd = get_file_fd("data_object_test_file.bin")
 
162
                (err, obj1) = data_object_file_new(fd)
 
163
                self.assertEqual(err, 0)
 
164
 
 
165
                (err, result) = data_object_compare(self.obj, obj1)
 
166
                self.assertEqual(err, 0)
 
167
                self.assertEqual(result, 0)
 
168
 
 
169
                data_object_free(obj1)
 
170
 
 
171
                fd = get_file_fd("data_object_file_tests.py")
 
172
                (err, obj1) = data_object_file_new(fd)
 
173
                self.assertEqual(err, 0)
 
174
 
 
175
                (err, result) = data_object_compare(self.obj, obj1)
 
176
                self.assertEqual(err, 0)
 
177
                self.assertEqual(result, 1)
 
178
 
 
179
                # Compare a file object with a memory object
 
180
 
 
181
                (err, obj2) = data_object_memory_new_ptr(0, get_max_size_t() - 1)
 
182
                self.assertEqual(err, 0)
 
183
 
 
184
                (err, result) = data_object_compare(obj1, obj2)
 
185
                self.assertEqual(err, 0)
 
186
                self.assertEqual(result, 1)
 
187
 
 
188
                data_object_free(obj1)
 
189
                data_object_free(obj2)
 
190
 
 
191
        def testTempFile(self):
 
192
                "Create a tempfile data object"
 
193
 
 
194
                (fd1, fd1_path) = get_tmp_copy_file_fd("data_object_test_file.bin", os.O_RDWR)
 
195
                (err, obj1) = data_object_tempfile_new(fd1, fd1_path)
 
196
                self.assertEqual(err, 0)
 
197
 
 
198
                # Read data from the file
 
199
                (err, buf) = data_object_get_data(obj1, 0, 10, DATA_OBJECT_READ)
 
200
                self.assertEqual(err, 0)
 
201
                self.assertEqual(len(buf), 10)
 
202
 
 
203
                expected_data = "1234567890"
 
204
 
 
205
                for i in range(len(buf)):
 
206
                        self.assertEqual(buf[i], expected_data[i])
 
207
 
 
208
                # Free the data object
 
209
                data_object_free(obj1)
 
210
 
 
211
                # Make sure that the temporary file has been deleted
 
212
                self.assertEqual(os.path.exists(fd1_path), False)
 
213
 
 
214
 
 
215
if __name__ == '__main__':
 
216
        unittest.main()