1 | import os |
---|
2 | import tempfile |
---|
3 | import shutil |
---|
4 | import unittest |
---|
5 | from StringIO import StringIO |
---|
6 | from hurry.file import HurryFile |
---|
7 | from hurry.file.interfaces import IFileRetrieval |
---|
8 | from zope.component import ( |
---|
9 | getUtility, provideUtility, queryUtility, provideAdapter) |
---|
10 | from zope.component.hooks import setSite |
---|
11 | from zope.interface.verify import verifyClass, verifyObject |
---|
12 | from waeup.sirp.app import University |
---|
13 | from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase |
---|
14 | from waeup.sirp.imagestorage import ( |
---|
15 | FileStoreNameChooser, ExtFileStore, DefaultFileStoreHandler, |
---|
16 | DefaultStorage) |
---|
17 | from waeup.sirp.interfaces import ( |
---|
18 | IFileStoreNameChooser, IExtFileStore, IFileStoreHandler,) |
---|
19 | |
---|
20 | class HelperFuncsTests(unittest.TestCase): |
---|
21 | |
---|
22 | def setUp(self): |
---|
23 | self.workdir = tempfile.mkdtemp() |
---|
24 | |
---|
25 | def tearDown(self): |
---|
26 | shutil.rmtree(self.workdir) |
---|
27 | |
---|
28 | class FileStoreNameChooserTests(FunctionalTestCase): |
---|
29 | |
---|
30 | layer = FunctionalLayer |
---|
31 | |
---|
32 | def test_iface(self): |
---|
33 | # we provide the interfaces we promise to do |
---|
34 | obj = FileStoreNameChooser(None) |
---|
35 | verifyClass(IFileStoreNameChooser, FileStoreNameChooser) |
---|
36 | verifyObject(IFileStoreNameChooser, obj) |
---|
37 | return |
---|
38 | |
---|
39 | def test_accessible_as_adapter(self): |
---|
40 | # we can get a file name chooser via adapter |
---|
41 | chooser = IFileStoreNameChooser(object()) |
---|
42 | self.assertTrue( |
---|
43 | isinstance(chooser, FileStoreNameChooser)) |
---|
44 | return |
---|
45 | |
---|
46 | def test_check_name(self): |
---|
47 | # default file name choosers accept any string |
---|
48 | chooser = FileStoreNameChooser(object()) |
---|
49 | self.assertEqual(chooser.checkName('Hi there!'), True) |
---|
50 | self.assertEqual(chooser.checkName(None), False) |
---|
51 | return |
---|
52 | |
---|
53 | def test_choose_name(self): |
---|
54 | # we get a simple string if we do not pass in a valid string |
---|
55 | chooser = FileStoreNameChooser(object()) |
---|
56 | self.assertEqual(chooser.chooseName('myname'), 'myname') |
---|
57 | self.assertEqual(chooser.chooseName(None), u'unknown_file') |
---|
58 | return |
---|
59 | |
---|
60 | class ExtFileStoreTests(unittest.TestCase): |
---|
61 | # Test external file store (non-functional mode) |
---|
62 | |
---|
63 | def setUp(self): |
---|
64 | self.workdir = tempfile.mkdtemp() |
---|
65 | self.root = None |
---|
66 | return |
---|
67 | |
---|
68 | def tearDown(self): |
---|
69 | shutil.rmtree(self.workdir) |
---|
70 | if self.root is not None: |
---|
71 | shutil.rmtree(self.root) |
---|
72 | return |
---|
73 | |
---|
74 | def test_iface(self): |
---|
75 | obj = ExtFileStore(None) |
---|
76 | verifyClass(IExtFileStore, ExtFileStore) |
---|
77 | verifyObject(IExtFileStore, obj) |
---|
78 | return |
---|
79 | |
---|
80 | def test_root_setup_wo_site(self): |
---|
81 | # if no site is available we can use a temporary root |
---|
82 | fs = ExtFileStore() |
---|
83 | self.root = fs.root |
---|
84 | self.assertTrue(isinstance(self.root, basestring)) |
---|
85 | self.assertTrue(os.path.exists(self.root)) |
---|
86 | return |
---|
87 | |
---|
88 | def test_create_instance(self): |
---|
89 | storage1 = ExtFileStore() |
---|
90 | storage2 = ExtFileStore(root=self.workdir) |
---|
91 | self.root = storage1.root |
---|
92 | self.assertTrue(storage1.root is not None) |
---|
93 | self.assertTrue(storage1.root != storage2.root) |
---|
94 | self.assertEqual(storage2.root, self.workdir) |
---|
95 | return |
---|
96 | |
---|
97 | def test_create_file(self): |
---|
98 | # We can store files |
---|
99 | storage = ExtFileStore(root=self.workdir) |
---|
100 | dummy_file = StringIO('sample file') |
---|
101 | image_file = storage.createFile('mysample', dummy_file) |
---|
102 | self.assertTrue('mysample' in os.listdir(storage.root)) |
---|
103 | self.assertEqual('mysample', image_file.data) |
---|
104 | return |
---|
105 | |
---|
106 | def test_create_file_w_ext(self): |
---|
107 | # We can store files with filename extension |
---|
108 | storage = ExtFileStore(root=self.workdir) |
---|
109 | dummy_file = StringIO('sample file') |
---|
110 | image_file = storage.createFile('mysample.txt', dummy_file) |
---|
111 | self.assertTrue('mysample.txt' in os.listdir(storage.root)) |
---|
112 | self.assertEqual('mysample.txt', image_file.data) |
---|
113 | return |
---|
114 | |
---|
115 | def test_get_file(self): |
---|
116 | # We can get files after having them stored |
---|
117 | storage = ExtFileStore(root=self.workdir) |
---|
118 | dummy_file = StringIO('sample file') |
---|
119 | image_file = storage.createFile('mysample', dummy_file) |
---|
120 | result = storage.getFile(image_file.data) |
---|
121 | self.assertEqual(result.read(), 'sample file') |
---|
122 | return |
---|
123 | |
---|
124 | def test_get_file_w_ext(self): |
---|
125 | # We can get files with filename extension after having them |
---|
126 | # stored |
---|
127 | storage = ExtFileStore(root=self.workdir) |
---|
128 | dummy_file = StringIO('sample file') |
---|
129 | image_file = storage.createFile('mysample.txt', dummy_file) |
---|
130 | result = storage.getFile('mysample') |
---|
131 | self.assertEqual(result.read(), 'sample file') |
---|
132 | return |
---|
133 | |
---|
134 | def test_replace_file_w_new_ext(self): |
---|
135 | # when we store a file with the same file_id but different |
---|
136 | # filename extension, the old file will be deleted |
---|
137 | storage = ExtFileStore(root=self.workdir) |
---|
138 | dummy_file = StringIO('sample_file') |
---|
139 | image_file = storage.createFile('mysample.jpg', dummy_file) |
---|
140 | file1_path = storage.getFile('mysample').name |
---|
141 | new_file = storage.createFile( |
---|
142 | 'mysample.png', StringIO('new file')) |
---|
143 | file2 = storage.getFile('mysample') |
---|
144 | self.assertEqual(file2.name[-12:], 'mysample.png') |
---|
145 | self.assertEqual(file2.read(), 'new file') |
---|
146 | # the old file was deleted |
---|
147 | self.assertFalse(os.path.exists(file1_path)) |
---|
148 | return |
---|
149 | |
---|
150 | def test_extract_marker(self): |
---|
151 | # file stores support extracting markers from filenames |
---|
152 | storage = ExtFileStore(root=self.workdir) |
---|
153 | result1 = storage.extractMarker(None) |
---|
154 | result2 = storage.extractMarker('') |
---|
155 | result3 = storage.extractMarker('no-marker') |
---|
156 | result4 = storage.extractMarker('no-marker.txt') |
---|
157 | result5 = storage.extractMarker('__MARKER__foo.jpg') |
---|
158 | result6 = storage.extractMarker('__MaRkEr__foo.jpg') |
---|
159 | result7 = storage.extractMarker('__THE_MARKER__foo.jpg') |
---|
160 | result8 = storage.extractMarker('__A_MARK__my__foo.jpg') |
---|
161 | |
---|
162 | self.assertEqual(result1, ('', '', '', '')) |
---|
163 | self.assertEqual(result2, ('', '', '', '')) |
---|
164 | self.assertEqual(result3, ('', 'no-marker', 'no-marker', '')) |
---|
165 | self.assertEqual(result4, ('', 'no-marker.txt', 'no-marker', '.txt')) |
---|
166 | self.assertEqual(result5, ('marker', 'foo.jpg', 'foo', '.jpg')) |
---|
167 | self.assertEqual(result6, ('marker', 'foo.jpg', 'foo', '.jpg')) |
---|
168 | self.assertEqual(result7, ('the_marker', 'foo.jpg', 'foo', '.jpg')) |
---|
169 | self.assertEqual(result8, ('a_mark', 'my__foo.jpg', 'my__foo', '.jpg')) |
---|
170 | return |
---|
171 | |
---|
172 | class DefaultFileStoreHandlerTests(unittest.TestCase): |
---|
173 | |
---|
174 | def test_iface(self): |
---|
175 | obj = DefaultFileStoreHandler() |
---|
176 | verifyClass(IFileStoreHandler, DefaultFileStoreHandler) |
---|
177 | verifyObject(IFileStoreHandler, obj) |
---|
178 | return |
---|
179 | |
---|
180 | class CustomizedFileHandler(object): |
---|
181 | def pathFromFileID(self, store, root, file_id): |
---|
182 | return os.path.join(root, file_id[12:]) |
---|
183 | |
---|
184 | def createFile(self, store, root, filename, file_id, f): |
---|
185 | ext = os.path.splitext(filename)[1] |
---|
186 | path = self.pathFromFileID(store, root, file_id) + ext |
---|
187 | return f, path, HurryFile(filename, file_id + ext) |
---|
188 | |
---|
189 | class CustomContext(object): |
---|
190 | pass |
---|
191 | |
---|
192 | class CustomContextFileChooser(object): |
---|
193 | def __init__(self, context): |
---|
194 | self.context = context |
---|
195 | |
---|
196 | def chooseName(self, name=None, attr=None): |
---|
197 | # this name chooser returns different file ids depending on |
---|
198 | # the `attr` parameter, a simple string. |
---|
199 | if attr=='img': |
---|
200 | return '__mymarker__mysample_img.jpg' |
---|
201 | elif attr=='doc': |
---|
202 | return '__mymarker__mysample_doc.doc' |
---|
203 | return '__mymarker__mysample.txt' |
---|
204 | |
---|
205 | class FunctionalExtFileStoreTests(FunctionalTestCase): |
---|
206 | |
---|
207 | layer = FunctionalLayer |
---|
208 | |
---|
209 | def setUp(self): |
---|
210 | super(FunctionalExtFileStoreTests, self).setUp() |
---|
211 | self.workdir = tempfile.mkdtemp() |
---|
212 | self.samplefile = os.path.join(self.workdir, 'sample') |
---|
213 | self.otherfile = os.path.join(self.workdir, 'other') |
---|
214 | open(self.samplefile, 'wb').write('Hi there!') |
---|
215 | open(self.otherfile, 'wb').write('Hi from other!') |
---|
216 | self.fd = open(self.samplefile, 'r') |
---|
217 | self.fd2 = open(self.otherfile, 'r') |
---|
218 | self.getRootFolder()['app'] = University() |
---|
219 | self.app = self.getRootFolder()['app'] |
---|
220 | self.app['datacenter'].setStoragePath(self.workdir) |
---|
221 | # register a custom filename mangler |
---|
222 | provideUtility( |
---|
223 | CustomizedFileHandler(), IFileStoreHandler, name=u'mymarker') |
---|
224 | # register a file chooser adapter for CustomContext |
---|
225 | provideAdapter( |
---|
226 | CustomContextFileChooser, |
---|
227 | (CustomContext,), IFileStoreNameChooser) |
---|
228 | return |
---|
229 | |
---|
230 | |
---|
231 | def tearDown(self): |
---|
232 | super(FunctionalExtFileStoreTests, self).tearDown() |
---|
233 | self.fd.close() |
---|
234 | self.fd2.close() |
---|
235 | shutil.rmtree(self.workdir) |
---|
236 | return |
---|
237 | |
---|
238 | def test_root_setup_w_site(self): |
---|
239 | # if a site is available we use it to determine the root dir |
---|
240 | fs = ExtFileStore() |
---|
241 | setSite(self.app) |
---|
242 | self.root = fs.root |
---|
243 | expected_root = os.path.join( |
---|
244 | self.app['datacenter'].storage, 'media') |
---|
245 | self.assertTrue(isinstance(self.root, basestring)) |
---|
246 | self.assertEqual(self.root, expected_root) |
---|
247 | return |
---|
248 | |
---|
249 | def test_get_utility(self): |
---|
250 | # we can get an ExtFileStore by global utility lookup |
---|
251 | fs1 = getUtility(IExtFileStore) |
---|
252 | fs2 = getUtility(IExtFileStore) |
---|
253 | self.assertTrue(isinstance(fs1, ExtFileStore)) |
---|
254 | self.assertTrue(fs1 is fs2) |
---|
255 | return |
---|
256 | |
---|
257 | def test_default_handler_create_file(self): |
---|
258 | # we can use the default handler to store files |
---|
259 | fs = ExtFileStore() |
---|
260 | result = fs.createFile('sample.txt', StringIO('sample text')) |
---|
261 | self.assertEqual(result.data, 'sample.txt') |
---|
262 | self.assertTrue('sample.txt' in os.listdir(fs.root)) |
---|
263 | return |
---|
264 | |
---|
265 | def test_default_handler_get_file(self): |
---|
266 | # we can get files stored by the default handler |
---|
267 | fs = ExtFileStore() |
---|
268 | fs.createFile('sample.txt', StringIO('sample text')) |
---|
269 | result1 = fs.getFile('sample.txt') |
---|
270 | result2 = fs.getFile('not-existent') |
---|
271 | self.assertEqual(result1.read(), 'sample text') |
---|
272 | self.assertTrue(result2 is None) |
---|
273 | return |
---|
274 | |
---|
275 | def test_customized_handler_create_file(self): |
---|
276 | # we can use registered filename handlers |
---|
277 | fs = ExtFileStore() |
---|
278 | result = fs.createFile( |
---|
279 | '__MYMARKER__sample.txt', StringIO('sample text')) |
---|
280 | self.assertEqual(result.data, '__MYMARKER__sample.txt') |
---|
281 | self.assertTrue('sample.txt' in os.listdir(fs.root)) |
---|
282 | return |
---|
283 | |
---|
284 | def test_customized_handler_create_file_w_ext(self): |
---|
285 | # when we create a file of img type, the filename ext is taken |
---|
286 | # from input file. |
---|
287 | fs = ExtFileStore() |
---|
288 | result = fs.createFile( |
---|
289 | '__MYMARKER__sample_img.png', StringIO('sample text')) |
---|
290 | self.assertEqual(result.data, '__MYMARKER__sample_img.png') |
---|
291 | self.assertTrue('sample_img.png' in os.listdir(fs.root)) |
---|
292 | return |
---|
293 | |
---|
294 | def test_customized_handler_get_file(self): |
---|
295 | # we consider registered filename handlers when asking for |
---|
296 | # stored files. |
---|
297 | fs = ExtFileStore() |
---|
298 | fs.createFile('__MYMARKER__sample.txt', StringIO('sample text')) |
---|
299 | result1 = fs.getFile('__MYMARKER__sample.txt') |
---|
300 | result2 = fs.getFile('__MYMARKER__not-existent') |
---|
301 | result3 = fs.getFile('not-existent') |
---|
302 | self.assertEqual(result1.read(), 'sample text') |
---|
303 | self.assertTrue(result2 is None) |
---|
304 | self.assertTrue(result3 is None) |
---|
305 | return |
---|
306 | |
---|
307 | def test_get_file_by_context_w_attr(self): |
---|
308 | # if we register a file name chooser, we can also get a file |
---|
309 | # by context and attribute |
---|
310 | fs = ExtFileStore() |
---|
311 | context = CustomContext() |
---|
312 | file_id1 = IFileStoreNameChooser(context).chooseName() |
---|
313 | file_id2 = IFileStoreNameChooser(context).chooseName(attr='img') |
---|
314 | file_id3 = IFileStoreNameChooser(context).chooseName(attr='doc') |
---|
315 | fs = ExtFileStore() |
---|
316 | # create three files for a single context, each which |
---|
317 | # different content |
---|
318 | fs.createFile(file_id1, StringIO('my sample 1')) |
---|
319 | fs.createFile(file_id2, StringIO('my sample 2')) |
---|
320 | fs.createFile(file_id3, StringIO('my sample 3')) |
---|
321 | # now get back all files indicated by different `attr` markers |
---|
322 | result1 = fs.getFileByContext(context) |
---|
323 | result2 = fs.getFileByContext(context, attr='img') |
---|
324 | result3 = fs.getFileByContext(context, attr='doc') |
---|
325 | # each file has a different file id |
---|
326 | self.assertEqual(file_id1, '__mymarker__mysample.txt') |
---|
327 | self.assertEqual(file_id2, '__mymarker__mysample_img.jpg') |
---|
328 | self.assertEqual(file_id3, '__mymarker__mysample_doc.doc') |
---|
329 | # each file has different content |
---|
330 | self.assertEqual(result1.read(), 'my sample 1') |
---|
331 | self.assertEqual(result2.read(), 'my sample 2') |
---|
332 | self.assertEqual(result3.read(), 'my sample 3') |
---|
333 | return |
---|
334 | |
---|
335 | def test_get_default_handler(self): |
---|
336 | # we can get a default handler |
---|
337 | result = queryUtility(IFileStoreHandler) |
---|
338 | self.assertTrue( |
---|
339 | isinstance(result, DefaultFileStoreHandler)) |
---|
340 | return |
---|
341 | |
---|
342 | def test_get_default_file_retrieval(self): |
---|
343 | # we get a file store when requesting a file retrieval |
---|
344 | result = queryUtility(IFileRetrieval) |
---|
345 | self.assertTrue( |
---|
346 | isinstance(result, DefaultStorage)) |
---|
347 | |
---|
348 | def test_delete_file(self): |
---|
349 | # we can remove stored files from storage |
---|
350 | fs = ExtFileStore() |
---|
351 | # First, we store a file in file store |
---|
352 | fs.createFile('sample.txt', StringIO('sample text')) |
---|
353 | # Then we delete it |
---|
354 | fs.deleteFile('sample.txt') |
---|
355 | # 'Deletion' means, next call to getFile should get None |
---|
356 | result = fs.getFile('sample.txt') |
---|
357 | self.assertTrue(result is None) |
---|
358 | # Hm, okay we can also check, whether it was really deleted |
---|
359 | self.assertTrue('sample.txt' not in os.listdir(fs.root)) |
---|
360 | return |
---|
361 | |
---|
362 | def test_delete_file_by_context_w_attr(self): |
---|
363 | # if we register a file name chooser, we can also delete a file |
---|
364 | # by context and attribute |
---|
365 | fs = ExtFileStore() |
---|
366 | context = CustomContext() |
---|
367 | file_id1 = IFileStoreNameChooser(context).chooseName() |
---|
368 | file_id2 = IFileStoreNameChooser(context).chooseName(attr='img') |
---|
369 | file_id3 = IFileStoreNameChooser(context).chooseName(attr='doc') |
---|
370 | fs = ExtFileStore() |
---|
371 | # create three files for a single context, each which |
---|
372 | # different content |
---|
373 | fs.createFile(file_id1, StringIO('my sample 1')) |
---|
374 | fs.createFile(file_id2, StringIO('my sample 2')) |
---|
375 | fs.createFile(file_id3, StringIO('my sample 3')) |
---|
376 | # now delete first two of these files |
---|
377 | fs.deleteFileByContext(context) |
---|
378 | fs.deleteFileByContext(context, attr='img') |
---|
379 | # Following getFile calls should give None for two of the |
---|
380 | # files. |
---|
381 | result1 = fs.getFileByContext(context) |
---|
382 | result2 = fs.getFileByContext(context, attr='img') |
---|
383 | result3 = fs.getFileByContext(context, attr='doc') |
---|
384 | self.assertEqual(result1, None) |
---|
385 | self.assertEqual(result2, None) |
---|
386 | self.assertEqual(result3.read(), 'my sample 3') |
---|
387 | return |
---|