218 lines
7.5 KiB
Python
218 lines
7.5 KiB
Python
"""Ссылки на загруженные файлы в веб-интерфейсе OpenCloud.
|
|
|
|
Цепочка: заголовок OC-FileId ответа на PUT -> PROPFIND (Depth: 0) за
|
|
oc:privatelink / oc:fileid -> None (запасную ссылку подставляет service).
|
|
Поведение реального сервера не проверено, поэтому ни один шаг не должен
|
|
ронять загрузку.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
import requests
|
|
|
|
from app.opencloud_access import (
|
|
build_opencloud_folder_url,
|
|
build_opencloud_private_link,
|
|
build_opencloud_web_folder_url,
|
|
resolve_opencloud_file_id,
|
|
resolve_opencloud_web_link,
|
|
resolve_personal_drive,
|
|
upload_local_file_to_opencloud,
|
|
)
|
|
|
|
DAV_URL = "https://docs.raisa.go.rshbank.ru/dav/spaces/abc/Out/report.xlsx"
|
|
|
|
|
|
def _propfind_body(privatelink: str = "", fileid: str = "") -> bytes:
|
|
found = []
|
|
missing = []
|
|
for name, value in (("privatelink", privatelink), ("fileid", fileid)):
|
|
if value:
|
|
found.append(f"<oc:{name}>{value}</oc:{name}>")
|
|
else:
|
|
missing.append(f"<oc:{name}/>")
|
|
return f"""<?xml version="1.0"?>
|
|
<d:multistatus xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">
|
|
<d:response>
|
|
<d:href>/dav/spaces/abc/Out/report.xlsx</d:href>
|
|
<d:propstat><d:prop>{''.join(found)}</d:prop>
|
|
<d:status>HTTP/1.1 200 OK</d:status></d:propstat>
|
|
<d:propstat><d:prop>{''.join(missing)}</d:prop>
|
|
<d:status>HTTP/1.1 404 Not Found</d:status></d:propstat>
|
|
</d:response>
|
|
</d:multistatus>""".encode()
|
|
|
|
|
|
def _upload(tmp_path: Path, put_response: Mock):
|
|
local_file = tmp_path / "report.xlsx"
|
|
local_file.write_bytes(b"xlsx")
|
|
with patch("app.opencloud_access.requests.put", return_value=put_response):
|
|
return upload_local_file_to_opencloud(
|
|
local_file,
|
|
space_id="abc",
|
|
folder_segments=("Out",),
|
|
file_name="report.xlsx",
|
|
username="raykovmse",
|
|
token="token-123",
|
|
)
|
|
|
|
|
|
def test_upload_takes_file_id_from_put_header_case_insensitively(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
put_response = Mock(
|
|
status_code=201, text="", headers={"oc-fileid": "store$space!opaque"}
|
|
)
|
|
with patch("app.opencloud_access.requests.request") as request:
|
|
result = _upload(tmp_path, put_response)
|
|
|
|
assert result.dav_url == DAV_URL
|
|
assert result.web_url == (
|
|
"https://docs.raisa.go.rshbank.ru/f/store%24space%21opaque"
|
|
)
|
|
request.assert_not_called()
|
|
|
|
|
|
def test_upload_without_header_uses_privatelink_from_propfind_as_is(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
put_response = Mock(status_code=204, text="", headers={"ETag": '"1"'})
|
|
propfind_response = Mock(
|
|
status_code=207,
|
|
content=_propfind_body(
|
|
privatelink="https://docs.raisa.go.rshbank.ru/f/from-server",
|
|
fileid="ignored",
|
|
),
|
|
)
|
|
with patch(
|
|
"app.opencloud_access.requests.request", return_value=propfind_response
|
|
) as request:
|
|
result = _upload(tmp_path, put_response)
|
|
|
|
assert result.web_url == "https://docs.raisa.go.rshbank.ru/f/from-server"
|
|
request.assert_called_once()
|
|
method, url = request.call_args.args
|
|
assert method == "PROPFIND"
|
|
assert url == DAV_URL
|
|
kwargs = request.call_args.kwargs
|
|
assert kwargs["headers"]["Depth"] == "0"
|
|
body = kwargs["data"].decode("utf-8")
|
|
assert 'xmlns:oc="http://owncloud.org/ns"' in body
|
|
assert "<oc:privatelink/>" in body
|
|
assert "<oc:fileid/>" in body
|
|
assert kwargs["auth"].username == "raykovmse"
|
|
|
|
|
|
def test_propfind_without_privatelink_builds_link_from_fileid() -> None:
|
|
propfind_response = Mock(status_code=207, content=_propfind_body(fileid="s$p!f"))
|
|
with patch("app.opencloud_access.requests.request", return_value=propfind_response):
|
|
link = resolve_opencloud_web_link(DAV_URL, "raykovmse", "token-123")
|
|
|
|
assert link == build_opencloud_private_link("s$p!f")
|
|
|
|
|
|
def test_upload_survives_when_no_link_can_be_obtained(tmp_path: Path) -> None:
|
|
put_response = Mock(status_code=201, text="", headers={})
|
|
with patch(
|
|
"app.opencloud_access.requests.request",
|
|
side_effect=requests.ConnectionError("down"),
|
|
):
|
|
result = _upload(tmp_path, put_response)
|
|
|
|
assert result.dav_url == DAV_URL
|
|
assert result.web_url is None
|
|
|
|
|
|
def test_propfind_http_error_gives_no_link() -> None:
|
|
with patch(
|
|
"app.opencloud_access.requests.request",
|
|
return_value=Mock(status_code=403, content=b""),
|
|
):
|
|
assert resolve_opencloud_web_link(DAV_URL, "u", "t") is None
|
|
|
|
|
|
def test_propfind_broken_xml_gives_no_link() -> None:
|
|
with patch(
|
|
"app.opencloud_access.requests.request",
|
|
return_value=Mock(status_code=207, content=b"<not xml"),
|
|
):
|
|
assert resolve_opencloud_web_link(DAV_URL, "u", "t") is None
|
|
|
|
|
|
def test_propfind_with_empty_properties_gives_no_link() -> None:
|
|
with patch(
|
|
"app.opencloud_access.requests.request",
|
|
return_value=Mock(status_code=207, content=_propfind_body()),
|
|
):
|
|
assert resolve_opencloud_web_link(DAV_URL, "u", "t") is None
|
|
|
|
|
|
def test_build_opencloud_folder_url_encodes_segments() -> None:
|
|
assert build_opencloud_folder_url("abc$def", ("Out", "2026", "Райков")) == (
|
|
"https://docs.raisa.go.rshbank.ru/dav/spaces/abc$def/Out/2026/"
|
|
"%D0%A0%D0%B0%D0%B9%D0%BA%D0%BE%D0%B2"
|
|
)
|
|
|
|
|
|
def test_web_folder_url_matches_browser_address_bar() -> None:
|
|
"""Эталон — адрес папки запуска из браузера на нашем OpenCloud (29.09),
|
|
без параметров отображения (items-per-page, view-mode, tiles-size)."""
|
|
file_id = (
|
|
"28c0cecf-1e20-4664-b165-5c9d5515e70f$1faf47bc-6134-4130-9f38-d4743818ac63"
|
|
"!0dfd7588-9b95-4a74-b477-e25be91eaf49"
|
|
)
|
|
|
|
url = build_opencloud_web_folder_url(
|
|
"personal/raykov-mse", ("Out", "2026", "9", "29", "1"), file_id
|
|
)
|
|
|
|
assert url == (
|
|
"https://docs.raisa.go.rshbank.ru/files/spaces/personal/raykov-mse/"
|
|
"Out/2026/9/29/1?fileId=28c0cecf-1e20-4664-b165-5c9d5515e70f"
|
|
"%241faf47bc-6134-4130-9f38-d4743818ac63"
|
|
"%210dfd7588-9b95-4a74-b477-e25be91eaf49"
|
|
)
|
|
|
|
|
|
def test_web_folder_url_without_file_id_is_plain_path() -> None:
|
|
assert build_opencloud_web_folder_url("personal/raykov-mse", ("Out", "Райков")) == (
|
|
"https://docs.raisa.go.rshbank.ru/files/spaces/personal/raykov-mse/"
|
|
"Out/%D0%A0%D0%B0%D0%B9%D0%BA%D0%BE%D0%B2"
|
|
)
|
|
|
|
|
|
def test_resolve_file_id_reads_oc_fileid() -> None:
|
|
with patch(
|
|
"app.opencloud_access.requests.request",
|
|
return_value=Mock(status_code=207, content=_propfind_body(fileid="s$p!d")),
|
|
):
|
|
assert resolve_opencloud_file_id(DAV_URL, "u", "t") == "s$p!d"
|
|
|
|
|
|
def test_resolve_file_id_survives_network_error() -> None:
|
|
with patch(
|
|
"app.opencloud_access.requests.request",
|
|
side_effect=requests.ConnectionError("down"),
|
|
):
|
|
assert resolve_opencloud_file_id(DAV_URL, "u", "t") is None
|
|
|
|
|
|
def test_personal_drive_carries_drive_alias() -> None:
|
|
response = Mock(
|
|
status_code=200,
|
|
json=lambda: {
|
|
"value": [
|
|
{
|
|
"id": "s$p",
|
|
"driveType": "personal",
|
|
"driveAlias": "personal/raykov-mse",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
with patch("app.opencloud_access.requests.get", return_value=response):
|
|
assert resolve_personal_drive("raykov-mse", "t").alias == "personal/raykov-mse"
|