Срез 0 [4.2] T8.1-T8.3: inbox_name заменяет символы Windows/SMB, poll ждёт занятый файл, put терпит отказ mkdir при существующем Input
T8.1: inbox_name заменяет `: < > " | ? *` и управляющие символы на `_`, убирает завершающие точки/пробелы; ValueError только если имени не осталось. validate_name/put остаются строгими. T8.2: SmbHotFolder.poll возвращает None на STATUS_SHARING_VIOLATION/STATUS_DELETE_PENDING (ИС ещё пишет ответ), монопольное открытие сохранено. T8.3: при ошибке mkdir, кроме EEXIST/ENOENT, put продолжает, если smbclient.path.isdir(Input); иначе исходная ошибка. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
1695573b35
commit
5fb046fced
@ -40,12 +40,17 @@ def _split_basename(name: str) -> str:
|
||||
def inbox_name(job_id: str, original_name: str) -> str:
|
||||
"""Имя входящего файла `<job_id>_<original_name>`.
|
||||
|
||||
`original_name` приходит из недоверенного запроса: берётся только последний компонент пути (без разделителей
|
||||
и `..`). Если после этого имени не осталось (пусто, `.`, `..`) или в нём есть символы, недопустимые в имени файла
|
||||
Windows/SMB, либо управляющие (см. `validate_name`) — `ValueError`. Пробелы и кириллица сохраняются как есть.
|
||||
`original_name` приходит из недоверенного запроса («как загрузил пользователь», контракт требует только
|
||||
непустую строку): берётся последний компонент пути (без разделителей и `..`), символы, недопустимые в имени файла
|
||||
Windows/SMB, и управляющие символы заменяются на `_`, завершающие точки и пробелы (их отбрасывает Windows)
|
||||
убираются. Пробелы внутри имени и кириллица сохраняются. Результат всегда проходит `validate_name`, если `job_id`
|
||||
допустим. `ValueError` — только если имени не осталось (пусто, `.`, `..`, одни точки/пробелы).
|
||||
"""
|
||||
base = _split_basename(original_name)
|
||||
if base in ("", ".", "..") or any(char in _ILLEGAL_CHARS for char in base):
|
||||
if base in (".", ".."):
|
||||
raise ValueError(f"недопустимое имя файла: {original_name!r}")
|
||||
base = "".join("_" if char in _ILLEGAL_CHARS else char for char in base).rstrip(". ")
|
||||
if not base:
|
||||
raise ValueError(f"недопустимое имя файла: {original_name!r}")
|
||||
return f"{job_id}_{base}"
|
||||
|
||||
|
||||
@ -15,9 +15,13 @@ import ntpath
|
||||
from typing import Any
|
||||
|
||||
import smbclient
|
||||
from smbprotocol.header import NtStatus
|
||||
|
||||
from src.gateway.hotfolder import validate_job_id, validate_name
|
||||
|
||||
# Ответ есть, но пока не читается: ИС держит файл открытым на запись или файл удаляется — опрос повторится позже
|
||||
_NOT_READY = frozenset({NtStatus.STATUS_SHARING_VIOLATION, NtStatus.STATUS_DELETE_PENDING})
|
||||
|
||||
|
||||
def _has_errno(err: OSError, code: int, exc_type: type[OSError]) -> bool:
|
||||
"""`SMBOSError` несёт `errno`, но не превращается в подкласс `OSError` по нему — проверяем и то, и другое."""
|
||||
@ -87,13 +91,16 @@ class SmbHotFolder:
|
||||
|
||||
def poll(self, job_id: str) -> bytes | None:
|
||||
r"""Байты `<output>\<job_id>\<job_id>.json` или `None`, пока его нет (нет файла, каталога задания или
|
||||
`Output`). Остальные ошибки SMB поднимаются."""
|
||||
`Output`) или он ещё не готов (ИС держит его открытым — `STATUS_SHARING_VIOLATION`, удаляется —
|
||||
`STATUS_DELETE_PENDING`). Остальные ошибки SMB поднимаются."""
|
||||
path = self.answer_path(job_id)
|
||||
try:
|
||||
# Монопольное открытие (share_access по умолчанию 0): пока ИС пишет файл, оно не проходит, и недописанный
|
||||
# JSON не читается
|
||||
with smbclient.open_file(path, mode="rb", **self._credentials) as fd:
|
||||
data: bytes = fd.read()
|
||||
except OSError as err:
|
||||
if _has_errno(err, errno.ENOENT, FileNotFoundError):
|
||||
if _has_errno(err, errno.ENOENT, FileNotFoundError) or getattr(err, "ntstatus", None) in _NOT_READY:
|
||||
return None
|
||||
raise
|
||||
return data
|
||||
@ -107,9 +114,22 @@ class SmbHotFolder:
|
||||
return True
|
||||
|
||||
def _ensure_input_dir(self) -> None:
|
||||
"""`mkdir` входящего каталога; «уже существует» (EEXIST) — не ошибка."""
|
||||
"""`mkdir` входящего каталога, как `os.makedirs(exist_ok=True)` без родителей.
|
||||
|
||||
«Уже существует» (EEXIST) — не ошибка. Другая ошибка (например, `STATUS_ACCESS_DENIED`: учётке нельзя
|
||||
создавать подкаталоги в корне, хотя писать в `Input` можно) — тоже не ошибка, если каталог уже есть
|
||||
(`smbclient.path.isdir`); иначе поднимается исходная ошибка `mkdir`. ENOENT (нет корня) — сразу наружу.
|
||||
"""
|
||||
try:
|
||||
smbclient.mkdir(self.input_dir, **self._credentials)
|
||||
except OSError as err:
|
||||
if not _has_errno(err, errno.EEXIST, FileExistsError):
|
||||
if _has_errno(err, errno.EEXIST, FileExistsError):
|
||||
return
|
||||
if _has_errno(err, errno.ENOENT, FileNotFoundError) or not self._input_dir_exists():
|
||||
raise
|
||||
|
||||
def _input_dir_exists(self) -> bool:
|
||||
try:
|
||||
return bool(smbclient.path.isdir(self.input_dir, **self._credentials))
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
@ -8,7 +8,7 @@ from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.gateway.hotfolder import HotFolder, LocalHotFolder, inbox_name
|
||||
from src.gateway.hotfolder import HotFolder, LocalHotFolder, inbox_name, validate_name
|
||||
|
||||
|
||||
def _job_id() -> str:
|
||||
@ -40,22 +40,60 @@ class TestInboxName:
|
||||
def test_sanitizes_to_basename(self, original: str, expected: str) -> None:
|
||||
assert inbox_name("job", original) == f"job_{expected}"
|
||||
|
||||
@pytest.mark.parametrize("original", ["", "/", "..", ".", "a/..", "dir/", "\\", "bad\x00name.pdf"])
|
||||
@pytest.mark.parametrize("original", ["", "/", "..", ".", "a/..", "dir/", "\\", "...", ". .", " ", "dir/. "])
|
||||
def test_rejects_unusable_names(self, original: str) -> None:
|
||||
with pytest.raises(ValueError, match="имя файла"):
|
||||
inbox_name("job", original)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"original", ["a:b.pdf", "отчёт?.pdf", "a*b.pdf", 'a"b.pdf', "a|b.pdf", "a<b>.pdf", "a\x01b.pdf"]
|
||||
("original", "expected"),
|
||||
[
|
||||
("a:b?.pdf", "a_b_.pdf"),
|
||||
("Отчёт: 2024.pdf", "Отчёт_ 2024.pdf"),
|
||||
('Баланс "итог".pdf', "Баланс _итог_.pdf"),
|
||||
("форма?.pdf", "форма_.pdf"),
|
||||
("a|b<c>*.pdf", "a_b_c__.pdf"),
|
||||
("C:report.pdf", "C_report.pdf"),
|
||||
("a\x01b.pdf", "a_b.pdf"),
|
||||
("bad\x00name.pdf", "bad_name.pdf"),
|
||||
("tab\tname.pdf", "tab_name.pdf"),
|
||||
],
|
||||
)
|
||||
def test_rejects_names_illegal_on_windows(self, original: str) -> None:
|
||||
with pytest.raises(ValueError, match="имя файла"):
|
||||
inbox_name("job", original)
|
||||
def test_replaces_chars_illegal_on_windows(self, original: str, expected: str) -> None:
|
||||
"""Имя от пользователя законно на Linux/S3/macOS: заменяем, а не отвергаем (контракт: только minLength 1)."""
|
||||
assert inbox_name("job", original) == f"job_{expected}"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("original", "expected"),
|
||||
[("report.pdf.", "report.pdf"), ("report.pdf . ", "report.pdf"), ("report .pdf", "report .pdf")],
|
||||
)
|
||||
def test_strips_trailing_dots_and_spaces(self, original: str, expected: str) -> None:
|
||||
assert inbox_name("job", original) == f"job_{expected}"
|
||||
|
||||
def test_keeps_spaces_and_cyrillic(self) -> None:
|
||||
job_id = _job_id()
|
||||
assert inbox_name(job_id, "Отчёт за 2024 год (копия).PDF") == f"{job_id}_Отчёт за 2024 год (копия).PDF"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"original",
|
||||
[
|
||||
"a:b?.pdf",
|
||||
'Баланс "итог".pdf',
|
||||
"Отчёт за 2024 год (копия).PDF",
|
||||
".hidden.pdf",
|
||||
"..pdf",
|
||||
"x\x00\x1f<>|*?.pdf. ",
|
||||
"C:\\dir\\..\\скан.png",
|
||||
" leading.pdf",
|
||||
],
|
||||
)
|
||||
def test_result_always_passes_validate_name_and_put(self, tmp_path: Path, original: str) -> None:
|
||||
hf = LocalHotFolder(tmp_path)
|
||||
name = inbox_name(_job_id(), original)
|
||||
validate_name(name)
|
||||
hf.put(name, b"x")
|
||||
assert (hf.input_dir / name).read_bytes() == b"x"
|
||||
|
||||
def test_result_is_accepted_by_put(self, tmp_path: Path) -> None:
|
||||
hf = LocalHotFolder(tmp_path)
|
||||
name = inbox_name(_job_id(), "../../scan.pdf")
|
||||
|
||||
@ -7,6 +7,7 @@
|
||||
import os
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import IO, Any
|
||||
|
||||
import pytest
|
||||
@ -29,6 +30,11 @@ class FakeSmbClient:
|
||||
self.base = base
|
||||
self.calls: list[tuple[str, tuple[str, ...], dict[str, Any]]] = []
|
||||
self.fail: dict[str, BaseException] = {} # имя функции -> исключение, которое она бросит
|
||||
self.path = SimpleNamespace(isdir=self._isdir) # подмодуль smbclient.path
|
||||
|
||||
def _isdir(self, path: str, **kwargs: Any) -> bool:
|
||||
self._record("isdir", (path,), kwargs)
|
||||
return self._local(path).is_dir()
|
||||
|
||||
def _record(self, func: str, args: tuple[str, ...], kwargs: dict[str, Any]) -> None:
|
||||
self.calls.append((func, args, kwargs))
|
||||
@ -185,11 +191,31 @@ class TestPut:
|
||||
assert fake.funcs() == ["mkdir"]
|
||||
assert not fake.base.exists()
|
||||
|
||||
def test_mkdir_other_error_propagates(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
def test_mkdir_denied_but_input_exists_is_fine(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
"""Урезанные ACL: нельзя создавать подкаталоги в корне, но `Input` уже есть и в него писать можно."""
|
||||
(share_root / "Input").mkdir()
|
||||
fake.fail["mkdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
with pytest.raises(SMBOSError):
|
||||
_hotfolder().put("a.pdf", b"1")
|
||||
assert (share_root / "Input" / "a.pdf").read_bytes() == b"1"
|
||||
assert fake.funcs() == ["mkdir", "isdir", "open_file", "replace"]
|
||||
assert fake.calls[1] == ("isdir", (INPUT,), {})
|
||||
|
||||
def test_mkdir_denied_and_no_input_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
fake.fail["mkdir"] = denied
|
||||
with pytest.raises(SMBOSError) as excinfo:
|
||||
_hotfolder().put("a.pdf", b"1")
|
||||
assert fake.funcs() == ["mkdir"]
|
||||
assert excinfo.value is denied
|
||||
assert fake.funcs() == ["mkdir", "isdir"]
|
||||
|
||||
def test_mkdir_denied_and_isdir_fails_raises_original(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
denied = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
fake.fail["mkdir"] = denied
|
||||
fake.fail["isdir"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, INPUT)
|
||||
with pytest.raises(SMBOSError) as excinfo:
|
||||
_hotfolder().put("a.pdf", b"1")
|
||||
assert excinfo.value is denied
|
||||
assert fake.funcs() == ["mkdir", "isdir"]
|
||||
|
||||
def test_overwrites_existing_file(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
hf = _hotfolder()
|
||||
@ -290,6 +316,16 @@ class TestPoll:
|
||||
fake.fail["open_file"] = FileNotFoundError("нет файла")
|
||||
assert _hotfolder().poll(_job_id()) is None
|
||||
|
||||
@pytest.mark.parametrize("status", [NtStatus.STATUS_SHARING_VIOLATION, NtStatus.STATUS_DELETE_PENDING])
|
||||
def test_answer_being_written_or_deleted_is_none(
|
||||
self, fake: FakeSmbClient, share_root: Path, status: NtStatus
|
||||
) -> None:
|
||||
"""ИС ещё держит файл (монопольное открытие не проходит) — ответа пока нет, опросим позже."""
|
||||
job_id = _job_id()
|
||||
_write_answer(share_root, job_id, b"{}")
|
||||
fake.fail["open_file"] = SMBOSError(status, OUTPUT)
|
||||
assert _hotfolder().poll(job_id) is None
|
||||
|
||||
def test_other_errors_propagate(self, fake: FakeSmbClient, share_root: Path) -> None:
|
||||
fake.fail["open_file"] = SMBOSError(NtStatus.STATUS_ACCESS_DENIED, OUTPUT)
|
||||
with pytest.raises(SMBOSError):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user