b5f7060a1f
Внесение данных из файлов приёма в реестр невостребованных документов и заполнение журнала. Дата внесения — текущая; журнал заполняется чисто (без строк-примеров шаблона). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
374 lines
14 KiB
Python
374 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Тесты граничных случаев и обработки исключений.
|
||
|
||
Запуск: pytest -q (из корня проекта)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from odf.opendocument import OpenDocumentSpreadsheet, load
|
||
from odf.style import Style, TableRowProperties
|
||
from odf.table import Table, TableCell, TableRow
|
||
from odf.text import P
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from priem_register.mapping import discover_sources # noqa: E402
|
||
from priem_register.models import Record # noqa: E402
|
||
from priem_register.readers import get_reader # noqa: E402
|
||
from priem_register.readers.base import SourceReader # noqa: E402
|
||
from priem_register.writers import JournalWriter, RegisterWriter # noqa: E402
|
||
|
||
|
||
# --- Record -------------------------------------------------------------
|
||
|
||
def test_record_value_out_of_range():
|
||
r = Record(values=["a", "b"])
|
||
assert r.value(1) == "a" # колонка B
|
||
assert r.value(2) == "b" # колонка C
|
||
assert r.value(12) is None # колонка M отсутствует -> None
|
||
assert r.value(99) is None
|
||
|
||
|
||
# --- get_reader ---------------------------------------------------------
|
||
|
||
def test_get_reader_unknown_extension(tmp_path):
|
||
bad = tmp_path / "file.txt"
|
||
bad.write_text("x")
|
||
with pytest.raises(ValueError):
|
||
get_reader(bad)
|
||
|
||
|
||
# --- mapping ------------------------------------------------------------
|
||
|
||
def test_discover_missing_dir(tmp_path):
|
||
with pytest.raises(FileNotFoundError):
|
||
discover_sources(tmp_path / "нет")
|
||
|
||
|
||
def test_discover_empty_dir(tmp_path):
|
||
assert discover_sources(tmp_path) == []
|
||
|
||
|
||
def test_discover_ignores_non_matching(tmp_path):
|
||
(tmp_path / "ТО1").mkdir()
|
||
(tmp_path / "ТО1" / "заметка.txt").write_text("x")
|
||
(tmp_path / "ТО1" / "отчёт_74_99.xlsx").write_bytes(b"") # нет даты
|
||
(tmp_path / "ТО1" / "09.04.2026_99_25.docx").write_bytes(b"") # чужой формат
|
||
assert discover_sources(tmp_path) == []
|
||
|
||
|
||
def test_discover_invalid_date_raises_clear_error(tmp_path):
|
||
(tmp_path / "ТО1").mkdir()
|
||
(tmp_path / "ТО1" / "32.04.2026_74_99.xlsx").write_bytes(b"") # дня 32 нет
|
||
with pytest.raises(ValueError, match="Некорректная дата"):
|
||
discover_sources(tmp_path)
|
||
|
||
|
||
def test_discover_matches_and_sorts(tmp_path):
|
||
d = tmp_path / "ТО2"
|
||
d.mkdir()
|
||
(d / "21.04.2026_74_03.xlsx").write_bytes(b"")
|
||
(d / "03.04.2026_74_03.xlsx").write_bytes(b"")
|
||
found = discover_sources(tmp_path)
|
||
assert [s.name for s in found] == ["03.04.2026_74_03", "21.04.2026_74_03"]
|
||
assert all(s.sheet == "ТО2_03" for s in found)
|
||
assert found[0].date < found[1].date
|
||
|
||
|
||
def test_discover_file_without_to_dir(tmp_path):
|
||
(tmp_path / "09.04.2026_74_25.ods").write_bytes(b"")
|
||
with pytest.raises(ValueError):
|
||
discover_sources(tmp_path)
|
||
|
||
|
||
# --- RegisterWriter на синтетическом .ods -------------------------------
|
||
|
||
def _make_register(tmp_path, sheet_name="ТО1_25", numbers=(1, 2),
|
||
trailing_block=True):
|
||
doc = OpenDocumentSpreadsheet()
|
||
rs = Style(name="ro1", family="table-row")
|
||
rs.addElement(TableRowProperties(rowheight="0.45cm"))
|
||
doc.automaticstyles.addElement(rs)
|
||
|
||
table = Table(name=sheet_name)
|
||
# шапка
|
||
head = TableRow()
|
||
head.addElement(_p_cell("№ п/п"))
|
||
table.addElement(head)
|
||
# строки данных
|
||
for n in numbers:
|
||
row = TableRow(stylename="ro1")
|
||
row.addElement(_float_cell(n))
|
||
for _ in range(12):
|
||
row.addElement(TableCell())
|
||
table.addElement(row)
|
||
if trailing_block:
|
||
empty = TableRow(numberrowsrepeated="100")
|
||
empty.addElement(TableCell(numbercolumnsrepeated="13"))
|
||
table.addElement(empty)
|
||
doc.spreadsheet.addElement(table)
|
||
|
||
path = tmp_path / "reg.ods"
|
||
doc.save(str(path))
|
||
return path
|
||
|
||
|
||
def _p_cell(text):
|
||
c = TableCell(valuetype="string")
|
||
c.addElement(P(text=text))
|
||
return c
|
||
|
||
|
||
def _float_cell(n):
|
||
c = TableCell(valuetype="float", value=str(n))
|
||
c.addElement(P(text=str(n)))
|
||
return c
|
||
|
||
|
||
def _read_numbers(path, sheet):
|
||
doc = load(str(path))
|
||
table = [t for t in doc.spreadsheet.getElementsByType(Table)
|
||
if t.getAttribute("name") == sheet][0]
|
||
nums = []
|
||
for tr in table.getElementsByType(TableRow):
|
||
if tr.getAttribute("numberrowsrepeated"):
|
||
continue
|
||
cells = tr.getElementsByType(TableCell)
|
||
if cells and cells[0].getAttribute("value"):
|
||
nums.append(int(float(cells[0].getAttribute("value"))))
|
||
return nums
|
||
|
||
|
||
def test_register_missing_sheet(tmp_path):
|
||
path = _make_register(tmp_path)
|
||
w = RegisterWriter(path)
|
||
assert w.has_sheet("ТО1_25")
|
||
assert not w.has_sheet("НЕТ")
|
||
|
||
|
||
def test_register_no_numbered_rows(tmp_path):
|
||
doc = OpenDocumentSpreadsheet()
|
||
table = Table(name="ТО1_25")
|
||
head = TableRow()
|
||
head.addElement(_p_cell("№ п/п"))
|
||
table.addElement(head)
|
||
doc.spreadsheet.addElement(table)
|
||
path = tmp_path / "empty.ods"
|
||
doc.save(str(path))
|
||
with pytest.raises(ValueError):
|
||
RegisterWriter(path).append("ТО1_25", [Record(["a"])])
|
||
|
||
|
||
def test_register_append_continues_numbering(tmp_path):
|
||
path = _make_register(tmp_path, numbers=(1, 2, 3))
|
||
w = RegisterWriter(path)
|
||
n = w.append("ТО1_25", [Record(["x"]), Record(["y"])])
|
||
out = tmp_path / "out.ods"
|
||
w.save(out)
|
||
assert n == 2
|
||
assert _read_numbers(out, "ТО1_25") == [1, 2, 3, 4, 5]
|
||
|
||
|
||
def test_register_append_empty_records(tmp_path):
|
||
path = _make_register(tmp_path, numbers=(1, 2))
|
||
w = RegisterWriter(path)
|
||
assert w.append("ТО1_25", []) == 0
|
||
out = tmp_path / "out.ods"
|
||
w.save(out)
|
||
assert _read_numbers(out, "ТО1_25") == [1, 2]
|
||
|
||
|
||
def test_register_template_is_last_element(tmp_path):
|
||
# нет хвостового пустого блока -> образец является последним элементом
|
||
path = _make_register(tmp_path, numbers=(1, 2), trailing_block=False)
|
||
w = RegisterWriter(path)
|
||
w.append("ТО1_25", [Record(["x"]), Record(["y"])])
|
||
out = tmp_path / "out.ods"
|
||
w.save(out)
|
||
assert _read_numbers(out, "ТО1_25") == [1, 2, 3, 4]
|
||
|
||
|
||
def test_register_preserves_order_and_types(tmp_path):
|
||
path = _make_register(tmp_path, numbers=(10,))
|
||
w = RegisterWriter(path)
|
||
rec = Record(["Адрес", "MFC-1", "объект", 5, datetime.date(2026, 4, 15)])
|
||
w.append("ТО1_25", [rec])
|
||
out = tmp_path / "out.ods"
|
||
w.save(out)
|
||
doc = load(str(out))
|
||
table = [t for t in doc.spreadsheet.getElementsByType(Table)
|
||
if t.getAttribute("name") == "ТО1_25"][0]
|
||
last = [tr for tr in table.getElementsByType(TableRow)
|
||
if not tr.getAttribute("numberrowsrepeated")][-1]
|
||
cells = last.getElementsByType(TableCell)
|
||
assert cells[0].getAttribute("value") == "11" # № продолжен
|
||
assert cells[4].getAttribute("valuetype") == "float" # кол-во
|
||
assert cells[5].getAttribute("valuetype") == "date" # дата
|
||
assert cells[5].getAttribute("datevalue") == "2026-04-15"
|
||
|
||
|
||
# --- JournalWriter ------------------------------------------------------
|
||
|
||
def test_journal_writes_under_header_and_clears_examples(tmp_path):
|
||
import openpyxl
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.append(["Дата внесения", "Файл внесения", "Количество записей"])
|
||
ws.append(["Например", None, None])
|
||
ws.append([datetime.date(2026, 4, 22), "пример_файл", 99])
|
||
path = tmp_path / "j.xlsx"
|
||
wb.save(path)
|
||
|
||
from priem_register.models import JournalEntry
|
||
w = JournalWriter(path)
|
||
w.write([JournalEntry(datetime.date(2026, 6, 8), "файл_1", 5)])
|
||
out = tmp_path / "jout.xlsx"
|
||
w.save(out)
|
||
|
||
wb2 = openpyxl.load_workbook(out)
|
||
ws2 = wb2.active
|
||
# шапка на месте
|
||
assert ws2.cell(row=1, column=1).value == "Дата внесения"
|
||
# реальная запись сразу под шапкой (строка 2), а не после примера
|
||
assert ws2.cell(row=2, column=2).value == "файл_1"
|
||
assert ws2.cell(row=2, column=3).value == 5
|
||
# строки-примеры затёрты
|
||
assert ws2.cell(row=3, column=2).value is None
|
||
assert ws2.cell(row=3, column=3).value is None
|
||
|
||
|
||
# --- аномалии чтения: данные не теряются молча ---------------------------
|
||
|
||
class _FakeReader(SourceReader):
|
||
def __init__(self, rows):
|
||
super().__init__("fake.xlsx")
|
||
self._fake = rows
|
||
|
||
def _rows(self):
|
||
return self._fake
|
||
|
||
|
||
def test_reader_enters_all_rows_from_second():
|
||
# задание: «внести все строки, начиная со второй» — без фильтра по №
|
||
rows = [
|
||
["№ п/п", "Адрес"], # шапка — пропускается
|
||
[1, "Адрес", "MFC-1"], # запись с №
|
||
[None, None, None, None, None, None, None, None, None, None,
|
||
"08.04", "покупатель"], # строка без № — тоже запись
|
||
]
|
||
res = _FakeReader(rows).read()
|
||
assert len(res.records) == 2
|
||
assert res.records[0].value(2) == "MFC-1"
|
||
assert res.records[1].value(10) == "08.04" # Дата выдачи
|
||
assert res.records[1].value(11) == "покупатель" # Примечание
|
||
assert res.warnings == []
|
||
assert res.notes == []
|
||
|
||
|
||
def test_reader_extra_columns_merged_into_note():
|
||
rows = [
|
||
["№ п/п"],
|
||
[1] + ["v%d" % i for i in range(1, 12)] + ["реестр", "ХВОСТ"], # 14 колонок
|
||
]
|
||
res = _FakeReader(rows).read()
|
||
assert len(res.records) == 1
|
||
assert res.records[0].value(12) == "реестр" # колонка M на месте
|
||
assert res.records[0].value(11).endswith("ХВОСТ") # N ушёл в Примечание
|
||
assert res.warnings == []
|
||
assert len(res.notes) == 1
|
||
|
||
|
||
def test_reader_skips_header_and_empty_rows():
|
||
rows = [
|
||
["№ п/п"], # шапка
|
||
[1, "a"], # запись
|
||
[None, None, None], # пусто — пропускается
|
||
["", "", ""], # пусто — пропускается
|
||
[2, "b"], # запись
|
||
]
|
||
res = _FakeReader(rows).read()
|
||
assert len(res.records) == 2
|
||
assert res.warnings == []
|
||
assert res.notes == []
|
||
|
||
|
||
def test_reader_header_not_flagged():
|
||
rows = [["№ п/п", "Адрес", "Рег"]] # только шапка
|
||
res = _FakeReader(rows).read()
|
||
assert res.records == []
|
||
assert res.warnings == []
|
||
|
||
|
||
# --- OdsReader: текст внутри span/кратные пробелы не теряется ------------
|
||
|
||
def test_ods_reader_extracts_span_and_spaces(tmp_path):
|
||
from odf.opendocument import OpenDocumentSpreadsheet
|
||
from odf.table import Table as OTable, TableCell as OCell, TableRow as ORow
|
||
from odf.text import P as OP, S as OS, Span as OSpan
|
||
from priem_register.readers.ods_reader import OdsReader
|
||
|
||
doc = OpenDocumentSpreadsheet()
|
||
table = OTable(name="ТО1_25")
|
||
# шапка
|
||
h = ORow(); hc = OCell(valuetype="string"); hc.addElement(OP(text="№ п/п"))
|
||
h.addElement(hc); table.addElement(h)
|
||
# строка данных: объект с span и тройным пробелом
|
||
r = ORow()
|
||
c0 = OCell(valuetype="float", value="1"); c0.addElement(OP(text="1"))
|
||
r.addElement(c0)
|
||
p = OP(text="74:25:1 ")
|
||
p.addElement(OS(c=3)) # три неразрывных пробела ODF
|
||
p.addElement(OSpan(text="ЗЛАТОУСТ")) # текст внутри span
|
||
cell = OCell(valuetype="string"); cell.addElement(p)
|
||
r.addElement(cell)
|
||
table.addElement(r)
|
||
doc.spreadsheet.addElement(table)
|
||
path = tmp_path / "spans.ods"
|
||
doc.save(str(path))
|
||
|
||
res = OdsReader(path).read()
|
||
assert len(res.records) == 1
|
||
assert res.records[0].value(1) == "74:25:1 ЗЛАТОУСТ" # span + 3 пробела (колонка B)
|
||
|
||
|
||
# --- интеграционные тесты на реальных данных ----------------------------
|
||
|
||
def test_full_pipeline_real_data(tmp_path):
|
||
from priem_register import Pipeline
|
||
reg_out = tmp_path / "reg_out.ods"
|
||
jour_out = tmp_path / "jour_out.xlsx"
|
||
p = Pipeline(register_out=reg_out, journal_out=jour_out)
|
||
res = p.run()
|
||
assert res.total == 523
|
||
assert len(res.entries) == 12
|
||
assert reg_out.exists() and jour_out.exists()
|
||
assert all(e.count > 0 for e in res.entries)
|
||
assert res.warnings == []
|
||
# все непустые строки внесены (включая «уведомление»/«покупатель» как записи)
|
||
counts = {e.file_name: e.count for e in res.entries}
|
||
assert counts["09.04.2026_74_34"] == 66 # 62 + 4×уведомление
|
||
assert counts["17.04.2026_74_25"] == 46 # 45 + покупатель
|
||
# единственная заметка — данные за колонкой M (закладная) в «Примечание»
|
||
assert len(res.notes) == 1
|
||
assert "закладная" in res.notes[0]
|
||
|
||
|
||
def test_pipeline_idempotent(tmp_path):
|
||
from priem_register import Pipeline
|
||
totals = []
|
||
for i in range(2):
|
||
p = Pipeline(register_out=tmp_path / ("r%d.ods" % i),
|
||
journal_out=tmp_path / ("j%d.xlsx" % i))
|
||
totals.append(p.run().total)
|
||
assert totals[0] == totals[1] == 523
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(pytest.main([__file__, "-q"]))
|