SFM/tests/unit/test_attachments.py

from __future__ import annotations

import tarfile
import zipfile
from pathlib import Path

import pytest

from app.pipeline.attachments import (
    AttachmentError,
    collect_xml_attachments,
    extract_archive,
)

XML_BODY = "<?xml version='1.0' encoding='UTF-8'?><СообщОперКО/>"


def _zip(path: Path, members: dict[str, str]) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(path, "w") as archive:
        for name, content in members.items():
            archive.writestr(name, content)
    return path


def _xml(path: Path) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(XML_BODY, encoding="utf-8")
    return path


def test_collects_loose_xml_and_archives_together(tmp_path: Path) -> None:
    loose = _xml(tmp_path / "loose" / "a.xml")
    archive = _zip(tmp_path / "batch.zip", {"b.xml": XML_BODY, "notes.txt": "skip me"})
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([loose, archive], extract_root)

    assert [item.file_name for item in result.files] == ["a.xml", "b.xml"]
    assert result.loose_xml_files == 1
    assert result.archives == 1
    assert result.problems == []
    # У каждого файла видно, из какого вложения он пришёл.
    assert {item.source for item in result.files} == {"a.xml", "batch.zip"}


def test_finds_xml_in_nested_folders(tmp_path: Path) -> None:
    archive = _zip(
        tmp_path / "batch.zip",
        {"one/two/three/deep.xml": XML_BODY, "one/shallow.xml": XML_BODY},
    )
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([archive], extract_root)

    assert sorted(item.file_name for item in result.files) == [
        "deep.xml",
        "shallow.xml",
    ]


def test_same_file_name_in_two_archives_is_collected_once(tmp_path: Path) -> None:
    first = _zip(tmp_path / "first.zip", {"same.xml": XML_BODY})
    second = _zip(tmp_path / "second.zip", {"same.xml": XML_BODY})
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([first, second], extract_root)

    assert [item.file_name for item in result.files] == ["same.xml"]
    assert len(result.duplicates) == 1
    assert "first.zip" in result.duplicates[0].reason
    assert result.duplicates[0].source == "second.zip"
    # Имя самого файла нужно протоколу: строка «пропущен» должна стоять
    # под тем же именем, что и обработанная.
    assert result.duplicates[0].file_name == "same.xml"


def test_archives_with_equal_names_do_not_overwrite_each_other(tmp_path: Path) -> None:
    """Два вложения могут называться одинаково — распаковываться они должны
    в разные каталоги, иначе одно затрёт другое."""
    first = _zip(tmp_path / "a" / "Файлы.zip", {"one.xml": XML_BODY})
    second = _zip(tmp_path / "b" / "Файлы.zip", {"two.xml": XML_BODY})
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([first, second], extract_root)

    assert sorted(item.file_name for item in result.files) == ["one.xml", "two.xml"]
    assert result.problems == []


def test_tar_gz_archive_is_supported(tmp_path: Path) -> None:
    xml_path = _xml(tmp_path / "src" / "in_tar.xml")
    archive = tmp_path / "batch.tar.gz"
    with tarfile.open(archive, "w:gz") as tar:
        tar.add(xml_path, arcname=xml_path.name)
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([archive], extract_root)

    assert [item.file_name for item in result.files] == ["in_tar.xml"]
    assert result.archives == 1


def test_format_is_detected_by_content_not_extension(tmp_path: Path) -> None:
    """Выгрузки приезжают и с неверным суффиксом — формат определяем по
    содержимому."""
    archive = _zip(tmp_path / "actually_a_zip.bin", {"x.xml": XML_BODY})
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([archive], extract_root)

    assert [item.file_name for item in result.files] == ["x.xml"]


def test_broken_attachment_is_reported_without_stopping_the_rest(
    tmp_path: Path,
) -> None:
    good = _zip(tmp_path / "good.zip", {"ok.xml": XML_BODY})
    broken = tmp_path / "broken.zip"
    broken.write_bytes(b"definitely not an archive")
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([broken, good], extract_root)

    assert [item.file_name for item in result.files] == ["ok.xml"]
    assert len(result.problems) == 1
    assert result.problems[0].source == "broken.zip"


def test_unsupported_archive_format_explains_itself(tmp_path: Path) -> None:
    archive = tmp_path / "batch.7z"
    archive.write_bytes(b"7z\xbc\xaf\x27\x1c not really")
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([archive], extract_root)

    assert result.files == []
    assert ".7z" in result.problems[0].reason


def test_zip_with_path_traversal_member_is_rejected(tmp_path: Path) -> None:
    """Защита от zip slip: член архива не должен попадать за пределы каталога
    распаковки."""
    archive = tmp_path / "evil.zip"
    with zipfile.ZipFile(archive, "w") as zip_file:
        zip_file.writestr("../../escaped.xml", XML_BODY)
    extract_root = tmp_path / "extract"
    extract_root.mkdir()

    result = collect_xml_attachments([archive], extract_root)

    assert not (tmp_path.parent / "escaped.xml").exists()
    # Имя очищается до безопасного, файл остаётся внутри каталога распаковки.
    for item in result.files:
        assert extract_root.resolve() in item.path.resolve().parents


def test_extract_archive_raises_for_plain_file(tmp_path: Path) -> None:
    plain = tmp_path / "notes.txt"
    plain.write_text("just text", encoding="utf-8")
    target = tmp_path / "out"
    target.mkdir()

    with pytest.raises(AttachmentError):
        extract_archive(plain, target)


def test_empty_attachment_list_gives_empty_result(tmp_path: Path) -> None:
    result = collect_xml_attachments([], tmp_path)

    assert result.files == []
    assert result.problems == []
    assert result.archives == 0
    assert result.loose_xml_files == 0