refactorization draft
authorAlain Leufroy <alain.leufroy@logilab.fr>
Tue, 02 Apr 2013 12:21:13 +0200
changeset 14 f4791bdd245a
parent 0 03d0140deb6c
child 15 981b5667158c
refactorization - separate fs specific code to hdf5 specific codes - user better names - use generators instead of list - move functions to class method as they should be overwriten for future implementations
h5fs.py
test/test_H5FS.py
--- a/h5fs.py	Tue Apr 03 20:34:00 2012 +0200
+++ b/h5fs.py	Tue Apr 02 12:21:13 2013 +0200
@@ -7,13 +7,15 @@
 #
 
 '''
-HDF% filesystem implementation.
+HDF5 filesystem implementation.
 '''
 
 import os
 import os.path as osp
 import stat
 import errno
+from itertools import imap, chain
+from functools import partial
 
 import h5py
 
@@ -59,7 +61,7 @@
 
     h5file = None
 
-    def _get_entry(self, path):
+    def get_entry(self, path):
         '''Return the entry recorded at ``path`` or raise H5IOError'''
         try:
             h5entry = self.h5file[path]
@@ -68,34 +70,34 @@
         return h5entry
 
     @staticmethod
-    def _readdir(h5entry, offset):
-        '''A fonction generator that yields entries from the h5 entry'''
-        for name in ('.', '..'):
-            yield fuse.Direntry(name)
-        for content in h5entry.itervalues():
-            yield fuse.Direntry(get_name(content), offset=offset)
+    def get_stat(h5entry):
+        '''Return the entry stat information (an instance of Stat)'''
+        if isinstance(h5entry, h5py.highlevel.Group):
+            return GroupStat()
+        if isinstance(h5entry, h5py.highlevel.Dataset):
+            return DatasetStat()
+        raise H5IOError(errno.ENOENT, 'Unknown entry type', h5entry.name)
+
+    @staticmethod
+    def get_name(h5entry):
+        '''Return the entry file name as string'''
+        # XXX unicode seems not to be allowed
+        return str(osp.basename(h5entry.name))
+
+    def list_group(self, group, specs=('.', '..')):
+        '''A fonction generator that yields name of the entries in the group.
+        The list is in arbritary order with specials entries ``specs``.'''
+        return chain(iter(specs), imap(self.get_name, group.itervalues()))
 
     def readdir(self, path, offset):
         '''Return a generator that yields entries from the given Group path'''
-        return self._readdir(self._get_entry(path), offset)
+        convert = partial(fuse.Direntry, offset=offset)
+        return imap(convert, self.list_group(self.get_entry(path)))
 
     def getattr(self, path):
         '''Return the stat attributes of the given ``path``'''
-        return get_stat(self._get_entry(path))
-
+        return self.get_stat(self.get_entry(path))
 
-def get_stat(h5entry):
-    '''Return the entry stat information (an instance of Stat)'''
-    if isinstance(h5entry, h5py.highlevel.Group):
-        return GroupStat()
-    if isinstance(h5entry, h5py.highlevel.Dataset):
-        return DatasetStat()
-    raise H5IOError(errno.ENOENT, 'Unknown entry type', h5entry.name)
-
-def get_name(h5entry):
-    '''Return the entry file name as string'''
-    # XXX unicode seems not to be allowed
-    return str(osp.basename(h5entry.name))
 
 def __main__():
     '''Main entry point that start the HDF5 filesystem server'''
--- a/test/test_H5FS.py	Tue Apr 03 20:34:00 2012 +0200
+++ b/test/test_H5FS.py	Tue Apr 02 12:21:13 2013 +0200
@@ -7,60 +7,69 @@
 from logilab.common.testlib import within_tempdir
 
 import h5fs
+FSM = h5fs.FsMixin()
 
 __folder__=osp.dirname(osp.abspath(__file__))
 
 def getdata(path):
     return osp.join(__folder__, 'data', path)
 
-class GetStat_TC(TC):
-
+def with_h5file(func):
     @within_tempdir
-    def test_group(self):
+    def wrapped(self):
         h5file = h5py.File('file.hdf5')
-        group = h5file.create_group(u'sam')
-        self.assertTrue(isinstance(h5fs.get_stat(group), h5fs.GroupStat))
+        out = func(self, h5file)
         h5file.close()
+        return out
+    return wrapped
 
-    @within_tempdir
-    def test_dataset(self):
-        h5file = h5py.File('file.hdf5')
-        group = h5file.create_dataset(u'sam', [])
-        self.assertTrue(isinstance(h5fs.get_stat(group), h5fs.DatasetStat))
-        h5file.close()
+def with_fsmixin(func):
+    @with_h5file
+    def wrapped(self, h5file):
+        h5file.create_group(u'sam')
+        h5file.create_dataset(u'melu', data=xrange(12), shape=(4, 3), dtype='u8')
+        fsmx = h5fs.FsMixin()
+        fsmx.h5file = h5file
+        return func(self, fsmx)
+    return wrapped
+
 
 class FsMixin_TC(TC):
-    def init(self):
-        h5file = h5py.File('file.hdf5')
-        h5file.create_group(u'sam')
-        h5file.create_dataset(u'melu', (0,))
-        h5mx = h5fs.FsMixin()
-        h5mx.h5file = h5file
-        return h5file, h5mx
+    @with_fsmixin
+    def test_get_stat_with_group(self, fsmx):
+        group = fsmx.h5file['sam']
+        self.assertTrue(isinstance(fsmx.get_stat(group), h5fs.GroupStat))
+
+    @with_fsmixin
+    def test_get_stat_with_dataset(self, fsmx):
+        dataset = fsmx.h5file['melu']
+        self.assertTrue(isinstance(fsmx.get_stat(dataset), h5fs.DatasetStat))
 
-    @within_tempdir
-    def test_readdir(self):
-        h5file, h5mx = self.init()
-        result = list(h5mx.readdir(u'/', 0))
-        expected = list(h5file[u'/'].keys()) + ['.', '..']
+    @with_fsmixin
+    def test_get_name(self, fsmx):
+        dataset = fsmx.h5file['melu']
+        self.assertEqual(FSM.get_name(dataset), 'melu')
+
+    @with_fsmixin
+    def test_readdir(self, fsmx):
+        result = list(fsmx.readdir(u'/', 0))
+        expected = list(fsmx.h5file[u'/'].keys()) + ['.', '..']
         self.assertEqual(len(expected), len(result))
 
-    @within_tempdir
-    def test_readdir_wrong(self):
-        h5file, h5mx = self.init()
-        self.assertRaises(h5fs.H5IOError, h5mx.readdir, u'/wrong data path', 0)
+    @with_fsmixin
+    def test_readdir_wrong(self, fsmx):
+        self.assertRaises(h5fs.H5IOError, fsmx.readdir, u'/wrong data path', 0)
 
-    @within_tempdir
-    def test_getattr(self):
-        h5file, h5mx = self.init()
-        self.assertTrue(isinstance(h5mx.getattr('/'), h5fs.GroupStat))
-        self.assertTrue(isinstance(h5mx.getattr('/sam'), h5fs.GroupStat))
-        self.assertTrue(isinstance(h5mx.getattr('/melu'), h5fs.DatasetStat))
+    @with_fsmixin
+    def test_getattr(self, fsmx):
+        self.assertTrue(isinstance(fsmx.getattr('/'), h5fs.GroupStat))
+        self.assertTrue(isinstance(fsmx.getattr('/sam'), h5fs.GroupStat))
+        self.assertTrue(isinstance(fsmx.getattr('/melu'), h5fs.DatasetStat))
 
-    @within_tempdir
-    def test_getattr_wrong(self):
-        h5file, h5mx = self.init()
-        self.assertRaises(h5fs.H5IOError, h5mx.getattr, u'/wrong data path')
+    @with_fsmixin
+    def test_getattr_wrong(self, fsmx):
+        self.assertRaises(h5fs.H5IOError, fsmx.getattr, u'/wrong data path')
+
 
 if __name__ == '__main__':
     main()