329 lines
9.9 KiB
Python
329 lines
9.9 KiB
Python
from __future__ import annotations
|
|
|
|
import tarfile
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from app.opencloud_access import OpenCloudUploadResult, PersonalDrive
|
|
from app.pipeline import service
|
|
from app.pipeline.service import run_batch
|
|
|
|
|
|
def _valid_xml() -> str:
|
|
return """<?xml version="1.0" encoding="UTF-8"?>
|
|
<СообщОперКО xmlns="urn:test">
|
|
<СлужЧасть><ИдФайл>ID1</ИдФайл></СлужЧасть>
|
|
<ИнформЧасть>
|
|
<СведКО>
|
|
<Операция>
|
|
<ИдентификаторЗаписи>OP1</ИдентификаторЗаписи>
|
|
<УчастникОп><Тип>01</Тип></УчастникОп>
|
|
</Операция>
|
|
</СведКО>
|
|
</ИнформЧасть>
|
|
</СообщОперКО>
|
|
"""
|
|
|
|
|
|
def _fake_opencloud_upload(
|
|
local_path: Path, space_id, folder_segments, file_name, username, token
|
|
) -> OpenCloudUploadResult:
|
|
assert local_path.exists()
|
|
return OpenCloudUploadResult(
|
|
dav_url=f"https://opencloud.test/{'/'.join(folder_segments)}/{file_name}",
|
|
web_url=f"https://opencloud.test/f/{file_name}",
|
|
)
|
|
|
|
|
|
def _fake_resolve_opencloud_file_id(dav_url, username, token) -> str:
|
|
_ = (dav_url, username, token)
|
|
return "store$space!folder"
|
|
|
|
|
|
def _fake_resolve_personal_drive(username, token) -> PersonalDrive:
|
|
_ = (username, token)
|
|
return PersonalDrive(id="personal-space-1", alias="personal/test-user")
|
|
|
|
|
|
def _fake_ensure_opencloud_folder(space_id, folder_segments, username, token) -> None:
|
|
_ = (space_id, folder_segments, username, token)
|
|
|
|
|
|
def _fake_next_opencloud_run_number(
|
|
space_id, day_folder_segments, username, token
|
|
) -> int:
|
|
_ = (space_id, day_folder_segments, username, token)
|
|
return 1
|
|
|
|
|
|
def _patch_opencloud(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
service, "upload_local_file_to_opencloud", _fake_opencloud_upload
|
|
)
|
|
monkeypatch.setattr(service, "resolve_personal_drive", _fake_resolve_personal_drive)
|
|
monkeypatch.setattr(
|
|
service, "ensure_opencloud_folder", _fake_ensure_opencloud_folder
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "next_opencloud_run_number", _fake_next_opencloud_run_number
|
|
)
|
|
monkeypatch.setattr(
|
|
service, "resolve_opencloud_file_id", _fake_resolve_opencloud_file_id
|
|
)
|
|
|
|
|
|
def test_run_batch_uses_run_number_from_opencloud(tmp_path: Path, monkeypatch) -> None:
|
|
_patch_opencloud(monkeypatch)
|
|
monkeypatch.setattr(service, "next_opencloud_run_number", lambda *args, **kwargs: 3)
|
|
|
|
zip_path = tmp_path / "batch.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.writestr("SKO115FZ_01_123456789_20260616_X00001.xml", _valid_xml())
|
|
|
|
result = run_batch(
|
|
input_paths=[zip_path],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert "/3/" in result.report_path
|
|
assert "_3.xlsx" in result.report_path
|
|
|
|
|
|
def test_run_batch_success(tmp_path: Path, monkeypatch) -> None:
|
|
_patch_opencloud(monkeypatch)
|
|
|
|
zip_path = tmp_path / "batch.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.writestr("SKO115FZ_01_123456789_20260616_X00001.xml", _valid_xml())
|
|
|
|
result = run_batch(
|
|
input_paths=[zip_path],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert result.processed == 1
|
|
assert result.errors == 0
|
|
assert result.report_path
|
|
assert result.protocol_path
|
|
assert result.log_path
|
|
|
|
|
|
def test_run_batch_bad_file_name_is_error(tmp_path: Path, monkeypatch) -> None:
|
|
_patch_opencloud(monkeypatch)
|
|
|
|
zip_path = tmp_path / "batch.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.writestr("bad_name.xml", _valid_xml())
|
|
|
|
result = run_batch(
|
|
input_paths=[zip_path],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert result.processed == 0
|
|
assert result.errors == 1
|
|
|
|
|
|
def test_run_batch_empty_archive(tmp_path: Path, monkeypatch) -> None:
|
|
_patch_opencloud(monkeypatch)
|
|
|
|
zip_path = tmp_path / "empty.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.writestr("readme.txt", "no xml here")
|
|
|
|
result = run_batch(
|
|
input_paths=[zip_path],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert result.processed == 0
|
|
|
|
|
|
def test_run_batch_missing_token(tmp_path: Path) -> None:
|
|
zip_path = tmp_path / "batch.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.writestr("SKO115FZ_01_123456789_20260616_X00001.xml", _valid_xml())
|
|
|
|
result = run_batch(
|
|
input_paths=[zip_path],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="",
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert not result.report_path
|
|
|
|
|
|
def test_run_batch_missing_username(tmp_path: Path) -> None:
|
|
zip_path = tmp_path / "batch.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.writestr("SKO115FZ_01_123456789_20260616_X00001.xml", _valid_xml())
|
|
|
|
result = run_batch(
|
|
input_paths=[zip_path],
|
|
launcher="tester",
|
|
opencloud_username="",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert not result.report_path
|
|
|
|
|
|
def test_run_batch_bad_zip_file(tmp_path: Path, monkeypatch) -> None:
|
|
_patch_opencloud(monkeypatch)
|
|
|
|
zip_path = tmp_path / "not_a_zip.zip"
|
|
zip_path.write_bytes(b"this is not a zip file")
|
|
|
|
result = run_batch(
|
|
input_paths=[zip_path],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert result.errors == 1
|
|
assert any("not_a_zip.zip" in message for message in result.messages)
|
|
|
|
|
|
def _write_zip(path: Path, members: dict[str, str]) -> Path:
|
|
with zipfile.ZipFile(path, "w") as archive:
|
|
for name, content in members.items():
|
|
archive.writestr(name, content)
|
|
return path
|
|
|
|
|
|
def _write_xml(path: Path, content: str | None = None) -> Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content or _valid_xml(), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def test_run_batch_accepts_several_archives_and_loose_xml(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
"""Основной сценарий правки: вложений несколько, среди них и архивы,
|
|
и отдельные .xml."""
|
|
_patch_opencloud(monkeypatch)
|
|
|
|
first = _write_zip(
|
|
tmp_path / "day1.zip",
|
|
{"SKO115FZ_01_123456789_20260616_X00001.xml": _valid_xml()},
|
|
)
|
|
second = _write_zip(
|
|
tmp_path / "day2.zip",
|
|
{
|
|
"nested/SKO115FZ_01_123456789_20260616_X00002.xml": _valid_xml(),
|
|
"nested/deeper/SKO115FZ_01_123456789_20260616_X00003.xml": _valid_xml(),
|
|
},
|
|
)
|
|
loose = _write_xml(tmp_path / "SKO115FZ_01_123456789_20260616_X00004.xml")
|
|
|
|
result = run_batch(
|
|
input_paths=[first, second, loose],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert result.processed == 4
|
|
assert result.errors == 0
|
|
|
|
|
|
def test_run_batch_keeps_going_when_one_attachment_is_broken(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
"""Одно нечитаемое вложение не отменяет обработку остальных."""
|
|
_patch_opencloud(monkeypatch)
|
|
|
|
good = _write_zip(
|
|
tmp_path / "good.zip",
|
|
{"SKO115FZ_01_123456789_20260616_X00001.xml": _valid_xml()},
|
|
)
|
|
broken = tmp_path / "broken.zip"
|
|
broken.write_bytes(b"not an archive at all")
|
|
|
|
result = run_batch(
|
|
input_paths=[good, broken],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.processed == 1
|
|
assert result.errors == 1
|
|
assert result.status == "error"
|
|
assert result.report_path
|
|
|
|
|
|
def test_run_batch_skips_duplicate_file_names_across_attachments(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
"""Один и тот же файл, прикреплённый дважды, не должен удваивать отчёт."""
|
|
_patch_opencloud(monkeypatch)
|
|
|
|
name = "SKO115FZ_01_123456789_20260616_X00001.xml"
|
|
first = _write_zip(tmp_path / "a.zip", {name: _valid_xml()})
|
|
second = _write_zip(tmp_path / "b.zip", {name: _valid_xml()})
|
|
|
|
result = run_batch(
|
|
input_paths=[first, second],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.processed == 1
|
|
assert result.errors == 0
|
|
assert any("Пропущено повторно" in message for message in result.messages)
|
|
|
|
|
|
def test_run_batch_accepts_tar_gz_archive(tmp_path: Path, monkeypatch) -> None:
|
|
_patch_opencloud(monkeypatch)
|
|
|
|
xml_path = _write_xml(
|
|
tmp_path / "src" / "SKO115FZ_01_123456789_20260616_X00001.xml"
|
|
)
|
|
archive_path = tmp_path / "batch.tar.gz"
|
|
with tarfile.open(archive_path, "w:gz") as archive:
|
|
archive.add(xml_path, arcname=xml_path.name)
|
|
|
|
result = run_batch(
|
|
input_paths=[archive_path],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert result.processed == 1
|
|
|
|
|
|
def test_run_batch_without_attachments_is_error(tmp_path: Path) -> None:
|
|
_ = tmp_path
|
|
result = run_batch(
|
|
input_paths=[],
|
|
launcher="tester",
|
|
opencloud_username="test-user",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert not result.report_path
|