more tests, docs. (#60)

Author: bigcat88

nc_py_api/__init__.py

@@ -13,6 +13,7 @@
     fs_filter_by,
     fs_list_directory,
     fs_node_info,
+    fs_nodes_info,
     fs_sort_by_id,
 )
 from .log import cpa_logger

nc_py_api/_version.py

@@ -1,3 +1,3 @@
 """ Version of Nc_Py_Api """
 
-__version__ = "0.0.10"
+__version__ = "0.0.11.dev0"

nc_py_api/files.py

@@ -52,18 +52,15 @@ class FsNodeInfo(TypedDict):
 """A value from the config that defines the maximum file size allowed to be requested from php."""
 
 
-def fs_node_info(obj: Union[list[int], int, str], user_id=USER_ID) -> Union[list[FsNodeInfo], Optional[FsNodeInfo]]:
-    """Gets `FsNodeInfo` by list of ids, id or path.
+def fs_node_info(obj: Union[int, str], user_id=USER_ID) -> Optional[FsNodeInfo]:
+    """Gets `FsNodeInfo` by `fileid` or path.
 
-    :param obj: for the list of ints or one int it is a `fileid` value. For ``str`` type it is the
-    relative path to file/directory. `path` field from NC DB, without `files/` prefix.
+    :param obj: `fileid` value or ``str`` type if it is the
+    relative path to file/directory(`path` field from NC DB, without `files/` prefix).
     :param user_id: `uid` of user. Optional, in most cases you should not specify it.
 
-    :returns: list of :py:data:`FsNodeInfo`, :py:data:`FsNodeInfo` or None in case of error.
-     Depends on the type of `obj` parameter."""
+    :returns: :py:data:`FsNodeInfo` or None in case of error."""
 
-    if isinstance(obj, list):
-        return [db_record_to_fs_node(i) for i in get_fileids_info(obj)]
     if isinstance(obj, int):
         raw_result = get_fileid_info(obj)
     else:
@@ -77,6 +74,16 @@ def fs_node_info(obj: Union[list[int], int, str], user_id=USER_ID) -> Union[list
     return None
 
 
+def fs_nodes_info(file_ids: list[int]) -> list[FsNodeInfo]:
+    """Gets list of `FsNodeInfo` by list of file ids.
+
+    :param file_ids: list of `fileid`s for which get info.
+
+    :returns: list of :py:data:`FsNodeInfo`."""
+
+    return [db_record_to_fs_node(i) for i in get_fileids_info(file_ids)]
+
+
 def fs_list_directory(file_id: Optional[Union[int, FsNodeInfo]] = None, user_id=USER_ID) -> list[FsNodeInfo]:
     """Gets listing of the directory.
 
@@ -107,7 +114,11 @@ def fs_list_directory(file_id: Optional[Union[int, FsNodeInfo]] = None, user_id=
 
 
 def fs_apply_exclude_lists(fs_objs: list[FsNodeInfo], excl_file_ids: list[int], excl_mask: list[str]) -> None:
-    """Purge all records according to exclude_(mask/fileid) from `where_to_purge`(or from fs_records)."""
+    """Purge all records according to exclude_(mask/fileid).
+
+    :param fs_objs: list of :py:data:`FsNodeInfo` to which exclusion lists apply.
+    :param excl_file_ids: list of int, representing exclude masks fileids.
+    :param excl_mask: list of str, representing exclude masks. Comparison done using `fnmatch`."""
 
     indexes_to_purge = []
     for index, fs_obj in enumerate(fs_objs):
@@ -120,6 +131,11 @@ def fs_apply_exclude_lists(fs_objs: list[FsNodeInfo], excl_file_ids: list[int],
 
 
 def fs_extract_sub_dirs(fs_objs: list[FsNodeInfo]) -> list[FsNodeInfo]:
+    """Extracts and return all records with ``mimetype`` equal to ``DIR``.
+
+    :param fs_objs: list of :py:data:`FsNodeInfo` from which extract directory records. Will be edite in place.
+    :return: list of :py:data:`FsNodeInfo` that are directories."""
+
     sub_dirs = []
     indexes_to_purge = []
     for index, fs_obj in enumerate(fs_objs):
@@ -132,13 +148,26 @@ def fs_extract_sub_dirs(fs_objs: list[FsNodeInfo]) -> list[FsNodeInfo]:
 
 
 def fs_apply_ignore_flags(fs_objs: list[FsNodeInfo]) -> None:
+    """Check for ``.noimage``/``.nomedia`` flag and removes all records with `mimetype` equal to ``IMAGE/VIDEO``.
+
+    The flag also will be removed. The list will be edited in place.
+
+    :param fs_objs: list of :py:data:`FsNodeInfo`."""
+
     ignore_flag = any(fs_obj["name"] in (".noimage", ".nomedia") for fs_obj in fs_objs)
     if ignore_flag:
         fs_filter_by(fs_objs, "mimepart", [mimetype.IMAGE, mimetype.VIDEO], reverse_filter=True)
         fs_apply_exclude_lists(fs_objs, [], [".noimage", ".nomedia"])
 
 
 def fs_filter_by(fs_objs: list[FsNodeInfo], field: FsNodeInfoField, values: list, reverse_filter=False) -> None:
+    """Filter elements in FS list(in-place) by specified mask.
+
+    :param fs_objs: directory listing returned by :py:func:`~nc_py_api.fs_list_directory`.
+    :param field: value from :py:data:`FsNodeInfoField` on which the filter will be applied.
+    :param values: list of values to compare.
+    :param reverse_filter: a boolean indicating that the entry should be filtered if the value is present."""
+
     indexes_to_purge = []
     if reverse_filter:
         for index, fs_obj in enumerate(fs_objs):
@@ -153,6 +182,11 @@ def fs_filter_by(fs_objs: list[FsNodeInfo], field: FsNodeInfoField, values: list
 
 
 def fs_sort_by_id(fs_objs: list[FsNodeInfo]) -> list[FsNodeInfo]:
+    """Helper function, to sort list of `FsNodeInfo` in ascending order by `fileid`.
+
+    :param fs_objs: list of :py:data:`FsNodeInfo`.
+    :return: list of :py:data:`FsNodeInfo`."""
+
     return sorted(fs_objs, key=lambda i: i["id"])
 
 

nc_py_api/mimetype.py

@@ -1,5 +1,7 @@
 from .db_requests import get_mimetype_id
 
 DIR = get_mimetype_id("httpd/unix-directory")
+AUDIO = get_mimetype_id("audio")
 IMAGE = get_mimetype_id("image")
+TEXT = get_mimetype_id("text")
 VIDEO = get_mimetype_id("video")

tests/nc_py_api/etc_test.py

@@ -0,0 +1,17 @@
+from unittest.mock import patch
+
+import nc_py_api
+
+
+def test_remote_filesize_limit():
+    assert nc_py_api.db_requests.get_non_direct_access_filesize_limit() == 512 * 1024 * 1024
+
+
+@patch("nc_py_api.db_requests.execute_fetchall", autospec=True)
+def test_no_remote_filesize_limit(mock_execute_fetchall):
+    def execute_fetchall_side_effect(*args, **kwargs):
+        query = str(args[0]).replace("remote_filesize_limit", "invalid_remote_filesize_limit")
+        return nc_py_api.db_api.execute_fetchall(query, *args[1:], **kwargs)
+
+    mock_execute_fetchall.side_effect = execute_fetchall_side_effect
+    assert nc_py_api.db_requests.get_non_direct_access_filesize_limit() == 256 * 1024 * 1024

tests/nc_py_api/fs_test.py

@@ -4,13 +4,21 @@
 import pytest
 
 from nc_py_api import (
+    fs_apply_exclude_lists,
+    fs_apply_ignore_flags,
+    fs_extract_sub_dirs,
     fs_file_data,
+    fs_filter_by,
     fs_list_directory,
     fs_node_info,
+    fs_nodes_info,
+    fs_sort_by_id,
     get_mimetype_id,
     mimetype,
 )
 
+INVALID_FILEID = 18446744073709551610
+
 
 @pytest.mark.parametrize("test_path", ["", "/"])
 @pytest.mark.parametrize("user_id", ["admin"])
@@ -20,11 +28,12 @@ def test_node_info_root(test_path, user_id):
     assert root_dir_info["mimetype"] == mimetype.DIR
     assert root_dir_info["is_dir"]
     assert root_dir_info["name"] == "files"
+    assert root_dir_info["ownerName"] == user_id
 
 
 @pytest.mark.parametrize("test_path", ["Documents", "Photos", "Templates", "test_dir"])
 def test_node_info_path_dirs(test_path):
-    dir_info = fs_node_info(test_path, user_id="admin")
+    dir_info = fs_node_info(test_path)
     assert dir_info
     assert dir_info["mimetype"] == mimetype.DIR
     assert dir_info["is_dir"]
@@ -33,15 +42,15 @@ def test_node_info_path_dirs(test_path):
 
 @pytest.mark.parametrize("test_path", ["test_dir/hopper.png", "/test_dir/复杂 目录 Í/empty_file.bin"])
 def test_node_info_path_files(test_path):
-    file_info = fs_node_info(test_path, user_id="admin")
+    file_info = fs_node_info(test_path)
     assert file_info
     assert not file_info["is_dir"]
     assert file_info["name"] == Path(test_path).parts[-1:][0]
 
 
 @pytest.mark.parametrize("test_path", ["/test_dir/empty_dir/", "test_dir/empty_dir/", "/test_dir/empty_dir"])
 def test_node_info_path_slashes(test_path):
-    dir_info = fs_node_info(test_path, user_id="admin")
+    dir_info = fs_node_info(test_path)
     assert dir_info
     assert dir_info["mimetype"] == mimetype.DIR
     assert dir_info["is_dir"]
@@ -50,14 +59,14 @@ def test_node_info_path_slashes(test_path):
 
 @pytest.mark.parametrize("test_path", ["/test_dir/复杂 目录 Í/", "test_dir/复杂 目录 Í/", "/test_dir/复杂 目录 Í"])
 def test_node_info_path_diff_symbols(test_path):
-    dir_info = fs_node_info(test_path, user_id="admin")
+    dir_info = fs_node_info(test_path)
     assert dir_info
     assert dir_info["mimetype"] == mimetype.DIR
     assert dir_info["is_dir"]
     assert dir_info["name"] == "复杂 目录 Í"
 
 
-@pytest.mark.parametrize("test_path", ["*-1", "no path", "/no path", "no path/"])
+@pytest.mark.parametrize("test_path", ["*-1", "no path", "/no path", "no path/", INVALID_FILEID])
 @pytest.mark.parametrize("user_id", ["", None, "non_exist"])
 def test_node_info_invalid_input(test_path, user_id):
     logging.disable(logging.CRITICAL)
@@ -66,6 +75,14 @@ def test_node_info_invalid_input(test_path, user_id):
     assert path_info is None
 
 
+@pytest.mark.parametrize("test_ids", [[], [0], [INVALID_FILEID], [0, INVALID_FILEID]])
+def test_nodes_info_invalid_input(test_ids):
+    logging.disable(logging.CRITICAL)
+    path_info = fs_nodes_info(test_ids)
+    logging.disable(logging.NOTSET)
+    assert len(path_info) == 0
+
+
 @pytest.mark.parametrize("user_id", ["admin"])
 def test_list_directory_root(user_id):
     root_dir_listing = fs_list_directory(user_id=user_id)
@@ -77,7 +94,13 @@ def test_list_directory_root(user_id):
     assert any(fs_obj["name"] == "test_dir" for fs_obj in root_dir_listing)
 
 
-@pytest.mark.parametrize("file_id", [None, 0, 18446744073709551610])
+def test_list_directory_by_fileid():
+    root_dir_listing = fs_list_directory()
+    test_dir_id = [fs_obj["id"] for fs_obj in root_dir_listing if fs_obj["name"] == "test_dir"][0]
+    assert fs_list_directory(test_dir_id) == fs_list_directory(fs_node_info("test_dir"))
+
+
+@pytest.mark.parametrize("file_id", [None, 0, INVALID_FILEID])
 @pytest.mark.parametrize("user_id", ["", None, "non_exist"])
 def test_list_directory_invalid_input(file_id, user_id):
     logging.disable(logging.CRITICAL)
@@ -88,46 +111,49 @@ def test_list_directory_invalid_input(file_id, user_id):
 
 
 def test_list_directory_test_dir():
-    test_dir_listing = fs_list_directory(fs_node_info("test_dir", user_id="admin"), user_id="admin")
-    assert len(test_dir_listing) == 4
+    test_dir_listing = fs_list_directory(fs_node_info("test_dir"))
+    n = len(test_dir_listing)
+    assert n == 5
     assert any(fs_obj["name"] == "empty_dir" for fs_obj in test_dir_listing)
     assert any(fs_obj["name"] == "复杂 目录 Í" for fs_obj in test_dir_listing)
     assert any(fs_obj["name"] == "hopper.png" for fs_obj in test_dir_listing)
     assert any(fs_obj["name"] == "test.txt" for fs_obj in test_dir_listing)
+    assert any(fs_obj["name"] == "відео та картинки" for fs_obj in test_dir_listing)
 
 
 @pytest.mark.parametrize("test_path", ["test_dir", "test_dir/hopper.png", "/test_dir/复杂 目录 Í/empty_file.bin"])
 def test_node_info_id(test_path):
-    file_info1 = fs_node_info(test_path, user_id="admin")
+    file_info1 = fs_node_info(test_path)
     file_info2 = fs_node_info(file_info1["id"])
     assert isinstance(file_info1, dict)
     assert file_info1 == file_info2
 
 
-def test_node_info_ids():
-    test_dir_listing = fs_list_directory(fs_node_info("test_dir", user_id="admin"), user_id="admin")
+def test_nodes_info_ids():
+    test_dir_listing = fs_list_directory(fs_node_info("test_dir"))
     ids = [i["id"] for i in test_dir_listing]
-    objs_info = fs_node_info(ids)
+    objs_info = fs_nodes_info(ids)
     assert isinstance(objs_info, list)
-    assert len(objs_info) == 4
+    n = len(objs_info)
+    assert n == 5
     assert all(obj["id"] for obj in objs_info)
 
 
 def test_parent_field():
-    dir_info = fs_node_info("test_dir/empty_dir", user_id="admin")
+    dir_info = fs_node_info("test_dir/empty_dir")
     parent_info = fs_node_info(dir_info["parent_id"])
     assert parent_info["name"] == "test_dir"
-    dir_info = fs_node_info("test_dir/hopper.png", user_id="admin")
+    dir_info = fs_node_info("test_dir/hopper.png")
     parent_info = fs_node_info(dir_info["parent_id"])
     assert parent_info["name"] == "test_dir"
-    dir_info = fs_node_info("test_dir/empty_dir/empty_file.bin", user_id="admin")
+    dir_info = fs_node_info("test_dir/empty_dir/empty_file.bin")
     parent_info = fs_node_info(dir_info["parent_id"])
     assert parent_info["name"] == "empty_dir"
 
 
 def test_node_dir_fields():
-    dir_info = fs_node_info("test_dir/empty_dir", user_id="admin")
-    _ = fs_list_directory(fs_node_info("test_dir", user_id="admin"), user_id="admin")
+    dir_info = fs_node_info("test_dir/empty_dir")
+    _ = fs_list_directory(fs_node_info("test_dir"))
     dir_info2 = [i for i in _ if i["name"] == "empty_dir"][0]
     assert dir_info == dir_info2
     assert dir_info["is_dir"]
@@ -141,8 +167,8 @@ def test_node_dir_fields():
 
 
 def test_node_file_fields():
-    file_info = fs_node_info("test_dir/hopper.png", user_id="admin")
-    _ = fs_list_directory(fs_node_info("test_dir", user_id="admin"), user_id="admin")
+    file_info = fs_node_info("test_dir/hopper.png")
+    _ = fs_list_directory(fs_node_info("test_dir"))
     file_info2 = [i for i in _ if i["name"] == "hopper.png"][0]
     assert file_info == file_info2
     assert not file_info["is_dir"]
@@ -157,10 +183,80 @@ def test_node_file_fields():
 
 @pytest.mark.parametrize("file", ["test_dir/hopper.png", "test_dir/test.txt", "/test_dir/复杂 目录 Í/empty_file.bin"])
 def test_fs_file_data(file):
-    node_info = fs_node_info(file, user_id="admin")
+    node_info = fs_node_info(file)
     file_data = fs_file_data(node_info)
     assert isinstance(file_data, bytes)
     if file.find("empty") == -1:
         assert file_data
     else:
         assert not len(file_data)
+
+
+def test_fs_filter_by():
+    def len_after_filter_by(field, values, reverse=False) -> int:
+        _ = fs_list_directory(fs_node_info("test_dir"))
+        fs_filter_by(_, field, values, reverse_filter=reverse)
+        return len(_)
+
+    # mimepart
+    assert len_after_filter_by("mimepart", [mimetype.IMAGE]) == 1
+    assert len_after_filter_by("mimetype", [mimetype.DIR], True) == 2
+    # is_dir
+    assert len_after_filter_by("is_dir", [True]) == 3
+    # is_local
+    assert len_after_filter_by("is_local", [True]) == 5
+    # name
+    assert len_after_filter_by("name", ["test.txt"]) == 1
+    assert len_after_filter_by("name", ["test.txt"], True) == 4
+    # direct_access
+    assert len_after_filter_by("direct_access", [True]) == 5
+
+
+def test_fs_apply_exclude_lists():
+    dir_list = fs_list_directory()
+    assert any(fs_obj["name"] == "test_dir" for fs_obj in dir_list)
+    fileid = dir_list[0]["id"]
+    ids = [i["id"] for i in dir_list]
+    assert fileid in ids
+    fs_apply_exclude_lists(dir_list, [fileid], [])
+    ids = [i["id"] for i in dir_list]
+    assert fileid not in ids
+    assert any(fs_obj["name"] == "test_dir" for fs_obj in dir_list)
+    fs_apply_exclude_lists(dir_list, [], ["test_dir"])
+    assert not any(fs_obj["name"] == "test_dir" for fs_obj in dir_list)
+
+
+def test_fs_apply_ignore_flags():
+    dir_list = fs_list_directory(fs_node_info("test_dir/відео та картинки"))
+    assert len(dir_list) == 3
+    fs_apply_ignore_flags(dir_list)
+    assert not len(dir_list)
+
+
+def test_not_initialized_user_get_files_root_node():
+    import nc_py_api
+
+    logging.disable(logging.CRITICAL)
+    assert nc_py_api.files.get_files_root_node(user_id="user") is None
+    logging.disable(logging.NOTSET)
+
+
+def test_fs_extract_sub_dirs():
+    dir_list = fs_list_directory()
+    assert any(fs_obj["is_dir"] for fs_obj in dir_list)
+    sub_dirs = fs_extract_sub_dirs(dir_list)
+    assert not any(fs_obj["is_dir"] for fs_obj in dir_list)
+    assert all(fs_obj["is_dir"] for fs_obj in sub_dirs)
+    dir_list = fs_list_directory(fs_node_info("test_dir/empty_dir"))
+    sub_dirs = fs_extract_sub_dirs(dir_list)
+    assert not sub_dirs
+
+
+def test_fs_sort_by_id():
+    dir1_list = fs_list_directory()
+    dir2_list = fs_list_directory(fs_node_info("test_dir"))
+    dir3_list = fs_list_directory(fs_node_info("test_dir/відео та картинки"))
+    united_dir_list = [*dir2_list, *dir3_list, *dir1_list]
+    sorted_dir_list = fs_sort_by_id(united_dir_list)
+    assert united_dir_list != sorted_dir_list
+    assert sorted_dir_list == fs_sort_by_id(sorted_dir_list)