132 lines
4.5 KiB
Python
132 lines
4.5 KiB
Python
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"]
|