5506286099

pending completion

Build # 5506286099

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #37 from deadc0de6/fix-36

size attr fix for #36

Run Details

72 of 72 new or added lines in 4 files covered. (100.0%)

1329 of 1751 relevant lines covered (75.9%)

3.79 hits per line

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

0.0

/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
try:
    import fuse  # type: ignore
except ModuleNotFoundError:
    pass

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

deadc0de6 / catcli / 5506286099

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