task1/src/memoryfs.py - master - georgiy/work

task1/src/memoryfs.py
262 строки9 KB
Georgiy Yankovskiy
Task1
17 ноя 2024, 01:47
17 ноя 2024, 01:47bdefd3a
import copy
import threading

import fuse
import os
import stat
import errno
import time


class MemoryFS(fuse.Operations):
    def __init__(self):
        self.data = {}
        self.permissions = {}
        self.ext = {}
        self.lock11 = threading.Lock()

        # self.create_entry("/", 777, is_dir=True)
        self.data["/"] = {}
        self.permissions["/"] = 493
        self.ext["/"] = {
            'st_atime': time.time(),
            'st_ctime': time.time(),
            'st_mtime': time.time(),
            'st_gid': os.getgid(),
            'st_uid': os.getuid(),
        }

    def utimens(self, path, times=None):
        if times is None:
            self.ext[path]['st_atime'] = time.time()
            self.ext[path]['st_mtime'] = time.time()
        else:
            self.ext[path]['st_atime'] = times[0]
            self.ext[path]['st_mtime'] = times[1]
        return 0

    def chmod(self, path, mode):
        self.permissions[path] = mode

    def chown(self, path, uid, gid):
        self.ext[path]['st_uid'] = uid
        self.ext[path]['st_gid'] = gid

    def rename(self, old, new):
        print('rename', old, new)
        with self.lock11:
            if old not in self.data:
                raise OSError(errno.ENFILE, "File or directory not found")

            # TODO: fixme
            if isinstance(self.data[old], dict) and self.data[old]:
                print('WARNING! renaming non-empty folders with files not yet implemented')
                print(self.data[old])
                raise OSError(errno.EIO, "Not yet implemented")

            path_split = old.split('/')
            entry_name = path_split[-1]
            path_parent = '/' + '/'.join(path_split[1:-1])
            print('>', path_split, entry_name, path_parent)

            if path_parent not in self.data:
                raise OSError(errno.EIO, "Directory not found")

            if not isinstance(self.data[path_parent], dict):
                raise OSError(errno.EIO, "Is file")

            path_split_new = new.split('/')
            entry_name_new = path_split_new[-1]
            path_parent_new = '/' + '/'.join(path_split_new[1:-1])
            print('>', path_split_new, entry_name_new, path_parent_new)

            if path_parent_new not in self.data:
                raise OSError(errno.EIO, "Directory not found")

            if not isinstance(self.data[path_parent_new], dict):
                raise OSError(errno.EIO, "Is file")

            self.data[new] = copy.deepcopy(self.data[old])
            self.data[path_parent_new][entry_name_new] = copy.deepcopy(self.data[path_parent][entry_name])
            self.permissions[new] = copy.deepcopy(self.permissions[old])
            self.ext[new] = copy.deepcopy(self.ext[old])

            del self.data[old]
            del self.data[path_parent][entry_name]
            del self.permissions[old]
            del self.ext[old]

    def create_entry(self, path, mode, is_dir=False):
        print('create_entry', path, mode, is_dir)
        with self.lock11:
            if path in self.data:
                raise OSError(errno.EEXIST, "File or directory already exists")

            path_split = path.split('/')
            entry_name = path_split[-1]
            path_parent = '/' + '/'.join(path_split[1:-1])
            print('>', path_split, entry_name, path_parent)

            if path_parent not in self.data:
                raise OSError(errno.EIO, "Directory not found")

            if not isinstance(self.data[path_parent], dict):
                raise OSError(errno.EIO, "Is file")

            self.data[path] = b"" if not is_dir else {}
            self.permissions[path] = mode

            self.ext[path] = {
                'st_atime': time.time(),
                'st_ctime': time.time(),
                'st_mtime': time.time(),
                'st_gid': os.getgid(),
                'st_uid': os.getuid(),
            }

            self.data[path_parent][entry_name] = b"" if not is_dir else {}
            # self.permissions[f'{path_parent}{entry_name}'] = mode

    def getattr(self, path, fh=None):
        print('getattr', path, fh)
        try:
            with self.lock11:
                if path not in self.data:
                    raise OSError(errno.ENOENT, "myerror: File or directory not found")
                ext = self.ext[path]
                print(self.permissions[path], stat.S_IFDIR, stat.S_IFREG)
                st = {
                    'st_atime': ext['st_atime'],
                    'st_ctime': ext['st_ctime'],
                    'st_mtime': ext['st_mtime'],
                    'st_mode': self.permissions[path] | (
                        stat.S_IFDIR if isinstance(self.data[path], dict) else stat.S_IFREG),
                    'st_nlink': 2 if path != '/' else 1,  # 2 for files and dirs (except root)
                    'st_size': len(self.data[path]) if not isinstance(self.data[path], dict) else 0,
                    'st_gid': ext['st_gid'],
                    'st_uid': ext['st_uid'],
                }
                print(st)
                return st
        except OSError as e:
            print(e)
            raise e

    def readdir(self, path, fh):
        with self.lock11:
            print('readdir', path, fh)
            if path not in self.data or not isinstance(self.data[path], dict):
                raise OSError(errno.ENOENT, "Directory not found")

            self.ext[path]['st_atime'] = time.time()

            result = ['.', '..'] + list(self.data[path].keys())
            print('>', self.data)
            print('>', self.data[path])
            print('>', self.data[path].keys())
            print('>', result)
            return result

    def mkdir(self, path, mode):
        print('mkdir', path, mode)
        self.create_entry(path, mode, is_dir=True)

    def rmdir(self, path):
        with self.lock11:
            print('rmdir', path)
            if path not in self.data or not isinstance(self.data[path], dict) or self.data[path]:  #check for empty
                raise OSError(errno.ENOTEMPTY if self.data[path] else errno.ENOENT, "Directory not empty or not found")

            path_split = path.split('/')
            entry_name = path_split[-1]
            path_parent = '/' + '/'.join(path_split[1:-1])
            print('>', path_split, entry_name, path_parent)

            if path_parent not in self.data:
                raise OSError(errno.EIO, "Directory not found")

            if not isinstance(self.data[path_parent], dict):
                raise OSError(errno.EIO, "Is file")

            del self.data[path]
            del self.data[path_parent][entry_name]
            del self.permissions[path]
            del self.ext[path]
            # self.data[path]
            # self.permissions[path]

    def unlink(self, path):
        with self.lock11:
            print('unlink', path)
            if path not in self.data:
                raise OSError(errno.ENOENT, "File not found")
            path_split = path.split('/')
            entry_name = path_split[-1]
            path_parent = '/' + '/'.join(path_split[1:-1])
            print('>', path_split, entry_name, path_parent)

            if path_parent not in self.data:
                raise OSError(errno.EIO, "Directory not found")

            if not isinstance(self.data[path_parent], dict):
                raise OSError(errno.EIO, "Parent instance is file for some reason")

            del self.data[path]
            del self.data[path_parent][entry_name]
            del self.permissions[path]
            del self.ext[path]

    def open(self, path, flags):
        with self.lock11:
            print('open', path, flags)
            if path not in self.data:
                raise OSError(errno.ENOENT, "File not found")
            if not isinstance(self.data[path], bytes):
                raise OSError(errno.EISDIR, "Is a directory")

            self.ext[path]['st_atime'] = time.time()

            return 0  # File handle

    def read(self, path, length, offset, fh):
        print('read', path, length, offset, fh)
        try:
            with self.lock11:
                if path not in self.data:
                    raise OSError(errno.ENOENT, "File not found")
                self.ext[path]['st_atime'] = time.time()
                return self.data[path][offset:offset + length]
        except OSError as e:
            raise e
        except IndexError:
            return b""

    def write(self, path, data, offset, fh):
        print('write', path, data, offset, fh)
        try:
            with self.lock11:
                if path not in self.data:
                    raise OSError(errno.ENOENT, "File not found")
                current_data = self.data[path]
                self.data[path] = current_data[:offset] + data + current_data[offset + len(data):]
                self.ext[path]['st_mtime'] = time.time()
                return len(data)
        except OSError as e:
            raise e

    def create(self, path, mode, fi=None):
        print('create', path, mode, fi)
        self.create_entry(path, mode)
        return 0


if __name__ == '__main__':
    import sys

    if len(sys.argv) != 2:
        print('usage: %s <mountpoint>' % sys.argv[0])
        exit(1)

    mountpoint = sys.argv[1]

    fuse.FUSE(MemoryFS(), mountpoint, foreground=True)  # , allow_other=True)