Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions nc_py_api/files/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,8 @@ def __init__(self, **kwargs):
self.creation_date = kwargs.get("creation_date", datetime.datetime(1970, 1, 1))
except (ValueError, TypeError):
self.creation_date = datetime.datetime(1970, 1, 1)
self._download_url: str = kwargs.get("download_url", "")
self._download_url_expiration: int = kwargs.get("download_url_expiration", 0)
Comment thread
oleksandr-nc marked this conversation as resolved.
self._trashbin: dict[str, str | int] = {}
for i in ("trashbin_filename", "trashbin_original_location", "trashbin_deletion_time"):
if i in kwargs:
Expand All @@ -132,6 +134,20 @@ def permissions(self) -> str:
"""Permissions for the object."""
return self._raw_data["permissions"]

@property
def download_url(self) -> str:
"""S3 presigned URL for direct download, bypassing Nextcloud.

Only available when the storage backend is S3 with ``use_presigned_url`` enabled.
Empty string when not available.
"""
return self._download_url

@property
def download_url_expiration(self) -> int:
"""Expiration timestamp for :py:attr:`download_url`. Zero when not available."""
return self._download_url_expiration

@property
def last_modified(self) -> datetime.datetime:
"""Time when the object was last modified.
Expand Down
9 changes: 8 additions & 1 deletion nc_py_api/files/_files.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""Helper functions for **FilesAPI** and **AsyncFilesAPI** classes."""

import contextlib
import enum
from datetime import datetime, timezone
from io import BytesIO
Expand All @@ -26,6 +27,7 @@
"oc:id",
"oc:fileid",
"oc:downloadURL",
"nc:download-url-expiration",
"oc:dDC",
"oc:permissions",
"oc:checksums",
Expand Down Expand Up @@ -306,6 +308,11 @@ def _parse_record(full_path: str, prop_stats: list[dict]) -> FsNode: # noqa pyl
fs_node_args["mimetype"] = prop["d:getcontenttype"]
if "oc:permissions" in prop_keys:
fs_node_args["permissions"] = prop["oc:permissions"]
if "oc:downloadURL" in prop_keys and prop["oc:downloadURL"]:
fs_node_args["download_url"] = prop["oc:downloadURL"]
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
if "nc:download-url-expiration" in prop_keys and prop["nc:download-url-expiration"]:
with contextlib.suppress(TypeError, ValueError):
fs_node_args["download_url_expiration"] = int(prop["nc:download-url-expiration"])
if "oc:favorite" in prop_keys:
fs_node_args["favorite"] = bool(int(prop["oc:favorite"]))
if "nc:trashbin-filename" in prop_keys:
Expand Down Expand Up @@ -367,7 +374,7 @@ def _webdav_response_to_records(webdav_res: Response, info: str) -> list[dict]:
if "d:error" in response_data:
err = response_data["d:error"]
raise NextcloudException(
reason=f'{err["s:exception"]}: {err["s:message"]}'.replace("\n", ""), info=info, response=webdav_res
reason=f"{err['s:exception']}: {err['s:message']}".replace("\n", ""), info=info, response=webdav_res
)
response = response_data["d:multistatus"].get("d:response", [])
return [response] if isinstance(response, dict) else response
92 changes: 92 additions & 0 deletions tests_unit/test_download_url.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""Tests for S3 presigned download URL properties on FsNode/FsNodeInfo."""

from nc_py_api.files import FsNode, FsNodeInfo
from nc_py_api.files._files import _parse_record


def test_fsnode_info_download_url_defaults():
info = FsNodeInfo()
assert info.download_url == ""
assert info.download_url_expiration == 0


def test_fsnode_info_download_url_with_values():
info = FsNodeInfo(download_url="https://s3.example.com/bucket/obj?sig=abc", download_url_expiration=1700000000)
assert info.download_url == "https://s3.example.com/bucket/obj?sig=abc"
assert info.download_url_expiration == 1700000000


def test_fsnode_passes_download_url_to_info():
node = FsNode(
"files/admin/test.txt", file_id="00000123", download_url="https://s3.test/f", download_url_expiration=9999
)
assert node.info.download_url == "https://s3.test/f"
assert node.info.download_url_expiration == 9999


def test_parse_record_with_download_url():
prop_stat = {
"d:status": "HTTP/1.1 200 OK",
"d:prop": {
"oc:id": "00000123",
"oc:fileid": "123",
"oc:permissions": "RGDNVW",
"d:getetag": '"abc123"',
"oc:downloadURL": "https://s3.example.com/bucket/urn:oid:123?X-Amz-Signature=abc",
"nc:download-url-expiration": "1700000000",
},
}
node = _parse_record("files/admin/test.txt", [prop_stat])
assert node.info.download_url == "https://s3.example.com/bucket/urn:oid:123?X-Amz-Signature=abc"
assert node.info.download_url_expiration == 1700000000


def test_parse_record_without_download_url():
prop_stat = {
"d:status": "HTTP/1.1 200 OK",
"d:prop": {
"oc:id": "00000123",
"oc:fileid": "123",
"oc:permissions": "RGDNVW",
"d:getetag": '"abc123"',
},
}
node = _parse_record("files/admin/test.txt", [prop_stat])
assert node.info.download_url == ""
assert node.info.download_url_expiration == 0


def test_parse_record_with_false_download_url():
"""When storage doesn't support presigned URLs, server returns 'false'."""
prop_stat = {
"d:status": "HTTP/1.1 200 OK",
"d:prop": {
"oc:id": "00000123",
"oc:fileid": "123",
"oc:permissions": "RGDNVW",
"d:getetag": '"abc123"',
"oc:downloadURL": False,
"nc:download-url-expiration": False,
},
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
node = _parse_record("files/admin/test.txt", [prop_stat])
assert node.info.download_url == ""
assert node.info.download_url_expiration == 0


def test_parse_record_with_malformed_expiration():
"""Malformed expiration should not crash parsing."""
prop_stat = {
"d:status": "HTTP/1.1 200 OK",
"d:prop": {
"oc:id": "00000123",
"oc:fileid": "123",
"oc:permissions": "RGDNVW",
"d:getetag": '"abc123"',
"oc:downloadURL": "https://s3.test/f",
"nc:download-url-expiration": "not-a-number",
},
}
node = _parse_record("files/admin/test.txt", [prop_stat])
assert node.info.download_url == "https://s3.test/f"
assert node.info.download_url_expiration == 0
Loading