shanxia2.0/backend/tests/test_pl_plan_archive_zip_export.py

import io
import zipfile

from openpyxl import load_workbook

from app.services.pl_plan_archive_zip_export import (
    sanitize_zip_segment,
    build_attachment_filename,
    first_row_no_by_plan_id,
    build_hyperlink_formula,
    parse_url_list,
    collect_hazard_cert_items,
    merge_plan_attachment_items,
    build_archive_zip_bytes,
)
from app.utils.excel_export import ExportColumn


def test_sanitize_zip_segment():
    assert sanitize_zip_segment("装货·司机/A") == "装货·司机_A"
    assert sanitize_zip_segment("  a  ") == "a"


def test_build_attachment_filename():
    name = build_attachment_filename(1, "装货·司机", "https://cdn.example/path/a.jpg")
    assert name.startswith("01_装货·司机_")
    assert name.endswith(".jpg")


def test_first_row_no_by_plan_id():
    rows = [
        {"planId": "p1", "rowNo": 1},
        {"planId": "p1", "rowNo": 2},
        {"planId": "p2", "rowNo": 3},
    ]
    assert first_row_no_by_plan_id(rows) == {"p1": 1, "p2": 3}


def test_build_hyperlink_formula():
    f = build_hyperlink_formula("附件/01_司机_ab12.jpg", "司机")
    assert f.startswith('=HYPERLINK(')
    assert "附件/01_司机_ab12.jpg" in f


def test_parse_url_list_csv_and_json():
    assert parse_url_list("a.jpg,b.jpg") == ["a.jpg", "b.jpg"]
    assert parse_url_list('["x.png","y.png"]') == ["x.png", "y.png"]


def test_collect_hazard_cert_items_labels():
    class P:
        product_name = "甲醇"
        purchase_cert_url = "https://x/p.pdf"
        transport_cert_url = None
        legal_use_cert_url = "https://x/l1.pdf,https://x/l2.pdf"

    items = collect_hazard_cert_items([P()])
    labels = [i["label"] for i in items]
    assert "购买证" in labels[0]
    assert any("合法使用证明" in x for x in labels)
    assert len(items) == 3


def test_merge_dedupes_url():
    doc = [{"url": "https://a/1.jpg", "label": "司机"}]
    haz = [{"url": "https://a/1.jpg", "label": "购买证"}]
    merged = merge_plan_attachment_items(doc, haz)
    assert len(merged) == 1
    assert merged[0]["label"] == "司机"


def test_build_archive_zip_bytes_structure_and_hyperlink():
    columns = [
        ExportColumn(title="序号", field="rowNo"),
        ExportColumn(title="计划编号", field="planNo"),
    ]
    rows = [
        {"planId": "p1", "rowNo": 1, "planNo": "PL001"},
        {"planId": "p1", "rowNo": 2, "planNo": "PL001"},
    ]
    attachments_by_plan = {
        "p1": [{"url": "https://cdn.example/a.jpg", "label": "司机"}],
    }

    def fake_download(url: str) -> bytes:
        return b"fake-image-bytes"

    stamp = "202609231230"
    data, failures = build_archive_zip_bytes(
        db=None,
        tenant_no="t1",
        columns=columns,
        rows=rows,
        attachments_by_plan=attachments_by_plan,
        stamp=stamp,
        download_fn=fake_download,
    )
    assert failures == []
    with zipfile.ZipFile(io.BytesIO(data)) as zf:
        names = zf.namelist()
        assert f"计划归档({stamp})/(总表)计划归档.xlsx" in names
        att = [n for n in names if n.startswith(f"计划归档({stamp})/附件/")]
        assert len(att) == 1
        assert "01_司机_" in att[0]
        xlsx_bytes = zf.read(f"计划归档({stamp})/(总表)计划归档.xlsx")
    wb = load_workbook(io.BytesIO(xlsx_bytes))
    ws = wb.active
    headers = [c.value for c in ws[1]]
    assert "附件_1" in headers
    attach_col = headers.index("附件_1") + 1
    assert str(ws.cell(2, attach_col).value).startswith("=HYPERLINK(")
    assert ws.cell(3, attach_col).value in (None, "")