diagntostic fix

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)

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 = (

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()

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"]