cceec9c4fb
- Drop SMB/SAMBA subsystem completely: app/samba_access.py, app/path_settings.py, app/path_store.py, app/pipeline/file_manager.py and their tests, plus the smbprotocol dependency and all related UI (login/password fields, path editor, SMB diagnostics) in main.py. - run_batch now takes a .zip attached via Streamlit's file_uploader (same as the local dev tool) instead of reading from SMB or a fictitious OpenCloud "Attach" folder; renamed from run_local_zip_batch since it's now the single canonical batch entrypoint used by both main.py and scripts/local_zip_test_app.py. - Fixed run number resolution: added next_opencloud_run_number() which lists the Out/year/month/day folder via WebDAV PROPFIND, so repeated runs land in incrementing subfolders instead of always overwriting "1". - Added a self-service "Добавить токен" popover in main.py so users can link their own OpenCloud App Token to their Keycloak account (opencloud_token_store) without an admin running the CLI script; reworded the missing-token error to point at it. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
249 lines
8.8 KiB
Python
249 lines
8.8 KiB
Python
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from unittest.mock import Mock, patch
|
||
|
||
import pytest
|
||
|
||
from app.opencloud_access import (
|
||
OpenCloudUploadError,
|
||
build_opencloud_url,
|
||
ensure_opencloud_folder,
|
||
next_opencloud_run_number,
|
||
resolve_personal_drive_id,
|
||
upload_local_file_to_opencloud,
|
||
)
|
||
|
||
|
||
def test_build_opencloud_url_encodes_cyrillic_and_spaces() -> None:
|
||
url = build_opencloud_url(
|
||
space_id="storage123$space456",
|
||
folder_segments=("Райков М.С.",),
|
||
file_name="report 2026.xlsx",
|
||
)
|
||
assert url == (
|
||
"https://docs.raisa.go.rshbank.ru/dav/spaces/storage123$space456/"
|
||
"%D0%A0%D0%B0%D0%B9%D0%BA%D0%BE%D0%B2%20%D0%9C.%D0%A1./"
|
||
"report%202026.xlsx"
|
||
)
|
||
|
||
|
||
def test_build_opencloud_url_without_folder() -> None:
|
||
url = build_opencloud_url(
|
||
space_id="abc", folder_segments=(), file_name="report.xlsx"
|
||
)
|
||
assert url == "https://docs.raisa.go.rshbank.ru/dav/spaces/abc/report.xlsx"
|
||
|
||
|
||
def test_build_opencloud_url_requires_space_id() -> None:
|
||
with pytest.raises(OpenCloudUploadError):
|
||
build_opencloud_url(space_id="", folder_segments=(), file_name="report.xlsx")
|
||
|
||
|
||
def test_upload_local_file_to_opencloud_success(tmp_path: Path) -> None:
|
||
local_file = tmp_path / "report.xlsx"
|
||
local_file.write_bytes(b"fake-xlsx-bytes")
|
||
|
||
fake_response = Mock(status_code=201, text="")
|
||
with patch("app.opencloud_access.requests.put", return_value=fake_response) as put:
|
||
url = upload_local_file_to_opencloud(
|
||
local_file,
|
||
space_id="abc$def",
|
||
folder_segments=("Райков М.С.",),
|
||
file_name="report.xlsx",
|
||
username="raykovmse",
|
||
token="coherent effects plausibly apple arming chef",
|
||
)
|
||
|
||
assert url.endswith("/report.xlsx")
|
||
_, kwargs = put.call_args
|
||
# Basic Auth (App Token), не Bearer — см. app/opencloud_access.py
|
||
assert "Authorization" not in kwargs["headers"]
|
||
auth = kwargs["auth"]
|
||
assert auth.username == "raykovmse"
|
||
assert auth.password == "coherent effects plausibly apple arming chef"
|
||
|
||
|
||
def test_upload_local_file_to_opencloud_missing_token(tmp_path: Path) -> None:
|
||
local_file = tmp_path / "report.xlsx"
|
||
local_file.write_bytes(b"data")
|
||
with pytest.raises(OpenCloudUploadError):
|
||
upload_local_file_to_opencloud(
|
||
local_file,
|
||
space_id="abc",
|
||
folder_segments=(),
|
||
file_name="report.xlsx",
|
||
username="raykovmse",
|
||
token="",
|
||
)
|
||
|
||
|
||
def test_upload_local_file_to_opencloud_missing_username(tmp_path: Path) -> None:
|
||
local_file = tmp_path / "report.xlsx"
|
||
local_file.write_bytes(b"data")
|
||
with pytest.raises(OpenCloudUploadError):
|
||
upload_local_file_to_opencloud(
|
||
local_file,
|
||
space_id="abc",
|
||
folder_segments=(),
|
||
file_name="report.xlsx",
|
||
username="",
|
||
token="token-123",
|
||
)
|
||
|
||
|
||
def test_upload_local_file_to_opencloud_missing_local_file(tmp_path: Path) -> None:
|
||
missing_file = tmp_path / "does-not-exist.xlsx"
|
||
with pytest.raises(OpenCloudUploadError):
|
||
upload_local_file_to_opencloud(
|
||
missing_file,
|
||
space_id="abc",
|
||
folder_segments=(),
|
||
file_name="report.xlsx",
|
||
username="raykovmse",
|
||
token="token-123",
|
||
)
|
||
|
||
|
||
def test_resolve_personal_drive_id_returns_personal_drive() -> None:
|
||
fake_response = Mock(
|
||
status_code=200,
|
||
json=lambda: {
|
||
"value": [
|
||
{"id": "storage1$shared1", "driveType": "project"},
|
||
{"id": "storage1$personal1", "driveType": "personal"},
|
||
]
|
||
},
|
||
)
|
||
with patch("app.opencloud_access.requests.get", return_value=fake_response) as get:
|
||
drive_id = resolve_personal_drive_id("raykovmse", "app token here")
|
||
|
||
assert drive_id == "storage1$personal1"
|
||
args, kwargs = get.call_args
|
||
assert args[0] == "https://docs.raisa.go.rshbank.ru/graph/v1.0/me/drives"
|
||
auth = kwargs["auth"]
|
||
assert auth.username == "raykovmse"
|
||
assert auth.password == "app token here"
|
||
|
||
|
||
def test_resolve_personal_drive_id_missing_personal_drive_raises() -> None:
|
||
fake_response = Mock(
|
||
status_code=200,
|
||
json=lambda: {"value": [{"id": "storage1$shared1", "driveType": "project"}]},
|
||
)
|
||
with patch("app.opencloud_access.requests.get", return_value=fake_response):
|
||
with pytest.raises(OpenCloudUploadError):
|
||
resolve_personal_drive_id("raykovmse", "app token here")
|
||
|
||
|
||
def test_resolve_personal_drive_id_http_error_raises() -> None:
|
||
fake_response = Mock(status_code=401, text="Unauthorized")
|
||
with patch("app.opencloud_access.requests.get", return_value=fake_response):
|
||
with pytest.raises(OpenCloudUploadError):
|
||
resolve_personal_drive_id("raykovmse", "app token here")
|
||
|
||
|
||
def test_resolve_personal_drive_id_requires_username_and_token() -> None:
|
||
with pytest.raises(OpenCloudUploadError):
|
||
resolve_personal_drive_id("", "app token here")
|
||
with pytest.raises(OpenCloudUploadError):
|
||
resolve_personal_drive_id("raykovmse", "")
|
||
|
||
|
||
def test_ensure_opencloud_folder_creates_each_segment_via_mkcol() -> None:
|
||
fake_response = Mock(status_code=201, text="")
|
||
with patch(
|
||
"app.opencloud_access.requests.request", return_value=fake_response
|
||
) as request:
|
||
ensure_opencloud_folder(
|
||
"abc$def",
|
||
("Out", "2026", "9", "17", "1"),
|
||
username="raykovmse",
|
||
token="app token here",
|
||
)
|
||
|
||
assert request.call_count == 5
|
||
method, url = request.call_args_list[-1].args
|
||
assert method == "MKCOL"
|
||
assert url == (
|
||
"https://docs.raisa.go.rshbank.ru/dav/spaces/abc$def/" "Out/2026/9/17/1"
|
||
)
|
||
|
||
|
||
def test_ensure_opencloud_folder_tolerates_already_existing_collection() -> None:
|
||
fake_response = Mock(status_code=405, text="")
|
||
with patch("app.opencloud_access.requests.request", return_value=fake_response):
|
||
ensure_opencloud_folder(
|
||
"abc", ("Out",), username="raykovmse", token="token-123"
|
||
)
|
||
|
||
|
||
def test_ensure_opencloud_folder_raises_on_unexpected_status() -> None:
|
||
fake_response = Mock(status_code=500, text="boom")
|
||
with patch("app.opencloud_access.requests.request", return_value=fake_response):
|
||
with pytest.raises(OpenCloudUploadError):
|
||
ensure_opencloud_folder(
|
||
"abc", ("Out",), username="raykovmse", token="token-123"
|
||
)
|
||
|
||
|
||
def test_next_opencloud_run_number_finds_max_plus_one() -> None:
|
||
body = """<?xml version="1.0"?>
|
||
<d:multistatus xmlns:d="DAV:">
|
||
<d:response><d:href>/dav/spaces/abc/Out/2026/9/17/</d:href></d:response>
|
||
<d:response><d:href>/dav/spaces/abc/Out/2026/9/17/1/</d:href></d:response>
|
||
<d:response><d:href>/dav/spaces/abc/Out/2026/9/17/2/</d:href></d:response>
|
||
</d:multistatus>"""
|
||
fake_response = Mock(status_code=207, content=body.encode("utf-8"))
|
||
with patch(
|
||
"app.opencloud_access.requests.request", return_value=fake_response
|
||
) as request:
|
||
run_number = next_opencloud_run_number(
|
||
"abc", ("Out", "2026", "9", "17"), username="raykovmse", token="token-123"
|
||
)
|
||
|
||
assert run_number == 3
|
||
method, url = request.call_args.args
|
||
assert method == "PROPFIND"
|
||
assert url == "https://docs.raisa.go.rshbank.ru/dav/spaces/abc/Out/2026/9/17"
|
||
assert request.call_args.kwargs["headers"] == {"Depth": "1"}
|
||
|
||
|
||
def test_next_opencloud_run_number_missing_day_folder_returns_one() -> None:
|
||
fake_response = Mock(status_code=404, text="Not Found")
|
||
with patch("app.opencloud_access.requests.request", return_value=fake_response):
|
||
run_number = next_opencloud_run_number(
|
||
"abc", ("Out", "2026", "9", "17"), username="raykovmse", token="token-123"
|
||
)
|
||
|
||
assert run_number == 1
|
||
|
||
|
||
def test_next_opencloud_run_number_http_error_raises() -> None:
|
||
fake_response = Mock(status_code=500, text="boom")
|
||
with patch("app.opencloud_access.requests.request", return_value=fake_response):
|
||
with pytest.raises(OpenCloudUploadError):
|
||
next_opencloud_run_number(
|
||
"abc",
|
||
("Out", "2026", "9", "17"),
|
||
username="raykovmse",
|
||
token="token-123",
|
||
)
|
||
|
||
|
||
def test_upload_local_file_to_opencloud_http_error(tmp_path: Path) -> None:
|
||
local_file = tmp_path / "report.xlsx"
|
||
local_file.write_bytes(b"data")
|
||
|
||
fake_response = Mock(status_code=401, text="Unauthorized")
|
||
with patch("app.opencloud_access.requests.put", return_value=fake_response):
|
||
with pytest.raises(OpenCloudUploadError):
|
||
upload_local_file_to_opencloud(
|
||
local_file,
|
||
space_id="abc",
|
||
folder_segments=(),
|
||
file_name="report.xlsx",
|
||
username="raykovmse",
|
||
token="token-123",
|
||
)
|