added test for MemProtocol.ls()

.pre-commit-config.yaml

@@ -17,7 +17,7 @@ repos:
   - id: mixed-line-ending
     args: ['--fix=no']
   - id: flake8
-    args: ['--max-line-length=88']  # default of Black
+    args: ['--max-line-length=120']  # default of Black
 
 - repo: https://github.com/pre-commit/mirrors-isort
   rev: v4.3.21

setup.cfg

@@ -12,7 +12,7 @@ long-description = file: README.rst
 long-description-content-type = text/x-rst; charset=UTF-8
 url = https://github.com/kamikaze/aiofm
 project-urls =
-    Documentation = https://pyscaffold.org/
+    Documentation = https://github.com/kamikaze/aiofm
 # Change if running only on Windows, Mac or Linux (comma-separated)
 platforms = any
 # Add here all kinds of additional classifiers as defined under
@@ -32,7 +32,11 @@ setup_requires = pyscaffold>=3.2a0,<3.3a0
 # Add here dependencies of your project (semicolon/line-separated), e.g.
 # install_requires = numpy; scipy
 # The usage of test_requires is discouraged, see `Dependency Management` docs
-# tests_require = pytest; pytest-cov
+#tests_require =
+#    pytest
+#    pytest-asyncio
+#    pytest-cov
+#    pytest-mock
 # Require a specific Python version, e.g. Python 2.7 or >= 3.4
 # python_requires = >=2.7,!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*
 
@@ -48,7 +52,9 @@ exclude =
 # Add here test requirements (semicolon/line-separated)
 testing =
     pytest
+    pytest-asyncio
     pytest-cov
+    pytest-mock
 
 [options.entry_points]
 # Add here console scripts like:

src/aiofm/protocols.py

@@ -1,6 +1,9 @@
+import operator
 import os.path
 from abc import ABCMeta, abstractmethod
-from typing import Sequence, Tuple
+from functools import reduce
+from pathlib import PurePath
+from typing import Mapping, Sequence, Tuple, Union
 
 
 class BaseProtocol(metaclass=ABCMeta):
@@ -57,10 +60,24 @@ class MemProtocol(BaseProtocol):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.path_sep = os.path.sep
+        self._tree = {}
 
     @staticmethod
-    async def ls(path: str, pattern: str = None, *args, **kwargs) -> Sequence:
-        pass
+    def _get_tree_item(tree: Mapping, path: Union[str, PurePath]):
+        try:
+            path_parts = path.parts
+        except AttributeError:
+            path_parts = PurePath(path).parts
+
+        try:
+            return reduce(operator.getitem, path_parts, tree)
+        except KeyError:
+            raise FileNotFoundError
+
+    async def ls(self, path: Union[str, PurePath], pattern: str = None, *args, **kwargs) -> Sequence:
+        item = self._get_tree_item(self._tree, path)
+
+        return tuple(item)
 
     async def open(self, path, *args, **kwargs):
         pass

tests/aiofm/test_mem_protocol.py

@@ -0,0 +1,20 @@
+import pytest
+
+from aiofm.protocols import MemProtocol
+
+
+@pytest.mark.asyncio
+async def test_ls_tmp_dir():
+    fs = MemProtocol()
+    fs._tree = {'/': {'tmp': {'xxx': {}, 'a.txt': b'data data data'}}}
+
+    assert await fs.ls('/tmp') == ('xxx', 'a.txt')
+
+
+@pytest.mark.asyncio
+async def test_ls_inextisting_dir_fails():
+    fs = MemProtocol()
+    fs._tree = {'/': {'tmp': {'xxx': {}, 'a.txt': b'data data data'}}}
+
+    with pytest.raises(FileNotFoundError):
+        assert await fs.ls('/pmt')