diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d07775a..9203f9d2 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 55fb335f..1b662f7e 100644 --- a/README.md +++ b/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. diff --git a/nc_py_api/__init__.py b/nc_py_api/__init__.py index 6676c641..7c9fe8d2 100644 --- a/nc_py_api/__init__.py +++ b/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 diff --git a/nc_py_api/db_requests.py b/nc_py_api/db_requests.py index 4e2fb7ef..273358da 100644 --- a/nc_py_api/db_requests.py +++ b/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.""" diff --git a/nc_py_api/files.py b/nc_py_api/files.py index 33bc8c22..645fc87b 100644 --- a/nc_py_api/files.py +++ b/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"], diff --git a/tests/nc_py_api/files_test.py b/tests/nc_py_api/files_test.py deleted file mode 100644 index 55caa203..00000000 --- a/tests/nc_py_api/files_test.py +++ /dev/null @@ -1,38 +0,0 @@ -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... diff --git a/tests/nc_py_api/fs_test.py b/tests/nc_py_api/fs_test.py new file mode 100644 index 00000000..4e8eaeea --- /dev/null +++ b/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) diff --git a/tests/nc_py_api/mimetype_test.py b/tests/nc_py_api/mimetype_test.py index ec31b9da..d8c1e899 100644 --- a/tests/nc_py_api/mimetype_test.py +++ b/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) diff --git a/tests/nc_py_api/occ_test.py b/tests/nc_py_api/occ_test.py index 972864b8..e00187db 100644 --- a/tests/nc_py_api/occ_test.py +++ b/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") diff --git a/tests/nc_py_api/test_dir/test.txt b/tests/nc_py_api/test_dir/test.txt new file mode 100644 index 00000000..524acfff --- /dev/null +++ b/tests/nc_py_api/test_dir/test.txt @@ -0,0 +1 @@ +Test file diff --git "a/tests/nc_py_api/test_dir/\345\244\215\346\235\202 \347\233\256\345\275\225 \303\215/empty_file.bin" "b/tests/nc_py_api/test_dir/\345\244\215\346\235\202 \347\233\256\345\275\225 \303\215/empty_file.bin" new file mode 100644 index 00000000..e69de29b