diff --git a/app/pipeline/file_manager.py b/app/pipeline/file_manager.py index b622c9d..5058b84 100644 --- a/app/pipeline/file_manager.py +++ b/app/pipeline/file_manager.py @@ -80,6 +80,10 @@ class SMBClientConnection: entries = smbclient.scandir(full_path) return [entry.name for entry in entries] + def stat(self, path: str) -> object: + smbclient = _import_smbclient() + return smbclient.stat(self._resolve_path(path)) + def is_file(self, path: str) -> bool: smbclient = _import_smbclient() full_path = self._resolve_path(path) @@ -153,6 +157,10 @@ class SMBClientConnection: full_path = self._resolve_path(path) smbclient.remove(full_path) + def rmdir(self, path: str) -> None: + smbclient = _import_smbclient() + smbclient.rmdir(self._resolve_path(path)) + def rename(self, path_from: str, path_to: str, replace: bool = False) -> None: smbclient = _import_smbclient() full_source = self._resolve_path(path_from) diff --git a/app/samba_access.py b/app/samba_access.py index 9ef5294..39d7adb 100644 --- a/app/samba_access.py +++ b/app/samba_access.py @@ -1,7 +1,13 @@ from __future__ import annotations +import re +from collections.abc import Callable +from dataclasses import dataclass from pathlib import Path +from typing import Literal +from uuid import uuid4 +from app.pipeline.config import join_path from app.pipeline.file_manager import ( check_input_dir_access, connect_samba, @@ -9,6 +15,19 @@ from app.pipeline.file_manager import ( ) +@dataclass(frozen=True) +class SMBDiagnosticStep: + stage: str + status: Literal["ok", "error", "skipped"] + details: str + + +@dataclass(frozen=True) +class SMBDiagnosticResult: + test_path: str + steps: tuple[SMBDiagnosticStep, ...] + + def build_user_params(samba_user: str, samba_password: str) -> dict[str, str]: return { "user_name": samba_user, @@ -70,6 +89,136 @@ def check_access_and_list_files( return True, "", files +def diagnose_smb_access( + smb_base_path: str, + parent_dir: str, + samba_user: str, + samba_password: str, +) -> SMBDiagnosticResult: + test_path = join_path(parent_dir, f"Input_diag_{uuid4().hex[:8]}") + steps: list[SMBDiagnosticStep] = [] + + if not samba_user: + steps.append( + SMBDiagnosticStep( + stage="Подключение к SMB", + status="error", + details="Введите логин SAMBA.", + ) + ) + return SMBDiagnosticResult(test_path=test_path, steps=tuple(steps)) + + try: + samba_conn = connect_samba( + smb_base_path=smb_base_path, + user_params=build_user_params(samba_user, samba_password), + ) + with samba_conn: + pass + except Exception as exc: # noqa: BLE001 + steps.append(_diagnostic_error("Подключение к SMB", exc)) + return SMBDiagnosticResult(test_path=test_path, steps=tuple(steps)) + + steps.append(_diagnostic_ok("Подключение к SMB")) + _run_diagnostic_step( + steps, + "Проверка родительской папки", + lambda: samba_conn.stat(parent_dir), + ) + _run_diagnostic_step( + steps, + "Чтение родительской папки", + lambda: samba_conn.listdir(parent_dir), + ) + + try: + created = _run_diagnostic_step( + steps, + "Создание тестовой папки", + lambda: samba_conn.mkdir(test_path), + ) + if created: + _run_diagnostic_step( + steps, + "Проверка созданной папки", + lambda: samba_conn.stat(test_path), + ) + _run_diagnostic_step( + steps, + "Чтение созданной папки", + lambda: samba_conn.listdir(test_path), + ) + _run_diagnostic_step( + steps, + "Проверка папки в родительском каталоге", + lambda: _require_directory_in_parent( + samba_conn.listdir(parent_dir), + test_path, + ), + ) + else: + for stage in ( + "Проверка созданной папки", + "Чтение созданной папки", + "Проверка папки в родительском каталоге", + ): + steps.append( + SMBDiagnosticStep( + stage=stage, + status="skipped", + details="Пропущено: создание тестовой папки завершилось ошибкой.", + ) + ) + finally: + _run_diagnostic_step( + steps, + "Удаление тестовой папки", + lambda: samba_conn.rmdir(test_path), + ) + + return SMBDiagnosticResult(test_path=test_path, steps=tuple(steps)) + + +def _run_diagnostic_step( + steps: list[SMBDiagnosticStep], + stage: str, + action: Callable[[], object], +) -> bool: + try: + action() + except Exception as exc: # noqa: BLE001 + steps.append(_diagnostic_error(stage, exc)) + return False + steps.append(_diagnostic_ok(stage)) + return True + + +def _require_directory_in_parent(entries: list[str], test_path: str) -> None: + directory_name = test_path.rstrip("/\\").rsplit("\\", 1)[-1] + if directory_name not in entries: + raise FileNotFoundError( + f"Созданная папка {directory_name} отсутствует в списке родителя." + ) + + +def _diagnostic_ok(stage: str) -> SMBDiagnosticStep: + return SMBDiagnosticStep(stage=stage, status="ok", details="Успешно.") + + +def _diagnostic_error(stage: str, exc: Exception) -> SMBDiagnosticStep: + details = str(exc) + status = getattr(exc, "ntstatus", None) or getattr(exc, "status", None) + if status is None: + match = re.search(r"\bSTATUS_[A-Z0-9_]+\b", details, flags=re.IGNORECASE) + status = match.group(0).upper() if match else None + suffix = f"; NTSTATUS: {status}" if status is not None else "" + return SMBDiagnosticStep( + stage=stage, + status="error", + details=f"{type(exc).__name__}: {details}{suffix}", + ) + + def _format_connect_error(details: str) -> str: normalized = details.lower() auth_hints = ( diff --git a/main.py b/main.py index 76f57a2..6cfb313 100644 --- a/main.py +++ b/main.py @@ -13,7 +13,7 @@ from app.path_settings import ( from app.path_store import set_user_path from app.pipeline.config import join_path from app.pipeline.service import run_batch -from app.samba_access import check_access_and_list_files +from app.samba_access import check_access_and_list_files, diagnose_smb_access def main() -> None: @@ -117,6 +117,24 @@ def main() -> None: else: st.info("Папка доступна, но пуста.") + if st.button("Диагностика SMB", use_container_width=True): + with st.spinner("Выполняется диагностика SMB..."): + diagnostic = diagnose_smb_access( + smb_base_path=SFM_SMB_BASE_PATH, + parent_dir=resolved_input_dir, + samba_user=samba_user, + samba_password=samba_password, + ) + st.caption(f"Тестовая папка: `{diagnostic.test_path}`") + for step in diagnostic.steps: + message = f"{step.stage}: {step.details}" + if step.status == "ok": + st.success(message) + elif step.status == "error": + st.error(message) + else: + st.info(message) + if st.button("Выйти", type="secondary"): clear_session() st.rerun() diff --git a/tests/unit/test_samba_access.py b/tests/unit/test_samba_access.py new file mode 100644 index 0000000..962cf36 --- /dev/null +++ b/tests/unit/test_samba_access.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +from unittest.mock import patch + +import pytest + +from app.samba_access import diagnose_smb_access + + +class FakeSambaConnection: + def __init__( + self, + *, + mkdir_error: BaseException | None = None, + child_stat_error: Exception | None = None, + ) -> None: + self.mkdir_error = mkdir_error + self.child_stat_error = child_stat_error + self.created_path = "" + self.removed_paths: list[str] = [] + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb) -> bool: # noqa: ANN001 + return False + + def stat(self, path: str) -> object: + if path == self.created_path and self.child_stat_error is not None: + raise self.child_stat_error + return object() + + def listdir(self, path: str) -> list[str]: + if self.created_path and path != self.created_path: + return [self.created_path.rsplit("\\", 1)[1]] + return [] + + def mkdir(self, path: str, parent: bool = False) -> None: + assert parent is False + if self.mkdir_error is not None: + raise self.mkdir_error + self.created_path = path + + def rmdir(self, path: str) -> None: + self.removed_paths.append(path) + + +def test_diagnose_smb_access_runs_all_stages_and_removes_test_directory() -> None: + connection = FakeSambaConnection() + + with ( + patch("app.samba_access.connect_samba", return_value=connection), + patch("app.samba_access.uuid4") as uuid4, + ): + uuid4.return_value.hex = "12345678abcdef" + result = diagnose_smb_access( + smb_base_path=r"\\server\share", + parent_dir=r"\\server\share\user", + samba_user="DOMAIN\\user", + samba_password="secret", + ) + + assert result.test_path == r"\\server\share\user\Input_diag_12345678" + assert all(step.status == "ok" for step in result.steps) + assert connection.removed_paths == [result.test_path] + + +def test_diagnose_smb_access_reports_mkdir_error_without_child_checks() -> None: + connection = FakeSambaConnection( + mkdir_error=PermissionError("STATUS_ACCESS_DENIED") + ) + + with ( + patch("app.samba_access.connect_samba", return_value=connection), + patch("app.samba_access.uuid4") as uuid4, + ): + uuid4.return_value.hex = "12345678abcdef" + result = diagnose_smb_access( + smb_base_path=r"\\server\share", + parent_dir=r"\\server\share\user", + samba_user="DOMAIN\\user", + samba_password="secret", + ) + + steps = {step.stage: step for step in result.steps} + assert steps["Создание тестовой папки"].status == "error" + assert "STATUS_ACCESS_DENIED" in steps["Создание тестовой папки"].details + assert steps["Проверка созданной папки"].status == "skipped" + assert connection.removed_paths == [result.test_path] + + +def test_diagnose_smb_access_cleans_up_after_child_check_error() -> None: + connection = FakeSambaConnection( + child_stat_error=PermissionError("STATUS_ACCESS_DENIED") + ) + + with ( + patch("app.samba_access.connect_samba", return_value=connection), + patch("app.samba_access.uuid4") as uuid4, + ): + uuid4.return_value.hex = "12345678abcdef" + result = diagnose_smb_access( + smb_base_path=r"\\server\share", + parent_dir=r"\\server\share\user", + samba_user="DOMAIN\\user", + samba_password="secret", + ) + + steps = {step.stage: step for step in result.steps} + assert steps["Проверка созданной папки"].status == "error" + assert steps["Удаление тестовой папки"].status == "ok" + assert connection.removed_paths == [result.test_path] + + +def test_diagnose_smb_access_attempts_cleanup_when_creation_is_interrupted() -> None: + connection = FakeSambaConnection(mkdir_error=KeyboardInterrupt()) + + with ( + patch("app.samba_access.connect_samba", return_value=connection), + patch("app.samba_access.uuid4") as uuid4, + ): + uuid4.return_value.hex = "12345678abcdef" + with pytest.raises(KeyboardInterrupt): + diagnose_smb_access( + smb_base_path=r"\\server\share", + parent_dir=r"\\server\share\user", + samba_user="DOMAIN\\user", + samba_password="secret", + ) + + assert connection.removed_paths == [r"\\server\share\user\Input_diag_12345678"]