import copy import hashlib import json import os import subprocess import sys import tempfile import unittest import zipfile from pathlib import Path from lxml import etree as ET from openpyxl import load_workbook sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from expense_template import (CONTENT, MAIN, PACKAGE, REL, Template, encoded, export_custom_expense, inspect_template, set_cell, tag, validate_template, xml) from exports import export_expense from test_domain import material, workspace class ExpenseTemplateTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name) self.builtin = Path(__file__).resolve().parents[1] / 'personal-expense-template.xlsx' self.state = workspace([], []) self.state['matches'] = [self.match('traffic-1', '交通', '30.10'), self.match('traffic-2', '交通', '70.20'), self.match('office', '办公用品', '444.00')] self.template = self.root / '旧报销单.xlsx' old = dict(matches=[self.match('old-' + str(index), '旧类别' + str(index), '999.00') for index in range(7)]) export_expense(old, self.template, self.builtin, {}, ['旧岗位'], dict(recipient='旧收款人', bankName='旧开户行', accountNumber='9999999', preparer='旧制单人')) self.mapping = inspect_template(self.template)['sheets'][0]['mapping'] def tearDown(self): self.temporary.cleanup() def match(self, identifier, category, amount): return dict(id=identifier, category=category, invoices=[material(identifier, 'invoice', amount)], payments=[]) def export(self, mapping=None, template=None, state=None, payee=None): output = self.root / 'output.xlsx' metadata = export_custom_expense(state or self.state, output, template or self.template, mapping or self.mapping, {}, ['经办人', '财务'], payee or {}, {'交通': '交通', '办公用品': '=办公用品'}) return output, metadata def rewrite(self, path, parts): with zipfile.ZipFile(path, 'w', zipfile.ZIP_DEFLATED) as archive: for name, data in parts.items(): archive.writestr(name, data) def alternate(self): template = Template(self.template) sheet, path, _ = template.sheet('个人报销单') sheet.set('name', '组 B 报销') document = ET.Element(tag('worksheet'), nsmap={None: MAIN}) ET.SubElement(document, tag('sheetData')) values = {'A1': '自定义组 B 报销单', 'A4': '序号', 'B4': '用途', 'D4': '单据数量', 'F4': '金额', 'G4': '备注', 'B5': '旧内容1', 'F5': 500, 'G5': '旧备注', 'B6': '旧内容2', 'F6': 800, 'A10': '收款人', 'B10': '旧收款人', 'A11': '开户行', 'B11': '旧银行', 'A12': '账号', 'B12': '旧账号', 'E2': '制单人', 'F2': '旧人', 'E8': '报销合计', 'A14': '要清空的旧值', 'A15': '旧签字', 'F15': '旧岗位', 'A16': '保留的自定义说明'} for reference, value in values.items(): set_cell(document, reference, value) set_cell(document, 'F8', 'SUM(F5:F6)', formula=True) merges = ET.SubElement(document, tag('mergeCells'), count='3') for reference in ['B4:C4', 'B5:C5', 'B6:C6']: ET.SubElement(merges, tag('mergeCell'), ref=reference) ET.SubElement(document, tag('pageMargins'), left='0.2', right='0.2', top='0.3', bottom='0.3', header='0', footer='0') ET.SubElement(document, tag('pageSetup'), paperSize='9', orientation='landscape', fitToWidth='1') template.parts[path] = encoded(document) names = template.workbook.find(tag('definedNames')) for item in list(names): names.remove(item) ET.SubElement(names, tag('definedName'), name='_xlnm.Print_Area', localSheetId='0').text = "'组 B 报销'!$A$1:$G$16" sheet_list = template.workbook.find(tag('sheets')) ET.SubElement(sheet_list, tag('sheet'), name='保留说明', sheetId='2', attrib={'{' + REL + '}id': 'rIdNotes'}) ET.SubElement(template.relations, '{' + PACKAGE + '}Relationship', Id='rIdNotes', Type=REL + '/worksheet', Target='worksheets/notes.xml') ET.SubElement(template.types, '{' + CONTENT + '}Override', PartName='/xl/worksheets/notes.xml', ContentType='application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml') notes = ET.Element(tag('worksheet'), nsmap={None: MAIN}) ET.SubElement(notes, tag('sheetData')) set_cell(notes, 'A1', '保持这个工作表不变') template.parts['xl/worksheets/notes.xml'] = encoded(notes) template.parts['xl/workbook.xml'] = encoded(template.workbook) template.parts['xl/_rels/workbook.xml.rels'] = encoded(template.relations) template.parts['[Content_Types].xml'] = encoded(template.types) target = self.root / 'different-layout.xlsx' self.rewrite(target, template.parts) mapping = inspect_template(target)['sheets'][0]['mapping'] mapping['signatureCells'] = 'A15,F15' mapping['clearCells'] = 'A14' return target, mapping def test_auto_detect_uploaded_style_and_merged_signature_slots(self): self.assertEqual(self.mapping['sheetName'], '个人报销单') self.assertEqual((self.mapping['startRow'], self.mapping['endRow']), (9, 21)) self.assertEqual(self.mapping['columns'], dict(purpose='C', amount='H', count='G', invoiceType='B', sequence='A', remarks='I')) self.assertEqual(self.mapping['cells'], dict(recipient='C23', bankName='C24', accountNumber='C25', preparer='H6', total='F27')) self.assertNotIn('D32', self.mapping['signatureCells']) validation = validate_template(self.template, self.mapping) self.assertEqual(validation['mapping'], self.mapping) self.assertEqual(len(validation['fingerprint']), 64) def test_replaces_old_seven_rows_with_two_aggregated_categories(self): original = self.template.read_bytes() output, metadata = self.export() self.assertEqual(metadata['expenseRowCount'], 2) self.assertEqual(metadata['expensePageCount'], 1) workbook = load_workbook(output) sheet = workbook.active self.assertEqual(sheet['C9'].value, '交通') self.assertEqual(sheet['H9'].value, 100.30) self.assertEqual(sheet['G9'].value, 2) self.assertEqual(sheet['C10'].value, '=办公用品') self.assertEqual(sheet['C10'].data_type, 's') self.assertEqual(sheet['H10'].value, 444) for row in range(11, 22): for column in ['A', 'B', 'C', 'G', 'H', 'I']: self.assertIn(sheet[f'{column}{row}'].value, ('', None)) for reference in ['C23', 'C24', 'C25', 'H6']: self.assertIn(sheet[reference].value, ('', None)) self.assertEqual(sheet['F27'].value, '=SUM(H9:H21)') self.assertEqual(sheet['A30'].value, '经办人：') self.assertEqual(sheet['D30'].value, '财务：') self.assertEqual(self.template.read_bytes(), original) workbook.close() def test_preserves_styles_merges_print_settings_and_media(self): output, _ = self.export() with zipfile.ZipFile(self.template) as original, zipfile.ZipFile(output) as generated: for name in original.namelist(): if name == 'xl/styles.xml' or name.startswith('xl/media/') or name.startswith('xl/drawings/'): self.assertEqual(original.read(name), generated.read(name), name) old, new = load_workbook(self.template), load_workbook(output) for reference in ['A1', 'A3', 'A4', 'C9', 'H9', 'C25']: self.assertEqual(old.active[reference].style_id, new.active[reference].style_id) self.assertEqual(str(old.active.merged_cells), str(new.active.merged_cells)) self.assertEqual(str(old.active.print_area), str(new.active.print_area)) self.assertEqual(old.active.page_setup, new.active.page_setup) self.assertEqual(old.active['A3'].value, new.active['A3'].value) self.assertEqual(old.active['A4'].value, new.active['A4'].value) self.assertEqual(old.active['D29'].value, new.active['D29'].value) old.close() new.close() def test_alternate_layout_preserves_other_sheet_and_paginates(self): template, mapping = self.alternate() self.assertEqual(mapping['columns']['purpose'], 'B') self.assertEqual((mapping['startRow'], mapping['endRow']), (5, 6)) self.assertEqual(mapping['cells']['total'], 'F8') state = dict(matches=self.state['matches'] + [self.match('hotel', '住宿', '9.99')]) output, result = self.export(mapping, template, state, dict(recipient='新收款人', bankName='=新银行', accountNumber='0001234567890123456789')) self.assertEqual(result['expensePageCount'], 2) workbook = load_workbook(output) self.assertEqual(workbook.sheetnames, ['组 B 报销', '保留说明', '组 B 报销-续2']) self.assertEqual(workbook['保留说明']['A1'].value, '保持这个工作表不变') for sheet in [workbook['组 B 报销'], workbook['组 B 报销-续2']]: self.assertEqual(sheet['B10'].value, '新收款人') self.assertEqual(sheet['F2'].value, '新收款人') self.assertEqual(sheet['B11'].data_type, 's') self.assertEqual(sheet['B12'].value, '0001234567890123456789') self.assertEqual(sheet['B12'].data_type, 's') self.assertEqual(sheet['F8'].value, '=SUM(F5:F6)') self.assertEqual(sheet['A16'].value, '保留的自定义说明') self.assertIn(sheet['A14'].value, ('', None)) self.assertEqual(sheet['A15'].value, '经办人：') self.assertEqual(sheet.page_setup.orientation, 'landscape') self.assertTrue(str(sheet.print_area).endswith('!$A$1:$G$16')) self.assertEqual(workbook['组 B 报销-续2']['B5'].value, '住宿') self.assertEqual(workbook['组 B 报销-续2']['A5'].value, 3) self.assertIn(workbook['组 B 报销-续2']['B6'].value, ('', None)) workbook.close() def test_unmapped_signature_cells_preserve_template(self): mapping = copy.deepcopy(self.mapping) mapping['signatureCells'] = '' output, _ = self.export(mapping) book = load_workbook(output) self.assertEqual(book.active['A30'].value, '旧岗位：') book.close() def test_mapping_normalizes_lowercase(self): mapping = copy.deepcopy(self.mapping) mapping['columns']['purpose'] = ' c ' mapping['cells']['recipient'] = ' c23 ' mapping['signatureCells'] = 'a30， d30' result = validate_template(self.template, mapping) self.assertEqual(result['mapping']['columns']['purpose'], 'C') self.assertEqual(result['mapping']['cells']['recipient'], 'C23') self.assertEqual(result['mapping']['signatureCells'], 'A30,D30') def test_invalid_mappings_do_not_write_output(self): mutations = [ lambda mapping: mapping.update(startRow=22, endRow=9), lambda mapping: mapping.update(startRow=0), lambda mapping: mapping.update(endRow=300), lambda mapping: mapping.update(sheetName='不存在'), lambda mapping: mapping['columns'].update(purpose='D'), lambda mapping: mapping['columns'].update(count='H'), lambda mapping: mapping['columns'].update(amount=''), lambda mapping: mapping['columns'].update(amount='A1'), lambda mapping: mapping['cells'].update(recipient='C9'), lambda mapping: mapping['cells'].update(bankName='C23'), lambda mapping: mapping.update(signatureCells='D32'), lambda mapping: mapping.update(clearCells='D23'), lambda mapping: mapping.update(clearCells='C9'), ] for mutate in mutations: mapping = copy.deepcopy(self.mapping) mutate(mapping) with self.subTest(mapping=mapping), self.assertRaises(ValueError): self.export(mapping) self.assertFalse((self.root / 'output.xlsx').exists()) def test_corrupt_and_non_xlsx_rejected(self): for name in ['bad.xlsx', 'old.xls', 'macro.xlsm']: path = self.root / name path.write_bytes(b'not an xlsx') with self.assertRaises(ValueError): inspect_template(path) def test_live_external_formula_rejected_but_orphans_removed(self): template = Template(self.template) _, path, document = template.sheet('个人报销单') set_cell(document, 'J1', "'[1]Sheet1'!A1", formula=True) template.parts[path] = encoded(document) live = self.root / 'external.xlsx' self.rewrite(live, template.parts) with self.assertRaisesRegex(ValueError, '外部'): inspect_template(live) output, _ = self.export() with zipfile.ZipFile(output) as archive: self.assertFalse(any(name.startswith('xl/externalLinks/') for name in archive.namelist())) def test_formula_caches_are_invalidated(self): output, _ = self.export() template = Template(output) _, _, document = template.sheet('个人报销单') self.assertTrue(list(document.iter(tag('f')))) for cell in document.iter(tag('c')): if cell.find(tag('f')) is not None: self.assertIsNone(cell.find(tag('v'))) self.assertEqual(template.workbook.find(tag('calcPr')).get('calcMode'), 'auto') self.assertNotIn('xl/calcChain.xml', template.parts) def engine(self, request): binary = os.environ.get('RECEIPT_ENGINE_BINARY') command = [binary] if binary else [sys.executable, str(Path(__file__).resolve().parents[1] / 'engine.py')] process = subprocess.run(command, input=json.dumps(request), capture_output=True, text=True, timeout=60) return process, json.loads(process.stdout.splitlines()[-1]) def test_engine_process_inspects_validates_and_uses_custom_template(self): process, inspected = self.engine(dict(operation='inspect-expense-template', templatePath=str(self.template))) self.assertEqual(process.returncode, 0, process.stderr) self.assertEqual(inspected['result']['sheets'][0]['mapping'], self.mapping) process, validated = self.engine(dict(operation='validate-expense-template', templatePath=str(self.template), templateMapping=self.mapping)) self.assertEqual(process.returncode, 0, process.stderr) output = self.root / 'protocol.xlsx' process, result = self.engine(dict(operation='expense', state=self.state, destination=str(output), templatePath=str(self.template), templateMapping=self.mapping, payee=dict(recipient='新用户', accountNumber='0009876543210987654321'))) self.assertEqual(process.returncode, 0, process.stderr) self.assertEqual(result['result']['templateFingerprint'], hashlib.sha256(self.template.read_bytes()).hexdigest()) self.assertEqual(result['result']['templateMappingDigest'], validated['result']['mappingDigest']) self.assertEqual(result['result']['expenseRowCount'], 2) book = load_workbook(output) self.assertEqual(book.active['C23'].value, '新用户') self.assertEqual(book.active['H9'].value, 100.30) book.close() def test_engine_process_missing_custom_template_never_falls_back(self): output = self.root / 'existing.xlsx' output.write_bytes(b'keep existing output') process, result = self.engine(dict(operation='expense', state=self.state, destination=str(output), templatePath=str(self.root / 'missing.xlsx'), templateMapping=self.mapping)) self.assertNotEqual(process.returncode, 0) self.assertEqual(result['event'], 'error') self.assertEqual(output.read_bytes(), b'keep existing output') self.assertFalse(list(self.root.glob('.*.xlsx'))) if __name__ == '__main__': unittest.main()