4399176145

Build Type

github

Committed by GitHub

Commit Message

Merge ee2cf80d9 into 7590ad02c

Pull Request Pull Request #30: Fuse

Run Details

474 of 474 new or added lines in 13 files covered. (100.0%)

1304 of 1691 relevant lines covered (77.11%)

3.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

26.47

/catcli/fuser.py

"""
author: deadc0de6 (https://github.com/deadc0de6)
Copyright (c) 2023, deadc0de6

fuse for catcli
"""

import os
from time import time
from stat import S_IFDIR, S_IFREG
from typing import List, Dict, Any, Optional
import fuse  # type: ignore

# local imports
from catcli.noder import Noder
from catcli.nodes import NodeTop, NodeAny
from catcli.utils import path_to_search_all, path_to_top
from catcli import nodes


class Fuser:
    """fuse filesystem mounter"""

    def __init__(self, mountpoint: str,
                 top: NodeTop,
                 noder: Noder,
                 debug: bool = False):
        """fuse filesystem"""
        filesystem = CatcliFilesystem(top, noder)
        fuse.FUSE(filesystem,
                  mountpoint,
                  foreground=debug,
                  allow_other=True,
                  nothreads=True,
                  debug=debug)


class CatcliFilesystem(fuse.LoggingMixIn, fuse.Operations):  # type: ignore
    """in-memory filesystem for catcli catalog"""

    def __init__(self, top: NodeTop,
                 noder: Noder):
        """init fuse filesystem"""
        self.top = top
        self.noder = noder

    def _get_entry(self, path: str) -> Optional[NodeAny]:
        """return the node pointed by path"""
        path = path_to_top(path)
        found = self.noder.list(self.top, path,
                                rec=False,
                                fmt='native',
                                raw=True)
        if found:
            return found[0]
        return None

    def _get_entries(self, path: str) -> List[NodeAny]:
        """return nodes pointed by path"""
        path = path_to_search_all(path)
        found = self.noder.list(self.top, path,
                                rec=False,
                                fmt='native',
                                raw=True)
        return found

    def _getattr(self, path: str) -> Dict[str, Any]:
        entry = self._get_entry(path)
        if not entry:
            return {}

        maccess = time()
        mode: Any = S_IFREG
        size: int = 0
        if entry.type == nodes.TYPE_ARCHIVED:
            mode = S_IFREG
            size = entry.size
        elif entry.type == nodes.TYPE_DIR:
            mode = S_IFDIR
            size = entry.size
            maccess = entry.maccess
        elif entry.type == nodes.TYPE_FILE:
            mode = S_IFREG
            size = entry.size
            maccess = entry.maccess
        elif entry.type == nodes.TYPE_STORAGE:
            mode = S_IFDIR
            size = entry.size
            maccess = entry.ts
        elif entry.type == nodes.TYPE_META:
            mode = S_IFREG
        elif entry.type == nodes.TYPE_TOP:
            mode = S_IFREG
        mode = mode | 0o777
        return {
            'st_mode': (mode),  # file type
            'st_nlink': 1,  # count hard link
            'st_size': size,
            'st_ctime': maccess,  # attr last modified
            'st_mtime': maccess,  # content last modified
            'st_atime': maccess,  # access time
            'st_uid': os.getuid(),
            'st_gid': os.getgid(),
        }

    def getattr(self, path: str, _fh: Any = None) -> Dict[str, Any]:
        """return attr of file pointed by path"""
        if path == '/':
            # mountpoint
            curt = time()
            meta = {
                'st_mode': (S_IFDIR | 0o777),
                'st_nlink': 1,
                'st_size': 0,
                'st_ctime': curt,
                'st_mtime': curt,
                'st_atime': curt,
                'st_uid': os.getuid(),
                'st_gid': os.getgid(),
            }
            return meta
        meta = self._getattr(path)
        return meta

    def readdir(self, path: str, _fh: Any) -> List[str]:
        """read directory content"""
        content = ['.', '..']
        entries = self._get_entries(path)
        for entry in entries:
            content.append(entry.name)
        return content

1	"""
2	author: deadc0de6 (https://github.com/deadc0de6)
3	Copyright (c) 2023, deadc0de6
4
5	fuse for catcli
6	"""
7
8	import os	5✔
9	from time import time	5✔
10	from stat import S_IFDIR, S_IFREG	5✔
11	from typing import List, Dict, Any, Optional	5✔
12	import fuse # type: ignore	5✔
13
14	# local imports
15	from catcli.noder import Noder	5✔
16	from catcli.nodes import NodeTop, NodeAny	5✔
17	from catcli.utils import path_to_search_all, path_to_top	5✔
18	from catcli import nodes	5✔
19
20
21	class Fuser:	5✔
22	"""fuse filesystem mounter"""
23
24	def __init__(self, mountpoint: str,	5✔
25	top: NodeTop,
26	noder: Noder,
27	debug: bool = False):
28	"""fuse filesystem"""
29	filesystem = CatcliFilesystem(top, noder)	×
30	fuse.FUSE(filesystem,	×
31	mountpoint,
32	foreground=debug,
33	allow_other=True,
34	nothreads=True,
35	debug=debug)
36
37
38	class CatcliFilesystem(fuse.LoggingMixIn, fuse.Operations): # type: ignore	5✔
39	"""in-memory filesystem for catcli catalog"""
40
41	def __init__(self, top: NodeTop,	5✔
42	noder: Noder):
43	"""init fuse filesystem"""
44	self.top = top	×
45	self.noder = noder	×
46
47	def _get_entry(self, path: str) -> Optional[NodeAny]:	5✔
48	"""return the node pointed by path"""
49	path = path_to_top(path)	×
50	found = self.noder.list(self.top, path,	×
51	rec=False,
52	fmt='native',
53	raw=True)
54	if found:	×
55	return found[0]	×
56	return None	×
57
58	def _get_entries(self, path: str) -> List[NodeAny]:	5✔
59	"""return nodes pointed by path"""
60	path = path_to_search_all(path)	×
61	found = self.noder.list(self.top, path,	×
62	rec=False,
63	fmt='native',
64	raw=True)
65	return found	×
66
67	def _getattr(self, path: str) -> Dict[str, Any]:	5✔
68	entry = self._get_entry(path)	×
69	if not entry:	×
70	return {}	×
71
72	maccess = time()	×
73	mode: Any = S_IFREG	×
74	size: int = 0	×
75	if entry.type == nodes.TYPE_ARCHIVED:	×
76	mode = S_IFREG	×
77	size = entry.size	×
78	elif entry.type == nodes.TYPE_DIR:	×
79	mode = S_IFDIR	×
80	size = entry.size	×
81	maccess = entry.maccess	×
82	elif entry.type == nodes.TYPE_FILE:	×
83	mode = S_IFREG	×
84	size = entry.size	×
85	maccess = entry.maccess	×
86	elif entry.type == nodes.TYPE_STORAGE:	×
87	mode = S_IFDIR	×
88	size = entry.size	×
89	maccess = entry.ts	×
90	elif entry.type == nodes.TYPE_META:	×
91	mode = S_IFREG	×
92	elif entry.type == nodes.TYPE_TOP:	×
93	mode = S_IFREG	×
94	mode = mode \| 0o777	×
95	return {	×
96	'st_mode': (mode), # file type
97	'st_nlink': 1, # count hard link
98	'st_size': size,
99	'st_ctime': maccess, # attr last modified
100	'st_mtime': maccess, # content last modified
101	'st_atime': maccess, # access time
102	'st_uid': os.getuid(),
103	'st_gid': os.getgid(),
104	}
105
106	def getattr(self, path: str, _fh: Any = None) -> Dict[str, Any]:	5✔
107	"""return attr of file pointed by path"""
108	if path == '/':	×
109	# mountpoint
110	curt = time()	×
111	meta = {	×
112	'st_mode': (S_IFDIR \| 0o777),
113	'st_nlink': 1,
114	'st_size': 0,
115	'st_ctime': curt,
116	'st_mtime': curt,
117	'st_atime': curt,
118	'st_uid': os.getuid(),
119	'st_gid': os.getgid(),
120	}
121	return meta	×
122	meta = self._getattr(path)	×
123	return meta	×
124
125	def readdir(self, path: str, _fh: Any) -> List[str]:	5✔
126	"""read directory content"""
127	content = ['.', '..']	×
128	entries = self._get_entries(path)	×
129	for entry in entries:	×
130	content.append(entry.name)	×
131	return content	×

deadc0de6 / catcli / 4399176145

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous