91 lines
3.9 KiB
Python
91 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import unittest
|
|
from dataclasses import replace
|
|
from io import BytesIO
|
|
from unittest.mock import patch
|
|
from zipfile import ZipFile
|
|
|
|
from openpyxl import Workbook
|
|
|
|
from govoplan_core.core.tabular_sources import TabularSourceValidationError
|
|
from govoplan_connectors.backend import tabular_adapters as adapters
|
|
|
|
|
|
def workbook_bytes(*, dimension: str | None = None, last_cell: str = "A2") -> bytes:
|
|
workbook = Workbook()
|
|
workbook.active.append(["name"])
|
|
workbook.active.append(["Ada"])
|
|
output = BytesIO()
|
|
workbook.save(output)
|
|
workbook.close()
|
|
result = BytesIO()
|
|
with ZipFile(BytesIO(output.getvalue())) as original, ZipFile(result, "w") as modified:
|
|
for entry in original.infolist():
|
|
content = original.read(entry)
|
|
if entry.filename == "xl/worksheets/sheet1.xml":
|
|
xml = content.decode()
|
|
if dimension is not None:
|
|
xml = re.sub(r'<dimension ref="[^"]+"\s*/>', f'<dimension ref="{dimension}"/>', xml)
|
|
xml = xml.replace('r="A2"', f'r="{last_cell}"')
|
|
xml = xml.replace('<row r="2">', f'<row r="{re.search(r"[0-9]+$", last_cell).group()}">')
|
|
content = xml.encode()
|
|
modified.writestr(entry, content)
|
|
return result.getvalue()
|
|
|
|
|
|
def parse(payload: bytes):
|
|
return adapters.parse_managed_tabular_content(payload, filename="fixture.xlsx", content_type=None, delimiter=",", sheet_name=None)
|
|
|
|
|
|
class XlsxSafetyBoundsTests(unittest.TestCase):
|
|
def test_sparse_rows_cannot_bypass_limit_by_not_counting_as_data(self):
|
|
with self.assertRaisesRegex(TabularSourceValidationError, "row"):
|
|
parse(workbook_bytes(last_cell=f"A{adapters.MAX_FILE_ROWS + 2}"))
|
|
|
|
def test_forged_small_dimensions_cannot_hide_far_away_cells(self):
|
|
with self.assertRaisesRegex(TabularSourceValidationError, "row"):
|
|
parse(workbook_bytes(dimension="A1:A2", last_cell="A1000000"))
|
|
|
|
def test_forged_small_dimensions_cannot_hide_out_of_range_columns(self):
|
|
with self.assertRaisesRegex(TabularSourceValidationError, "column"):
|
|
parse(workbook_bytes(dimension="A1:A2", last_cell="XFD2"))
|
|
|
|
def test_declared_dimensions_do_not_expand_or_truncate_real_rows(self):
|
|
for dimension in ("A1:XFD1048576", "A1:A1"):
|
|
with self.subTest(dimension=dimension):
|
|
rows, sheet = parse(workbook_bytes(dimension=dimension))
|
|
self.assertEqual(({"name": "Ada"},), rows)
|
|
self.assertEqual("Sheet", sheet)
|
|
|
|
def test_small_blank_gaps_and_exact_limit_remain_usable(self):
|
|
for cell in ("A4", f"A{adapters.MAX_FILE_ROWS + 1}"):
|
|
rows, _sheet = parse(workbook_bytes(last_cell=cell))
|
|
self.assertEqual(({"name": "Ada"},), rows)
|
|
|
|
def test_workbook_is_parsed_in_a_fresh_child_not_the_parent(self):
|
|
with patch.object(adapters, "_parse_xlsx_content", side_effect=AssertionError("parent parser ran")):
|
|
rows, sheet = parse(workbook_bytes())
|
|
self.assertEqual(({"name": "Ada"},), rows)
|
|
self.assertEqual("Sheet", sheet)
|
|
|
|
def test_real_worker_timeout_fails_without_parent_fallback(self):
|
|
limits = replace(adapters.XLSX_PROCESS_LIMITS, wall_seconds=0.001)
|
|
with patch.object(adapters, "XLSX_PROCESS_LIMITS", limits), patch.object(
|
|
adapters, "_parse_xlsx_content", side_effect=AssertionError("parent parser ran")
|
|
):
|
|
with self.assertRaisesRegex(TabularSourceValidationError, "timeout"):
|
|
parse(workbook_bytes())
|
|
|
|
def test_child_validation_keeps_missing_sheet_message(self):
|
|
with self.assertRaisesRegex(TabularSourceValidationError, "worksheet 'Missing' was not found"):
|
|
adapters.parse_managed_tabular_content(
|
|
workbook_bytes(), filename="fixture.xlsx", content_type=None,
|
|
delimiter=",", sheet_name="Missing",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|