289 lines
16 KiB
Python
289 lines
16 KiB
Python
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()
|