Implemented dictionnary based file system to speed up the tests

sdp/sheerkaDataProviderIO.py

@@ -0,0 +1,189 @@
+import io
+from os import path
+import os
+from fs.memoryfs import MemoryFS
+import logging
+
+
+class SheerkaDataProviderIO:
+
+    def __init__(self, root):
+        self.root = root
+
+    def exists(self, file_path):
+        pass
+
+    def open(self, file_path, mode):
+        pass
+
+    def read_text(self, file_path):
+        pass
+
+    def read_binary(self, file_path):
+        pass
+
+    def write_text(self, file_path, content):
+        pass
+
+    def write_binary(self, file_path, content):
+        pass
+
+    def remove(self, file_path):
+        pass
+
+    @staticmethod
+    def get(root):
+        if root == "mem://":
+            return SheerkaDataProviderDictionaryIO()
+        else:
+            return SheerkaDataProviderFileIO(root)
+
+    def get_obj_path(self, object_type, digest):
+        return path.join(self.root, object_type, digest[:24], digest)
+
+    def path_join(self, *paths):
+        return path.join(self.root, *paths)
+
+
+class SheerkaDataProviderFileIO(SheerkaDataProviderIO):
+    log = logging.getLogger("FileIO")
+
+    def __init__(self, root):
+        root = path.abspath(path.join(path.expanduser("~"), ".sheerka")) \
+            if root is None \
+            else path.abspath(root)
+        super().__init__(root)
+
+        self.log.debug("root is set to '" + self.root + "'")
+
+        if not path.exists(self.root):
+            self.log.debug("root folder not found. Creating it.")
+            os.makedirs(self.root)
+            self.first_time = True
+        else:
+            self.first_time = False
+
+    def open(self, file_path, mode):
+        return open(file_path, mode)
+
+    def read_text(self, file_path):
+        with open(file_path) as f:
+            return f.read()
+
+    def read_binary(self, file_path):
+        with open(file_path, "rb") as f:
+            return f.read()
+
+    def write_text(self, file_path, content):
+        self._write(file_path, content, "w")
+
+    def write_binary(self, file_path, content):
+        self._write(file_path, content, "wb")
+
+    def exists(self, file_path):
+        return path.exists(file_path)
+
+    def remove(self, file_path):
+        os.remove(file_path)
+
+    @staticmethod
+    def _write(file_path, content, mode):
+        if not path.exists(path.dirname(file_path)):
+            os.makedirs(path.dirname(file_path))
+
+        with open(file_path, mode) as f:
+            f.write(content)
+
+
+class SheerkaDataProviderMemoryIO(SheerkaDataProviderIO):
+    log = logging.getLogger("MemoryIO")
+
+    def __init__(self):
+        super().__init__("")
+        self.mem_fs = MemoryFS()
+        self.log.debug("Initializing memory file.")
+        self.first_time = True
+
+    def open(self, file_path, mode):
+        return self.mem_fs.open(file_path, mode)
+
+    def exists(self, file_path):
+        return self.mem_fs.exists(file_path)
+
+    def read_text(self, file_path):
+        return self.mem_fs.readtext(file_path)
+
+    def read_binary(self, file_path):
+        return self.mem_fs.readbytes(file_path)
+
+    def write_binary(self, file_path, content):
+        self._ensure_parent_folder(file_path)
+        self.mem_fs.writebytes(file_path, content)
+
+    def write_text(self, file_path, content):
+        self._ensure_parent_folder(file_path)
+        self.mem_fs.writetext(file_path, content)
+
+    def remove(self, file_path):
+        self.mem_fs.remove(file_path)
+
+    def _ensure_parent_folder(self, file_path):
+        if not self.mem_fs.exists(path.dirname(file_path)):
+            self.mem_fs.makedirs(path.dirname(file_path))
+
+
+class SheerkaDataProviderDictionaryIO(SheerkaDataProviderIO):
+    log = logging.getLogger("DictionaryIO")
+
+    def __init__(self):
+        super().__init__("")
+        self.cache = {}
+        self.log.debug("Initializing dictionary file.")
+        self.first_time = True
+
+    def exists(self, file_path):
+        if file_path == "":
+            return True
+
+        return file_path in self.cache
+
+    def read_text(self, file_path):
+        return self.cache[file_path]
+
+    def read_binary(self, file_path):
+        return self.cache[file_path]
+
+    def write_binary(self, file_path, content):
+        self.cache[file_path] = content
+
+    def write_text(self, file_path, content):
+        self.cache[file_path] = content
+
+    def remove(self, file_path):
+        del (self.cache[file_path])
+
+    def open(self, file_path, mode):
+        if "w" in mode:
+            stream = io.BytesIO() if "b" in mode else io.StringIO()
+            stream.close = on_close(self, file_path, stream)(stream.close)
+            return stream
+
+        return io.BytesIO(self.cache[file_path]) if "b" in mode else io.StringIO(self.cache[file_path])
+
+
+def on_close(dictionary_io, file_path, stream):
+    """
+    Decorator to intercept the close.
+    I guess that there are solution that are more elegant
+    :param dictionary_io:
+    :param file_path:
+    :param stream:
+    :return:
+    """
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            stream.seek(0)
+            dictionary_io.cache[file_path] = stream.read()
+            func(*args, **kwargs)
+        return wrapper
+    return decorator