This commit is contained in:
csj
2026-09-13 15:20:40 +08:00
parent 8bc9ae8046
commit da6cb24249
8 changed files with 566 additions and 4 deletions
+101
View File
@@ -0,0 +1,101 @@
import copy
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from PIL import Image
from pptx import Presentation
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from engine import dispatch
from exports import export_ppt
class ApprovedPPTTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.root = Path(self.temporary.name)
self.image = self.root / 'image.png'
Image.new('RGB', (600, 400), 'white').save(self.image)
self.record = dict(id='2', projectName='项目甲', team='摄影组', name='张三',
date='2026-09-13', type='道具耗材', amountText='123.45', note='采购材料', materials=[])
def tearDown(self):
self.temporary.cleanup()
def material(self, kind, path=None):
return dict(kind=kind, path=str(path or self.image), name='测试材料')
def export(self, records):
destination = self.root / 'approved.pptx'
dispatch(dict(operation='approved-ppt', state=dict(approvedRecords=records), destination=str(destination)))
return Presentation(destination)
def test_all_materials_and_payment_pagination(self):
self.record['materials'] = [self.material('invoice')] + [self.material('payment')] * 5 + [self.material('receipt'), self.material('custom')]
deck = self.export([self.record])
self.assertEqual(len(deck.slides), 6)
pictures = [shape for page in deck.slides for shape in page.shapes if shape.shape_type == 13]
self.assertEqual(len(pictures), 8)
self.assertTrue(all('#2' in '\n'.join(shape.text for shape in page.shapes if shape.has_text_frame) for page in deck.slides))
self.assertFalse(any('请手动粘贴' in shape.text for page in deck.slides for shape in page.shapes if shape.has_text_frame))
def test_multipage_pdf_exports_every_page(self):
pdf = self.root / 'invoice.pdf'
first = Image.new('RGB', (300, 200), 'white')
second = Image.new('RGB', (300, 200), 'blue')
first.save(pdf, save_all=True, append_images=[second])
self.record['materials'] = [self.material('invoice', pdf)]
self.assertEqual(len(self.export([self.record]).slides), 3)
def test_preserves_selected_order_without_merging_people(self):
other = copy.deepcopy(self.record)
other.update(id='1', name='李四', date='2026-09-01')
self.record['materials'] = [self.material('payment')]
other['materials'] = [self.material('payment')]
deck = self.export([self.record, other])
self.assertEqual(len(deck.slides), 4)
self.assertIn('#2', deck.slides[0].shapes[0].text)
self.assertIn('#1', deck.slides[2].shapes[0].text)
def test_empty_materials_and_long_note_not_silently_lost(self):
self.record['note'] = '采购备注' * 250
deck = self.export([self.record])
text = '\n'.join(shape.text for page in deck.slides for shape in page.shapes if shape.has_text_frame)
self.assertEqual(text.count('采购备注'), 250)
self.assertIn('该申请没有上传附件', text)
def test_failure_keeps_existing_output(self):
destination = self.root / 'original.pptx'
destination.write_bytes(b'original')
self.record['materials'] = [self.material('invoice', self.root / 'missing.pdf')]
with self.assertRaisesRegex(ValueError, '申请 #2'):
dispatch(dict(operation='approved-ppt', state=dict(approvedRecords=[self.record]), destination=str(destination)))
self.assertEqual(destination.read_bytes(), b'original')
self.assertEqual(list(self.root.glob('.*.pptx')), [])
def test_rejects_empty_and_excessive_batches(self):
for records in ([], [self.record] * 101):
with self.assertRaises(ValueError):
export_ppt(dict(approvedRecords=records), self.root / 'invalid.pptx', False)
def test_engine_process_protocol(self):
binary = os.environ.get('RECEIPT_ENGINE_BINARY')
command = [binary] if binary else [sys.executable, str(Path(__file__).resolve().parents[1] / 'engine.py')]
self.record['materials'] = [self.material('invoice'), self.material('payment')]
destination = self.root / 'process.pptx'
request = dict(operation='approved-ppt', state=dict(approvedRecords=[self.record]), destination=str(destination))
process = subprocess.run(command, input=json.dumps(request) + '\n', capture_output=True, text=True, timeout=60)
self.assertEqual(process.returncode, 0, process.stdout + process.stderr)
events = [json.loads(line) for line in process.stdout.splitlines()]
self.assertEqual(events[-1]['event'], 'result')
self.assertEqual(events[-1]['result']['destination'], str(destination))
self.assertEqual(len(Presentation(destination).slides), 3)
if __name__ == '__main__':
unittest.main()