15 # |
15 # |
16 # You should have received a copy of the GNU Lesser General Public License along |
16 # You should have received a copy of the GNU Lesser General Public License along |
17 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
17 # with CubicWeb. If not, see <http://www.gnu.org/licenses/>. |
18 """unit tests for module cubicweb.server.sources.storages""" |
18 """unit tests for module cubicweb.server.sources.storages""" |
19 |
19 |
|
20 from six import PY2 |
|
21 |
20 from logilab.common.testlib import unittest_main, tag, Tags |
22 from logilab.common.testlib import unittest_main, tag, Tags |
21 from cubicweb.devtools.testlib import CubicWebTC |
23 from cubicweb.devtools.testlib import CubicWebTC |
22 |
24 |
23 from glob import glob |
25 from glob import glob |
24 import os |
26 import os |
25 import os.path as osp |
27 import os.path as osp |
|
28 import sys |
26 import shutil |
29 import shutil |
27 import tempfile |
30 import tempfile |
28 |
31 |
29 from cubicweb import Binary, QueryError |
32 from cubicweb import Binary, QueryError |
30 from cubicweb.predicates import is_instance |
33 from cubicweb.predicates import is_instance |
55 tags = CubicWebTC.tags | Tags('Storage', 'BFSS') |
58 tags = CubicWebTC.tags | Tags('Storage', 'BFSS') |
56 |
59 |
57 def setup_database(self): |
60 def setup_database(self): |
58 self.tempdir = tempfile.mkdtemp() |
61 self.tempdir = tempfile.mkdtemp() |
59 bfs_storage = storages.BytesFileSystemStorage(self.tempdir) |
62 bfs_storage = storages.BytesFileSystemStorage(self.tempdir) |
|
63 self.bfs_storage = bfs_storage |
60 storages.set_attribute_storage(self.repo, 'File', 'data', bfs_storage) |
64 storages.set_attribute_storage(self.repo, 'File', 'data', bfs_storage) |
61 storages.set_attribute_storage(self.repo, 'BFSSTestable', 'opt_attr', bfs_storage) |
65 storages.set_attribute_storage(self.repo, 'BFSSTestable', 'opt_attr', bfs_storage) |
62 |
66 |
63 def tearDown(self): |
67 def tearDown(self): |
64 super(StorageTC, self).tearDown() |
68 super(StorageTC, self).tearDown() |
65 storages.unset_attribute_storage(self.repo, 'File', 'data') |
69 storages.unset_attribute_storage(self.repo, 'File', 'data') |
|
70 del self.bfs_storage |
66 shutil.rmtree(self.tempdir) |
71 shutil.rmtree(self.tempdir) |
67 |
72 |
68 |
73 |
69 def create_file(self, cnx, content=b'the-data'): |
74 def create_file(self, cnx, content=b'the-data'): |
70 return cnx.create_entity('File', data=Binary(content), |
75 return cnx.create_entity('File', data=Binary(content), |
71 data_format=u'text/plain', |
76 data_format=u'text/plain', |
72 data_name=u'foo.pdf') |
77 data_name=u'foo.pdf') |
73 |
78 |
74 def fspath(self, cnx, entity): |
79 def fspath(self, cnx, entity): |
75 fspath = cnx.execute('Any fspath(D) WHERE F eid %(f)s, F data D', |
80 fspath = cnx.execute('Any fspath(D) WHERE F eid %(f)s, F data D', |
76 {'f': entity.eid})[0][0] |
81 {'f': entity.eid})[0][0].getvalue() |
77 return fspath.getvalue() |
82 return fspath if PY2 else fspath.decode('utf-8') |
78 |
83 |
79 def test_bfss_wrong_fspath_usage(self): |
84 def test_bfss_wrong_fspath_usage(self): |
80 with self.admin_access.repo_cnx() as cnx: |
85 with self.admin_access.repo_cnx() as cnx: |
81 f1 = self.create_file(cnx) |
86 f1 = self.create_file(cnx) |
82 cnx.execute('Any fspath(D) WHERE F eid %(f)s, F data D', {'f': f1.eid}) |
87 cnx.execute('Any fspath(D) WHERE F eid %(f)s, F data D', {'f': f1.eid}) |