johannes/einvoice
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: create validated ZUGFeRD invoices in one console command

Browse Source
This commit is contained in:
admin committed 2026-10-16 09:29:08 +02:00
1 parent e0c15fc7f2
commit a643cf5227
30 files changed
+1509 -207
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_create.py
+118
View File
@@ -0,0 +1,118 @@
import io
import logging
import os
from datetime import date
from decimal import Decimal
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
import yaml
from jr_einvoice.create import decimal_input, save_customer, find_customer, service_period, create_invoice
from jr_einvoice.quick_cli import main
from jr_einvoice.pipeline import BuildResult
from jr_einvoice.presets import load_yaml
ROOT = Path(__file__).resolve().parent.parent
class CreateTest(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name)
self.presets = self.root / 'presets'
self.presets.mkdir()
sample = load_yaml(ROOT / 'examples/invoice.sample.yaml')
(self.presets / 'seller.default.yaml').write_text(yaml.safe_dump({
'seller': sample['seller'], 'payment': sample['payment']}))
self.key, _ = save_customer(self.presets, 'Neuer / alter Kunde GmbH', sample['buyer'] | {'name': 'Neuer / alter Kunde GmbH'})
self.jar = self.root / 'test.jar'
self.jar.touch()
self.kwargs = dict(root=self.root / 'invoices', resources=ROOT, presets=self.presets,
customer='Neuer / alter Kunde GmbH', entry='Geleistete Stunden Oktober 2026',
volume=Decimal('120'), price=Decimal('85'), jar=self.jar,
issued=date(2026, 10, 31), logger=logging.getLogger('test'))
def fake_build(self, target, *args):
out = target / 'out'
(out / 'validation').mkdir(parents=True, exist_ok=True)
paths = [out / 'invoice.pdf', out / 'factur-x.xml', out / 'invoice.facturx.pdf', out / 'validation/report.xml']
for path in paths:
path.write_bytes(b'test artifact')
return BuildResult(target, out, *paths)
def test_german_numbers(self):
for text, expected in [('85,00 €', '85'), ('1.234,56 EUR', '1234.56'),
('120', '120'), ('72,5', '72.5'), ('85.00', '85'), ('0.125', '0.125')]:
self.assertEqual(decimal_input(text), Decimal(expected))
for text in ['NaN', 'Infinity', '1,2,3', '12.34,56', '85 USD', '', '1e9']:
with self.assertRaises(ValueError):
decimal_input(text)
def test_customer_reuse_and_no_overwrite(self):
self.assertEqual(find_customer(self.presets, 'NEUER / ALTER KUNDE GMBH')[0], self.key)
self.assertEqual(find_customer(self.presets, self.key)[0], self.key)
with self.assertRaises(FileExistsError):
save_customer(self.presets, 'Neuer / alter Kunde GmbH', find_customer(self.presets, self.key)[1]['buyer'])
self.assertEqual((self.presets / 'customers' / f'{self.key}.yaml').stat().st_mode & 0o777, 0o600)
def test_service_period_inference_and_override(self):
self.assertEqual(service_period('Geleistete Stunden Februar 2028', None, None, date(2028, 3, 1)),
(date(2028, 2, 29), date(2028, 2, 1), date(2028, 2, 29)))
self.assertEqual(service_period('Oktober 2026', None, date(2026, 10, 15), date.today()),
(date(2026, 10, 15), None, None))
with self.assertRaises(ValueError):
service_period('Test', '2026-13', None, date.today())
with self.assertRaises(ValueError):
service_period('Test', '2026-10', date.today(), date.today())
def test_create_allocates_numbers_and_attachment(self):
with patch('jr_einvoice.create.build_in_folder', side_effect=self.fake_build):
first = create_invoice(**self.kwargs)
second = create_invoice(**self.kwargs)
with self.assertRaises(ValueError):
create_invoice(**self.kwargs, number=first.invoice.meta.invoice_number)
self.assertEqual(first.invoice.meta.invoice_number, 'JR-2026-0001')
self.assertEqual(second.invoice.meta.invoice_number, 'JR-2026-0002')
self.assertEqual(first.invoice.meta.service_period_end, date(2026, 10, 31))
self.assertEqual(first.invoice.payment.due_date, date(2026, 11, 30))
self.assertEqual(first.attachment.read_bytes(), first.build.facturx_pdf.read_bytes())
self.assertEqual(first.build.invoice_dir.stat().st_mode & 0o777, 0o700)
self.assertTrue((first.build.out_dir / 'manifest.json').is_file())
def test_failed_build_retains_number_but_publishes_no_attachment(self):
with patch('jr_einvoice.create.build_in_folder', side_effect=RuntimeError('invalid XML')):
with self.assertRaises(RuntimeError):
create_invoice(**self.kwargs)
self.assertEqual(len(list((self.root / 'invoices').glob('*/invoice.yaml'))), 1)
self.assertEqual(list((self.root / 'invoices').glob('*/Rechnung_*.pdf')), [])
with patch('jr_einvoice.create.build_in_folder', side_effect=self.fake_build):
second = create_invoice(**self.kwargs)
self.assertEqual(second.invoice.meta.invoice_number, 'JR-2026-0002')
def test_unknown_customer_noninteractive_is_actionable(self):
with patch.dict(os.environ, {'JR_EINVOICE_PRESETS': str(self.presets)}), patch('sys.stderr', new_callable=io.StringIO) as err:
with self.assertRaises(SystemExit) as result:
main(['--create', '--Customer', 'Unbekannt', '--Entry', 'Beratung', '--Volume', '1', '--Price', '85,00 €', '--non-interactive'])
self.assertEqual(result.exception.code, 1)
self.assertIn('--Street', err.getvalue())
self.assertIsNone(find_customer(self.presets, 'Unbekannt'))
def test_interactive_customer_is_saved_once(self):
with patch.dict(os.environ, {'JR_EINVOICE_PRESETS': str(self.presets)}), \
patch('sys.stdin.isatty', return_value=True), \
patch('builtins.input', side_effect=['Neuer Weg 5', '12345', 'Berlin']) as ask, \
patch('jr_einvoice.quick_cli.create_invoice', side_effect=RuntimeError('stop before build')), \
patch('sys.stderr', new_callable=io.StringIO), patch('sys.stdout', new_callable=io.StringIO):
with self.assertRaises(SystemExit):
main(['--create', '--Customer', 'Neue Firma', '--Entry', 'Beratung', '--Volume', '1', '--Price', '85'])
self.assertEqual(ask.call_count, 3)
self.assertEqual(find_customer(self.presets, 'Neue Firma')[1]['buyer']['address']['street'], 'Neuer Weg 5')
def test_invalid_amount_creates_no_invoice(self):
with self.assertRaises(ValueError):
create_invoice(**(self.kwargs | {'volume': Decimal('-1')}))
self.assertEqual(list((self.root / 'invoices').glob('*/invoice.yaml')), [])
tests/test_invoices.py
+121
View File
@@ -0,0 +1,121 @@
import logging
import tempfile
import unittest
from datetime import date
from decimal import Decimal
from pathlib import Path
from unittest.mock import patch
from subprocess import CompletedProcess
import xml.etree.ElementTree as ET
from jr_einvoice.calc import compute_totals
from jr_einvoice.io import load_invoice_yaml
from jr_einvoice.models import Invoice
from jr_einvoice.facturx import generate_facturx_xml, NS
from jr_einvoice.mustang import validate, MustangError
from jr_einvoice.monthly import monthly_invoice
from jr_einvoice.render import render_html
from jr_einvoice.delivery import prepare_email
ROOT = Path(__file__).resolve().parent.parent
LOGGER = logging.getLogger('test')
class InvoicesTest(unittest.TestCase):
def setUp(self):
self.inv = load_invoice_yaml(ROOT / 'examples/invoice.sample.yaml')
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name)
def test_tax_rounding_by_group_and_line(self):
self.inv.items = [self.inv.items[0].model_copy(update={
'quantity': Decimal('1'), 'unit_price_net': Decimal('0.03')}) for _ in range(3)]
total = compute_totals(self.inv)
self.assertEqual(total.net_total, Decimal('0.09'))
self.assertEqual(total.tax_total, Decimal('0.02'))
self.inv.items[0].unit_price_net = Decimal('1.005')
self.assertEqual(compute_totals(self.inv).net_total, Decimal('1.07'))
def test_multiple_rates_and_complete_xml(self):
self.inv.items[1].tax_rate = Decimal('7')
total = compute_totals(self.inv)
self.assertEqual(total.gross_total, Decimal('1335.00'))
xml = generate_facturx_xml(self.inv, self.root / 'invoice.xml', LOGGER)
root = ET.parse(xml).getroot()
self.assertEqual(len(root.findall('.//ram:IncludedSupplyChainTradeLineItem', NS)), 2)
self.assertEqual(root.findtext('.//ram:SellerTradeParty/ram:Name', namespaces=NS), self.inv.seller.name)
self.assertEqual(root.findtext('.//ram:BuyerTradeParty/ram:Name', namespaces=NS), self.inv.buyer.name)
self.assertEqual(root.findtext('.//ram:GrandTotalAmount', namespaces=NS), '1335.00')
def test_invalid_inputs(self):
for change in [lambda d: d.update(items=[]),
lambda d: d['items'][0].update(tax_rate=0),
lambda d: d['items'][0].update(quantity='NaN'),
lambda d: d['items'][0].update(unit_price_net='Infinity'),
lambda d: d['meta'].update(unknown_field='typo'),
lambda d: d['meta'].update(service_period_start='2026-01-01'),
lambda d: d['payment'].update(iban='DE00000000000000000000'),
lambda d: d['payment'].update(due_date='2020-01-01'),
lambda d: d['seller'].update(email='a@example.com\nBcc: x@example.com')]:
data = self.inv.model_dump()
change(data)
with self.assertRaises(ValueError):
Invoice.model_validate(data)
def test_fail_closed_validator_retains_report(self):
for content, rc in [('', 0), ('noise', 0), ('<validation/>', 0),
('<validation><summary status="valid"/></validation>', 0),
('<validation><summary status="invalid"/></validation>', 0),
('<validation><summary status="valid"/></validation>', 1),
('<validation><pdf><summary status="invalid"/></pdf><summary status="valid"/></validation>', 0)]:
with patch('jr_einvoice.mustang._run', return_value=CompletedProcess([], rc, content, '')):
with self.assertRaises(MustangError):
validate(Path('fake.jar'), Path('fake.pdf'), self.root / 'report.xml', LOGGER)
self.assertTrue((self.root / 'report.xml').exists())
def test_validator_accepts_valid_summary(self):
content = '<validation><xml><summary status="valid"/></xml><summary status="valid"/></validation>'
with patch('jr_einvoice.mustang._run', return_value=CompletedProcess([], 0, content, '')):
validate(Path('fake.jar'), Path('fake.xml'), self.root / 'report.xml', LOGGER)
def test_display_preserves_price_and_quantity_precision(self):
self.inv.items[0].quantity = Decimal('1.125')
self.inv.items[0].unit_price_net = Decimal('0.125')
html = render_html(self.inv, ROOT / 'templates/invoice.html.j2')
self.assertIn('1,125', html)
self.assertIn('0,125 EUR', html)
self.assertIn('0,14 EUR', html)
def test_html_escapes_user_text(self):
self.inv.items[0].name = '<script>alert(1)</script>'
html = render_html(self.inv, ROOT / 'templates/invoice.html.j2')
self.assertNotIn('<script>', html)
self.assertIn('&lt;script&gt;', html)
def test_monthly_period_rate_and_duplicate_number(self):
# Explicit repository presets, no personal data required.
import yaml
repo = self.root / 'repo'
(repo / 'examples').mkdir(parents=True)
(repo / 'examples/invoice.init.yaml').write_text((ROOT / 'examples/invoice.sample.yaml').read_text())
(repo / 'presets/customers').mkdir(parents=True)
(repo / 'presets/customers/test.yaml').write_text(yaml.safe_dump({'items': [self.inv.items[0].model_dump(mode='json')]}))
target = monthly_invoice(self.root / 'invoices', repo, LOGGER, 'test', '2028-02', Decimal('75'), '2028-08', date(2028, 3, 1), Decimal('85'))
inv = load_invoice_yaml(target / 'invoice.yaml')
self.assertEqual(inv.meta.service_period_end, date(2028, 2, 29))
self.assertEqual(inv.payment.due_date, date(2028, 3, 31))
self.assertEqual(compute_totals(inv).gross_total, Decimal('7586.25'))
with self.assertRaises(FileExistsError):
monthly_invoice(self.root / 'invoices', repo, LOGGER, 'test', '2028-02', Decimal('75'), '2028-08')
def test_missing_recipient_cannot_leave_stale_draft(self):
import shutil
shutil.copy(ROOT / 'examples/invoice.sample.yaml', self.root / 'invoice.yaml')
(self.root / 'out').mkdir()
(self.root / 'out/invoice.eml').write_text('stale')
with self.assertRaises(ValueError):
prepare_email(self.root, Path('fake.jar'), LOGGER)
self.assertFalse((self.root / 'out/invoice.eml').exists())
if __name__ == '__main__':
unittest.main()
tests/test_mustang_integration.py
+84
View File
@@ -0,0 +1,84 @@
"""Opt in: JR_EINVOICE_INTEGRATION=1 python -m unittest discover -s tests -v."""
import os
import io
import json
from unittest.mock import patch
import logging
import tempfile
import unittest
from pathlib import Path
from email import policy
from email.parser import BytesParser
import xml.etree.ElementTree as ET
import yaml
from jr_einvoice.pipeline import build_in_folder
from jr_einvoice.delivery import prepare_email
from jr_einvoice.mustang import validate, MustangError
from jr_einvoice.facturx import NS
ROOT = Path(__file__).resolve().parent.parent
JAR = Path(os.environ.get('JR_EINVOICE_MUSTANG_JAR', ROOT / 'tools/mustang/Mustang-CLI.jar'))
LOGGER = logging.getLogger('integration')
@unittest.skipUnless(os.environ.get('JR_EINVOICE_INTEGRATION') == '1', 'Opt in with JR_EINVOICE_INTEGRATION=1')
class MustangIntegrationTest(unittest.TestCase):
def test_build_email_and_reject_bad_xml_and_plain_pdf(self):
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
data = yaml.safe_load((ROOT / 'examples/invoice.sample.yaml').read_text())
data['buyer']['email'] = 'rechnung@example.invalid'
data['items'][1]['tax_rate'] = 7
(root / 'invoice.yaml').write_text(yaml.safe_dump(data))
result = build_in_folder(root, 'out', ROOT / 'templates/invoice.html.j2', LOGGER, JAR, True, True)
self.assertIn('status="valid"', result.validate_report.read_text())
path = prepare_email(root, JAR, LOGGER)
message = BytesParser(policy=policy.default).parsebytes(path.read_bytes())
self.assertEqual(message['From'], data['seller']['email'])
self.assertEqual(message['To'], 'rechnung@example.invalid')
attachments = list(message.iter_attachments())
self.assertEqual(len(attachments), 1)
self.assertEqual(attachments[0].get_payload(decode=True), result.facturx_pdf.read_bytes())
self.assertTrue((root / 'out/manifest.json').is_file())
# A valid PDF/A without invoice XML must not be an accepted e-invoice.
with self.assertRaises(MustangError):
validate(JAR, result.invoice_pdf, root / 'plain-report.xml', LOGGER)
xml = ET.parse(result.facturx_xml)
xml.getroot().find('.//ram:GrandTotalAmount', NS).text = '0.01'
xml.write(root / 'broken.xml', encoding='utf-8', xml_declaration=True)
with self.assertRaises(MustangError):
validate(JAR, root / 'broken.xml', root / 'bad-report.xml', LOGGER)
self.assertTrue((root / 'bad-report.xml').is_file())
def test_one_call_new_existing_customer_and_rejected_invalid_unit(self):
from jr_einvoice.quick_cli import main
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
presets = root / 'presets'
presets.mkdir()
sample = yaml.safe_load((ROOT / 'examples/invoice.sample.yaml').read_text())
(presets / 'seller.default.yaml').write_text(yaml.safe_dump({
'seller': sample['seller'], 'payment': sample['payment']}))
base = ['--create', '--Customer', 'Beliebiger neuer / alter Kunde GmbH',
'--Entry', 'Geleistete Stunden Oktober 2026', '--Volume', '120',
'--Price', '85,00 €', '--Date', '2026-10-31', '--root', str(root / 'invoices'),
'--mustang-jar', str(JAR), '--json']
with patch.dict(os.environ, {'JR_EINVOICE_PRESETS': str(presets)}):
for expected, extra in [('JR-2026-0001', ['--Street', 'Testweg 1', '--PostalCode', '10115', '--City', 'Berlin']),
('JR-2026-0002', [])]:
with patch('sys.stdout', new_callable=io.StringIO) as output:
main(base + extra)
result = json.loads(output.getvalue())
self.assertEqual(result['invoice_number'], expected)
self.assertEqual(result['technical_validation'], 'valid')
self.assertTrue(Path(result['pdf']).is_file())
xml = ET.parse(Path(result['invoice_dir']) / 'out/factur-x.xml')
self.assertEqual(xml.getroot().findtext('.//ram:GrandTotalAmount', namespaces=NS), '12138.00')
with patch('sys.stdout', new_callable=io.StringIO) as output, patch('sys.stderr', new_callable=io.StringIO):
with self.assertRaises(SystemExit) as failure:
main(base + ['--Unit', 'ZZZ'])
self.assertEqual(failure.exception.code, 1)
self.assertNotIn('technical_validation', output.getvalue())
failed = next((root / 'invoices').glob('*JR-2026-0003*'))
self.assertFalse((failed / 'Rechnung_JR-2026-0003.pdf').exists())
self.assertTrue((failed / 'out/validation/validate_report.xml').exists())