Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_memory_path_io_repr():
from aioftp.pathio import Node
pio = aioftp.MemoryPathIO(loop="loop")
pio.fs = [Node("dir", "/", 1, 1, content=[])]
nose.tools.eq_(
repr(pio),
"[Node(type='dir', name='/', ctime=1, mtime=1, content=[])]"
)
if parent is None:
raise FileNotFoundError
elif parent.type != "dir":
raise NotADirectoryError
node = Node("dir", path.name, content=[])
parent.content.append(node)
else:
nodes = self.fs
for part in path.parts:
if isinstance(nodes, list):
for node in nodes:
if node.name == part:
nodes = node.content
break
else:
node = Node("dir", part, content=[])
nodes.append(node)
nodes = node.content
else:
raise NotADirectoryError
async def _open(self, path, mode="rb", *args, **kwargs):
if mode == "rb":
node = self.get_node(path)
if node is None:
raise FileNotFoundError
file_like = node.content
file_like.seek(0, io.SEEK_SET)
elif mode in ("wb", "ab", "r+b"):
node = self.get_node(path)
if node is None:
parent = self.get_node(path.parent)
if parent is None or parent.type != "dir":
raise FileNotFoundError
new_node = Node("file", path.name, content=io.BytesIO())
parent.content.append(new_node)
file_like = new_node.content
elif node.type != "file":
raise IsADirectoryError
else:
if mode == "wb":
file_like = node.content = io.BytesIO()
elif mode == "ab":
file_like = node.content
file_like.seek(0, io.SEEK_END)
elif mode == "r+b":
file_like = node.content
file_like.seek(0, io.SEEK_SET)
else:
raise ValueError(f"invalid mode: {mode}")
return file_like
async def mkdir(self, path, *, parents=False, exist_ok=False):
path = self._absolute(path)
node = self.get_node(path)
if node:
if node.type != "dir" or not exist_ok:
raise FileExistsError
elif not parents:
parent = self.get_node(path.parent)
if parent is None:
raise FileNotFoundError
elif parent.type != "dir":
raise NotADirectoryError
node = Node("dir", path.name, content=[])
parent.content.append(node)
else:
nodes = self.fs
for part in path.parts:
if isinstance(nodes, list):
for node in nodes:
if node.name == part:
nodes = node.content
break
else:
node = Node("dir", part, content=[])
nodes.append(node)
nodes = node.content
else:
raise NotADirectoryError
def __init__(self, *args, state=None, cwd=None, **kwargs):
super().__init__(*args, **kwargs)
self.cwd = pathlib.PurePosixPath(cwd or "/")
if state is None:
self.fs = [Node("dir", "/", content=[])]
else:
self.fs = state