Add ability to list root catalogue for fs_list_directory

.gitattributes

@@ -5,7 +5,18 @@
 js/* binary
 screenshots/* binary
 lib/TProto/* binary
-img/* binary
+*.bin binary
+*.heif binary
+*.heic binary
+*.hif binary
+*.avif binary
+*.png binary
+*.gif binary
+*.webp binary
+*.tiff binary
+*.jpeg binary
+*.jpg binary
+*.svg binary
 
 # Files to exclude from GitHub Languages statistics
 *.Dockerfile linguist-vendored=true

.github/workflows/py_analysis-coverage.yml

@@ -113,8 +113,11 @@ jobs:
         with:
           path: apps/${{ env.APP_NAME }}
 
-      - name: Enable App
-        run: php occ app:enable ${{ env.APP_NAME }}
+      - name: Enable App & Test Data
+        run: |
+          php occ app:enable ${{ env.APP_NAME }}
+          cp -R apps/${{ env.APP_NAME }}/tests/nc_py_api/test_dir ./data/admin/files/
+          php occ files:scan admin
 
       - name: Generate coverage report
         working-directory: apps/${{ env.APP_NAME }}
@@ -202,8 +205,11 @@ jobs:
         with:
           path: apps/${{ env.APP_NAME }}
 
-      - name: Enable App
-        run: php occ app:enable ${{ env.APP_NAME }}
+      - name: Enable App & Test Data
+        run: |
+          php occ app:enable ${{ env.APP_NAME }}
+          cp -R apps/${{ env.APP_NAME }}/tests/nc_py_api/test_dir ./data/admin/files/
+          php occ files:scan admin
 
       - name: Generate coverage report
         working-directory: apps/${{ env.APP_NAME }}
@@ -291,8 +297,11 @@ jobs:
         with:
           path: apps/${{ env.APP_NAME }}
 
-      - name: Enable App
-        run: php occ app:enable ${{ env.APP_NAME }}
+      - name: Enable App & Test Data
+        run: |
+          php occ app:enable ${{ env.APP_NAME }}
+          cp -R apps/${{ env.APP_NAME }}/tests/nc_py_api/test_dir ./data/admin/files/
+          php occ files:scan admin
 
       - name: Generate coverage report
         working-directory: apps/${{ env.APP_NAME }}

.pre-commit-config.yaml

@@ -17,20 +17,20 @@ repos:
     rev: 5.10.1
     hooks:
     -   id: isort
-        exclude: (^3rdparty)
+        files: nc_py_api/
 
 -   repo: https://github.com/psf/black
     rev: 22.10.0
     hooks:
     -   id: black
-        exclude: (^3rdparty)
+        files: nc_py_api/
 
 -   repo: https://github.com/PyCQA/flake8
     rev: 6.0.0
     hooks:
     -   id: flake8
         types: [file, python]
-        exclude: (^3rdparty)
+        files: nc_py_api/
 
 -   repo: https://github.com/pre-commit/mirrors-mypy
     rev: v0.991

nc_py_api/db_requests.py

@@ -11,7 +11,7 @@
 )
 
 
-def get_paths_by_ids(file_ids: list) -> list:
+def get_paths_by_ids(file_ids: list[int]) -> list:
     """For each element of list in file_ids return [path, fileid, storage]. Order of file_ids is not preserved."""
 
     query = (

nc_py_api/files.py

@@ -4,7 +4,7 @@
 from fnmatch import fnmatch
 from os import environ, path
 from pathlib import Path
-from typing import Literal, Optional, TypedDict
+from typing import Literal, Optional, TypedDict, Union
 
 from . import mimetype
 from .config import CONFIG
@@ -62,12 +62,31 @@ def fs_get_objs_info(file_ids: list[int]) -> list[FsNodeInfo]:
     return [db_record_to_fs_node(i) for i in raw_result]
 
 
-def fs_list_directory(file_id: int, user_id=USER_ID) -> list[FsNodeInfo]:
-    _ = user_id  # noqa # will be used in 0.4.0 version
-    dir_info = get_paths_by_ids([file_id])
+def fs_list_directory(file_id: Optional[Union[int, FsNodeInfo]] = None, user_id=USER_ID) -> list[FsNodeInfo]:
+    """Get 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.
+
+    :returns: list of :py:data:`FsNodeInfo` dictionaries."""
+
+    storage_id = internal_path = None
+    if file_id is None:  # get user root `files` folder
+        file_id = get_files_root_node(user_id)
+        if file_id is None:
+            return []
+    if not isinstance(file_id, int):  # FsNodeInfo
+        storage_id = file_id["storageId"]
+        internal_path = file_id["internal_path"]
+        file_id = file_id["id"]
+    else:
+        dir_info = get_paths_by_ids([file_id])
+        if dir_info:
+            storage_id = dir_info[0]["storage"]
+            internal_path = dir_info[0]["path"]
     file_mounts = []
-    if dir_info:
-        file_mounts = get_mounts_to(dir_info[0]["storage"], dir_info[0]["path"])
+    if storage_id and internal_path:
+        file_mounts = get_mounts_to(storage_id, internal_path)
     raw_result = get_directory_list(file_id, file_mounts)
     return [db_record_to_fs_node(i) for i in raw_result]
 
@@ -132,29 +151,36 @@ def fs_get_file_data(file_info: FsNodeInfo) -> bytes:
     return request_file_from_php(file_info)
 
 
-def get_storage_info(storage_id: int) -> dict:
+def get_storage_by_id(storage_id: int) -> dict:
     for storage_info in STORAGES_INFO:
         if storage_info["numeric_id"] == storage_id:
             return storage_info
     return {}
 
 
+def get_storage_by_user_id(user_id: str) -> dict:
+    for storage_info in STORAGES_INFO:
+        if storage_info["user_id"] == user_id:
+            return storage_info
+    return {}
+
+
 def get_storage_mount_point(storage_id: int) -> str:
-    storage_info = get_storage_info(storage_id)
+    storage_info = get_storage_by_id(storage_id)
     if storage_info:
         return storage_info["mount_point"]
     return ""
 
 
 def get_storage_user_id(storage_id: int) -> str:
-    storage_info = get_storage_info(storage_id)
+    storage_info = get_storage_by_id(storage_id)
     if storage_info:
         return storage_info["user_id"]
     return ""
 
 
 def get_storage_root_id(storage_id: int) -> int:
-    storage_info = get_storage_info(storage_id)
+    storage_info = get_storage_by_id(storage_id)
     if storage_info:
         return storage_info["root_id"]
     return 0
@@ -171,7 +197,7 @@ def request_file_from_php(file_info: FsNodeInfo) -> bytes:
 
 
 def get_file_full_path(storage_id: int, relative_path: str) -> str:
-    storage_info = get_storage_info(storage_id)
+    storage_info = get_storage_by_id(storage_id)
     if not storage_info:
         return ""
     path_data = storage_info["id"].split(sep="::", maxsplit=1)
@@ -186,7 +212,7 @@ def get_file_full_path(storage_id: int, relative_path: str) -> str:
 
 
 def is_local_storage(storage_id: int) -> bool:
-    storage_info = get_storage_info(storage_id)
+    storage_info = get_storage_by_id(storage_id)
     if not storage_info:
         return False
     if storage_info["available"] == 0:
@@ -249,3 +275,15 @@ def is_path_in_exclude(fs_path: str, exclude_patterns: list[str]) -> bool:
         if fnmatch(name, pattern):
             return True
     return False
+
+
+def get_files_root_node(user_id: str) -> Union[FsNodeInfo, None]:
+    root_id = get_storage_by_user_id(user_id).get("root_id", 0)
+    if not root_id:
+        log.debug("can not find storage for specified user: %s", user_id)
+        return None
+    for i in get_directory_list(root_id, []):
+        if i["name"] == "files" and i["mimetype"] == mimetype.DIR:
+            return db_record_to_fs_node(i)
+    log.debug("can not find `files` directory inside root_id dir")
+    return None

tests/nc_py_api/files_test.py

@@ -0,0 +1,38 @@
+import pytest
+
+import nc_py_api
+
+
+@pytest.mark.parametrize("user_id", ["admin"])
+def test_fs_list_directory(user_id):
+    root_dir_listing = nc_py_api.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)
+    test_dir = [fs_obj for fs_obj in root_dir_listing if fs_obj["name"] == "test_dir"][0]
+    assert test_dir["is_dir"]
+    assert test_dir["is_local"]
+    assert test_dir["mimetype"] == nc_py_api.mimetype.DIR
+    assert test_dir["mimepart"] == nc_py_api.get_mimetype_id("httpd")
+    assert test_dir["internal_path"] == "files/test_dir"
+    assert test_dir["permissions"] == 31
+    assert test_dir["ownerName"] == user_id
+    test_dir_listing = nc_py_api.fs_list_directory(test_dir["id"])
+    assert test_dir_listing == nc_py_api.fs_list_directory(test_dir)  # results should be the same
+    empty_dir = [fs_obj for fs_obj in test_dir_listing if fs_obj["name"] == "empty_dir"][0]
+    assert empty_dir["size"] == 0
+    # directory should be with one empty file
+    assert len(nc_py_api.fs_list_directory(empty_dir)) == 1  # pass FsNodeInfo as fileid
+    assert len(nc_py_api.fs_list_directory(empty_dir["id"])) == 1  # pass fileid as fileid
+    hopper_img = [fs_obj for fs_obj in test_dir_listing if fs_obj["name"] == "hopper.png"][0]
+    assert not hopper_img["is_dir"]
+    assert hopper_img["is_local"]
+    assert hopper_img["mimetype"] == nc_py_api.get_mimetype_id("image/png")
+    assert hopper_img["mimepart"] == nc_py_api.mimetype.IMAGE
+    assert hopper_img["internal_path"] == "files/test_dir/hopper.png"
+    assert hopper_img["permissions"] == 27
+    assert hopper_img["ownerName"] == user_id
+    # probably tests should be divided into smaller parts, need api for getting FsNode by internal_path + user_id...

tests/nc_py_api/occ_test.py

@@ -1,3 +1,4 @@
+import logging
 from unittest import mock
 
 import nc_py_api
@@ -8,7 +9,9 @@ def test_occ_call():
 
 
 def test_occ_call_invalid_command():
+    logging.disable(logging.CRITICAL)
     assert nc_py_api.occ_call("invalid command") is None
+    logging.disable(logging.NOTSET)
 
 
 def test_occ_call_with_param():
@@ -20,7 +23,9 @@ def test_occ_call_decode():
 
 
 def test_occ_call_decode_invalid_command():
+    logging.disable(logging.CRITICAL)
     assert nc_py_api.occ_call_decode("invalid command") is None
+    logging.disable(logging.NOTSET)
 
 
 def test_occ_call_decode_with_param():
@@ -32,7 +37,9 @@ def test_get_cloud_app_config_value():
 
 
 def test_get_cloud_app_config_invalid_name():
+    logging.disable(logging.CRITICAL)
     assert nc_py_api.get_cloud_app_config_value("core", "invalid_name") is None
+    logging.disable(logging.NOTSET)
 
 
 def test_get_cloud_app_config_default_value():