diff --git a/app/app_config.py b/app/app_config.py index 463aa44..0afdbbe 100644 --- a/app/app_config.py +++ b/app/app_config.py @@ -10,14 +10,6 @@ SFM_SMB_BASE_PATH = "//sgo-fc01-r06/sfm/ОТЧЕТ_7081У" SFM_DEFAULT_INPUT_DIR = r"//sgo-fc01-r06/sfm/ОТЧЕТ_7081У" ACCESS_SUPPORT_CONTACT = "администратору СФМ" -# TODO(BIQ-36593): подтвердить SPACE_ID для прода (сейчас — значение, -# проверенное вручную на тестовом пространстве "AURORA"). По наблюдениям -# ID может быть составным ("storageId$spaceId") — храним как есть, без -# разбора, через env-переменную, чтобы не хардкодить прод-значение в коде. -OPENCLOUD_SPACE_ID = os.getenv( - "SFM_OPENCLOUD_SPACE_ID", "28c0cecf-1e20-4664-b165-5c9d5515e70f" -) - EDIT_PATH_OPEN_KEY = "edit_path_open" EDIT_PATH_VALUE_KEY = "edit_path_value" # Runtime-only user paths storage. In bank contour we use tmpfs. @@ -81,24 +73,3 @@ def resolve_opencloud_access_token() -> str | None: except Exception: # noqa: BLE001 # Локальный тест без поднятой БД/переменных окружения — fallback. return os.getenv("SFM_OPENCLOUD_ACCESS_TOKEN") or None - - -def resolve_opencloud_user_folder() -> str | None: - """Возвращает предполагаемое имя личной папки пользователя в OpenCloud. - - ОТКРЫТЫЙ ВОПРОС: неизвестно, откуда в проде берётся имя папки вида - "Райков М.С." — это предположение, что claim "name" из Keycloak уже - приходит в этом формате (совпадает с тем, что руками использовалось в - open_cloud_upload_fixed.py). Альтернатива — папки создаются заранее - администратором по другому принципу (email/preferred_username), это - нужно сверить в проде. - - Для локального теста — fallback на переменную окружения - SFM_OPENCLOUD_FOLDER из .env. - """ - profile = st.session_state.get("user_profile", {}) - if isinstance(profile, dict): - value = profile.get("name") - if value: - return str(value) - return os.getenv("SFM_OPENCLOUD_FOLDER") or None diff --git a/app/opencloud_access.py b/app/opencloud_access.py index 4934331..6546e9e 100644 --- a/app/opencloud_access.py +++ b/app/opencloud_access.py @@ -24,6 +24,63 @@ class OpenCloudUploadError(RuntimeError): """Ошибка загрузки файла в OpenCloud по WebDAV.""" +def resolve_personal_drive_id(username: str, token: str) -> str: + """Находит ID личного пространства пользователя через Graph API. + + Больше не грузим файлы в общее тестовое пространство по фиксированному + SPACE_ID — у каждого заказчика есть личное пространство (driveType == + "personal" в GET /graph/v1.0/me/drives), и его ID нужно получать заново + перед каждой загрузкой: пространство может быть пересоздано, а хранить + потенциально устаревший ID рискованнее, чем сделать лишний HTTP-запрос. + + Авторизация — та же Basic Auth (username + App Token), что и для самой + загрузки по WebDAV, см. upload_local_file_to_opencloud. + """ + if not username: + raise OpenCloudUploadError( + "Не задан username для поиска личного пространства в OpenCloud." + ) + if not token: + raise OpenCloudUploadError( + "Отсутствует App Token для поиска личного пространства в OpenCloud." + ) + + url = f"https://{OPENCLOUD_DOMAIN}/graph/v1.0/me/drives" + try: + response = requests.get( + url, + auth=HTTPBasicAuth(username, token), + timeout=OPENCLOUD_UPLOAD_TIMEOUT_SECONDS, + ) + except requests.RequestException as exc: + raise OpenCloudUploadError( + f"Ошибка сети при поиске личного пространства OpenCloud: {exc}" + ) from exc + + if response.status_code != 200: + raise OpenCloudUploadError( + "Не удалось получить список пространств OpenCloud: " + f"HTTP {response.status_code}, body={response.text[:500]!r}" + ) + + drives = response.json().get("value", []) + personal_drives = [ + drive for drive in drives if drive.get("driveType") == "personal" + ] + if not personal_drives: + raise OpenCloudUploadError( + f"Не найдено личное пространство (driveType='personal') для " + f"пользователя '{username}' в OpenCloud." + ) + + drive_id = personal_drives[0].get("id") + if not drive_id: + raise OpenCloudUploadError( + "Личное пространство найдено в OpenCloud, но у него отсутствует id." + ) + return str(drive_id) + + def build_opencloud_url( space_id: str, folder_segments: tuple[str, ...] | list[str], file_name: str ) -> str: @@ -54,6 +111,47 @@ def build_opencloud_url( return f"https://{OPENCLOUD_DOMAIN}{path}" +def ensure_opencloud_folder( + space_id: str, + folder_segments: tuple[str, ...] | list[str], + username: str, + token: str, +) -> None: + """Создаёт папку folder_segments в личном пространстве через MKCOL. + + WebDAV PUT возвращает 409 Conflict, если хотя бы одна промежуточная + коллекция ещё не существует, поэтому перед первой загрузкой в новый + путь (Out/год/месяц/день/номер_запуска) создаём его сегмент за + сегментом — так же, как ensure_dir_exists делает это для SMB. + """ + segments = [segment for segment in folder_segments if segment] + accumulated: list[str] = [] + for segment in segments: + accumulated.append(segment) + encoded_path = "/".join(quote(part, safe="") for part in accumulated) + url = f"https://{OPENCLOUD_DOMAIN}/dav/spaces/{space_id}/{encoded_path}" + try: + response = requests.request( + "MKCOL", + url, + auth=HTTPBasicAuth(username, token), + timeout=OPENCLOUD_UPLOAD_TIMEOUT_SECONDS, + ) + except requests.RequestException as exc: + raise OpenCloudUploadError( + f"Ошибка сети при создании папки в OpenCloud: url='{url}', {exc}" + ) from exc + + # 201 — папка создана, 405 — уже существует (MKCOL на существующую + # коллекцию запрещён спецификацией WebDAV, это не ошибка для нас). + if response.status_code not in (201, 405): + raise OpenCloudUploadError( + "Не удалось создать папку в OpenCloud: " + f"HTTP {response.status_code}, url='{url}', " + f"body={response.text[:500]!r}" + ) + + def upload_local_file_to_opencloud( local_path: Path, space_id: str, @@ -74,11 +172,9 @@ def upload_local_file_to_opencloud( OpenCloud (несовпадение audience) — App Token через Basic Auth работает, т.к. идёт отдельным от OIDC путём авторизации. - ВРЕМЕННО (тестовый прогон): папку НЕ создаём (без MKCOL) — просто - кладём файл прямо в личную папку участника (folder_segments — как - правило один сегмент, например "Райков М.С."). Структуру Out/год/ - месяц/день/номер_запуска вернём отдельным шагом после того, как - подтвердим на тесте базовый сценарий "файл долетает в нужную папку". + Промежуточные папки (folder_segments) должны быть уже созданы через + ensure_opencloud_folder — сам PUT их не создаёт и вернёт 409, если + хотя бы одна коллекция на пути отсутствует. """ if not username: raise OpenCloudUploadError( diff --git a/app/pipeline/file_manager.py b/app/pipeline/file_manager.py index 5058b84..51ed81c 100644 --- a/app/pipeline/file_manager.py +++ b/app/pipeline/file_manager.py @@ -26,6 +26,9 @@ class RunContext: processed_dir: str run_number: int date_key: str + year: str + month: str + day: str @dataclass @@ -273,7 +276,7 @@ def create_run_context( run_number = next_run_number(out_day_dir, samba_conn) run_dir = join_path(out_day_dir, str(run_number)) ensure_dir_exists(run_dir, samba_conn) - _, _, _, date_key = build_date_parts(now) + year, month, day, date_key = build_date_parts(now) return RunContext( run_dir=run_dir, out_day_dir=out_day_dir, @@ -281,6 +284,9 @@ def create_run_context( processed_dir=processed_dir, run_number=run_number, date_key=date_key, + year=year, + month=month, + day=day, ) diff --git a/app/pipeline/service.py b/app/pipeline/service.py index 4d2e809..9a6eeca 100644 --- a/app/pipeline/service.py +++ b/app/pipeline/service.py @@ -7,8 +7,14 @@ from dataclasses import dataclass, field from datetime import datetime from pathlib import Path -from ..opencloud_access import OpenCloudUploadError, upload_local_file_to_opencloud +from ..opencloud_access import ( + OpenCloudUploadError, + ensure_opencloud_folder, + resolve_personal_drive_id, + upload_local_file_to_opencloud, +) from .config import ( + DEFAULT_OUTPUT_DIR_NAME, build_date_parts, build_process_log_file_name, build_protocol_file_name, @@ -52,8 +58,6 @@ def run_batch( samba_user: str, samba_password: str, launcher: str, - opencloud_space_id: str, - opencloud_folder: str | None, opencloud_username: str | None, opencloud_access_token: str | None, artifacts_dir: str | None = None, @@ -96,17 +100,6 @@ def run_batch( "(проверьте таблицу токенов)." ], ) - if not opencloud_folder: - return BatchResult( - status="error", - processed=0, - partial=0, - errors=1, - report_path="", - protocol_path="", - log_path="", - messages=["Не удалось определить личную папку пользователя в OpenCloud."], - ) samba_conn = connect_samba( smb_base_path=smb_base_path, @@ -256,10 +249,12 @@ def run_batch( report_file_name=report_file_name, protocol_file_name=protocol_file_name, log_file_name=log_file_name, - opencloud_space_id=opencloud_space_id, - opencloud_folder=opencloud_folder, opencloud_username=opencloud_username, opencloud_access_token=opencloud_access_token, + year=run_context.year, + month=run_context.month, + day=run_context.day, + run_number=run_context.run_number, ) except OpenCloudUploadError as exc: return BatchResult( @@ -312,8 +307,6 @@ def run_batch( def run_local_zip_batch( zip_path: Path, launcher: str, - opencloud_space_id: str, - opencloud_folder: str | None, opencloud_username: str | None, opencloud_access_token: str | None, ) -> BatchResult: @@ -352,17 +345,6 @@ def run_local_zip_batch( "(проверьте таблицу токенов)." ], ) - if not opencloud_folder: - return BatchResult( - status="error", - processed=0, - partial=0, - errors=1, - report_path="", - protocol_path="", - log_path="", - messages=["Не удалось определить личную папку пользователя в OpenCloud."], - ) protocol_rows: list[ProtocolRow] = [] log_lines: list[str] = [f"{_now()} | local zip batch start: {zip_path.name}"] @@ -452,7 +434,7 @@ def run_local_zip_batch( ) log_lines.append(f"{_now()} | empty batch") - _, _, _, date_key = build_date_parts() + year, month, day, date_key = build_date_parts() run_number = 1 report_file_name = build_report_file_name(date_key, run_number) protocol_file_name = build_protocol_file_name(date_key, run_number) @@ -476,10 +458,12 @@ def run_local_zip_batch( report_file_name=report_file_name, protocol_file_name=protocol_file_name, log_file_name=log_file_name, - opencloud_space_id=opencloud_space_id, - opencloud_folder=opencloud_folder, opencloud_username=opencloud_username, opencloud_access_token=opencloud_access_token, + year=year, + month=month, + day=day, + run_number=run_number, ) except OpenCloudUploadError as exc: return BatchResult( @@ -691,22 +675,30 @@ def _upload_artifacts( report_file_name: str, protocol_file_name: str, log_file_name: str, - opencloud_space_id: str, - opencloud_folder: str, opencloud_username: str, opencloud_access_token: str, + year: str, + month: str, + day: str, + run_number: int, ) -> tuple[str, str, str]: - # ВРЕМЕННО (тестовый прогон): грузим плоско, прямо в личную папку - # участника, без MKCOL и без структуры Out/год/месяц/день/номер запуска. - # Сначала проверяем на тесте, что файл вообще долетает и попадает в - # нужную папку — структуру подпапок вернём отдельным шагом. + # Личное пространство ищем заново перед каждой загрузкой (не кешируем + # ID) — см. resolve_personal_drive_id. Кладём файлы по той же структуре + # Out/год/месяц/день/номер_запуска, что и на SMB (RunContext.run_dir). # # opencloud_access_token здесь — App Token заказчика (не Bearer/ID-токен), # авторизация идёт через HTTP Basic Auth вместе с opencloud_username. - folder_segments = (opencloud_folder,) + space_id = resolve_personal_drive_id(opencloud_username, opencloud_access_token) + folder_segments = (DEFAULT_OUTPUT_DIR_NAME, year, month, day, str(run_number)) + ensure_opencloud_folder( + space_id, + folder_segments, + username=opencloud_username, + token=opencloud_access_token, + ) report_url = upload_local_file_to_opencloud( local_report, - space_id=opencloud_space_id, + space_id=space_id, folder_segments=folder_segments, file_name=report_file_name, username=opencloud_username, @@ -714,7 +706,7 @@ def _upload_artifacts( ) protocol_url = upload_local_file_to_opencloud( local_protocol, - space_id=opencloud_space_id, + space_id=space_id, folder_segments=folder_segments, file_name=protocol_file_name, username=opencloud_username, @@ -722,7 +714,7 @@ def _upload_artifacts( ) log_url = upload_local_file_to_opencloud( local_log, - space_id=opencloud_space_id, + space_id=space_id, folder_segments=folder_segments, file_name=log_file_name, username=opencloud_username, diff --git a/main.py b/main.py index 61f84cd..d938ffd 100644 --- a/main.py +++ b/main.py @@ -3,10 +3,8 @@ from dotenv import load_dotenv from app.app_config import ( ACCESS_SUPPORT_CONTACT, - OPENCLOUD_SPACE_ID, SFM_SMB_BASE_PATH, resolve_opencloud_access_token, - resolve_opencloud_user_folder, resolve_opencloud_username, resolve_user_id, ) @@ -87,8 +85,6 @@ def main() -> None: samba_user=samba_user, samba_password=samba_password, launcher=launcher_name, - opencloud_space_id=OPENCLOUD_SPACE_ID, - opencloud_folder=resolve_opencloud_user_folder(), opencloud_username=resolve_opencloud_username(), opencloud_access_token=resolve_opencloud_access_token(), artifacts_dir=resolved_input_dir, diff --git a/scripts/local_opencloud_smoke.py b/scripts/local_opencloud_smoke.py deleted file mode 100644 index 0efd1dc..0000000 --- a/scripts/local_opencloud_smoke.py +++ /dev/null @@ -1,55 +0,0 @@ -"""Локальный ручной тест: чисто attach в OpenCloud, без Streamlit. - -Использование: - 1. Впишите ниже ACCESS_TOKEN, SPACE_ID (если отличается от дефолта), - FOLDER и путь к тестовому файлу. - 2. Запустите: python scripts/local_opencloud_smoke.py - -Использует тот же app/opencloud_access.py, что и продакшн-код в -app/pipeline/service.py — так что если тут сработает, значит и в -приложении сработает (сама логика загрузки не отличается). -""" - -from __future__ import annotations - -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).resolve().parents[1])) - -from app.opencloud_access import OpenCloudUploadError, upload_local_file_to_opencloud - -# --- Заполните перед запуском --- -USERNAME = "PASTE_USERNAME_HERE" # логин, под которым создан App Token в OpenCloud -APP_TOKEN = "PASTE_APP_TOKEN_HERE" # App Token из веб-интерфейса OpenCloud (6 слов через пробел) -SPACE_ID = "28c0cecf-1e20-4664-b165-5c9d5515e70f" # тестовое пространство AURORA -FOLDER = "Райков М.С." # ваша личная папка в OpenCloud -LOCAL_FILE = Path(__file__).resolve().parent / "test_upload.xlsx" -REMOTE_FILE_NAME = "test_upload.xlsx" -# --------------------------------- - - -def main() -> None: - if not LOCAL_FILE.exists(): - LOCAL_FILE.write_bytes(b"local opencloud smoke test") - print(f"Тестовый файл создан: {LOCAL_FILE}") - - try: - url = upload_local_file_to_opencloud( - LOCAL_FILE, - space_id=SPACE_ID, - folder_segments=(FOLDER,), - file_name=REMOTE_FILE_NAME, - username=USERNAME, - token=APP_TOKEN, - ) - except OpenCloudUploadError as exc: - print(f"ОШИБКА: {exc}") - raise SystemExit(1) from exc - - print(f"OK, загружено: {url}") - print("Проверьте файл в веб-интерфейсе OpenCloud, в папке", FOLDER) - - -if __name__ == "__main__": - main() diff --git a/scripts/local_zip_test_app.py b/scripts/local_zip_test_app.py index 68e9ad7..9ee3d20 100644 --- a/scripts/local_zip_test_app.py +++ b/scripts/local_zip_test_app.py @@ -8,8 +8,10 @@ Запуск: streamlit run scripts/local_zip_test_app.py -Токен/папку можно один раз прописать в .env (см. .env.example) — поля -на странице предзаполнятся, но их всегда можно поменять руками. +Username/токен можно один раз прописать в .env (см. .env.example) — поля +на странице предзаполнятся, но их всегда можно поменять руками. Папка не +запрашивается — файлы летят в личное пространство пользователя (id +находится через GET /graph/v1.0/me/drives, см. app/opencloud_access.py). """ from __future__ import annotations @@ -24,7 +26,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import streamlit as st from dotenv import load_dotenv -from app.app_config import OPENCLOUD_SPACE_ID from app.pipeline.service import run_local_zip_batch load_dotenv() @@ -53,15 +54,6 @@ access_token = st.text_input( "SFM_OPENCLOUD_ACCESS_TOKEN." ), ) -space_id = st.text_input( - "SPACE_ID", - value=os.getenv("SFM_OPENCLOUD_SPACE_ID", OPENCLOUD_SPACE_ID), -) -folder = st.text_input( - "Папка (личная папка участника в OpenCloud)", - value=os.getenv("SFM_OPENCLOUD_FOLDER", ""), - placeholder="Райков М.С.", -) launcher = st.text_input("Запустил (для протокола)", value="local-test") uploaded_zip = st.file_uploader("Архив с XML-файлами (.zip)", type=["zip"]) @@ -71,8 +63,6 @@ if st.button("Начать обработку", type="primary", use_container_wi st.error("Укажите username.") elif not access_token: st.error("Укажите App Token.") - elif not folder: - st.error("Укажите папку.") elif uploaded_zip is None: st.error("Прикрепите .zip-архив.") else: @@ -83,8 +73,6 @@ if st.button("Начать обработку", type="primary", use_container_wi result = run_local_zip_batch( zip_path=zip_path, launcher=launcher or "local-test", - opencloud_space_id=space_id, - opencloud_folder=folder, opencloud_username=username, opencloud_access_token=access_token, ) diff --git a/tests/smoke/test_batch_smoke.py b/tests/smoke/test_batch_smoke.py index 6a344ba..696b97d 100644 --- a/tests/smoke/test_batch_smoke.py +++ b/tests/smoke/test_batch_smoke.py @@ -93,6 +93,15 @@ def _fake_opencloud_upload( return f"https://opencloud.test/{'/'.join(folder_segments)}/{file_name}" +def _fake_resolve_personal_drive_id(username, token) -> str: + _ = (username, token) + return "personal-space-1" + + +def _fake_ensure_opencloud_folder(space_id, folder_segments, username, token) -> None: + _ = (space_id, folder_segments, username, token) + + def test_run_batch_smoke_and_idempotency(tmp_path: Path, monkeypatch) -> None: samba = FakeSamba(tmp_path) in_dir_local = tmp_path / "Exchange" / "SMBDEMO" / "test" @@ -110,6 +119,12 @@ def test_run_batch_smoke_and_idempotency(tmp_path: Path, monkeypatch) -> None: monkeypatch.setattr( service, "upload_local_file_to_opencloud", _fake_opencloud_upload ) + monkeypatch.setattr( + service, "resolve_personal_drive_id", _fake_resolve_personal_drive_id + ) + monkeypatch.setattr( + service, "ensure_opencloud_folder", _fake_ensure_opencloud_folder + ) first = run_batch( smb_base_path="//test-server/share", @@ -117,8 +132,6 @@ def test_run_batch_smoke_and_idempotency(tmp_path: Path, monkeypatch) -> None: samba_user="user", samba_password="pass", launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="test-user", opencloud_access_token="token-123", ) @@ -132,8 +145,6 @@ def test_run_batch_smoke_and_idempotency(tmp_path: Path, monkeypatch) -> None: samba_user="user", samba_password="pass", launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="test-user", opencloud_access_token="token-123", ) @@ -158,6 +169,12 @@ def test_run_batch_success_when_delete_denied_after_copy( monkeypatch.setattr( service, "upload_local_file_to_opencloud", _fake_opencloud_upload ) + monkeypatch.setattr( + service, "resolve_personal_drive_id", _fake_resolve_personal_drive_id + ) + monkeypatch.setattr( + service, "ensure_opencloud_folder", _fake_ensure_opencloud_folder + ) result = run_batch( smb_base_path="//test-server/share", @@ -165,8 +182,6 @@ def test_run_batch_success_when_delete_denied_after_copy( samba_user="user", samba_password="pass", launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="test-user", opencloud_access_token="token-123", ) diff --git a/tests/smoke/test_local_zip_batch_smoke.py b/tests/smoke/test_local_zip_batch_smoke.py index 85c6905..ab4f463 100644 --- a/tests/smoke/test_local_zip_batch_smoke.py +++ b/tests/smoke/test_local_zip_batch_smoke.py @@ -30,10 +30,29 @@ def _fake_opencloud_upload( return f"https://opencloud.test/{'/'.join(folder_segments)}/{file_name}" -def test_run_local_zip_batch_success(tmp_path: Path, monkeypatch) -> None: +def _fake_resolve_personal_drive_id(username, token) -> str: + _ = (username, token) + return "personal-space-1" + + +def _fake_ensure_opencloud_folder(space_id, folder_segments, username, token) -> None: + _ = (space_id, folder_segments, username, token) + + +def _patch_opencloud(monkeypatch) -> None: monkeypatch.setattr( service, "upload_local_file_to_opencloud", _fake_opencloud_upload ) + monkeypatch.setattr( + service, "resolve_personal_drive_id", _fake_resolve_personal_drive_id + ) + monkeypatch.setattr( + service, "ensure_opencloud_folder", _fake_ensure_opencloud_folder + ) + + +def test_run_local_zip_batch_success(tmp_path: Path, monkeypatch) -> None: + _patch_opencloud(monkeypatch) zip_path = tmp_path / "batch.zip" with zipfile.ZipFile(zip_path, "w") as archive: @@ -42,8 +61,6 @@ def test_run_local_zip_batch_success(tmp_path: Path, monkeypatch) -> None: result = run_local_zip_batch( zip_path=zip_path, launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="test-user", opencloud_access_token="token-123", ) @@ -59,9 +76,7 @@ def test_run_local_zip_batch_success(tmp_path: Path, monkeypatch) -> None: def test_run_local_zip_batch_bad_file_name_is_error( tmp_path: Path, monkeypatch ) -> None: - monkeypatch.setattr( - service, "upload_local_file_to_opencloud", _fake_opencloud_upload - ) + _patch_opencloud(monkeypatch) zip_path = tmp_path / "batch.zip" with zipfile.ZipFile(zip_path, "w") as archive: @@ -70,8 +85,6 @@ def test_run_local_zip_batch_bad_file_name_is_error( result = run_local_zip_batch( zip_path=zip_path, launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="test-user", opencloud_access_token="token-123", ) @@ -82,9 +95,7 @@ def test_run_local_zip_batch_bad_file_name_is_error( def test_run_local_zip_batch_empty_archive(tmp_path: Path, monkeypatch) -> None: - monkeypatch.setattr( - service, "upload_local_file_to_opencloud", _fake_opencloud_upload - ) + _patch_opencloud(monkeypatch) zip_path = tmp_path / "empty.zip" with zipfile.ZipFile(zip_path, "w") as archive: @@ -93,8 +104,6 @@ def test_run_local_zip_batch_empty_archive(tmp_path: Path, monkeypatch) -> None: result = run_local_zip_batch( zip_path=zip_path, launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="test-user", opencloud_access_token="token-123", ) @@ -111,8 +120,6 @@ def test_run_local_zip_batch_missing_token(tmp_path: Path) -> None: result = run_local_zip_batch( zip_path=zip_path, launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="test-user", opencloud_access_token="", ) @@ -129,8 +136,6 @@ def test_run_local_zip_batch_missing_username(tmp_path: Path) -> None: result = run_local_zip_batch( zip_path=zip_path, launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="", opencloud_access_token="token-123", ) @@ -146,8 +151,6 @@ def test_run_local_zip_batch_bad_zip_file(tmp_path: Path) -> None: result = run_local_zip_batch( zip_path=zip_path, launcher="tester", - opencloud_space_id="space-1", - opencloud_folder="Тестов Т.Т.", opencloud_username="test-user", opencloud_access_token="token-123", ) diff --git a/tests/unit/test_opencloud_access.py b/tests/unit/test_opencloud_access.py index 0cab090..c65e787 100644 --- a/tests/unit/test_opencloud_access.py +++ b/tests/unit/test_opencloud_access.py @@ -8,6 +8,8 @@ import pytest from app.opencloud_access import ( OpenCloudUploadError, build_opencloud_url, + ensure_opencloud_folder, + resolve_personal_drive_id, upload_local_file_to_opencloud, ) @@ -102,6 +104,88 @@ def test_upload_local_file_to_opencloud_missing_local_file(tmp_path: Path) -> No ) +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_upload_local_file_to_opencloud_http_error(tmp_path: Path) -> None: local_file = tmp_path / "report.xlsx" local_file.write_bytes(b"data")