157 lines
4.7 KiB
Python
157 lines
4.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, username, 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_folder="Тестов Т.Т.",
|
||
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_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_folder="Тестов Т.Т.",
|
||
opencloud_username="test-user",
|
||
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_folder="Тестов Т.Т.",
|
||
opencloud_username="test-user",
|
||
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_folder="Тестов Т.Т.",
|
||
opencloud_username="test-user",
|
||
opencloud_access_token="",
|
||
)
|
||
|
||
assert result.status == "error"
|
||
assert not result.report_path
|
||
|
||
|
||
def test_run_local_zip_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_local_zip_batch(
|
||
zip_path=zip_path,
|
||
launcher="tester",
|
||
opencloud_space_id="space-1",
|
||
opencloud_folder="Тестов Т.Т.",
|
||
opencloud_username="",
|
||
opencloud_access_token="token-123",
|
||
)
|
||
|
||
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_folder="Тестов Т.Т.",
|
||
opencloud_username="test-user",
|
||
opencloud_access_token="token-123",
|
||
)
|
||
|
||
assert result.status == "error"
|
||
assert any("распаковать" in message for message in result.messages)
|