"""IMM5709 stored-XFA tests; Reader validation remains manual.""" import copy import importlib.util import json from pathlib import Path import unittest from unittest.mock import patch import hashlib from uuid import uuid4 import pikepdf from lxml import etree as E from prefill_imm_xfa_pdf import fill BASE = Path(__file__).resolve().parents[2] NS = 'http://www.xfa.org/schema/xfa-data/1.0/' def packets(path): with pikepdf.open(path) as pdf: a = pdf.Root.AcroForm.XFA return {str(a[i]): a[i + 1].read_bytes() for i in range(0, len(a), 2)} def resolve(root, path, named=False): import re nodes = [root] for part in path.split('.'): match = re.fullmatch(r'([^\[\]]+)(?:\[(\d+)\])?', part) name, index = match.groups() candidates = [c for n in nodes for c in n if isinstance(c.tag, str) and (c.get('name') if named else E.QName(c).localname) == name] nodes = candidates if index is None else candidates[int(index):int(index)+1] return nodes class IMM5709WriterTest(unittest.TestCase): def setUp(self): self.work = BASE / 'storage/app/imm/tmp' / ('5709-test-' + uuid4().hex) self.work.mkdir(parents=True) def contract(self, language): return json.loads((BASE / f'resources/imm/imm5709-contract.{language}.json').read_text(encoding='utf-8')) def payload(self, language): contract = self.contract(language) fields = {} for index, (path, meta) in enumerate(contract['fields'].items()): value = 'VALUE' + str(index) if meta.get('allowed_values'): value = meta['allowed_values'][0] elif meta.get('choices'): value = next(k for k in meta['choices'] if k) elif meta.get('max_chars'): value = value[:meta['max_chars']] fields[path] = value fields['form1.Page2.ContactInformation.Resi.SameAsAddr.SameAsMailingInd'] = 'N' return {'form': 'IMM5709', 'language': language, 'fields': fields}, contract def write(self, payload, blank, name='filled.pdf'): source = self.work / 'payload.json' source.write_text(json.dumps(payload), encoding='utf-8') output = self.work / name fill(str(source), str(output), str(blank)) return output def test_isolated_writer_exists(self): self.assertIsNotNone(importlib.util.find_spec('imm5709_pdf'), 'IMM5709 requires isolated allowlisted writer') def test_bilingual_complete_payload_and_saved_state(self): for language, suffix in [('en', 'e'), ('fr', 'f')]: payload, contract = self.payload(language) snapshot = copy.deepcopy(payload) blank = BASE / f'storage/app/imm/imm5709{suffix}.pdf' before = packets(blank) output = self.write(payload, blank, language + '.pdf') actual = packets(output) self.assertEqual(payload, snapshot) for name, raw in before.items(): if name not in ('datasets', 'form'): self.assertEqual(actual[name], raw, name) data = E.fromstring(actual['datasets']).find('{' + NS + '}data') saved = E.fromstring(actual['form']) for path, value in payload['fields'].items(): nodes = resolve(data, path) self.assertEqual(len(nodes), 1, path) self.assertEqual(nodes[0].text or '', value, path) self.assertEqual(len(nodes[0]), 0, path) widgets = resolve(saved, contract['fields'][path]['template_path'], named=True) self.assertEqual(len(widgets), 1, path) texts = widgets[0].xpath('./*[local-name()="value"]/*') from imm5709_pdf import PHONE_PREFIXES if path.startswith(PHONE_PREFIXES): self.assertFalse(texts, path) else: self.assertEqual(''.join(texts[0].itertext()) if texts else '', value, path) self.assertEqual(packets(blank), before) def test_unknown_target_rejects_without_output(self): payload, _ = self.payload('en') payload['fields']['form1.Page1.UnknownTarget'] = 'INVALID' output = self.work / 'rejected.pdf' with self.assertRaisesRegex(ValueError, 'allowlist|allowlisted'): self.write(payload, BASE / 'storage/app/imm/imm5709e.pdf', output.name) self.assertFalse(output.exists()) def test_missing_application_dataset_created_without_touching_lookups(self): import imm5709_pdf payload, contract = self.payload('en') blank = self.work / 'lov-only.pdf' source = BASE / 'storage/app/imm/imm5709e.pdf' with pikepdf.open(source) as pdf: a = pdf.Root.AcroForm.XFA streams = {str(a[i]): a[i+1] for i in range(0, len(a), 2)} root = E.fromstring(streams['datasets'].read_bytes()) data = root.find('{' + NS + '}data') root.remove(data) streams['datasets'].write(E.tostring(root)) pdf.save(blank) contract['sha256'] = hashlib.sha256(blank.read_bytes()).hexdigest() before = E.fromstring(packets(blank)['datasets']) with patch.object(imm5709_pdf, 'load_contract', return_value=contract): output = self.write(payload, blank) after = E.fromstring(packets(output)['datasets']) self.assertEqual(E.tostring(before[0]), E.tostring(after[0])) data = after.find('{' + NS + '}data') for path, value in payload['fields'].items(): self.assertEqual(resolve(data, path)[0].text or '', value, path) def test_allowlisted_but_unresolved_template_target_rejects_without_output(self): import imm5709_pdf payload, contract = self.payload('en') key = next(iter(payload['fields'])) contract['fields'][key]['template_path'] = 'form1.Page1.Unknown' with patch.object(imm5709_pdf, 'load_contract', return_value=contract): with self.assertRaisesRegex(ValueError, 'Unresolved'): self.write(payload, BASE / 'storage/app/imm/imm5709e.pdf', 'unresolved.pdf') self.assertFalse((self.work / 'unresolved.pdf').exists()) def test_invalid_exports_and_values_reject_without_partial_writes(self): payload, contract = self.payload('en') checkbox = next(k for k, v in contract['fields'].items() if v.get('allowed_values')) choice = next(k for k, v in contract['fields'].items() if v.get('choices')) for key, value in [(checkbox, 'yes'), (choice, 'NOT AN EXPORT'), (checkbox, 1)]: current = copy.deepcopy(payload) current['fields'][key] = value output = self.work / 'invalid.pdf' with self.assertRaisesRegex(ValueError, 'Invalid'): self.write(current, BASE / 'storage/app/imm/imm5709e.pdf', output.name) self.assertFalse(output.exists()) payload['fields']['form1.Page2.ContactInformation.Resi.SameAsAddr.SameAsMailingInd'] = 'Y' payload['fields']['form1.Page2.ContactInformation.Resi.AddrLine1.Streetname'] = 123 with self.assertRaisesRegex(ValueError, 'Invalid'): self.write(payload, BASE / 'storage/app/imm/imm5709e.pdf', 'invalid-inactive.pdf') def test_writer_does_not_mutate_payload_or_stamp_signature_date(self): from imm5709_pdf import fill_imm5709 payload, _ = self.payload('en') snapshot = copy.deepcopy(payload) blank = BASE / 'storage/app/imm/imm5709e.pdf' output = self.work / 'direct.pdf' fill_imm5709(payload, output, blank) self.assertEqual(snapshot, payload) data = E.fromstring(packets(output)['datasets']) for node in data.iter(): if isinstance(node.tag, str) and E.QName(node).localname in ('ApplicationValidatedFlag', 'DateLastValidated', 'C1CertificateIssueDate'): self.assertFalse(node.text) def test_stale_validation_and_manual_signature_values_cleared(self): import imm5709_pdf contract = self.contract('en') blank = self.work / 'stale.pdf' with pikepdf.open(BASE / 'storage/app/imm/imm5709e.pdf') as pdf: a = pdf.Root.AcroForm.XFA streams = {str(a[i]): a[i+1] for i in range(0, len(a), 2)} for packet in ('datasets', 'form'): root = E.fromstring(streams[packet].read_bytes()) for node in root.iter(): name = node.get('name') if packet == 'form' else (E.QName(node).localname if isinstance(node.tag, str) else '') if name in ('ApplicationValidatedFlag', 'DateLastValidated', 'C1CertificateIssueDate'): if packet == 'form': imm5709_pdf.form_value(node, 'STALE') if name == 'DateLastValidated': for child in node.iter(): if child.get('name') in ('Year', 'Month', 'Day', 'DateCalc'): imm5709_pdf.form_value(child, 'STALE') else: node.text = 'STALE' if name == 'DateLastValidated': for child in node: child.text = 'STALE' streams[packet].write(E.tostring(root)) pdf.save(blank) contract['sha256'] = hashlib.sha256(blank.read_bytes()).hexdigest() with patch.object(imm5709_pdf, 'load_contract', return_value=contract): output = self.write({'form': 'IMM5709', 'fields': {}}, blank) for packet in ('datasets', 'form'): self.assertFalse(b'STALE' in packets(output)[packet], packet) def test_address_and_phone_saved_state_matches_actual_selections(self): for language, suffix in [('en', 'e'), ('fr', 'f')]: for choice in ('Y', 'N'): prefix = 'form1.Page2.ContactInformation.' fields = {prefix + 'Resi.SameAsAddr.SameAsMailingInd': choice, prefix + 'q3-4.Phone.CanOtherInd.CanadaUS': '0', prefix + 'q3-4.Phone.CanOtherInd.Other': '1', prefix + 'q3-4.Phone.IntlNumber.IntlNumber': '441234567890'} output = self.write({'form': 'IMM5709', 'language': language, 'fields': fields}, BASE / f'storage/app/imm/imm5709{suffix}.pdf') actual = packets(output) data = E.fromstring(actual['datasets']).find('{' + NS + '}data') saved = E.fromstring(actual['form']) address = resolve(data, prefix + 'Resi.SameAsAddr.SameAsMailingInd')[0] self.assertEqual(address.text, choice) self.assertEqual(len(address), 0) group = resolve(saved, prefix + 'Resi.SameAsAddr.SameAsMailingInd', named=True)[0] self.assertEqual(group.xpath('./*[local-name()="field"][@name="Yes"]/*[local-name()="value"]/*/text()'), ['Y'] if choice == 'Y' else []) residential = resolve(saved, prefix + 'Resi.AddrLine1.Streetname', named=True)[0] self.assertEqual(residential.get('access'), 'readOnly' if choice == 'Y' else 'open') self.assertEqual(residential.xpath('./*[local-name()="validate"]/@nullTest'), ['disabled' if choice == 'Y' else 'error']) if choice == 'Y': self.assertFalse(resolve(data, prefix + 'Resi.AddrLine2.Country')[0].text) self.assertEqual(resolve(saved, prefix + 'q3-4.Phone.NANumber', named=True)[0].get('presence'), 'invisible') self.assertEqual(resolve(saved, prefix + 'q3-4.Phone.IntlNumber', named=True)[0].get('presence'), 'visible') phone_checkbox = resolve(saved, prefix + 'q3-4.Phone.CanOtherInd.Other', named=True)[0] self.assertFalse(phone_checkbox.xpath('./*[local-name()="value"]')) template = E.fromstring(actual['template']) self.assertEqual(resolve(template, prefix + 'q3-4.Phone.NANumber', named=True)[0].get('presence'), 'invisible') self.assertEqual(resolve(template, prefix + 'q3-4.Phone.IntlNumber', named=True)[0].get('presence'), 'visible') def test_phone_merge_matches_working_5708_without_stale_restore_deltas(self): prefix = 'form1.Page2.ContactInformation.q3-4.Phone.' fields = {prefix + 'CanOtherInd.CanadaUS': '1', prefix + 'CanOtherInd.Other': '0', prefix + 'NANumber.AreaCode': '416', prefix + 'NANumber.FirstThree': '555', prefix + 'NANumber.LastFive': '0100', prefix + 'NumberCountry': '1', prefix + 'ActualNumber': '4165550100', prefix + 'Type': '02'} blank = BASE / 'storage/app/imm/imm5709e.pdf' output = self.write({'form': 'IMM5709', 'fields': fields}, blank) actual = packets(output) template = E.fromstring(actual['template']) data = E.fromstring(actual['datasets']).find('{' + NS + '}data') saved = E.fromstring(actual['form']) self.assertEqual(resolve(template, prefix + 'NANumber', named=True)[0].get('presence'), 'visible') for path, value in fields.items(): node = resolve(data, path)[0] self.assertEqual(node.text, value) self.assertNotIn('{' + NS + '}dataNode', node.attrib) self.assertFalse(resolve(saved, path, named=True)[0].xpath('./*[local-name()="value"]')) from imm5708_pdf_state import contact_phone_state original = packets(blank) contact = {key.removeprefix('form1.Page2.'): value for key, value in fields.items()} self.assertEqual(actual['template'], contact_phone_state(original['template'], contact)) before_scripts = E.fromstring(original['template']).xpath('//*[local-name()="script"]/text()') after_scripts = template.xpath('//*[local-name()="script"]/text()') self.assertEqual(before_scripts, after_scripts) if __name__ == '__main__': unittest.main()