more consistent FS API and tests

CHANGELOG.md

@@ -2,19 +2,18 @@ _# Changelog
 
 All notable changes to this project will be documented in this file.
 
-## [0.0.7 - 2022-12-12]
+## [0.0.7 - 2022-12-14]
 
 ### Added
 
 - FS functions:
+  * `fs_node_info`
+  * `fs_list_directory`
+  * `fs_file_data`
   * `fs_apply_exclude_lists`
   * `fs_apply_ignore_flags`
   * `fs_extract_sub_dirs`
   * `fs_filter_by`
-  * `fs_get_file_data`
-  * `fs_get_obj_info`
-  * `fs_get_objs_info`
-  * `fs_list_directory`
   * `fs_sort_by_id`
 
 ### Changed

README.md

@@ -4,6 +4,7 @@
 ![PythonVersion](https://img.shields.io/badge/python-3.9%20%7C%203.10%20%7C%203.11-blue)
 ![impl](https://img.shields.io/pypi/implementation/nc_py_api)
 ![pypi](https://img.shields.io/pypi/v/nc_py_api.svg)
+[![codecov](https://codecov.io/gh/cloud-py-api/cloud_py_api/branch/main/graph/badge.svg?token=6IHPKUYUU9)](https://codecov.io/gh/cloud-py-api/cloud_py_api)
 
 Framework(App) for Nextcloud to develop apps, that using Python.
 

nc_py_api/__init__.py

@@ -9,11 +9,10 @@
     fs_apply_exclude_lists,
     fs_apply_ignore_flags,
     fs_extract_sub_dirs,
+    fs_file_data,
     fs_filter_by,
-    fs_get_file_data,
-    fs_get_obj_info,
-    fs_get_objs_info,
     fs_list_directory,
+    fs_node_info,
     fs_sort_by_id,
 )
 from .log import cpa_logger

nc_py_api/db_requests.py

@@ -6,7 +6,7 @@
 
 FIELD_NAME_LIST = (
     "fcache.fileid, fcache.storage, fcache.path, fcache.storage, fcache.name, "
-    "fcache.mimetype, fcache.mimepart, "
+    "fcache.mimetype, fcache.mimepart, fcache.parent, "
     "fcache.size, fcache.mtime, fcache.encrypted, fcache.etag, fcache.permissions, fcache.checksum"
 )
 
@@ -77,6 +77,19 @@ def get_fileid_info(file_id: int) -> dict:
     return {}
 
 
+def get_fs_obj_info_by_path(obj_path: str, storage_numeric_id: int) -> dict:
+    """Returns dictionary with information for given userid:path."""
+
+    query = (
+        f"SELECT {FIELD_NAME_LIST} FROM {TABLES.file_cache} AS fcache "
+        f"WHERE fcache.path = '{obj_path}' AND fcache.storage = {storage_numeric_id};"
+    )
+    result = execute_fetchall(query)
+    if result:
+        return result[0]
+    return {}
+
+
 def get_fileids_info(file_ids: list[int]) -> list[dict]:
     """Returns dictionaries with information for given file ids."""
 

nc_py_api/files.py

@@ -12,6 +12,7 @@
     get_directory_list,
     get_fileid_info,
     get_fileids_info,
+    get_fs_obj_info_by_path,
     get_non_direct_access_filesize_limit,
     get_paths_by_ids,
     get_storages_info,
@@ -30,6 +31,7 @@ class FsNodeInfo(TypedDict):
     internal_path: str
     abs_path: str
     size: int
+    parent_id: int
     permissions: int
     mtime: int
     checksum: str
@@ -50,20 +52,33 @@ class FsNodeInfo(TypedDict):
 """A value from the config that defines the maximum file size allowed to be requested from php."""
 
 
-def fs_get_obj_info(file_id: int) -> Optional[FsNodeInfo]:
-    raw_result = get_fileid_info(file_id)
+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.
+
+    :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 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."""
+
+    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:
+        numeric_id = get_storage_by_user_id(user_id).get("numeric_id", 0)
+        if not numeric_id:
+            log.debug("can not find storage for specified user: %s", user_id)
+            return None
+        raw_result = get_fs_obj_info_by_path(path.join("files", obj.lstrip("/")).rstrip("/"), numeric_id)
     if raw_result:
         return db_record_to_fs_node(raw_result)
     return None
 
 
-def fs_get_objs_info(file_ids: list[int]) -> list[FsNodeInfo]:
-    raw_result = get_fileids_info(file_ids)
-    return [db_record_to_fs_node(i) for i in raw_result]
-
-
 def fs_list_directory(file_id: Optional[Union[int, FsNodeInfo]] = None, user_id=USER_ID) -> list[FsNodeInfo]:
-    """Get listing of the directory.
+    """Gets listing of the directory.
 
     :param file_id: `fileid` or :py:data:`FsNodeInfo` of the directory. Can be `None` to list `root` directory.
     :param user_id: `uid` of user. Optional, in most cases you should not specify it.
@@ -141,7 +156,7 @@ def fs_sort_by_id(fs_objs: list[FsNodeInfo]) -> list[FsNodeInfo]:
     return sorted(fs_objs, key=lambda i: i["id"])
 
 
-def fs_get_file_data(file_info: FsNodeInfo) -> bytes:
+def fs_file_data(file_info: FsNodeInfo) -> bytes:
     if file_info["direct_access"]:
         try:
             with open(file_info["abs_path"], "rb") as h_file:
@@ -255,6 +270,7 @@ def db_record_to_fs_node(fs_record: dict) -> FsNodeInfo:
         "internal_path": fs_record["path"],
         "abs_path": get_file_full_path(fs_record["storage"], fs_record["path"]),
         "size": fs_record["size"],
+        "parent_id": fs_record["parent"],
         "permissions": fs_record["permissions"],
         "mtime": fs_record["mtime"],
         "checksum": fs_record["checksum"],

tests/nc_py_api/fs_test.py

@@ -0,0 +1,166 @@
+import logging
+from pathlib import Path
+
+import pytest
+
+from nc_py_api import (
+    fs_file_data,
+    fs_list_directory,
+    fs_node_info,
+    get_mimetype_id,
+    mimetype,
+)
+
+
+@pytest.mark.parametrize("test_path", ["", "/"])
+@pytest.mark.parametrize("user_id", ["admin"])
+def test_node_info_root(test_path, user_id):
+    root_dir_info = fs_node_info(test_path, user_id=user_id)
+    assert root_dir_info
+    assert root_dir_info["mimetype"] == mimetype.DIR
+    assert root_dir_info["is_dir"]
+    assert root_dir_info["name"] == "files"
+
+
+@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")
+    assert dir_info
+    assert dir_info["mimetype"] == mimetype.DIR
+    assert dir_info["is_dir"]
+    assert dir_info["name"] == 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")
+    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")
+    assert dir_info
+    assert dir_info["mimetype"] == mimetype.DIR
+    assert dir_info["is_dir"]
+    assert dir_info["name"] == "empty_dir"
+
+
+@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")
+    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("user_id", ["", None, "non_exist"])
+def test_node_info_invalid_input(test_path, user_id):
+    logging.disable(logging.CRITICAL)
+    path_info = fs_node_info(test_path, user_id=user_id)
+    logging.disable(logging.NOTSET)
+    assert path_info is None
+
+
+@pytest.mark.parametrize("user_id", ["admin"])
+def test_list_directory_root(user_id):
+    root_dir_listing = fs_list_directory(user_id=user_id)
+    # `Documents`, `Photos`, `Templates`, `test_files` folders
+    assert len(root_dir_listing) >= 4
+    assert any(fs_obj["name"] == "Documents" for fs_obj in root_dir_listing)
+    assert any(fs_obj["name"] == "Photos" for fs_obj in root_dir_listing)
+    assert any(fs_obj["name"] == "Templates" for fs_obj in root_dir_listing)
+    assert any(fs_obj["name"] == "test_dir" for fs_obj in root_dir_listing)
+
+
+@pytest.mark.parametrize("file_id", [None, 0, 18446744073709551610])
+@pytest.mark.parametrize("user_id", ["", None, "non_exist"])
+def test_list_directory_invalid_input(file_id, user_id):
+    logging.disable(logging.CRITICAL)
+    root_dir_listing = fs_list_directory(file_id=file_id, user_id=user_id)
+    logging.disable(logging.NOTSET)
+    assert isinstance(root_dir_listing, list)
+    assert not root_dir_listing
+
+
+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
+    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)
+
+
+@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_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")
+    ids = [i["id"] for i in test_dir_listing]
+    objs_info = fs_node_info(ids)
+    assert isinstance(objs_info, list)
+    assert len(objs_info) == 4
+    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")
+    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")
+    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")
+    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_info2 = [i for i in _ if i["name"] == "empty_dir"][0]
+    assert dir_info == dir_info2
+    assert dir_info["is_dir"]
+    assert dir_info["is_local"]
+    assert dir_info["mimetype"] == mimetype.DIR
+    assert dir_info["mimepart"] == get_mimetype_id("httpd")
+    assert dir_info["internal_path"] == "files/test_dir/empty_dir"
+    assert dir_info["permissions"] == 31
+    assert dir_info["ownerName"] == "admin"
+    assert dir_info["size"] == 0
+
+
+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_info2 = [i for i in _ if i["name"] == "hopper.png"][0]
+    assert file_info == file_info2
+    assert not file_info["is_dir"]
+    assert file_info["is_local"]
+    assert file_info["mimetype"] == get_mimetype_id("image/png")
+    assert file_info["mimepart"] == mimetype.IMAGE
+    assert file_info["internal_path"] == "files/test_dir/hopper.png"
+    assert file_info["permissions"] == 27
+    assert file_info["ownerName"] == "admin"
+    assert file_info["size"] == 30605
+
+
+@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")
+    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)

tests/nc_py_api/mimetype_test.py

@@ -1,6 +1,8 @@
+import logging
+
 import pytest
 
-from nc_py_api import get_mimetype_id
+from nc_py_api import fs_node_info, get_mimetype_id
 
 
 @pytest.mark.parametrize("mimetype", ["httpd", "httpd/unix-directory", "application", "text", "image", "video"])
@@ -10,4 +12,15 @@ def test_get_mimetype_id(mimetype):
 
 @pytest.mark.parametrize("mimetype", ["", "invalid_mime", None, "'invalid_mime'"])
 def test_get_mimetype_id_invalid(mimetype):
+    logging.disable(logging.CRITICAL)
     assert not get_mimetype_id(mimetype)
+    logging.disable(logging.NOTSET)
+
+
+def test_mimetype_other():
+    for key, value in {
+        "test_dir/empty_dir/empty_file.bin": "application/x-bin",
+        "test_dir/hopper.png": "image/png",
+        "test_dir/test.txt": "text/plain",
+    }.items():
+        assert fs_node_info(key, user_id="admin")["mimetype"] == get_mimetype_id(value)

tests/nc_py_api/occ_test.py

@@ -43,7 +43,9 @@ def test_get_cloud_app_config_invalid_name():
 
 
 def test_get_cloud_app_config_default_value():
+    logging.disable(logging.CRITICAL)
     assert nc_py_api.get_cloud_app_config_value("core", "invalid_name", default=3) == 3
+    logging.disable(logging.NOTSET)
 
 
 @mock.patch("nc_py_api.occ._PHP_PATH", "no_php")

tests/nc_py_api/test_dir/test.txt

@@ -0,0 +1 @@
+Test file