129 lines
3.7 KiB
Python
129 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from app.pipeline import service
|
|
from app.pipeline.service import run_local_zip_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, access_token
|
|
) -> str:
|
|
assert local_path.exists()
|
|
return f"https://opencloud.test/{'/'.join(folder_segments)}/{file_name}"
|
|
|
|
|
|
def test_run_local_zip_batch_success(tmp_path: Path, monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
service, "upload_local_file_to_opencloud", _fake_opencloud_upload
|
|
)
|
|
|
|
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_local_zip_batch(
|
|
zip_path=zip_path,
|
|
launcher="tester",
|
|
opencloud_space_id="space-1",
|
|
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_local_zip_batch_bad_file_name_is_error(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
service, "upload_local_file_to_opencloud", _fake_opencloud_upload
|
|
)
|
|
|
|
zip_path = tmp_path / "batch.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.writestr("bad_name.xml", _valid_xml())
|
|
|
|
result = run_local_zip_batch(
|
|
zip_path=zip_path,
|
|
launcher="tester",
|
|
opencloud_space_id="space-1",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert result.processed == 0
|
|
assert result.errors == 1
|
|
|
|
|
|
def test_run_local_zip_batch_empty_archive(tmp_path: Path, monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
service, "upload_local_file_to_opencloud", _fake_opencloud_upload
|
|
)
|
|
|
|
zip_path = tmp_path / "empty.zip"
|
|
with zipfile.ZipFile(zip_path, "w") as archive:
|
|
archive.writestr("readme.txt", "no xml here")
|
|
|
|
result = run_local_zip_batch(
|
|
zip_path=zip_path,
|
|
launcher="tester",
|
|
opencloud_space_id="space-1",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "success"
|
|
assert result.processed == 0
|
|
|
|
|
|
def test_run_local_zip_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_local_zip_batch(
|
|
zip_path=zip_path,
|
|
launcher="tester",
|
|
opencloud_space_id="space-1",
|
|
opencloud_access_token="",
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert not result.report_path
|
|
|
|
|
|
def test_run_local_zip_batch_bad_zip_file(tmp_path: Path) -> None:
|
|
zip_path = tmp_path / "not_a_zip.zip"
|
|
zip_path.write_bytes(b"this is not a zip file")
|
|
|
|
result = run_local_zip_batch(
|
|
zip_path=zip_path,
|
|
launcher="tester",
|
|
opencloud_space_id="space-1",
|
|
opencloud_access_token="token-123",
|
|
)
|
|
|
|
assert result.status == "error"
|
|
assert any("распаковать" in message for message in result.messages)
|