116 lines
3.4 KiB
Python
116 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
from app.pipeline.service import run_batch
|
|
|
|
|
|
class FakeSamba:
|
|
def __init__(self, root: Path) -> None:
|
|
self.root = root
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb): # noqa: ANN001
|
|
return False
|
|
|
|
def _to_local(self, remote_path: str) -> Path:
|
|
cleaned = remote_path.replace("/", "\\").strip("\\")
|
|
return self.root / Path(cleaned)
|
|
|
|
def path_exists(self, path: str) -> bool:
|
|
return self._to_local(path).exists()
|
|
|
|
def listdir(self, path: str) -> list[str]:
|
|
target = self._to_local(path)
|
|
if not target.exists():
|
|
return []
|
|
return [item.name for item in target.iterdir()]
|
|
|
|
def is_file(self, path: str) -> bool:
|
|
return self._to_local(path).is_file()
|
|
|
|
def mkdir(self, path: str, parent: bool = False) -> None:
|
|
target = self._to_local(path)
|
|
if parent:
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
else:
|
|
target.mkdir(exist_ok=True)
|
|
|
|
def copy(self, path_from: str, path_to: str, replace: bool = False) -> None:
|
|
source = self._to_local(path_from)
|
|
target = self._to_local(path_to)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
if target.exists() and not replace:
|
|
raise FileExistsError(path_to)
|
|
shutil.copy2(source, target)
|
|
|
|
def delete(self, path: str) -> None:
|
|
target = self._to_local(path)
|
|
if target.exists():
|
|
target.unlink()
|
|
|
|
def open(self, path: str, mode: str):
|
|
return self._to_local(path).open(mode)
|
|
|
|
def upload(self, path_from: str, path_to: str) -> None:
|
|
source = Path(path_from)
|
|
target = self._to_local(path_to)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(source, target)
|
|
|
|
|
|
def _valid_xml() -> str:
|
|
return """<?xml version="1.0" encoding="UTF-8"?>
|
|
<СообщОперКО xmlns="urn:test">
|
|
<СлужЧасть><ИдФайл>ID1</ИдФайл></СлужЧасть>
|
|
<ИнформЧасть>
|
|
<СведКО>
|
|
<Операция>
|
|
<ИдентификаторЗаписи>OP1</ИдентификаторЗаписи>
|
|
<УчастникОп><Тип>01</Тип></УчастникОп>
|
|
</Операция>
|
|
</СведКО>
|
|
</ИнформЧасть>
|
|
</СообщОперКО>
|
|
"""
|
|
|
|
|
|
def test_run_batch_smoke_and_idempotency(tmp_path: Path, monkeypatch) -> None:
|
|
samba = FakeSamba(tmp_path)
|
|
in_dir_local = tmp_path / "Exchange" / "SMBDEMO" / "test"
|
|
in_dir_local.mkdir(parents=True)
|
|
(in_dir_local / "SKO115FZ_01_123456789_20260616_X00001.xml").write_text(
|
|
_valid_xml(),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
from app.pipeline import service
|
|
|
|
monkeypatch.setattr(
|
|
service, "connect_samba", lambda smb_base_path, user_params: samba
|
|
)
|
|
|
|
first = run_batch(
|
|
smb_base_path="//test-server/share",
|
|
input_dir="Exchange\\SMBDEMO\\test",
|
|
samba_user="user",
|
|
samba_password="pass",
|
|
launcher="tester",
|
|
)
|
|
assert first.status in {"success", "partial"}
|
|
assert first.protocol_path
|
|
assert first.report_path
|
|
|
|
second = run_batch(
|
|
smb_base_path="//test-server/share",
|
|
input_dir="Exchange\\SMBDEMO\\test",
|
|
samba_user="user",
|
|
samba_password="pass",
|
|
launcher="tester",
|
|
)
|
|
assert second.status == "success"
|
|
assert second.processed == 0
|