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()