120 lines
4.0 KiB
Python
120 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
from app.pipeline.mapping import FIXED_REPORT_COLUMNS
|
|
from app.pipeline.report import TOP_GROUP_HEADERS, ReportRow, write_report
|
|
|
|
|
|
def test_write_report_streaming_preserves_header_and_formats(tmp_path: Path) -> None:
|
|
destination = tmp_path / "report.xlsx"
|
|
rows = [
|
|
ReportRow(
|
|
file_name="SKO115FZ_01_123456789_20260616_X00001.xml",
|
|
record_id="RID-1",
|
|
operation_index=1,
|
|
operation_fields={"СумОперации": "100.50", "ДатаСообщения": "2026-06-16"},
|
|
participant_fields={},
|
|
)
|
|
]
|
|
|
|
write_report(rows, destination)
|
|
|
|
workbook = load_workbook(destination)
|
|
sheet = workbook["Отчет"]
|
|
headers = [cell.value for cell in sheet[2]]
|
|
assert headers == list(FIXED_REPORT_COLUMNS)
|
|
assert sheet.freeze_panes == "A3"
|
|
row_values = [
|
|
sheet.cell(row=3, column=index).value
|
|
for index in range(1, len(FIXED_REPORT_COLUMNS) + 1)
|
|
]
|
|
assert "SKO115FZ_01_123456789_20260616_X00001.xml" in row_values
|
|
|
|
date_col_idx = FIXED_REPORT_COLUMNS.index("Дата сообщения") + 1
|
|
assert isinstance(sheet.cell(row=3, column=date_col_idx).value, datetime)
|
|
assert sheet.cell(row=3, column=date_col_idx).number_format == "DD.MM.YYYY"
|
|
|
|
sum_col_idx = FIXED_REPORT_COLUMNS.index("Сумма в валюте проведения") + 1
|
|
assert sheet.cell(row=3, column=sum_col_idx).number_format == "#,##0.00"
|
|
|
|
|
|
def test_write_report_contains_top_group_header_row(tmp_path: Path) -> None:
|
|
destination = tmp_path / "report.xlsx"
|
|
write_report([], destination)
|
|
|
|
workbook = load_workbook(destination)
|
|
sheet = workbook["Отчет"]
|
|
assert sheet.cell(row=1, column=1).value == "Служебные поля"
|
|
merged_ranges = {str(rng) for rng in sheet.merged_cells.ranges}
|
|
expected_ranges = set()
|
|
for header in TOP_GROUP_HEADERS:
|
|
if header.start_col > len(FIXED_REPORT_COLUMNS):
|
|
continue
|
|
end_col = min(header.end_col, len(FIXED_REPORT_COLUMNS))
|
|
if header.start_col == end_col:
|
|
continue
|
|
expected_ranges.add(
|
|
f"{_column_letter(header.start_col)}1:{_column_letter(end_col)}1"
|
|
)
|
|
assert merged_ranges == expected_ranges
|
|
|
|
|
|
def test_write_report_converts_date_string_to_excel_date(tmp_path: Path) -> None:
|
|
destination = tmp_path / "report.xlsx"
|
|
rows = [
|
|
ReportRow(
|
|
file_name="f.xml",
|
|
record_id="RID",
|
|
operation_index=1,
|
|
operation_fields={"ДатаСообщения": "23/08/2005"},
|
|
participant_fields={},
|
|
)
|
|
]
|
|
|
|
write_report(rows, destination)
|
|
|
|
workbook = load_workbook(destination)
|
|
sheet = workbook["Отчет"]
|
|
date_col_idx = FIXED_REPORT_COLUMNS.index("Дата сообщения") + 1
|
|
cell = sheet.cell(row=3, column=date_col_idx)
|
|
assert isinstance(cell.value, datetime)
|
|
assert cell.value.date() == datetime(2005, 8, 23).date()
|
|
assert cell.number_format == "DD.MM.YYYY"
|
|
|
|
|
|
def test_write_report_keeps_invalid_date_as_text_without_date_format(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
destination = tmp_path / "report.xlsx"
|
|
rows = [
|
|
ReportRow(
|
|
file_name="f.xml",
|
|
record_id="RID",
|
|
operation_index=1,
|
|
operation_fields={"ДатаСообщения": "не дата"},
|
|
participant_fields={},
|
|
)
|
|
]
|
|
|
|
write_report(rows, destination)
|
|
|
|
workbook = load_workbook(destination)
|
|
sheet = workbook["Отчет"]
|
|
date_col_idx = FIXED_REPORT_COLUMNS.index("Дата сообщения") + 1
|
|
cell = sheet.cell(row=3, column=date_col_idx)
|
|
assert cell.value == "не дата"
|
|
assert cell.number_format != "DD.MM.YYYY"
|
|
|
|
|
|
def _column_letter(index: int) -> str:
|
|
value = index
|
|
result = ""
|
|
while value:
|
|
value, remainder = divmod(value - 1, 26)
|
|
result = chr(65 + remainder) + result
|
|
return result
|