|
1 """unit tests for module cubicweb.server.sources.storages |
|
2 |
|
3 :organization: Logilab |
|
4 :copyright: 2010 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2. |
|
5 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr |
|
6 :license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses |
|
7 """ |
|
8 |
|
9 from logilab.common.testlib import unittest_main |
|
10 from cubicweb.devtools.testlib import CubicWebTC |
|
11 |
|
12 import os.path as osp |
|
13 import shutil |
|
14 import tempfile |
|
15 |
|
16 from cubicweb import Binary |
|
17 from cubicweb.selectors import implements |
|
18 from cubicweb.server.sources import storages |
|
19 from cubicweb.server.hook import Hook, Operation |
|
20 |
|
21 class DummyBeforeHook(Hook): |
|
22 __regid__ = 'dummy-before-hook' |
|
23 __select__ = Hook.__select__ & implements('File') |
|
24 events = ('before_add_entity',) |
|
25 |
|
26 def __call__(self): |
|
27 self._cw.transaction_data['orig_file_value'] = self.entity.data.getvalue() |
|
28 |
|
29 |
|
30 class DummyAfterHook(Hook): |
|
31 __regid__ = 'dummy-after-hook' |
|
32 __select__ = Hook.__select__ & implements('File') |
|
33 events = ('after_add_entity',) |
|
34 |
|
35 def __call__(self): |
|
36 # new value of entity.data should be the same as before |
|
37 oldvalue = self._cw.transaction_data['orig_file_value'] |
|
38 assert oldvalue == self.entity.data.getvalue() |
|
39 |
|
40 |
|
41 class StorageTC(CubicWebTC): |
|
42 |
|
43 def setup_database(self): |
|
44 self.tempdir = tempfile.mkdtemp() |
|
45 bfs_storage = storages.BytesFileSystemStorage(self.tempdir) |
|
46 storages.set_attribute_storage(self.repo, 'File', 'data', bfs_storage) |
|
47 |
|
48 def tearDown(self): |
|
49 super(CubicWebTC, self).tearDown() |
|
50 storages.unset_attribute_storage(self.repo, 'File', 'data') |
|
51 shutil.rmtree(self.tempdir) |
|
52 |
|
53 |
|
54 def create_file(self, content): |
|
55 req = self.request() |
|
56 return req.create_entity('File', data=Binary(content), |
|
57 data_format=u'text/plain', data_name=u'foo') |
|
58 |
|
59 def test_bfs_storage(self): |
|
60 f1 = self.create_file(content='the-data') |
|
61 expected_filepath = osp.join(self.tempdir, '%s_data' % f1.eid) |
|
62 self.failUnless(osp.isfile(expected_filepath)) |
|
63 self.assertEquals(file(expected_filepath).read(), 'the-data') |
|
64 |
|
65 def test_sqlite_fspath(self): |
|
66 f1 = self.create_file(content='the-data') |
|
67 expected_filepath = osp.join(self.tempdir, '%s_data' % f1.eid) |
|
68 fspath = self.execute('Any fspath(F, "File", "data") WHERE F eid %(f)s', |
|
69 {'f': f1.eid})[0][0] |
|
70 self.assertEquals(fspath.getvalue(), expected_filepath) |
|
71 |
|
72 def test_fs_importing_doesnt_touch_path(self): |
|
73 self.session.transaction_data['fs_importing'] = True |
|
74 f1 = self.session.create_entity('File', data=Binary('/the/path'), |
|
75 data_format=u'text/plain', data_name=u'foo') |
|
76 fspath = self.execute('Any fspath(F, "File", "data") WHERE F eid %(f)s', |
|
77 {'f': f1.eid})[0][0] |
|
78 self.assertEquals(fspath.getvalue(), '/the/path') |
|
79 |
|
80 def test_storage_transparency(self): |
|
81 self.vreg._loadedmods[__name__] = {} |
|
82 self.vreg.register(DummyBeforeHook) |
|
83 self.vreg.register(DummyAfterHook) |
|
84 try: |
|
85 self.create_file(content='the-data') |
|
86 finally: |
|
87 self.vreg.unregister(DummyBeforeHook) |
|
88 self.vreg.unregister(DummyAfterHook) |
|
89 |
|
90 if __name__ == '__main__': |
|
91 unittest_main() |