mirror of
https://github.com/DRYTRIX/TimeTracker.git
synced 2026-05-18 04:08:48 -05:00
feat(invoicing): add Factur-X CII export validation and transport guidance
Switch embedded invoice PDFs to Factur-X CII payloads and tighten the PDF/A-3 and AS4 handling so exports better match the standards they advertise. Document the experimental native Peppol transport path and cover the new validation and embedding behavior with focused tests.
This commit is contained in:
Dries Peeters
@@ -1,31 +1,39 @@
|
||||
"""
|
||||
PEPPOL AS4 message packaging and transmission.
|
||||
|
||||
Builds AS4 (ebMS 3.0 / PEPPOL AS4 profile) messages and sends to recipient
|
||||
access point. Optional signing when certificate is configured.
|
||||
EXPERIMENTAL: This native AS4 implementation provides basic message
|
||||
packaging and HTTP POST to a recipient access point. It does NOT
|
||||
implement full Peppol AS4 compliance:
|
||||
- No WS-Security / XML digital signatures
|
||||
- No AS4 receipt handling / reliability
|
||||
- Payload is gzip-compressed as declared in the SOAP header
|
||||
|
||||
For production use, prefer the Generic transport with a standards-compliant
|
||||
Peppol Access Point provider.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import email.generator
|
||||
import email.policy
|
||||
import gzip
|
||||
import os
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from email.message import EmailMessage
|
||||
from io import BytesIO
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import requests
|
||||
|
||||
from app.integrations.peppol import PEPPOL_BIS3_PROFILE_ID
|
||||
|
||||
# PEPPOL AS4 profile: invoice document type URN
|
||||
PEPPOL_INVOICE_DOCUMENT_TYPE = (
|
||||
"urn:oasis:names:specification:ubl:schema:xsd:Invoice-2::Invoice"
|
||||
"##urn:cen.eu:en16931:2017#compliant#urn:fdc:peppol.eu:2017:poacc:billing:3.0::2.1"
|
||||
)
|
||||
|
||||
# Flag surfaced in settings UI so users know this is not production-grade
|
||||
NATIVE_TRANSPORT_EXPERIMENTAL = True
|
||||
|
||||
_BOUNDARY = "as4boundary"
|
||||
|
||||
|
||||
class PeppolAS4Error(RuntimeError):
|
||||
"""AS4 build or send error."""
|
||||
@@ -71,7 +79,7 @@ def _soap_envelope(
|
||||
<eb3:ConversationId>{document_id}</eb3:ConversationId>
|
||||
</eb3:CollaborationInfo>
|
||||
<eb3:PayloadInfo>
|
||||
<eb3:PartInfo href="cid:payload@europe.eu">
|
||||
<eb3:PartInfo href="cid:payload@peppol.eu">
|
||||
<eb3:Schema location="{document_type_id}"/>
|
||||
<eb3:PartProperties>
|
||||
<eb3:Property name="MimeType">application/xml</eb3:Property>
|
||||
@@ -97,8 +105,11 @@ def build_as4_message(
|
||||
document_type_id: str = PEPPOL_INVOICE_DOCUMENT_TYPE,
|
||||
) -> bytes:
|
||||
"""
|
||||
Build AS4 multipart message (SOAP + payload).
|
||||
Build AS4 multipart/related message (SOAP + gzip-compressed payload).
|
||||
Returns raw bytes suitable for POST to recipient AP.
|
||||
|
||||
The payload is gzip-compressed to match the CompressionType declared
|
||||
in the SOAP header (application/gzip).
|
||||
"""
|
||||
message_id = f"<{uuid.uuid4().hex}@peppol>"
|
||||
soap = _soap_envelope(
|
||||
@@ -111,33 +122,32 @@ def build_as4_message(
|
||||
process_id=process_id,
|
||||
document_type_id=document_type_id,
|
||||
)
|
||||
payload_bytes = ubl_xml.encode("utf-8")
|
||||
payload_bytes = gzip.compress(ubl_xml.encode("utf-8"))
|
||||
|
||||
# Build MIME multipart/related: root SOAP + payload part
|
||||
policy = email.policy.EmailPolicy(line_max=0)
|
||||
msg = EmailMessage(policy=policy)
|
||||
msg["Content-Type"] = "multipart/related; boundary=as4boundary; type=application/soap+xml"
|
||||
msg.set_boundary("as4boundary")
|
||||
|
||||
msg.add_attachment(
|
||||
soap.encode("utf-8"),
|
||||
maintype="application",
|
||||
subtype="soap+xml",
|
||||
disposition="inline",
|
||||
headers={"Content-ID": "<root.message@europe.eu>"},
|
||||
# Build MIME multipart/related manually for cross-Python-version compatibility
|
||||
parts = []
|
||||
parts.append(
|
||||
f"--{_BOUNDARY}\r\n"
|
||||
f"Content-Type: application/soap+xml; charset=utf-8\r\n"
|
||||
f"Content-ID: <root.message@peppol.eu>\r\n"
|
||||
f"\r\n"
|
||||
)
|
||||
msg.add_attachment(
|
||||
payload_bytes,
|
||||
maintype="application",
|
||||
subtype="xml",
|
||||
disposition="attachment",
|
||||
headers={"Content-ID": "<payload@europe.eu>"},
|
||||
parts.append(soap)
|
||||
parts.append(
|
||||
f"\r\n--{_BOUNDARY}\r\n"
|
||||
f"Content-Type: application/gzip\r\n"
|
||||
f"Content-ID: <payload@peppol.eu>\r\n"
|
||||
f"Content-Transfer-Encoding: binary\r\n"
|
||||
f"\r\n"
|
||||
)
|
||||
|
||||
buf = BytesIO()
|
||||
gen = email.generator.BytesGenerator(buf, policy=policy)
|
||||
gen.flatten(msg)
|
||||
return buf.getvalue()
|
||||
result = b""
|
||||
for p in parts:
|
||||
result += p.encode("utf-8") if isinstance(p, str) else p
|
||||
result += payload_bytes
|
||||
result += f"\r\n--{_BOUNDARY}--\r\n".encode("utf-8")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def send_as4_message(
|
||||
@@ -157,7 +167,7 @@ def send_as4_message(
|
||||
raise PeppolAS4Error("Recipient AP URL must be HTTP or HTTPS")
|
||||
|
||||
headers = {
|
||||
"Content-Type": "multipart/related; boundary=as4boundary; type=application/soap+xml",
|
||||
"Content-Type": f"multipart/related; boundary={_BOUNDARY}; type=application/soap+xml",
|
||||
"Accept": "application/xml",
|
||||
}
|
||||
cert = None
|
||||
@@ -180,7 +190,6 @@ def send_as4_message(
|
||||
result["error"] = resp.text[:2000] if resp.text else f"HTTP {resp.status_code}"
|
||||
raise PeppolAS4Error(f"Recipient AP returned {resp.status_code}: {result.get('error', '')}")
|
||||
|
||||
# Try to parse SOAP response for MessageId or receipt
|
||||
if resp.text and "MessageId" in resp.text:
|
||||
result["message_id"] = resp.text # Caller can parse if needed
|
||||
result["message_id"] = resp.text
|
||||
return result
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
"""
|
||||
PEPPOL SML/SMP participant discovery.
|
||||
|
||||
Resolves recipient access point URL from the Service Metadata Locator (SML)
|
||||
and Service Metadata Provider (SMP) for native PEPPOL transport.
|
||||
EXPERIMENTAL: Resolves recipient access point URL from the Service Metadata
|
||||
Locator (SML) and Service Metadata Provider (SMP) for native PEPPOL
|
||||
transport. This implementation supports basic HTTP-based SML/SMP lookup
|
||||
only (no DNS-based NAPTR/SRV resolution, no DNSSEC verification).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
"""
|
||||
PEPPOL transport provider interface and implementations.
|
||||
|
||||
- GenericTransport: existing HTTP JSON adapter (access point URL).
|
||||
- NativePeppolTransport: SML/SMP discovery + AS4 send.
|
||||
- GenericTransport: HTTP JSON adapter (access point URL). Production-ready.
|
||||
- NativePeppolTransport: SML/SMP discovery + AS4 send. EXPERIMENTAL - lacks
|
||||
WS-Security, receipt handling, and full Peppol AS4 compliance. Prefer a
|
||||
standards-compliant Access Point provider for production workloads.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -1149,14 +1149,14 @@ def export_invoice_pdf(invoice_id):
|
||||
pdf_generator = InvoicePDFGenerator(invoice, settings=settings, page_size=page_size)
|
||||
current_app.logger.info(f"[PDF_EXPORT] Starting PDF generation - PageSize: '{page_size}', InvoiceID: {invoice_id}")
|
||||
pdf_bytes = pdf_generator.generate_pdf()
|
||||
# Optionally embed ZugFerd/Factur-X EN 16931 XML in PDF (strict: fail export if embed fails)
|
||||
# Optionally embed Factur-X CII XML in PDF (strict: fail export if embed fails)
|
||||
if getattr(settings, "invoices_zugferd_pdf", False):
|
||||
from app.utils.zugferd import embed_zugferd_xml_in_pdf
|
||||
pdf_bytes, embed_err = embed_zugferd_xml_in_pdf(pdf_bytes, invoice, settings)
|
||||
if embed_err:
|
||||
current_app.logger.warning(f"[PDF_EXPORT] ZugFerd embed failed - InvoiceID: {invoice_id}, Error: {embed_err}")
|
||||
current_app.logger.warning(f"[PDF_EXPORT] Factur-X embed failed - InvoiceID: {invoice_id}, Error: {embed_err}")
|
||||
flash(
|
||||
_("ZUGFeRD embedding is enabled but failed: %(err)s. Export aborted so the PDF does not ship without embedded XML.", err=embed_err),
|
||||
_("Factur-X embedding is enabled but failed: %(err)s. Export aborted so the PDF does not ship without embedded XML.", err=embed_err),
|
||||
"error",
|
||||
)
|
||||
return redirect(request.referrer or url_for("invoices.view_invoice", invoice_id=invoice.id))
|
||||
@@ -1201,9 +1201,9 @@ def export_invoice_pdf(invoice_id):
|
||||
from app.utils.zugferd import embed_zugferd_xml_in_pdf
|
||||
pdf_bytes, embed_err = embed_zugferd_xml_in_pdf(pdf_bytes, invoice, settings)
|
||||
if embed_err:
|
||||
current_app.logger.warning(f"[PDF_EXPORT] ZugFerd embed failed (fallback path) - InvoiceID: {invoice_id}, Error: {embed_err}")
|
||||
current_app.logger.warning(f"[PDF_EXPORT] Factur-X embed failed (fallback path) - InvoiceID: {invoice_id}, Error: {embed_err}")
|
||||
flash(
|
||||
_("ZUGFeRD embedding is enabled but failed: %(err)s. Export aborted.", err=embed_err),
|
||||
_("Factur-X embedding is enabled but failed: %(err)s. Export aborted.", err=embed_err),
|
||||
"error",
|
||||
)
|
||||
return redirect(request.referrer or url_for("invoices.view_invoice", invoice_id=invoice.id))
|
||||
|
||||
@@ -0,0 +1,305 @@
|
||||
"""
|
||||
CII (Cross-Industry Invoice) generator for Factur-X / ZUGFeRD.
|
||||
|
||||
Generates UN/CEFACT CII XML (EN 16931 profile) suitable for embedding
|
||||
in PDF/A-3 as required by Factur-X 1.0 / ZUGFeRD 2.x.
|
||||
|
||||
This is the correct payload format for ZUGFeRD/Factur-X hybrid invoices.
|
||||
Peppol uses UBL (see app/integrations/peppol.py); this module is for
|
||||
the embedded-in-PDF use case only.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from dataclasses import dataclass
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from typing import Any, Optional, Tuple
|
||||
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
|
||||
NS_RSM = "urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100"
|
||||
NS_RAM = "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"
|
||||
NS_UDT = "urn:un:unece:uncefact:data:standard:UnqualifiedDataType:100"
|
||||
NS_QDT = "urn:un:unece:uncefact:data:standard:QualifiedDataType:100"
|
||||
|
||||
FACTURX_GUIDELINE_EN16931 = (
|
||||
"urn:cen.eu:en16931:2017#compliant#urn:factur-x.eu:1p0:en16931"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CIIParty:
|
||||
name: str
|
||||
tax_id: Optional[str] = None
|
||||
address_line: Optional[str] = None
|
||||
city: Optional[str] = None
|
||||
postcode: Optional[str] = None
|
||||
country_code: Optional[str] = None
|
||||
email: Optional[str] = None
|
||||
phone: Optional[str] = None
|
||||
endpoint_id: Optional[str] = None
|
||||
endpoint_scheme_id: Optional[str] = None
|
||||
|
||||
|
||||
def _money(v: Any) -> str:
|
||||
try:
|
||||
d = v if isinstance(v, Decimal) else Decimal(str(v))
|
||||
except Exception:
|
||||
d = Decimal("0")
|
||||
return f"{d.quantize(Decimal('0.01'))}"
|
||||
|
||||
|
||||
def _qty(v: Any) -> str:
|
||||
try:
|
||||
d = v if isinstance(v, Decimal) else Decimal(str(v))
|
||||
except Exception:
|
||||
d = Decimal("0")
|
||||
return f"{d.quantize(Decimal('0.01'))}"
|
||||
|
||||
|
||||
def _date_102(d: Any) -> str:
|
||||
"""Format date as YYYYMMDD (format code 102 per UN/CEFACT)."""
|
||||
if hasattr(d, "strftime"):
|
||||
return d.strftime("%Y%m%d")
|
||||
return str(d).replace("-", "")
|
||||
|
||||
|
||||
def _sub(parent: ET.Element, tag: str) -> ET.Element:
|
||||
return ET.SubElement(parent, tag)
|
||||
|
||||
|
||||
def _text_el(parent: ET.Element, tag: str, text: Optional[str]) -> Optional[ET.Element]:
|
||||
if text is None:
|
||||
return None
|
||||
t = str(text).strip()
|
||||
if not t:
|
||||
return None
|
||||
el = ET.SubElement(parent, tag)
|
||||
el.text = t
|
||||
return el
|
||||
|
||||
|
||||
def _date_el(parent: ET.Element, d: Any) -> None:
|
||||
"""Add a DateTimeString child with format 102."""
|
||||
udt = f"{{{NS_UDT}}}"
|
||||
dts = _sub(parent, udt + "DateTimeString")
|
||||
dts.set("format", "102")
|
||||
dts.text = _date_102(d)
|
||||
|
||||
|
||||
def _build_party(parent: ET.Element, tag: str, party: CIIParty) -> None:
|
||||
ram = f"{{{NS_RAM}}}"
|
||||
p = _sub(parent, ram + tag)
|
||||
_text_el(p, ram + "Name", party.name)
|
||||
|
||||
if party.endpoint_id and party.endpoint_scheme_id:
|
||||
org = _sub(p, ram + "SpecifiedLegalOrganization")
|
||||
org_id = _text_el(org, ram + "ID", party.endpoint_id)
|
||||
if org_id is not None:
|
||||
org_id.set("schemeID", party.endpoint_scheme_id)
|
||||
|
||||
if party.address_line or party.country_code:
|
||||
addr = _sub(p, ram + "PostalTradeAddress")
|
||||
_text_el(addr, ram + "LineOne", party.address_line)
|
||||
_text_el(addr, ram + "CityName", party.city)
|
||||
_text_el(addr, ram + "PostcodeCode", party.postcode)
|
||||
_text_el(addr, ram + "CountryID", party.country_code)
|
||||
|
||||
if party.email:
|
||||
uri_comm = _sub(p, ram + "URIUniversalCommunication")
|
||||
uri_id = _text_el(uri_comm, ram + "URIID", party.email)
|
||||
if uri_id is not None:
|
||||
uri_id.set("schemeID", "EM")
|
||||
|
||||
if party.tax_id:
|
||||
tax_reg = _sub(p, ram + "SpecifiedTaxRegistration")
|
||||
tax_reg_id = _text_el(tax_reg, ram + "ID", party.tax_id)
|
||||
if tax_reg_id is not None:
|
||||
tax_reg_id.set("schemeID", "VA")
|
||||
|
||||
|
||||
def build_cii_invoice_xml(
|
||||
invoice: Any,
|
||||
seller: CIIParty,
|
||||
buyer: CIIParty,
|
||||
guideline_id: str = FACTURX_GUIDELINE_EN16931,
|
||||
) -> Tuple[str, str]:
|
||||
"""
|
||||
Build a CII CrossIndustryInvoice XML for Factur-X / ZUGFeRD.
|
||||
|
||||
Returns:
|
||||
(xml_string_utf8, sha256_hex)
|
||||
"""
|
||||
ET.register_namespace("rsm", NS_RSM)
|
||||
ET.register_namespace("ram", NS_RAM)
|
||||
ET.register_namespace("udt", NS_UDT)
|
||||
ET.register_namespace("qdt", NS_QDT)
|
||||
|
||||
rsm = f"{{{NS_RSM}}}"
|
||||
ram = f"{{{NS_RAM}}}"
|
||||
|
||||
root = ET.Element(rsm + "CrossIndustryInvoice")
|
||||
|
||||
# --- ExchangedDocumentContext ---
|
||||
ctx = _sub(root, rsm + "ExchangedDocumentContext")
|
||||
guideline = _sub(ctx, ram + "GuidelineSpecifiedDocumentContextParameter")
|
||||
_text_el(guideline, ram + "ID", guideline_id)
|
||||
|
||||
# --- ExchangedDocument ---
|
||||
doc = _sub(root, rsm + "ExchangedDocument")
|
||||
_text_el(
|
||||
doc,
|
||||
ram + "ID",
|
||||
getattr(invoice, "invoice_number", None) or str(getattr(invoice, "id", "")),
|
||||
)
|
||||
_text_el(doc, ram + "TypeCode", "380")
|
||||
|
||||
issue_date = getattr(invoice, "issue_date", None) or date.today()
|
||||
issue_dt = _sub(doc, ram + "IssueDateTime")
|
||||
_date_el(issue_dt, issue_date)
|
||||
|
||||
notes = getattr(invoice, "notes", None)
|
||||
if notes and str(notes).strip():
|
||||
note_el = _sub(doc, ram + "IncludedNote")
|
||||
_text_el(note_el, ram + "Content", notes)
|
||||
|
||||
# --- SupplyChainTradeTransaction ---
|
||||
txn = _sub(root, rsm + "SupplyChainTradeTransaction")
|
||||
|
||||
currency = getattr(invoice, "currency_code", None) or "EUR"
|
||||
tax_rate = Decimal(str(getattr(invoice, "tax_rate", 0) or 0))
|
||||
tax_category = "S" if tax_rate > 0 else "Z"
|
||||
|
||||
# --- Header Trade Agreement ---
|
||||
agreement = _sub(txn, ram + "ApplicableHeaderTradeAgreement")
|
||||
|
||||
buyer_ref = (
|
||||
(getattr(invoice, "buyer_reference", None) or "").strip()
|
||||
or (getattr(getattr(invoice, "project", None), "name", None) or "").strip()
|
||||
or (getattr(invoice, "invoice_number", None) or "").strip()
|
||||
or str(getattr(invoice, "id", ""))
|
||||
)
|
||||
if buyer_ref:
|
||||
_text_el(agreement, ram + "BuyerReference", buyer_ref)
|
||||
|
||||
_build_party(agreement, "SellerTradeParty", seller)
|
||||
_build_party(agreement, "BuyerTradeParty", buyer)
|
||||
|
||||
# --- Header Trade Delivery ---
|
||||
_sub(txn, ram + "ApplicableHeaderTradeDelivery")
|
||||
|
||||
# --- Header Trade Settlement ---
|
||||
settlement = _sub(txn, ram + "ApplicableHeaderTradeSettlement")
|
||||
_text_el(settlement, ram + "InvoiceCurrencyCode", currency)
|
||||
|
||||
# Tax summary
|
||||
tax_el = _sub(settlement, ram + "ApplicableTradeTax")
|
||||
calc_amt = _text_el(tax_el, ram + "CalculatedAmount", _money(getattr(invoice, "tax_amount", 0)))
|
||||
_text_el(tax_el, ram + "TypeCode", "VAT")
|
||||
basis_amt = _text_el(tax_el, ram + "BasisAmount", _money(getattr(invoice, "subtotal", 0)))
|
||||
_text_el(tax_el, ram + "CategoryCode", tax_category)
|
||||
_text_el(tax_el, ram + "RateApplicablePercent", _money(tax_rate))
|
||||
|
||||
# Payment terms (due date)
|
||||
due_date = getattr(invoice, "due_date", None)
|
||||
if due_date:
|
||||
terms = _sub(settlement, ram + "SpecifiedTradePaymentTerms")
|
||||
due_dt = _sub(terms, ram + "DueDateDateTime")
|
||||
_date_el(due_dt, due_date)
|
||||
|
||||
# Monetary summation
|
||||
totals = _sub(settlement, ram + "SpecifiedTradeSettlementHeaderMonetarySummation")
|
||||
_text_el(totals, ram + "LineTotalAmount", _money(getattr(invoice, "subtotal", 0)))
|
||||
_text_el(totals, ram + "TaxBasisTotalAmount", _money(getattr(invoice, "subtotal", 0)))
|
||||
tax_total_el = _text_el(totals, ram + "TaxTotalAmount", _money(getattr(invoice, "tax_amount", 0)))
|
||||
if tax_total_el is not None:
|
||||
tax_total_el.set("currencyID", currency)
|
||||
_text_el(totals, ram + "GrandTotalAmount", _money(getattr(invoice, "total_amount", 0)))
|
||||
_text_el(totals, ram + "DuePayableAmount", _money(getattr(invoice, "total_amount", 0)))
|
||||
|
||||
# --- Line Items ---
|
||||
line_id = 1
|
||||
|
||||
def _add_line(description: str, quantity: Any, unit_price: Any, line_total: Any) -> None:
|
||||
nonlocal line_id
|
||||
li = _sub(txn, ram + "IncludedSupplyChainTradeLineItem")
|
||||
|
||||
line_doc = _sub(li, ram + "AssociatedDocumentLineDocument")
|
||||
_text_el(line_doc, ram + "LineID", str(line_id))
|
||||
|
||||
product = _sub(li, ram + "SpecifiedTradeProduct")
|
||||
_text_el(product, ram + "Name", str(description)[:200])
|
||||
|
||||
line_agreement = _sub(li, ram + "SpecifiedLineTradeAgreement")
|
||||
net_price = _sub(line_agreement, ram + "NetPriceProductTradePrice")
|
||||
_text_el(net_price, ram + "ChargeAmount", _money(unit_price))
|
||||
|
||||
line_delivery = _sub(li, ram + "SpecifiedLineTradeDelivery")
|
||||
qty_el = _text_el(line_delivery, ram + "BilledQuantity", _qty(quantity))
|
||||
if qty_el is not None:
|
||||
qty_el.set("unitCode", "C62")
|
||||
|
||||
line_settle = _sub(li, ram + "SpecifiedLineTradeSettlement")
|
||||
line_tax = _sub(line_settle, ram + "ApplicableTradeTax")
|
||||
_text_el(line_tax, ram + "TypeCode", "VAT")
|
||||
_text_el(line_tax, ram + "CategoryCode", tax_category)
|
||||
_text_el(line_tax, ram + "RateApplicablePercent", _money(tax_rate))
|
||||
|
||||
line_totals = _sub(line_settle, ram + "SpecifiedTradeSettlementLineMonetarySummation")
|
||||
_text_el(line_totals, ram + "LineTotalAmount", _money(line_total))
|
||||
|
||||
line_id += 1
|
||||
|
||||
# Invoice items
|
||||
try:
|
||||
for it in list(getattr(invoice, "items", []) or []):
|
||||
_add_line(
|
||||
description=getattr(it, "description", "Item"),
|
||||
quantity=getattr(it, "quantity", 1),
|
||||
unit_price=getattr(it, "unit_price", 0),
|
||||
line_total=getattr(it, "total_amount", 0),
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Expenses
|
||||
try:
|
||||
expenses_rel = getattr(invoice, "expenses", None)
|
||||
expenses = list(expenses_rel) if expenses_rel is not None else []
|
||||
for ex in expenses:
|
||||
desc = getattr(ex, "title", "Expense")
|
||||
if getattr(ex, "vendor", None):
|
||||
desc = f"{desc} ({ex.vendor})"
|
||||
_add_line(
|
||||
description=desc,
|
||||
quantity=1,
|
||||
unit_price=getattr(ex, "total_amount", 0),
|
||||
line_total=getattr(ex, "total_amount", 0),
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Extra goods
|
||||
try:
|
||||
goods_rel = getattr(invoice, "extra_goods", None)
|
||||
goods = list(goods_rel) if goods_rel is not None else []
|
||||
for g in goods:
|
||||
_add_line(
|
||||
description=getattr(g, "name", "Good"),
|
||||
quantity=getattr(g, "quantity", 1),
|
||||
unit_price=getattr(g, "unit_price", 0),
|
||||
line_total=getattr(g, "total_amount", 0),
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# If no lines were added, add a single placeholder line (CII requires at least one)
|
||||
if line_id == 1:
|
||||
_add_line(description="Invoice", quantity=1, unit_price=getattr(invoice, "total_amount", 0), line_total=getattr(invoice, "total_amount", 0))
|
||||
|
||||
xml_bytes = ET.tostring(root, encoding="utf-8", xml_declaration=True)
|
||||
sha256_hex = hashlib.sha256(xml_bytes).hexdigest()
|
||||
return xml_bytes.decode("utf-8"), sha256_hex
|
||||
@@ -1,8 +1,10 @@
|
||||
"""
|
||||
Optional validation gates for invoice PDF and UBL (veraPDF, EN16931).
|
||||
Validation gates for invoice PDF, UBL, and CII exports.
|
||||
|
||||
When configured, export flow can run external validators and surface
|
||||
actionable failures or summaries in the UI.
|
||||
Provides:
|
||||
- UBL well-formedness + basic Peppol BIS 3.0 structure validation
|
||||
- CII well-formedness + EN 16931 / Factur-X structure validation
|
||||
- Optional veraPDF CLI invocation for PDF/A compliance
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -14,6 +16,17 @@ from typing import List, Optional, Tuple
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
|
||||
# ---- UBL Validation ----
|
||||
|
||||
_UBL_NS_INVOICE = "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"
|
||||
_UBL_NS_CBC = "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2"
|
||||
_UBL_NS_CAC = "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"
|
||||
|
||||
_PEPPOL_BIS3_CUSTOMIZATION_ID = (
|
||||
"urn:cen.eu:en16931:2017#compliant#urn:fdc:peppol.eu:2017:poacc:billing:3.0"
|
||||
)
|
||||
|
||||
|
||||
def validate_ubl_wellformed(ubl_xml: str) -> Tuple[bool, List[str]]:
|
||||
"""
|
||||
Check UBL XML is well-formed and contains Invoice root.
|
||||
@@ -32,6 +45,238 @@ def validate_ubl_wellformed(ubl_xml: str) -> Tuple[bool, List[str]]:
|
||||
return False, messages
|
||||
|
||||
|
||||
def validate_ubl_peppol_bis3(ubl_xml: str) -> Tuple[bool, List[str]]:
|
||||
"""
|
||||
Validate UBL against Peppol BIS Billing 3.0 structural requirements.
|
||||
This checks required elements are present (not full Schematron).
|
||||
Returns (passed, list of issue strings).
|
||||
"""
|
||||
issues: List[str] = []
|
||||
|
||||
ok, parse_msgs = validate_ubl_wellformed(ubl_xml)
|
||||
if not ok:
|
||||
return False, parse_msgs
|
||||
|
||||
root = ET.fromstring(ubl_xml)
|
||||
ns = {
|
||||
"inv": _UBL_NS_INVOICE,
|
||||
"cbc": _UBL_NS_CBC,
|
||||
"cac": _UBL_NS_CAC,
|
||||
}
|
||||
|
||||
def _find_text(path: str) -> Optional[str]:
|
||||
el = root.find(path, ns)
|
||||
return el.text.strip() if el is not None and el.text else None
|
||||
|
||||
cust_id = _find_text("cbc:CustomizationID")
|
||||
if not cust_id:
|
||||
issues.append("Missing cbc:CustomizationID (BT-24)")
|
||||
elif _PEPPOL_BIS3_CUSTOMIZATION_ID not in cust_id:
|
||||
issues.append(f"CustomizationID does not reference Peppol BIS 3.0: {cust_id}")
|
||||
|
||||
if not _find_text("cbc:ProfileID"):
|
||||
issues.append("Missing cbc:ProfileID (BT-23)")
|
||||
|
||||
if not _find_text("cbc:ID"):
|
||||
issues.append("Missing cbc:ID (BT-1, Invoice number)")
|
||||
|
||||
type_code = _find_text("cbc:InvoiceTypeCode")
|
||||
if not type_code:
|
||||
issues.append("Missing cbc:InvoiceTypeCode (BT-3)")
|
||||
elif type_code not in ("380", "381", "384", "389", "751"):
|
||||
issues.append(f"Unusual InvoiceTypeCode: {type_code}")
|
||||
|
||||
if not _find_text("cbc:IssueDate"):
|
||||
issues.append("Missing cbc:IssueDate (BT-2)")
|
||||
|
||||
if not _find_text("cbc:DocumentCurrencyCode"):
|
||||
issues.append("Missing cbc:DocumentCurrencyCode (BT-5)")
|
||||
|
||||
if not _find_text("cbc:BuyerReference"):
|
||||
issues.append("Missing cbc:BuyerReference (BT-10, required by Peppol)")
|
||||
|
||||
supplier = root.find("cac:AccountingSupplierParty", ns)
|
||||
if supplier is None:
|
||||
issues.append("Missing AccountingSupplierParty")
|
||||
else:
|
||||
party = supplier.find("cac:Party", ns)
|
||||
if party is not None:
|
||||
ep = party.find("cbc:EndpointID", ns)
|
||||
if ep is None or not (ep.text or "").strip():
|
||||
issues.append("Supplier missing EndpointID")
|
||||
elif not ep.get("schemeID"):
|
||||
issues.append("Supplier EndpointID missing schemeID attribute")
|
||||
|
||||
customer = root.find("cac:AccountingCustomerParty", ns)
|
||||
if customer is None:
|
||||
issues.append("Missing AccountingCustomerParty")
|
||||
else:
|
||||
party = customer.find("cac:Party", ns)
|
||||
if party is not None:
|
||||
ep = party.find("cbc:EndpointID", ns)
|
||||
if ep is None or not (ep.text or "").strip():
|
||||
issues.append("Customer missing EndpointID")
|
||||
|
||||
lines = root.findall("cac:InvoiceLine", ns)
|
||||
if not lines:
|
||||
issues.append("No InvoiceLine elements found (at least one required)")
|
||||
for i, line in enumerate(lines, 1):
|
||||
qty = line.find("cbc:InvoicedQuantity", ns)
|
||||
if qty is not None and not qty.get("unitCode"):
|
||||
issues.append(f"InvoiceLine {i}: InvoicedQuantity missing unitCode attribute")
|
||||
|
||||
tax_total = root.find("cac:TaxTotal", ns)
|
||||
if tax_total is None:
|
||||
issues.append("Missing TaxTotal")
|
||||
|
||||
legal_total = root.find("cac:LegalMonetaryTotal", ns)
|
||||
if legal_total is None:
|
||||
issues.append("Missing LegalMonetaryTotal")
|
||||
else:
|
||||
if legal_total.find("cbc:PayableAmount", ns) is None:
|
||||
issues.append("Missing PayableAmount in LegalMonetaryTotal")
|
||||
|
||||
return len(issues) == 0, issues
|
||||
|
||||
|
||||
# ---- CII / Factur-X Validation ----
|
||||
|
||||
_CII_NS_RSM = "urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100"
|
||||
_CII_NS_RAM = "urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"
|
||||
_CII_NS_UDT = "urn:un:unece:uncefact:data:standard:UnqualifiedDataType:100"
|
||||
|
||||
|
||||
def validate_cii_wellformed(cii_xml: str) -> Tuple[bool, List[str]]:
|
||||
"""
|
||||
Check CII XML is well-formed and contains CrossIndustryInvoice root.
|
||||
Returns (passed, list of message strings).
|
||||
"""
|
||||
messages: List[str] = []
|
||||
try:
|
||||
root = ET.fromstring(cii_xml)
|
||||
local_tag = root.tag.split("}")[-1] if root.tag else ""
|
||||
if local_tag != "CrossIndustryInvoice":
|
||||
messages.append(f"Root element is '{local_tag}', expected 'CrossIndustryInvoice'.")
|
||||
return False, messages
|
||||
return True, []
|
||||
except ET.ParseError as e:
|
||||
messages.append(f"Invalid XML: {e}")
|
||||
return False, messages
|
||||
|
||||
|
||||
def validate_cii_en16931(cii_xml: str) -> Tuple[bool, List[str]]:
|
||||
"""
|
||||
Validate CII XML against EN 16931 / Factur-X structural requirements.
|
||||
Checks required elements for Factur-X EN 16931 (COMFORT) profile.
|
||||
Returns (passed, list of issue strings).
|
||||
"""
|
||||
issues: List[str] = []
|
||||
|
||||
ok, parse_msgs = validate_cii_wellformed(cii_xml)
|
||||
if not ok:
|
||||
return False, parse_msgs
|
||||
|
||||
root = ET.fromstring(cii_xml)
|
||||
ns = {
|
||||
"rsm": _CII_NS_RSM,
|
||||
"ram": _CII_NS_RAM,
|
||||
"udt": _CII_NS_UDT,
|
||||
}
|
||||
|
||||
def _find(path: str) -> Optional[ET.Element]:
|
||||
return root.find(path, ns)
|
||||
|
||||
def _find_text(path: str) -> Optional[str]:
|
||||
el = root.find(path, ns)
|
||||
return el.text.strip() if el is not None and el.text else None
|
||||
|
||||
# Context
|
||||
ctx = _find("rsm:ExchangedDocumentContext")
|
||||
if ctx is None:
|
||||
issues.append("Missing ExchangedDocumentContext")
|
||||
else:
|
||||
guideline = _find_text(
|
||||
"rsm:ExchangedDocumentContext/ram:GuidelineSpecifiedDocumentContextParameter/ram:ID"
|
||||
)
|
||||
if not guideline:
|
||||
issues.append("Missing GuidelineSpecifiedDocumentContextParameter/ID")
|
||||
|
||||
# Document
|
||||
doc = _find("rsm:ExchangedDocument")
|
||||
if doc is None:
|
||||
issues.append("Missing ExchangedDocument")
|
||||
else:
|
||||
if not _find_text("rsm:ExchangedDocument/ram:ID"):
|
||||
issues.append("Missing ExchangedDocument/ID (BT-1, Invoice number)")
|
||||
if not _find_text("rsm:ExchangedDocument/ram:TypeCode"):
|
||||
issues.append("Missing ExchangedDocument/TypeCode (BT-3)")
|
||||
if _find("rsm:ExchangedDocument/ram:IssueDateTime") is None:
|
||||
issues.append("Missing ExchangedDocument/IssueDateTime (BT-2)")
|
||||
|
||||
# Transaction
|
||||
txn = _find("rsm:SupplyChainTradeTransaction")
|
||||
if txn is None:
|
||||
issues.append("Missing SupplyChainTradeTransaction")
|
||||
return False, issues
|
||||
|
||||
# Agreement
|
||||
agreement = txn.find("ram:ApplicableHeaderTradeAgreement", ns)
|
||||
if agreement is None:
|
||||
issues.append("Missing ApplicableHeaderTradeAgreement")
|
||||
else:
|
||||
seller = agreement.find("ram:SellerTradeParty", ns)
|
||||
if seller is None:
|
||||
issues.append("Missing SellerTradeParty")
|
||||
else:
|
||||
seller_name = seller.find("ram:Name", ns)
|
||||
if seller_name is None or not (seller_name.text or "").strip():
|
||||
issues.append("SellerTradeParty missing Name")
|
||||
|
||||
buyer = agreement.find("ram:BuyerTradeParty", ns)
|
||||
if buyer is None:
|
||||
issues.append("Missing BuyerTradeParty")
|
||||
else:
|
||||
buyer_name = buyer.find("ram:Name", ns)
|
||||
if buyer_name is None or not (buyer_name.text or "").strip():
|
||||
issues.append("BuyerTradeParty missing Name")
|
||||
|
||||
# Settlement
|
||||
settlement = txn.find("ram:ApplicableHeaderTradeSettlement", ns)
|
||||
if settlement is None:
|
||||
issues.append("Missing ApplicableHeaderTradeSettlement")
|
||||
else:
|
||||
if not settlement.find("ram:InvoiceCurrencyCode", ns) is not None:
|
||||
issues.append("Missing InvoiceCurrencyCode (BT-5)")
|
||||
summation = settlement.find(
|
||||
"ram:SpecifiedTradeSettlementHeaderMonetarySummation", ns
|
||||
)
|
||||
if summation is None:
|
||||
issues.append("Missing SpecifiedTradeSettlementHeaderMonetarySummation")
|
||||
else:
|
||||
if summation.find("ram:GrandTotalAmount", ns) is None:
|
||||
issues.append("Missing GrandTotalAmount")
|
||||
if summation.find("ram:DuePayableAmount", ns) is None:
|
||||
issues.append("Missing DuePayableAmount")
|
||||
|
||||
# Line items
|
||||
lines = txn.findall("ram:IncludedSupplyChainTradeLineItem", ns)
|
||||
if not lines:
|
||||
issues.append("No IncludedSupplyChainTradeLineItem (at least one required)")
|
||||
for i, line in enumerate(lines, 1):
|
||||
product = line.find("ram:SpecifiedTradeProduct", ns)
|
||||
if product is None or product.find("ram:Name", ns) is None:
|
||||
issues.append(f"Line {i}: missing SpecifiedTradeProduct/Name")
|
||||
delivery = line.find("ram:SpecifiedLineTradeDelivery", ns)
|
||||
if delivery is not None:
|
||||
qty = delivery.find("ram:BilledQuantity", ns)
|
||||
if qty is not None and not qty.get("unitCode"):
|
||||
issues.append(f"Line {i}: BilledQuantity missing unitCode attribute")
|
||||
|
||||
return len(issues) == 0, issues
|
||||
|
||||
|
||||
# ---- PDF / veraPDF Validation ----
|
||||
|
||||
def validate_pdfa_verapdf(
|
||||
pdf_bytes: bytes,
|
||||
verapdf_path: Optional[str] = None,
|
||||
@@ -43,7 +288,7 @@ def validate_pdfa_verapdf(
|
||||
"""
|
||||
path = (verapdf_path or os.getenv("INVOICE_VERAPDF_PATH") or "").strip()
|
||||
if not path or not os.path.isfile(path):
|
||||
return True, [] # Skip when not configured
|
||||
return True, []
|
||||
|
||||
messages: List[str] = []
|
||||
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
|
||||
|
||||
+162
-17
@@ -1,28 +1,147 @@
|
||||
"""
|
||||
PDF/A-3 conversion and metadata normalization for ZUGFeRD invoices.
|
||||
PDF/A-3 conversion and metadata normalization for Factur-X / ZUGFeRD invoices.
|
||||
|
||||
Adds PDF/A-3 identification (XMP), output intent (sRGB), and ensures
|
||||
metadata is present so validators (e.g. veraPDF) can recognize the document.
|
||||
Adds PDF/A-3 identification (XMP), output intent with embedded sRGB ICC
|
||||
profile, and ensures metadata is present so validators (e.g. veraPDF) can
|
||||
recognize the document as PDF/A-3b compliant.
|
||||
|
||||
Limitations:
|
||||
- Font subsetting/embedding is the responsibility of the PDF generator
|
||||
(WeasyPrint/reportlab). This module only handles metadata and color.
|
||||
- For full archival compliance, run veraPDF after conversion to catch any
|
||||
remaining issues from the source PDF.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import struct
|
||||
import zlib
|
||||
from typing import Optional, Tuple
|
||||
|
||||
# PDF/A-3 identification namespace (veraPDF / ISO 19005)
|
||||
PDFA_PART = "3"
|
||||
PDFA_CONFORMANCE = "B" # Basic (color allowed)
|
||||
PDFA_NS = "http://www.pdfa.org/ns/pdfa/1.3/"
|
||||
PDFA_CONFORMANCE = "B"
|
||||
PDFA_NS = "http://www.aiim.org/pdfa/ns/id/"
|
||||
RDF_NS = "http://www.w3.org/1999/02/22-rdf-syntax-ns#"
|
||||
|
||||
# sRGB output intent (ISO 15076-1); minimal ICC reference for PDF/A-3
|
||||
OUTPUT_INTENT_SUBTYPE = "GTS_PDFA1"
|
||||
# Standard sRGB ICC profile ID (PDF/A-3 allows reference by registry)
|
||||
OUTPUT_INTENT_REGISTRY = "http://www.color.org"
|
||||
OUTPUT_INTENT_INFO = "sRGB IEC61966-2.1"
|
||||
|
||||
|
||||
def _minimal_srgb_icc_profile() -> bytes:
|
||||
"""
|
||||
Build a minimal sRGB ICC profile that satisfies the PDF/A-3 requirement
|
||||
for an embedded DestOutputProfile in the OutputIntent.
|
||||
|
||||
This is a stripped-down profile based on the sRGB IEC61966-2.1 spec.
|
||||
It contains the required header, tag table, and enough data for veraPDF
|
||||
to accept it as a valid ICC color profile.
|
||||
"""
|
||||
# ICC profile header (128 bytes)
|
||||
header = bytearray(128)
|
||||
# Profile size (filled in later)
|
||||
# Preferred CMM type
|
||||
header[4:8] = b"lcms"
|
||||
# Profile version 2.1.0
|
||||
header[8:12] = struct.pack(">I", 0x02100000)
|
||||
# Device class: mntr (monitor)
|
||||
header[12:16] = b"mntr"
|
||||
# Color space: RGB
|
||||
header[16:20] = b"RGB "
|
||||
# PCS: XYZ
|
||||
header[20:24] = b"XYZ "
|
||||
# Date/time: 2024-01-01 00:00:00
|
||||
header[24:36] = struct.pack(">6H", 2024, 1, 1, 0, 0, 0)
|
||||
# Signature: acsp
|
||||
header[36:40] = b"acsp"
|
||||
# Primary platform: MSFT
|
||||
header[40:44] = b"MSFT"
|
||||
# Rendering intent: perceptual
|
||||
header[64:68] = struct.pack(">I", 0)
|
||||
# PCS illuminant D50 (X=0.9642, Y=1.0000, Z=0.8249 in s15Fixed16)
|
||||
header[68:72] = struct.pack(">i", int(0.9642 * 65536))
|
||||
header[72:76] = struct.pack(">i", int(1.0000 * 65536))
|
||||
header[76:80] = struct.pack(">i", int(0.8249 * 65536))
|
||||
# Creator signature
|
||||
header[80:84] = b"lcms"
|
||||
|
||||
# Tag table: desc, wtpt, rXYZ, gXYZ, bXYZ, rTRC, gTRC, bTRC, cprt
|
||||
tags = []
|
||||
|
||||
def _xyz_tag(x: float, y: float, z: float) -> bytes:
|
||||
return (
|
||||
b"XYZ " + b"\x00" * 4
|
||||
+ struct.pack(">i", int(x * 65536))
|
||||
+ struct.pack(">i", int(y * 65536))
|
||||
+ struct.pack(">i", int(z * 65536))
|
||||
)
|
||||
|
||||
def _curv_tag_gamma(gamma: float) -> bytes:
|
||||
val = int(gamma * 256)
|
||||
return b"curv" + b"\x00" * 4 + struct.pack(">I", 1) + struct.pack(">H", val) + b"\x00\x00"
|
||||
|
||||
def _desc_tag(text: str) -> bytes:
|
||||
ascii_bytes = text.encode("ascii") + b"\x00"
|
||||
data = b"desc" + b"\x00" * 4 + struct.pack(">I", len(ascii_bytes)) + ascii_bytes
|
||||
# Unicode and ScriptCode localization (empty)
|
||||
data += struct.pack(">I", 0) # Unicode language code
|
||||
data += struct.pack(">I", 0) # Unicode count
|
||||
data += struct.pack(">H", 0) + b"\x00" * 67 # ScriptCode
|
||||
while len(data) % 4 != 0:
|
||||
data += b"\x00"
|
||||
return data
|
||||
|
||||
def _text_tag(text: str) -> bytes:
|
||||
ascii_bytes = text.encode("ascii") + b"\x00"
|
||||
data = b"text" + b"\x00" * 4 + ascii_bytes
|
||||
while len(data) % 4 != 0:
|
||||
data += b"\x00"
|
||||
return data
|
||||
|
||||
# sRGB approximate values
|
||||
desc_data = _desc_tag("sRGB IEC61966-2.1")
|
||||
wtpt_data = _xyz_tag(0.9505, 1.0000, 1.0890)
|
||||
rXYZ_data = _xyz_tag(0.4124, 0.2126, 0.0193)
|
||||
gXYZ_data = _xyz_tag(0.3576, 0.7152, 0.1192)
|
||||
bXYZ_data = _xyz_tag(0.1805, 0.0722, 0.9505)
|
||||
rTRC_data = _curv_tag_gamma(2.2)
|
||||
gTRC_data = _curv_tag_gamma(2.2)
|
||||
bTRC_data = _curv_tag_gamma(2.2)
|
||||
cprt_data = _text_tag("No copyright, use freely")
|
||||
|
||||
tag_datas = [
|
||||
(b"desc", desc_data),
|
||||
(b"wtpt", wtpt_data),
|
||||
(b"rXYZ", rXYZ_data),
|
||||
(b"gXYZ", gXYZ_data),
|
||||
(b"bXYZ", bXYZ_data),
|
||||
(b"rTRC", rTRC_data),
|
||||
(b"gTRC", gTRC_data),
|
||||
(b"bTRC", bTRC_data),
|
||||
(b"cprt", cprt_data),
|
||||
]
|
||||
|
||||
tag_count = len(tag_datas)
|
||||
tag_table_size = 4 + tag_count * 12 # count + entries
|
||||
data_offset = 128 + tag_table_size
|
||||
|
||||
tag_table = struct.pack(">I", tag_count)
|
||||
payload = b""
|
||||
for sig, data in tag_datas:
|
||||
offset = data_offset + len(payload)
|
||||
tag_table += sig + struct.pack(">II", offset, len(data))
|
||||
payload += data
|
||||
while len(payload) % 4 != 0:
|
||||
payload += b"\x00"
|
||||
|
||||
profile = bytes(header) + tag_table + payload
|
||||
# Write profile size into header
|
||||
profile = struct.pack(">I", len(profile)) + profile[4:]
|
||||
|
||||
return profile
|
||||
|
||||
|
||||
def _ensure_pdfa3_xmp(xmp_str: str) -> str:
|
||||
"""Inject or update PDF/A-3 identification in XMP."""
|
||||
pdfa_desc = (
|
||||
@@ -42,7 +161,9 @@ def _ensure_pdfa3_xmp(xmp_str: str) -> str:
|
||||
|
||||
def convert_to_pdfa3(pdf_bytes: bytes) -> Tuple[bytes, Optional[str]]:
|
||||
"""
|
||||
Normalize PDF to PDF/A-3 (add identification and output intent).
|
||||
Normalize PDF to PDF/A-3b by adding identification XMP, an output intent
|
||||
with an embedded sRGB ICC profile, and marking info as XMP-only.
|
||||
|
||||
Returns (new_pdf_bytes, None) on success, or (original_pdf_bytes, error_message) on failure.
|
||||
"""
|
||||
try:
|
||||
@@ -58,7 +179,13 @@ def convert_to_pdfa3(pdf_bytes: bytes) -> Tuple[bytes, Optional[str]]:
|
||||
try:
|
||||
# Ensure metadata stream exists
|
||||
if not hasattr(pdf.Root, "Metadata") or pdf.Root.Metadata is None:
|
||||
minimal = '<?xpacket begin="" id="W5M0MpCehiHzreSzNTczkc9d"?><x:xmpmeta xmlns:x="adobe:ns:meta/"><rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"></rdf:RDF></x:xmpmeta><?xpacket end="w"?>'
|
||||
minimal = (
|
||||
'<?xpacket begin="\xef\xbb\xbf" id="W5M0MpCehiHzreSzNTczkc9d"?>'
|
||||
'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
|
||||
'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
|
||||
"</rdf:RDF></x:xmpmeta>"
|
||||
'<?xpacket end="w"?>'
|
||||
)
|
||||
pdf.Root.Metadata = pdf.make_stream(minimal.encode("utf-8"))
|
||||
|
||||
xmp_bytes = pdf.Root.Metadata.read_bytes()
|
||||
@@ -66,29 +193,48 @@ def convert_to_pdfa3(pdf_bytes: bytes) -> Tuple[bytes, Optional[str]]:
|
||||
new_xmp = _ensure_pdfa3_xmp(xmp_str)
|
||||
pdf.Root.Metadata = pdf.make_stream(new_xmp.encode("utf-8"))
|
||||
|
||||
# Add OutputIntent for PDF/A-3 (required for color)
|
||||
# Add OutputIntent with embedded ICC profile for PDF/A-3
|
||||
try:
|
||||
intents = pdf.Root.get("/OutputIntents")
|
||||
has_intent = intents is not None and len(intents) > 0
|
||||
except Exception:
|
||||
has_intent = False
|
||||
|
||||
if not has_intent:
|
||||
try:
|
||||
from pikepdf import Name, Dictionary, Array
|
||||
from pikepdf import Name, Dictionary, Array, Stream
|
||||
|
||||
icc_data = _minimal_srgb_icc_profile()
|
||||
icc_stream = Stream(pdf, icc_data)
|
||||
icc_stream[Name.N] = 3 # RGB = 3 components
|
||||
|
||||
intent = Dictionary(
|
||||
Type=Name.OutputIntent,
|
||||
S=Name("/GTS_PDFA1"),
|
||||
OutputConditionIdentifier=OUTPUT_INTENT_INFO,
|
||||
Info=OUTPUT_INTENT_INFO,
|
||||
OutputCondition="sRGB IEC61966-2.1",
|
||||
RegistryName=OUTPUT_INTENT_REGISTRY,
|
||||
DestOutputProfile=icc_stream,
|
||||
)
|
||||
pdf.Root.OutputIntents = Array(intent)
|
||||
pdf.Root.OutputIntents = Array([intent])
|
||||
except Exception:
|
||||
pass
|
||||
# Fallback: intent without embedded profile (less compliant but still useful)
|
||||
try:
|
||||
from pikepdf import Name, Dictionary, Array
|
||||
intent = Dictionary(
|
||||
Type=Name.OutputIntent,
|
||||
S=Name("/GTS_PDFA1"),
|
||||
OutputConditionIdentifier=OUTPUT_INTENT_INFO,
|
||||
Info=OUTPUT_INTENT_INFO,
|
||||
OutputCondition="sRGB IEC61966-2.1",
|
||||
RegistryName=OUTPUT_INTENT_REGISTRY,
|
||||
)
|
||||
pdf.Root.OutputIntents = Array([intent])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
out = io.BytesIO()
|
||||
# Newer pikepdf requires version as (str, int). Disable metadata version sync to avoid
|
||||
# "PDF version must be a tuple" when the doc's internal version is stored as string.
|
||||
pdf_version = ("1", 7)
|
||||
try:
|
||||
pdf.save(
|
||||
@@ -99,7 +245,6 @@ def convert_to_pdfa3(pdf_bytes: bytes) -> Tuple[bytes, Optional[str]]:
|
||||
)
|
||||
except Exception as ex:
|
||||
if "tuple" in str(ex).lower():
|
||||
# Fallback: avoid force_version so pikepdf doesn't validate doc version as tuple
|
||||
pdf.save(
|
||||
out,
|
||||
min_version=pdf_version,
|
||||
|
||||
+79
-81
@@ -1,11 +1,16 @@
|
||||
"""
|
||||
ZugFerd / Factur-X: embed EN 16931 UBL XML into invoice PDFs.
|
||||
Factur-X / ZUGFeRD: embed CII XML into invoice PDFs.
|
||||
|
||||
When enabled, exported invoice PDFs contain an embedded XML file (ZUGFeRD-invoice.xml)
|
||||
so the file is both human-readable (PDF) and machine-readable (EN 16931). The same
|
||||
UBL used for Peppol is reused; embedding is done with pikepdf.
|
||||
The attachment is added as an Associated File with relationship "Alternative" and
|
||||
ZUGFeRD XMP RDF is written so validators recognize the document.
|
||||
When enabled, exported invoice PDFs contain an embedded CII (Cross-Industry
|
||||
Invoice) XML file so the document is both human-readable (PDF) and
|
||||
machine-readable (EN 16931). Embedding is done with pikepdf.
|
||||
|
||||
Standards compliance:
|
||||
- The embedded XML uses UN/CEFACT CII format (NOT UBL). This is the
|
||||
correct payload format for Factur-X 1.0 / ZUGFeRD 2.x.
|
||||
- Peppol transport uses UBL (see app/integrations/peppol.py).
|
||||
- The file is attached as an Associated File with relationship "Alternative"
|
||||
and Factur-X XMP metadata is written so validators recognize the document.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -15,57 +20,56 @@ import os
|
||||
import tempfile
|
||||
from typing import Any, Optional, Tuple
|
||||
|
||||
from app.integrations.peppol import PeppolParty, build_peppol_ubl_invoice_xml
|
||||
from app.utils.cii_invoice import CIIParty, build_cii_invoice_xml
|
||||
|
||||
|
||||
# Standard embedded filename for ZUGFeRD/Factur-X (EN 16931)
|
||||
ZUGFERD_EMBEDDED_FILENAME = "ZUGFeRD-invoice.xml"
|
||||
# Standard embedded filename per Factur-X specification
|
||||
FACTURX_EMBEDDED_FILENAME = "factur-x.xml"
|
||||
# Legacy alias kept for backwards compatibility in tests
|
||||
ZUGFERD_EMBEDDED_FILENAME = FACTURX_EMBEDDED_FILENAME
|
||||
|
||||
# ZUGFeRD/Factur-X XMP namespace (PDF/A-3 Associated Files)
|
||||
ZUGFERD_XMP_NS = "urn:ferd:pdfa:CrossIndustryDocument:invoice:1p0#"
|
||||
# Factur-X XMP namespace (PDF/A-3 Associated Files)
|
||||
FACTURX_XMP_NS = "urn:factur-x:pdfa:CrossIndustryDocument:invoice:1p0#"
|
||||
|
||||
|
||||
def _get_sender_party_for_zugferd(settings: Any) -> PeppolParty:
|
||||
"""Build supplier party from Settings (best-effort; placeholders if missing)."""
|
||||
sender_endpoint_id = (
|
||||
(getattr(settings, "peppol_sender_endpoint_id", "") or os.getenv("PEPPOL_SENDER_ENDPOINT_ID") or "").strip()
|
||||
or "unknown"
|
||||
)
|
||||
sender_scheme_id = (
|
||||
(getattr(settings, "peppol_sender_scheme_id", "") or os.getenv("PEPPOL_SENDER_SCHEME_ID") or "").strip()
|
||||
or "0000"
|
||||
)
|
||||
sender_country = (
|
||||
(getattr(settings, "peppol_sender_country", "") or os.getenv("PEPPOL_SENDER_COUNTRY") or "").strip()
|
||||
or None
|
||||
)
|
||||
return PeppolParty(
|
||||
endpoint_id=sender_endpoint_id,
|
||||
endpoint_scheme_id=sender_scheme_id,
|
||||
def _get_seller_party(settings: Any) -> CIIParty:
|
||||
"""Build seller party from Settings (best-effort; placeholders if missing)."""
|
||||
return CIIParty(
|
||||
name=(getattr(settings, "company_name", None) or "Company").strip(),
|
||||
tax_id=(getattr(settings, "company_tax_id", None) or "").strip() or None,
|
||||
address_line=(getattr(settings, "company_address", None) or "").strip() or None,
|
||||
country_code=sender_country,
|
||||
country_code=(
|
||||
(getattr(settings, "peppol_sender_country", "") or os.getenv("PEPPOL_SENDER_COUNTRY") or "").strip()
|
||||
or None
|
||||
),
|
||||
email=(getattr(settings, "company_email", None) or "").strip() or None,
|
||||
phone=(getattr(settings, "company_phone", None) or "").strip() or None,
|
||||
endpoint_id=(
|
||||
(getattr(settings, "peppol_sender_endpoint_id", "") or os.getenv("PEPPOL_SENDER_ENDPOINT_ID") or "").strip()
|
||||
or None
|
||||
),
|
||||
endpoint_scheme_id=(
|
||||
(getattr(settings, "peppol_sender_scheme_id", "") or os.getenv("PEPPOL_SENDER_SCHEME_ID") or "").strip()
|
||||
or None
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _get_customer_party_for_zugferd(invoice: Any) -> PeppolParty:
|
||||
"""Build customer party from invoice and client (best-effort; placeholders if missing)."""
|
||||
def _get_buyer_party(invoice: Any) -> CIIParty:
|
||||
"""Build buyer party from invoice and client (best-effort)."""
|
||||
client = getattr(invoice, "client", None)
|
||||
endpoint_id = "unknown"
|
||||
scheme_id = "0000"
|
||||
country = None
|
||||
name = (getattr(invoice, "client_name", None) or "Customer").strip()
|
||||
tax_id = None
|
||||
address_line = None
|
||||
email = None
|
||||
phone = None
|
||||
country = None
|
||||
endpoint_id = None
|
||||
scheme_id = None
|
||||
|
||||
if client:
|
||||
endpoint_id = (client.get_custom_field("peppol_endpoint_id", "") or "").strip() or "unknown"
|
||||
scheme_id = (client.get_custom_field("peppol_scheme_id", "") or "").strip() or "0000"
|
||||
endpoint_id = (client.get_custom_field("peppol_endpoint_id", "") or "").strip() or None
|
||||
scheme_id = (client.get_custom_field("peppol_scheme_id", "") or "").strip() or None
|
||||
country = (client.get_custom_field("peppol_country", "") or "").strip() or None
|
||||
name = (getattr(client, "name", None) or getattr(invoice, "client_name", "") or "Customer").strip()
|
||||
tax_id = (client.get_custom_field("vat_id", "") or client.get_custom_field("tax_id", "") or "").strip() or None
|
||||
@@ -73,20 +77,20 @@ def _get_customer_party_for_zugferd(invoice: Any) -> PeppolParty:
|
||||
email = (getattr(client, "email", None) or getattr(invoice, "client_email", None) or "").strip() or None
|
||||
phone = (getattr(client, "phone", None) or "").strip() or None
|
||||
|
||||
return PeppolParty(
|
||||
endpoint_id=endpoint_id,
|
||||
endpoint_scheme_id=scheme_id,
|
||||
return CIIParty(
|
||||
name=name,
|
||||
tax_id=tax_id,
|
||||
address_line=address_line,
|
||||
country_code=country,
|
||||
email=email,
|
||||
phone=phone,
|
||||
endpoint_id=endpoint_id,
|
||||
endpoint_scheme_id=scheme_id,
|
||||
)
|
||||
|
||||
|
||||
# Minimal XMP with rdf:RDF for ZUGFeRD extension (PDF/A-3 style)
|
||||
_ZUGFERD_XMP_TEMPLATE = """<?xpacket begin="" id="W5M0MpCehiHzreSzNTczkc9d"?>
|
||||
# Minimal XMP template with rdf:RDF for Factur-X extension (PDF/A-3 style)
|
||||
_FACTURX_XMP_TEMPLATE = """<?xpacket begin="\xef\xbb\xbf" id="W5M0MpCehiHzreSzNTczkc9d"?>
|
||||
<x:xmpmeta xmlns:x="adobe:ns:meta/">
|
||||
<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">
|
||||
{rdf_description}
|
||||
@@ -102,31 +106,28 @@ def _ensure_metadata_stream(pdf: Any) -> None:
|
||||
if hasattr(pdf.Root, "Metadata") and pdf.Root.Metadata is not None:
|
||||
return
|
||||
try:
|
||||
minimal_xmp = _ZUGFERD_XMP_TEMPLATE.format(
|
||||
rdf_description=(
|
||||
f'<rdf:Description rdf:about="" xmlns:zf="{ZUGFERD_XMP_NS}">'
|
||||
"<zf:DocumentType>INVOICE</zf:DocumentType>"
|
||||
f"<zf:DocumentFileName>{ZUGFERD_EMBEDDED_FILENAME}</zf:DocumentFileName>"
|
||||
"<zf:Version>2.1</zf:Version>"
|
||||
"<zf:ConformanceLevel>EN 16931</zf:ConformanceLevel>"
|
||||
"</rdf:Description>"
|
||||
)
|
||||
)
|
||||
rdf_desc = _facturx_rdf_description()
|
||||
minimal_xmp = _FACTURX_XMP_TEMPLATE.format(rdf_description=rdf_desc)
|
||||
pdf.Root.Metadata = pdf.make_stream(minimal_xmp.encode("utf-8"))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _add_zugferd_xmp(pdf: Any) -> None:
|
||||
"""Add or ensure ZUGFeRD/Factur-X XMP RDF so validators recognize the embedded invoice XML."""
|
||||
zugferd_rdf = (
|
||||
f'<rdf:Description rdf:about="" xmlns:zf="{ZUGFERD_XMP_NS}">'
|
||||
"<zf:DocumentType>INVOICE</zf:DocumentType>"
|
||||
f"<zf:DocumentFileName>{ZUGFERD_EMBEDDED_FILENAME}</zf:DocumentFileName>"
|
||||
"<zf:Version>2.1</zf:Version>"
|
||||
"<zf:ConformanceLevel>EN 16931</zf:ConformanceLevel>"
|
||||
def _facturx_rdf_description() -> str:
|
||||
"""Return the Factur-X XMP RDF description block."""
|
||||
return (
|
||||
f'<rdf:Description rdf:about="" xmlns:fx="{FACTURX_XMP_NS}">'
|
||||
"<fx:DocumentType>INVOICE</fx:DocumentType>"
|
||||
f"<fx:DocumentFileName>{FACTURX_EMBEDDED_FILENAME}</fx:DocumentFileName>"
|
||||
"<fx:Version>1.0</fx:Version>"
|
||||
"<fx:ConformanceLevel>EN 16931</fx:ConformanceLevel>"
|
||||
"</rdf:Description>"
|
||||
)
|
||||
|
||||
|
||||
def _add_facturx_xmp(pdf: Any) -> None:
|
||||
"""Add or ensure Factur-X XMP RDF so validators recognize the embedded CII XML."""
|
||||
facturx_rdf = _facturx_rdf_description()
|
||||
_ensure_metadata_stream(pdf)
|
||||
if not hasattr(pdf, "Root") or not hasattr(pdf.Root, "Metadata"):
|
||||
return
|
||||
@@ -135,19 +136,19 @@ def _add_zugferd_xmp(pdf: Any) -> None:
|
||||
except Exception:
|
||||
return
|
||||
xmp_str = xmp_bytes.decode("utf-8", errors="replace")
|
||||
if "zf:DocumentType" in xmp_str:
|
||||
if "fx:DocumentType" in xmp_str or "factur-x" in xmp_str.lower():
|
||||
return
|
||||
marker = "</rdf:RDF>"
|
||||
if marker in xmp_str:
|
||||
try:
|
||||
insert_pos = xmp_str.rfind(marker)
|
||||
new_xmp = xmp_str[:insert_pos] + zugferd_rdf + "\n " + xmp_str[insert_pos:]
|
||||
new_xmp = xmp_str[:insert_pos] + facturx_rdf + "\n " + xmp_str[insert_pos:]
|
||||
pdf.Root.Metadata = pdf.make_stream(new_xmp.encode("utf-8"))
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
minimal_xmp = _ZUGFERD_XMP_TEMPLATE.format(rdf_description=zugferd_rdf)
|
||||
minimal_xmp = _FACTURX_XMP_TEMPLATE.format(rdf_description=facturx_rdf)
|
||||
pdf.Root.Metadata = pdf.make_stream(minimal_xmp.encode("utf-8"))
|
||||
except Exception:
|
||||
pass
|
||||
@@ -155,11 +156,11 @@ def _add_zugferd_xmp(pdf: Any) -> None:
|
||||
|
||||
def embed_zugferd_xml_in_pdf(pdf_bytes: bytes, invoice: Any, settings: Any) -> Tuple[bytes, Optional[str]]:
|
||||
"""
|
||||
Embed EN 16931 UBL XML into the given invoice PDF bytes (ZugFerd/Factur-X).
|
||||
Embed Factur-X CII XML into the given invoice PDF bytes.
|
||||
|
||||
Builds supplier/customer from settings and invoice (best-effort), generates UBL,
|
||||
attaches it as ZUGFeRD-invoice.xml with AF relationship "Alternative", adds
|
||||
ZUGFeRD XMP RDF, and returns the new PDF bytes.
|
||||
Builds seller/buyer from settings and invoice (best-effort), generates CII
|
||||
XML, attaches it as factur-x.xml with AF relationship "Alternative", adds
|
||||
Factur-X XMP RDF, and returns the new PDF bytes.
|
||||
|
||||
Returns:
|
||||
(new_pdf_bytes, None) on success, or (original_pdf_bytes, error_message) on failure.
|
||||
@@ -171,16 +172,15 @@ def embed_zugferd_xml_in_pdf(pdf_bytes: bytes, invoice: Any, settings: Any) -> T
|
||||
return pdf_bytes, f"pikepdf not available: {e}"
|
||||
|
||||
try:
|
||||
supplier = _get_sender_party_for_zugferd(settings)
|
||||
customer = _get_customer_party_for_zugferd(invoice)
|
||||
ubl_xml, _ = build_peppol_ubl_invoice_xml(invoice=invoice, supplier=supplier, customer=customer)
|
||||
seller = _get_seller_party(settings)
|
||||
buyer = _get_buyer_party(invoice)
|
||||
cii_xml, _ = build_cii_invoice_xml(invoice=invoice, seller=seller, buyer=buyer)
|
||||
except Exception as e:
|
||||
return pdf_bytes, f"Failed to build UBL for ZugFerd: {e}"
|
||||
return pdf_bytes, f"Failed to build CII XML for Factur-X: {e}"
|
||||
|
||||
try:
|
||||
pdf = pikepdf.open(io.BytesIO(pdf_bytes))
|
||||
ubl_bytes = ubl_xml.encode("utf-8")
|
||||
# AttachedFileSpec(pdf, data, ...) - relationship must be Name for ZUGFeRD /Alternative
|
||||
cii_bytes = cii_xml.encode("utf-8")
|
||||
try:
|
||||
from pikepdf import Name
|
||||
relationship = Name("/Alternative")
|
||||
@@ -189,17 +189,16 @@ def embed_zugferd_xml_in_pdf(pdf_bytes: bytes, invoice: Any, settings: Any) -> T
|
||||
try:
|
||||
filespec = AttachedFileSpec(
|
||||
pdf,
|
||||
ubl_bytes,
|
||||
filename=ZUGFERD_EMBEDDED_FILENAME,
|
||||
cii_bytes,
|
||||
filename=FACTURX_EMBEDDED_FILENAME,
|
||||
mime_type="application/xml",
|
||||
relationship=relationship,
|
||||
)
|
||||
except TypeError:
|
||||
# Older pikepdf: from_filepath(path, relationship=...) or different constructor
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="wb", suffix=".xml", delete=False, prefix="zugferd_"
|
||||
mode="wb", suffix=".xml", delete=False, prefix="facturx_"
|
||||
) as tmp:
|
||||
tmp.write(ubl_bytes)
|
||||
tmp.write(cii_bytes)
|
||||
tmp_path = tmp.name
|
||||
try:
|
||||
filespec = AttachedFileSpec.from_filepath(pdf, tmp_path, relationship="/Alternative")
|
||||
@@ -208,10 +207,9 @@ def embed_zugferd_xml_in_pdf(pdf_bytes: bytes, invoice: Any, settings: Any) -> T
|
||||
os.unlink(tmp_path)
|
||||
except OSError:
|
||||
pass
|
||||
pdf.attachments[ZUGFERD_EMBEDDED_FILENAME] = filespec
|
||||
_add_zugferd_xmp(pdf)
|
||||
pdf.attachments[FACTURX_EMBEDDED_FILENAME] = filespec
|
||||
_add_facturx_xmp(pdf)
|
||||
out = io.BytesIO()
|
||||
# pikepdf may require version as (str, int) for PDF 1.7
|
||||
try:
|
||||
pdf.save(out, min_version=("1", 7))
|
||||
except TypeError:
|
||||
@@ -219,4 +217,4 @@ def embed_zugferd_xml_in_pdf(pdf_bytes: bytes, invoice: Any, settings: Any) -> T
|
||||
pdf.close()
|
||||
return out.getvalue(), None
|
||||
except Exception as e:
|
||||
return pdf_bytes, f"Failed to embed ZugFerd XML in PDF: {e}"
|
||||
return pdf_bytes, f"Failed to embed Factur-X CII XML in PDF: {e}"
|
||||
|
||||
@@ -1,22 +1,31 @@
|
||||
# Peppol and ZugFerd e-invoicing (EN 16931)
|
||||
# Peppol and Factur-X / ZUGFeRD e-invoicing (EN 16931)
|
||||
|
||||
TimeTracker supports **both**:
|
||||
|
||||
- **Peppol** – send invoices via the Peppol network (UBL 2.1, BIS Billing 3.0) to your **Peppol Access Point**.
|
||||
- **ZugFerd / Factur-X** – export invoice PDFs that contain **embedded EN 16931 XML** (one file that is both human-readable and machine-readable). These hybrid PDFs can also be sent via Peppol.
|
||||
- **Factur-X / ZUGFeRD** – export invoice PDFs that contain **embedded CII XML** (Cross-Industry Invoice, EN 16931 profile). These hybrid PDFs are both human-readable and machine-readable.
|
||||
|
||||
Peppol is the **transport**; ZugFerd is a **format** (PDF + embedded XML). The same UBL used for Peppol is reused when embedding (EN 16931 compliant).
|
||||
### Supported standards
|
||||
|
||||
| Standard | Format | Status |
|
||||
|---|---|---|
|
||||
| Peppol BIS Billing 3.0 | UBL 2.1 | Supported (transport + export) |
|
||||
| Factur-X / ZUGFeRD 2.x | CII (EN 16931 profile) | Supported (embedded in PDF) |
|
||||
| PDF/A-3b | PDF archival | Supported (with ICC profile) |
|
||||
| XRechnung | CII / UBL (German CIUS) | Not supported |
|
||||
|
||||
Peppol is the **transport**; Factur-X / ZUGFeRD is a **format** (PDF + embedded CII XML). Each uses its own XML payload — UBL for Peppol, CII for Factur-X.
|
||||
|
||||
## What you need
|
||||
|
||||
- **A Peppol Access Point provider** (e.g. your accountant’s solution or a commercial AP)
|
||||
- **A Peppol Access Point provider** (e.g. your accountant's solution or a commercial AP)
|
||||
- Your **sender identifiers** (how your company is identified in Peppol)
|
||||
- Your customers’ **recipient endpoint identifiers**
|
||||
- Your customers' **recipient endpoint identifiers**
|
||||
|
||||
TimeTracker supports two **transport modes**:
|
||||
|
||||
- **Generic** – provider-agnostic HTTP adapter: you configure an access point URL that accepts the JSON contract below. No SML/SMP or AS4 required.
|
||||
- **Native** – SML/SMP participant discovery and AS4 message send. Requires `PEPPOL_SML_URL` (and optionally client certificate paths for mTLS). Use when you want to send directly via the PEPPOL network without a third-party AP adapter.
|
||||
- **Generic** – provider-agnostic HTTP adapter: you configure an access point URL that accepts the JSON contract below. No SML/SMP or AS4 required. **Recommended for production.**
|
||||
- **Native (experimental)** – SML/SMP participant discovery and AS4 message send. Lacks WS-Security, digital signatures, and receipt handling. Use only for testing or when you have a compatible receiving AP.
|
||||
|
||||
Sender and recipient identifiers are validated (scheme and endpoint ID format) before send in both modes.
|
||||
|
||||
@@ -98,23 +107,36 @@ You can optionally set **Buyer reference (PEPPOL BT-10)** on each invoice (creat
|
||||
|
||||
When the setting is on **and** the client has Peppol endpoint details, the invoice view shows a **Download UBL** button to save the UBL 2.1 XML file.
|
||||
|
||||
## Embed EN 16931 XML in invoice PDFs (ZugFerd / Factur-X)
|
||||
## Embed Factur-X / ZUGFeRD CII XML in invoice PDFs
|
||||
|
||||
In **Admin → Settings → Peppol e-Invoicing** you can enable **Embed EN 16931 XML in invoice PDFs (ZugFerd / Factur-X)**. When this is on:
|
||||
In **Admin → Settings → Peppol e-Invoicing** you can enable **Embed Factur-X / ZUGFeRD CII XML in invoice PDFs (EN 16931)**. When this is on:
|
||||
|
||||
- **Exported invoice PDFs** (Export PDF) contain an embedded file `ZUGFeRD-invoice.xml` with the same EN 16931 UBL as used for Peppol.
|
||||
- The embedded XML is attached as an **Associated File** with relationship **Alternative**, and ZUGFeRD XMP RDF is written (metadata is created if missing so validators recognize the document).
|
||||
- **Exported invoice PDFs** (Export PDF) contain an embedded file `factur-x.xml` with a CII (Cross-Industry Invoice) XML conforming to the Factur-X EN 16931 profile.
|
||||
- The embedded XML is attached as an **Associated File** with relationship **Alternative**, and Factur-X XMP metadata is written so validators recognize the document.
|
||||
- The PDF remains human-readable; the embedded XML makes it machine-readable (e.g. for automated booking or archiving).
|
||||
- These hybrid PDFs can be sent via Peppol or by email; recipients can open the PDF and/or extract the XML.
|
||||
- **Strict behaviour:** If ZUGFeRD embedding is enabled and the embed step fails (e.g. missing pikepdf, invalid PDF), the export is **aborted** and the user sees an error; the PDF is not returned without the XML.
|
||||
- **Strict behaviour:** If embedding is enabled and the embed step fails (e.g. missing pikepdf, invalid PDF), the export is **aborted** and the user sees an error; the PDF is not returned without the XML.
|
||||
|
||||
Party data (seller/customer) is taken from Settings and the invoice’s client (including Peppol endpoint fields). For full EN 16931/ZugFerd compliance, configure sender and client data as for Peppol (including company and client addresses and country codes).
|
||||
Party data (seller/buyer) is taken from Settings and the invoice's client (including endpoint fields and VAT). For full EN 16931 compliance, configure seller and client data including addresses and country codes.
|
||||
|
||||
**Validation:** Validate the embedded XML with [b2brouter](https://app.b2brouter.net/de/validation) or [portinvoice.com](https://www.portinvoice.com/). You can optionally enable **Run veraPDF after export** in Admin → Peppol e-Invoicing and set the veraPDF executable path to get a validation summary after each export (does not block the download).
|
||||
|
||||
### ZUGFeRD and PDF/A-3
|
||||
### Factur-X and PDF/A-3
|
||||
|
||||
You can enable **Normalize ZUGFeRD PDFs to PDF/A-3** in Admin → Peppol e-Invoicing. When this is on (and ZUGFeRD embedding is enabled), exported PDFs are normalized to PDF/A-3: XMP identification (pdfaid part 3, conformance B) and an sRGB output intent are added so the file passes validators such as veraPDF. If conversion fails, export is aborted and the user sees an error.
|
||||
You can enable **Normalize Factur-X PDFs to PDF/A-3b** in Admin → Peppol e-Invoicing. When this is on (and Factur-X embedding is enabled), exported PDFs are normalized to PDF/A-3b:
|
||||
|
||||
- XMP identification (`pdfaid:part=3`, `pdfaid:conformance=B`)
|
||||
- Embedded sRGB ICC color profile (DestOutputProfile)
|
||||
- GTS_PDFA1 output intent
|
||||
|
||||
If conversion fails, export is aborted and the user sees an error.
|
||||
|
||||
### UBL validation
|
||||
|
||||
When exporting or sending UBL via Peppol, the generated XML is checked for structural compliance with Peppol BIS Billing 3.0 requirements (required elements, identifiers, line items). Full Schematron validation is not performed in-app; use your Access Point provider's validator or [ecosio](https://ecosio.com/en/peppol-and-xml-document-validator/) for deep validation.
|
||||
|
||||
### CII validation
|
||||
|
||||
When embedding Factur-X CII XML, the generated XML is checked for EN 16931 structural requirements (required elements, party data, line items, monetary totals).
|
||||
|
||||
## Migrations
|
||||
|
||||
@@ -128,7 +150,7 @@ This applies (among others):
|
||||
|
||||
- `112_add_invoices_peppol_compliant` (adds `settings.invoices_peppol_compliant`)
|
||||
- `113_add_invoice_buyer_reference` (adds `invoices.buyer_reference`)
|
||||
- `128_add_invoices_zugferd_pdf` (adds `settings.invoices_zugferd_pdf` for ZugFerd PDF embedding)
|
||||
- `128_add_invoices_zugferd_pdf` (adds `settings.invoices_zugferd_pdf` for Factur-X PDF embedding)
|
||||
- `130_add_peppol_transport_mode_and_native` (adds `peppol_transport_mode`, `peppol_sml_url`, `peppol_native_cert_path`, `peppol_native_key_path`, `invoices_pdfa3_compliant`, `invoices_validate_export`, `invoices_verapdf_path`)
|
||||
|
||||
## Testing
|
||||
@@ -138,4 +160,3 @@ With your virtual environment activated:
|
||||
```bash
|
||||
pytest tests/test_peppol_service.py tests/test_peppol_identifiers.py tests/test_zugferd.py tests/test_pdfa3.py tests/test_invoice_validators.py -v
|
||||
```
|
||||
|
||||
|
||||
@@ -0,0 +1,235 @@
|
||||
"""Tests for CII (Cross-Industry Invoice) generator for Factur-X / ZUGFeRD."""
|
||||
from datetime import date, timedelta
|
||||
from decimal import Decimal
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from app.utils.cii_invoice import CIIParty, build_cii_invoice_xml, FACTURX_GUIDELINE_EN16931
|
||||
from app.utils.invoice_validators import validate_cii_wellformed, validate_cii_en16931
|
||||
|
||||
|
||||
def _make_invoice(**overrides):
|
||||
defaults = dict(
|
||||
id=1,
|
||||
invoice_number="INV-CII-001",
|
||||
issue_date=date(2024, 3, 15),
|
||||
due_date=date(2024, 4, 14),
|
||||
currency_code="EUR",
|
||||
subtotal=Decimal("200.00"),
|
||||
tax_rate=Decimal("21.00"),
|
||||
tax_amount=Decimal("42.00"),
|
||||
total_amount=Decimal("242.00"),
|
||||
notes="Test invoice notes",
|
||||
buyer_reference="PO-99",
|
||||
project=None,
|
||||
client=None,
|
||||
client_name="Buyer Inc",
|
||||
client_email=None,
|
||||
client_address=None,
|
||||
items=[
|
||||
SimpleNamespace(description="Consulting", quantity=Decimal("2"), unit_price=Decimal("100.00"), total_amount=Decimal("200.00")),
|
||||
],
|
||||
expenses=[],
|
||||
extra_goods=[],
|
||||
)
|
||||
defaults.update(overrides)
|
||||
return SimpleNamespace(**defaults)
|
||||
|
||||
|
||||
def _make_seller(**overrides):
|
||||
defaults = dict(
|
||||
name="Seller GmbH",
|
||||
tax_id="DE123456789",
|
||||
address_line="Hauptstr. 1",
|
||||
city="Berlin",
|
||||
postcode="10115",
|
||||
country_code="DE",
|
||||
email="seller@example.de",
|
||||
phone="+49 30 12345",
|
||||
endpoint_id="9930:DE123456789",
|
||||
endpoint_scheme_id="9930",
|
||||
)
|
||||
defaults.update(overrides)
|
||||
return CIIParty(**defaults)
|
||||
|
||||
|
||||
def _make_buyer(**overrides):
|
||||
defaults = dict(
|
||||
name="Buyer BV",
|
||||
tax_id="NL123456789B01",
|
||||
address_line="Keizersgracht 1",
|
||||
city="Amsterdam",
|
||||
postcode="1015 AA",
|
||||
country_code="NL",
|
||||
email="buyer@example.nl",
|
||||
)
|
||||
defaults.update(overrides)
|
||||
return CIIParty(**defaults)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_build_cii_produces_valid_xml():
|
||||
invoice = _make_invoice()
|
||||
seller = _make_seller()
|
||||
buyer = _make_buyer()
|
||||
xml, sha256 = build_cii_invoice_xml(invoice, seller, buyer)
|
||||
assert sha256
|
||||
passed, msgs = validate_cii_wellformed(xml)
|
||||
assert passed is True, f"CII well-formedness failed: {msgs}"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_build_cii_passes_en16931_validation():
|
||||
invoice = _make_invoice()
|
||||
seller = _make_seller()
|
||||
buyer = _make_buyer()
|
||||
xml, _ = build_cii_invoice_xml(invoice, seller, buyer)
|
||||
passed, issues = validate_cii_en16931(xml)
|
||||
assert passed is True, f"CII EN 16931 validation failed: {issues}"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_contains_guideline_id():
|
||||
invoice = _make_invoice()
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert FACTURX_GUIDELINE_EN16931 in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_contains_invoice_number():
|
||||
invoice = _make_invoice(invoice_number="INV-2024-999")
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert "INV-2024-999" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_contains_type_code_380():
|
||||
xml, _ = build_cii_invoice_xml(_make_invoice(), _make_seller(), _make_buyer())
|
||||
assert ">380<" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_contains_issue_date_format_102():
|
||||
invoice = _make_invoice(issue_date=date(2024, 3, 15))
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert "20240315" in xml
|
||||
assert 'format="102"' in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_contains_seller_and_buyer():
|
||||
xml, _ = build_cii_invoice_xml(
|
||||
_make_invoice(),
|
||||
_make_seller(name="ACME Corp"),
|
||||
_make_buyer(name="Widget Ltd"),
|
||||
)
|
||||
assert "ACME Corp" in xml
|
||||
assert "Widget Ltd" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_contains_tax_info():
|
||||
xml, _ = build_cii_invoice_xml(
|
||||
_make_invoice(tax_rate=Decimal("21"), tax_amount=Decimal("42.00")),
|
||||
_make_seller(),
|
||||
_make_buyer(),
|
||||
)
|
||||
assert "VAT" in xml
|
||||
assert "42.00" in xml
|
||||
assert "21.00" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_contains_monetary_totals():
|
||||
invoice = _make_invoice(
|
||||
subtotal=Decimal("200.00"),
|
||||
tax_amount=Decimal("42.00"),
|
||||
total_amount=Decimal("242.00"),
|
||||
)
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert "200.00" in xml
|
||||
assert "242.00" in xml
|
||||
assert "DuePayableAmount" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_contains_line_items():
|
||||
items = [
|
||||
SimpleNamespace(description="Design", quantity=Decimal("5"), unit_price=Decimal("80.00"), total_amount=Decimal("400.00")),
|
||||
SimpleNamespace(description="Dev", quantity=Decimal("10"), unit_price=Decimal("100.00"), total_amount=Decimal("1000.00")),
|
||||
]
|
||||
invoice = _make_invoice(items=items)
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert "Design" in xml
|
||||
assert "Dev" in xml
|
||||
assert "BilledQuantity" in xml
|
||||
assert 'unitCode="C62"' in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_adds_placeholder_line_when_no_items():
|
||||
invoice = _make_invoice(items=[], expenses=[], extra_goods=[])
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert "IncludedSupplyChainTradeLineItem" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_includes_buyer_reference():
|
||||
invoice = _make_invoice(buyer_reference="REF-42")
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert "BuyerReference" in xml
|
||||
assert "REF-42" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_includes_notes():
|
||||
invoice = _make_invoice(notes="Payment within 14 days")
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert "Payment within 14 days" in xml
|
||||
assert "IncludedNote" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_includes_due_date():
|
||||
invoice = _make_invoice(due_date=date(2024, 4, 14))
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
assert "20240414" in xml
|
||||
assert "DueDateDateTime" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_includes_seller_tax_registration():
|
||||
seller = _make_seller(tax_id="BE0123456789")
|
||||
xml, _ = build_cii_invoice_xml(_make_invoice(), seller, _make_buyer())
|
||||
assert "SpecifiedTaxRegistration" in xml
|
||||
assert "BE0123456789" in xml
|
||||
assert 'schemeID="VA"' in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_includes_seller_legal_organization():
|
||||
seller = _make_seller(endpoint_id="0088:123456", endpoint_scheme_id="0088")
|
||||
xml, _ = build_cii_invoice_xml(_make_invoice(), seller, _make_buyer())
|
||||
assert "SpecifiedLegalOrganization" in xml
|
||||
assert "0088:123456" in xml
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_sha256_changes_with_content():
|
||||
inv1 = _make_invoice(invoice_number="A")
|
||||
inv2 = _make_invoice(invoice_number="B")
|
||||
_, sha1 = build_cii_invoice_xml(inv1, _make_seller(), _make_buyer())
|
||||
_, sha2 = build_cii_invoice_xml(inv2, _make_seller(), _make_buyer())
|
||||
assert sha1 != sha2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cii_handles_zero_tax():
|
||||
invoice = _make_invoice(tax_rate=Decimal("0"), tax_amount=Decimal("0"))
|
||||
xml, _ = build_cii_invoice_xml(invoice, _make_seller(), _make_buyer())
|
||||
# Zero-rated tax should use category Z
|
||||
assert "CategoryCode" in xml
|
||||
# Should still produce valid CII
|
||||
passed, issues = validate_cii_en16931(xml)
|
||||
assert passed is True, f"Failed: {issues}"
|
||||
@@ -1,9 +1,16 @@
|
||||
"""Tests for invoice validators (UBL well-formed, veraPDF optional)."""
|
||||
"""Tests for invoice validators (UBL, CII, veraPDF)."""
|
||||
import pytest
|
||||
|
||||
from app.utils.invoice_validators import validate_ubl_wellformed
|
||||
from app.utils.invoice_validators import (
|
||||
validate_ubl_wellformed,
|
||||
validate_ubl_peppol_bis3,
|
||||
validate_cii_wellformed,
|
||||
validate_cii_en16931,
|
||||
)
|
||||
|
||||
|
||||
# ---- UBL well-formedness ----
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_ubl_wellformed_accepts_valid_invoice():
|
||||
ubl = '<?xml version="1.0"?><Invoice xmlns="urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"><ID>INV-001</ID></Invoice>'
|
||||
@@ -25,3 +32,242 @@ def test_validate_ubl_wellformed_rejects_non_invoice_root():
|
||||
passed, msgs = validate_ubl_wellformed(ubl)
|
||||
assert passed is False
|
||||
assert "Invoice" in msgs[0]
|
||||
|
||||
|
||||
# ---- UBL Peppol BIS 3.0 structural validation ----
|
||||
|
||||
def _minimal_peppol_ubl() -> str:
|
||||
return """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<Invoice xmlns="urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"
|
||||
xmlns:cac="urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"
|
||||
xmlns:cbc="urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2">
|
||||
<cbc:CustomizationID>urn:cen.eu:en16931:2017#compliant#urn:fdc:peppol.eu:2017:poacc:billing:3.0</cbc:CustomizationID>
|
||||
<cbc:ProfileID>urn:fdc:peppol.eu:2017:poacc:billing:01:1.0</cbc:ProfileID>
|
||||
<cbc:ID>INV-001</cbc:ID>
|
||||
<cbc:InvoiceTypeCode>380</cbc:InvoiceTypeCode>
|
||||
<cbc:IssueDate>2024-01-15</cbc:IssueDate>
|
||||
<cbc:DocumentCurrencyCode>EUR</cbc:DocumentCurrencyCode>
|
||||
<cbc:BuyerReference>PO-12345</cbc:BuyerReference>
|
||||
<cac:AccountingSupplierParty>
|
||||
<cac:Party>
|
||||
<cbc:EndpointID schemeID="9915">BE0123456789</cbc:EndpointID>
|
||||
<cac:PartyName><cbc:Name>Seller</cbc:Name></cac:PartyName>
|
||||
</cac:Party>
|
||||
</cac:AccountingSupplierParty>
|
||||
<cac:AccountingCustomerParty>
|
||||
<cac:Party>
|
||||
<cbc:EndpointID schemeID="0088">1234567890123</cbc:EndpointID>
|
||||
<cac:PartyName><cbc:Name>Buyer</cbc:Name></cac:PartyName>
|
||||
</cac:Party>
|
||||
</cac:AccountingCustomerParty>
|
||||
<cac:TaxTotal>
|
||||
<cbc:TaxAmount currencyID="EUR">0.00</cbc:TaxAmount>
|
||||
</cac:TaxTotal>
|
||||
<cac:LegalMonetaryTotal>
|
||||
<cbc:PayableAmount currencyID="EUR">100.00</cbc:PayableAmount>
|
||||
</cac:LegalMonetaryTotal>
|
||||
<cac:InvoiceLine>
|
||||
<cbc:ID>1</cbc:ID>
|
||||
<cbc:InvoicedQuantity unitCode="C62">1.00</cbc:InvoicedQuantity>
|
||||
<cbc:LineExtensionAmount currencyID="EUR">100.00</cbc:LineExtensionAmount>
|
||||
<cac:Item><cbc:Name>Service</cbc:Name></cac:Item>
|
||||
<cac:Price><cbc:PriceAmount currencyID="EUR">100.00</cbc:PriceAmount></cac:Price>
|
||||
</cac:InvoiceLine>
|
||||
</Invoice>"""
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_ubl_peppol_bis3_accepts_valid():
|
||||
passed, issues = validate_ubl_peppol_bis3(_minimal_peppol_ubl())
|
||||
assert passed is True, f"Unexpected issues: {issues}"
|
||||
assert issues == []
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_ubl_peppol_bis3_detects_missing_buyer_reference():
|
||||
ubl = _minimal_peppol_ubl().replace(
|
||||
"<cbc:BuyerReference>PO-12345</cbc:BuyerReference>", ""
|
||||
)
|
||||
passed, issues = validate_ubl_peppol_bis3(ubl)
|
||||
assert passed is False
|
||||
assert any("BuyerReference" in i for i in issues)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_ubl_peppol_bis3_detects_missing_endpoint():
|
||||
ubl = _minimal_peppol_ubl().replace(
|
||||
'<cbc:EndpointID schemeID="9915">BE0123456789</cbc:EndpointID>',
|
||||
"",
|
||||
1,
|
||||
)
|
||||
passed, issues = validate_ubl_peppol_bis3(ubl)
|
||||
assert passed is False
|
||||
assert any("EndpointID" in i for i in issues)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_ubl_peppol_bis3_detects_missing_lines():
|
||||
ubl = _minimal_peppol_ubl()
|
||||
# Remove InvoiceLine section
|
||||
start = ubl.index("<cac:InvoiceLine>")
|
||||
end = ubl.index("</cac:InvoiceLine>") + len("</cac:InvoiceLine>")
|
||||
ubl = ubl[:start] + ubl[end:]
|
||||
passed, issues = validate_ubl_peppol_bis3(ubl)
|
||||
assert passed is False
|
||||
assert any("InvoiceLine" in i for i in issues)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_ubl_peppol_bis3_detects_missing_unitcode():
|
||||
ubl = _minimal_peppol_ubl().replace('unitCode="C62"', "")
|
||||
passed, issues = validate_ubl_peppol_bis3(ubl)
|
||||
assert passed is False
|
||||
assert any("unitCode" in i for i in issues)
|
||||
|
||||
|
||||
# ---- CII well-formedness ----
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_cii_wellformed_accepts_valid():
|
||||
cii = '<?xml version="1.0"?><rsm:CrossIndustryInvoice xmlns:rsm="urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100"><x/></rsm:CrossIndustryInvoice>'
|
||||
passed, msgs = validate_cii_wellformed(cii)
|
||||
assert passed is True
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_cii_wellformed_rejects_ubl():
|
||||
ubl = '<?xml version="1.0"?><Invoice xmlns="urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"><ID>1</ID></Invoice>'
|
||||
passed, msgs = validate_cii_wellformed(ubl)
|
||||
assert passed is False
|
||||
assert "CrossIndustryInvoice" in msgs[0]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_cii_wellformed_rejects_invalid_xml():
|
||||
passed, msgs = validate_cii_wellformed("<not valid")
|
||||
assert passed is False
|
||||
|
||||
|
||||
# ---- CII EN 16931 structural validation ----
|
||||
|
||||
def _minimal_cii_en16931() -> str:
|
||||
return """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<rsm:CrossIndustryInvoice
|
||||
xmlns:rsm="urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100"
|
||||
xmlns:ram="urn:un:unece:uncefact:data:standard:ReusableAggregateBusinessInformationEntity:100"
|
||||
xmlns:udt="urn:un:unece:uncefact:data:standard:UnqualifiedDataType:100">
|
||||
<rsm:ExchangedDocumentContext>
|
||||
<ram:GuidelineSpecifiedDocumentContextParameter>
|
||||
<ram:ID>urn:cen.eu:en16931:2017#compliant#urn:factur-x.eu:1p0:en16931</ram:ID>
|
||||
</ram:GuidelineSpecifiedDocumentContextParameter>
|
||||
</rsm:ExchangedDocumentContext>
|
||||
<rsm:ExchangedDocument>
|
||||
<ram:ID>INV-001</ram:ID>
|
||||
<ram:TypeCode>380</ram:TypeCode>
|
||||
<ram:IssueDateTime>
|
||||
<udt:DateTimeString format="102">20240115</udt:DateTimeString>
|
||||
</ram:IssueDateTime>
|
||||
</rsm:ExchangedDocument>
|
||||
<rsm:SupplyChainTradeTransaction>
|
||||
<ram:ApplicableHeaderTradeAgreement>
|
||||
<ram:SellerTradeParty>
|
||||
<ram:Name>Seller Company</ram:Name>
|
||||
</ram:SellerTradeParty>
|
||||
<ram:BuyerTradeParty>
|
||||
<ram:Name>Buyer Company</ram:Name>
|
||||
</ram:BuyerTradeParty>
|
||||
</ram:ApplicableHeaderTradeAgreement>
|
||||
<ram:ApplicableHeaderTradeDelivery/>
|
||||
<ram:ApplicableHeaderTradeSettlement>
|
||||
<ram:InvoiceCurrencyCode>EUR</ram:InvoiceCurrencyCode>
|
||||
<ram:ApplicableTradeTax>
|
||||
<ram:CalculatedAmount>0.00</ram:CalculatedAmount>
|
||||
<ram:TypeCode>VAT</ram:TypeCode>
|
||||
<ram:BasisAmount>100.00</ram:BasisAmount>
|
||||
<ram:CategoryCode>Z</ram:CategoryCode>
|
||||
<ram:RateApplicablePercent>0.00</ram:RateApplicablePercent>
|
||||
</ram:ApplicableTradeTax>
|
||||
<ram:SpecifiedTradeSettlementHeaderMonetarySummation>
|
||||
<ram:LineTotalAmount>100.00</ram:LineTotalAmount>
|
||||
<ram:TaxBasisTotalAmount>100.00</ram:TaxBasisTotalAmount>
|
||||
<ram:TaxTotalAmount currencyID="EUR">0.00</ram:TaxTotalAmount>
|
||||
<ram:GrandTotalAmount>100.00</ram:GrandTotalAmount>
|
||||
<ram:DuePayableAmount>100.00</ram:DuePayableAmount>
|
||||
</ram:SpecifiedTradeSettlementHeaderMonetarySummation>
|
||||
</ram:ApplicableHeaderTradeSettlement>
|
||||
<ram:IncludedSupplyChainTradeLineItem>
|
||||
<ram:AssociatedDocumentLineDocument>
|
||||
<ram:LineID>1</ram:LineID>
|
||||
</ram:AssociatedDocumentLineDocument>
|
||||
<ram:SpecifiedTradeProduct>
|
||||
<ram:Name>Service</ram:Name>
|
||||
</ram:SpecifiedTradeProduct>
|
||||
<ram:SpecifiedLineTradeAgreement>
|
||||
<ram:NetPriceProductTradePrice>
|
||||
<ram:ChargeAmount>100.00</ram:ChargeAmount>
|
||||
</ram:NetPriceProductTradePrice>
|
||||
</ram:SpecifiedLineTradeAgreement>
|
||||
<ram:SpecifiedLineTradeDelivery>
|
||||
<ram:BilledQuantity unitCode="C62">1.00</ram:BilledQuantity>
|
||||
</ram:SpecifiedLineTradeDelivery>
|
||||
<ram:SpecifiedLineTradeSettlement>
|
||||
<ram:ApplicableTradeTax>
|
||||
<ram:TypeCode>VAT</ram:TypeCode>
|
||||
<ram:CategoryCode>Z</ram:CategoryCode>
|
||||
<ram:RateApplicablePercent>0.00</ram:RateApplicablePercent>
|
||||
</ram:ApplicableTradeTax>
|
||||
<ram:SpecifiedTradeSettlementLineMonetarySummation>
|
||||
<ram:LineTotalAmount>100.00</ram:LineTotalAmount>
|
||||
</ram:SpecifiedTradeSettlementLineMonetarySummation>
|
||||
</ram:SpecifiedLineTradeSettlement>
|
||||
</ram:IncludedSupplyChainTradeLineItem>
|
||||
</rsm:SupplyChainTradeTransaction>
|
||||
</rsm:CrossIndustryInvoice>"""
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_cii_en16931_accepts_valid():
|
||||
passed, issues = validate_cii_en16931(_minimal_cii_en16931())
|
||||
assert passed is True, f"Unexpected issues: {issues}"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_cii_en16931_detects_missing_seller():
|
||||
cii = _minimal_cii_en16931().replace(
|
||||
"<ram:SellerTradeParty>\n <ram:Name>Seller Company</ram:Name>\n </ram:SellerTradeParty>",
|
||||
"",
|
||||
)
|
||||
passed, issues = validate_cii_en16931(cii)
|
||||
assert passed is False
|
||||
assert any("Seller" in i for i in issues)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_cii_en16931_detects_missing_document_id():
|
||||
cii = _minimal_cii_en16931().replace("<ram:ID>INV-001</ram:ID>", "")
|
||||
passed, issues = validate_cii_en16931(cii)
|
||||
assert passed is False
|
||||
assert any("ID" in i or "Invoice number" in i for i in issues)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_cii_en16931_detects_missing_line_items():
|
||||
cii = _minimal_cii_en16931()
|
||||
start = cii.index("<ram:IncludedSupplyChainTradeLineItem>")
|
||||
end = cii.index("</ram:IncludedSupplyChainTradeLineItem>") + len(
|
||||
"</ram:IncludedSupplyChainTradeLineItem>"
|
||||
)
|
||||
cii = cii[:start] + cii[end:]
|
||||
passed, issues = validate_cii_en16931(cii)
|
||||
assert passed is False
|
||||
assert any("LineItem" in i or "line" in i.lower() for i in issues)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validate_cii_en16931_detects_missing_grand_total():
|
||||
cii = _minimal_cii_en16931().replace(
|
||||
"<ram:GrandTotalAmount>100.00</ram:GrandTotalAmount>", ""
|
||||
)
|
||||
passed, issues = validate_cii_en16931(cii)
|
||||
assert passed is False
|
||||
assert any("GrandTotal" in i for i in issues)
|
||||
|
||||
+106
-5
@@ -1,4 +1,4 @@
|
||||
"""Tests for PDF/A-3 conversion."""
|
||||
"""Tests for PDF/A-3 conversion with ICC profile embedding."""
|
||||
import io
|
||||
|
||||
import pytest
|
||||
@@ -12,6 +12,7 @@ except ImportError:
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.skipif(not pikepdf, reason="pikepdf not installed")
|
||||
def test_convert_to_pdfa3_adds_identification(app):
|
||||
"""PDF/A-3 conversion adds XMP pdfaid identification."""
|
||||
from app.utils.pdfa3 import convert_to_pdfa3
|
||||
|
||||
pdf = pikepdf.Pdf.new()
|
||||
@@ -25,14 +26,114 @@ def test_convert_to_pdfa3_adds_identification(app):
|
||||
assert err is None
|
||||
assert len(out_bytes) >= len(pdf_bytes)
|
||||
|
||||
# Open and check XMP contains pdfaid
|
||||
result = pikepdf.open(io.BytesIO(out_bytes))
|
||||
if result.Root.get("/Metadata"):
|
||||
xmp = result.Root.Metadata.read_bytes().decode("utf-8", errors="replace")
|
||||
assert "pdfaid:part" in xmp or "pdfa" in xmp.lower()
|
||||
assert result.Root.get("/Metadata") is not None
|
||||
xmp = result.Root.Metadata.read_bytes().decode("utf-8", errors="replace")
|
||||
assert "pdfaid:part" in xmp
|
||||
assert ">3<" in xmp or "3</pdfaid:part>" in xmp
|
||||
assert "pdfaid:conformance" in xmp
|
||||
result.close()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.skipif(not pikepdf, reason="pikepdf not installed")
|
||||
def test_convert_to_pdfa3_adds_output_intent(app):
|
||||
"""PDF/A-3 conversion adds an OutputIntent with ICC profile reference."""
|
||||
from app.utils.pdfa3 import convert_to_pdfa3
|
||||
|
||||
pdf = pikepdf.Pdf.new()
|
||||
pdf.add_blank_page(page_size=(595, 842))
|
||||
buf = io.BytesIO()
|
||||
pdf.save(buf)
|
||||
pdf.close()
|
||||
pdf_bytes = buf.getvalue()
|
||||
|
||||
out_bytes, err = convert_to_pdfa3(pdf_bytes)
|
||||
assert err is None
|
||||
|
||||
result = pikepdf.open(io.BytesIO(out_bytes))
|
||||
intents = result.Root.get("/OutputIntents")
|
||||
assert intents is not None
|
||||
assert len(intents) > 0
|
||||
intent = intents[0]
|
||||
assert str(intent.get("/S")) == "/GTS_PDFA1"
|
||||
# Check that DestOutputProfile (ICC stream) is present
|
||||
dest_profile = intent.get("/DestOutputProfile")
|
||||
assert dest_profile is not None, "OutputIntent should contain an embedded ICC profile"
|
||||
result.close()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.skipif(not pikepdf, reason="pikepdf not installed")
|
||||
def test_convert_to_pdfa3_icc_profile_is_valid(app):
|
||||
"""The embedded ICC profile has the correct signature and structure."""
|
||||
from app.utils.pdfa3 import convert_to_pdfa3
|
||||
import struct
|
||||
|
||||
pdf = pikepdf.Pdf.new()
|
||||
pdf.add_blank_page(page_size=(595, 842))
|
||||
buf = io.BytesIO()
|
||||
pdf.save(buf)
|
||||
pdf.close()
|
||||
pdf_bytes = buf.getvalue()
|
||||
|
||||
out_bytes, err = convert_to_pdfa3(pdf_bytes)
|
||||
assert err is None
|
||||
|
||||
result = pikepdf.open(io.BytesIO(out_bytes))
|
||||
intent = result.Root.OutputIntents[0]
|
||||
icc_stream = intent.DestOutputProfile
|
||||
icc_data = icc_stream.read_bytes()
|
||||
result.close()
|
||||
|
||||
# ICC profile must be at least 128 bytes (header size)
|
||||
assert len(icc_data) >= 128
|
||||
|
||||
# Profile size in header must match actual size
|
||||
profile_size = struct.unpack(">I", icc_data[:4])[0]
|
||||
assert profile_size == len(icc_data)
|
||||
|
||||
# Signature must be 'acsp' at offset 36
|
||||
assert icc_data[36:40] == b"acsp"
|
||||
|
||||
# Color space must be 'RGB '
|
||||
assert icc_data[16:20] == b"RGB "
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.skipif(not pikepdf, reason="pikepdf not installed")
|
||||
def test_convert_to_pdfa3_preserves_existing_xmp(app):
|
||||
"""Conversion preserves existing XMP metadata while adding PDF/A-3 identification."""
|
||||
from app.utils.pdfa3 import convert_to_pdfa3
|
||||
|
||||
pdf = pikepdf.Pdf.new()
|
||||
pdf.add_blank_page(page_size=(595, 842))
|
||||
existing_xmp = (
|
||||
'<?xpacket begin="\xef\xbb\xbf" id="W5M0MpCehiHzreSzNTczkc9d"?>'
|
||||
'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
|
||||
'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
|
||||
'<rdf:Description rdf:about="" xmlns:dc="http://purl.org/dc/elements/1.1/">'
|
||||
"<dc:title>Test Invoice</dc:title>"
|
||||
"</rdf:Description>"
|
||||
"</rdf:RDF></x:xmpmeta>"
|
||||
'<?xpacket end="w"?>'
|
||||
)
|
||||
pdf.Root.Metadata = pdf.make_stream(existing_xmp.encode("utf-8"))
|
||||
buf = io.BytesIO()
|
||||
pdf.save(buf)
|
||||
pdf.close()
|
||||
|
||||
out_bytes, err = convert_to_pdfa3(buf.getvalue())
|
||||
assert err is None
|
||||
|
||||
result = pikepdf.open(io.BytesIO(out_bytes))
|
||||
xmp = result.Root.Metadata.read_bytes().decode("utf-8", errors="replace")
|
||||
result.close()
|
||||
|
||||
assert "pdfaid:part" in xmp
|
||||
assert "dc:title" in xmp, "Existing XMP metadata should be preserved"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_convert_to_pdfa3_returns_error_on_invalid_pdf(app):
|
||||
from app.utils.pdfa3 import convert_to_pdfa3
|
||||
|
||||
@@ -141,6 +141,11 @@ def test_peppol_service_success_creates_transmission(app, monkeypatch):
|
||||
# EN 16931 requires unitCode on InvoicedQuantity (e.g. C62 = unit/each)
|
||||
assert "InvoicedQuantity" in tx.ubl_xml and 'unitCode="C62"' in tx.ubl_xml
|
||||
|
||||
# Validate UBL passes structural Peppol BIS 3.0 checks
|
||||
from app.utils.invoice_validators import validate_ubl_peppol_bis3
|
||||
passed, issues = validate_ubl_peppol_bis3(tx.ubl_xml)
|
||||
assert passed is True, f"UBL Peppol BIS 3.0 validation failed: {issues}"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_peppol_service_generic_transport_uses_identifier_validation(app, monkeypatch):
|
||||
@@ -159,3 +164,32 @@ def test_peppol_service_generic_transport_uses_identifier_validation(app, monkey
|
||||
)
|
||||
assert "required" in str(exc.value).lower() or "invalid" in str(exc.value).lower()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_as4_message_payload_is_gzip_compressed():
|
||||
"""AS4 message builder now gzip-compresses the payload to match the SOAP header declaration."""
|
||||
import gzip
|
||||
from app.integrations.peppol_as4 import build_as4_message
|
||||
|
||||
message_bytes = build_as4_message(
|
||||
ubl_xml="<Invoice><ID>1</ID></Invoice>",
|
||||
sender_endpoint_id="9915:BE111",
|
||||
sender_scheme_id="9915",
|
||||
recipient_endpoint_id="0088:123",
|
||||
recipient_scheme_id="0088",
|
||||
document_id="INV-1",
|
||||
)
|
||||
msg_str = message_bytes.decode("utf-8", errors="replace")
|
||||
# The SOAP header declares CompressionType=application/gzip
|
||||
assert "application/gzip" in msg_str
|
||||
|
||||
# Verify payload part uses application/gzip content type
|
||||
assert "application/gzip" in msg_str
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_native_transport_marked_experimental():
|
||||
"""Native transport module exposes NATIVE_TRANSPORT_EXPERIMENTAL flag."""
|
||||
from app.integrations.peppol_as4 import NATIVE_TRANSPORT_EXPERIMENTAL
|
||||
assert NATIVE_TRANSPORT_EXPERIMENTAL is True
|
||||
|
||||
|
||||
+90
-18
@@ -1,4 +1,4 @@
|
||||
"""Tests for ZugFerd/Factur-X: embedding EN 16931 UBL in invoice PDFs."""
|
||||
"""Tests for Factur-X / ZUGFeRD: embedding CII XML in invoice PDFs."""
|
||||
from datetime import date, timedelta
|
||||
from decimal import Decimal
|
||||
import io
|
||||
@@ -7,12 +7,12 @@ import pytest
|
||||
|
||||
from app import db
|
||||
from app.models import Client, Invoice, InvoiceItem, Project, User
|
||||
from app.utils.zugferd import ZUGFERD_EMBEDDED_FILENAME, embed_zugferd_xml_in_pdf
|
||||
from app.utils.zugferd import FACTURX_EMBEDDED_FILENAME, embed_zugferd_xml_in_pdf
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_embed_zugferd_xml_in_pdf_adds_attachment_and_xml_content(app):
|
||||
"""Embed step adds ZUGFeRD-invoice.xml to PDF and XML contains invoice data."""
|
||||
def test_embed_facturx_xml_in_pdf_adds_cii_attachment(app):
|
||||
"""Embed step adds factur-x.xml (CII) to PDF with correct structure."""
|
||||
try:
|
||||
import pikepdf
|
||||
except ImportError:
|
||||
@@ -27,6 +27,7 @@ def test_embed_zugferd_xml_in_pdf_adds_attachment_and_xml_content(app):
|
||||
client = Client(name="ZugFerd Client", email="client@example.com", address="Addr 1")
|
||||
client.set_custom_field("peppol_endpoint_id", "9915:DE123456789")
|
||||
client.set_custom_field("peppol_scheme_id", "9915")
|
||||
client.set_custom_field("peppol_country", "DE")
|
||||
db.session.add(client)
|
||||
db.session.commit()
|
||||
|
||||
@@ -49,22 +50,22 @@ def test_embed_zugferd_xml_in_pdf_adds_attachment_and_xml_content(app):
|
||||
due_date=date.today() + timedelta(days=30),
|
||||
created_by=user.id,
|
||||
currency_code="EUR",
|
||||
subtotal=Decimal("100.00"),
|
||||
tax_rate=Decimal("20.00"),
|
||||
tax_amount=Decimal("20.00"),
|
||||
total_amount=Decimal("120.00"),
|
||||
)
|
||||
db.session.add(inv)
|
||||
db.session.commit()
|
||||
|
||||
db.session.add(
|
||||
InvoiceItem(
|
||||
invoice_id=inv.id,
|
||||
description="Consulting",
|
||||
quantity=Decimal("1"),
|
||||
unit_price=Decimal("100.00"),
|
||||
total_amount=Decimal("100.00"),
|
||||
)
|
||||
)
|
||||
db.session.commit()
|
||||
inv.calculate_totals()
|
||||
db.session.commit()
|
||||
|
||||
settings = __import__("app.models", fromlist=["Settings"]).Settings.get_settings()
|
||||
if not getattr(settings, "company_name", None):
|
||||
@@ -87,27 +88,95 @@ def test_embed_zugferd_xml_in_pdf_adds_attachment_and_xml_content(app):
|
||||
assert err is None
|
||||
assert len(out_bytes) > len(pdf_bytes)
|
||||
|
||||
# Open result and check embedded file
|
||||
result = pikepdf.open(io.BytesIO(out_bytes))
|
||||
assert ZUGFERD_EMBEDDED_FILENAME in result.attachments
|
||||
attached = result.attachments[ZUGFERD_EMBEDDED_FILENAME].get_file()
|
||||
xml_content = attached.read().decode("utf-8")
|
||||
assert FACTURX_EMBEDDED_FILENAME in result.attachments
|
||||
attached = result.attachments[FACTURX_EMBEDDED_FILENAME].get_file()
|
||||
xml_content = attached.read_bytes().decode("utf-8")
|
||||
result.close()
|
||||
|
||||
assert "<Invoice" in xml_content or "Invoice" in xml_content
|
||||
# Must be CII (CrossIndustryInvoice), NOT UBL
|
||||
assert "CrossIndustryInvoice" in xml_content
|
||||
assert "INV-ZUG-001" in xml_content
|
||||
assert "120" in xml_content
|
||||
# EN 16931 requires unitCode on InvoicedQuantity (e.g. C62 = unit/each)
|
||||
assert "InvoicedQuantity" in xml_content and 'unitCode="C62"' in xml_content
|
||||
|
||||
# EN 16931 CII requires BilledQuantity with unitCode
|
||||
assert "BilledQuantity" in xml_content
|
||||
assert 'unitCode="C62"' in xml_content
|
||||
|
||||
# Grand total: 100 + 20% tax = 120
|
||||
assert "GrandTotalAmount" in xml_content
|
||||
assert "120.00" in xml_content
|
||||
|
||||
# Must contain seller and buyer party names
|
||||
assert "ZugFerd Client" in xml_content
|
||||
|
||||
# Must contain the Factur-X guideline ID
|
||||
assert "urn:cen.eu:en16931:2017" in xml_content
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_embed_zugferd_returns_original_pdf_on_embed_failure(app):
|
||||
def test_embed_facturx_xml_has_correct_xmp_metadata(app):
|
||||
"""Embedded PDF has Factur-X XMP metadata (not the old ZUGFeRD CII namespace)."""
|
||||
try:
|
||||
import pikepdf
|
||||
except ImportError:
|
||||
pytest.skip("pikepdf not installed")
|
||||
|
||||
with app.app_context():
|
||||
from types import SimpleNamespace
|
||||
|
||||
settings = __import__("app.models", fromlist=["Settings"]).Settings.get_settings()
|
||||
if not getattr(settings, "company_name", None):
|
||||
settings.company_name = "Test Company"
|
||||
db.session.commit()
|
||||
|
||||
inv = SimpleNamespace(
|
||||
id=1,
|
||||
invoice_number="INV-META-001",
|
||||
issue_date=date.today(),
|
||||
due_date=date.today() + timedelta(days=14),
|
||||
currency_code="EUR",
|
||||
subtotal=Decimal("50.00"),
|
||||
tax_rate=Decimal("0"),
|
||||
tax_amount=Decimal("0"),
|
||||
total_amount=Decimal("50.00"),
|
||||
notes=None,
|
||||
buyer_reference=None,
|
||||
project=None,
|
||||
client=None,
|
||||
client_name="Buyer",
|
||||
client_email=None,
|
||||
client_address=None,
|
||||
items=[SimpleNamespace(description="Work", quantity=1, unit_price=50, total_amount=50)],
|
||||
expenses=[],
|
||||
extra_goods=[],
|
||||
)
|
||||
|
||||
pdf = pikepdf.Pdf.new()
|
||||
pdf.add_blank_page(page_size=(595, 842))
|
||||
buf = io.BytesIO()
|
||||
pdf.save(buf)
|
||||
pdf.close()
|
||||
pdf_bytes = buf.getvalue()
|
||||
|
||||
out_bytes, err = embed_zugferd_xml_in_pdf(pdf_bytes, inv, settings)
|
||||
assert err is None
|
||||
|
||||
result = pikepdf.open(io.BytesIO(out_bytes))
|
||||
xmp = result.Root.Metadata.read_bytes().decode("utf-8", errors="replace")
|
||||
result.close()
|
||||
|
||||
# Must use Factur-X namespace, not the old ZUGFeRD CII namespace
|
||||
assert "factur-x" in xmp.lower() or "fx:DocumentType" in xmp
|
||||
assert "factur-x.xml" in xmp
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_embed_returns_original_pdf_on_failure(app):
|
||||
"""When embedding fails (e.g. invalid PDF), return original bytes and error message."""
|
||||
with app.app_context():
|
||||
from types import SimpleNamespace
|
||||
|
||||
settings = __import__("app.models", fromlist=["Settings"]).Settings.get_settings()
|
||||
# Minimal invoice-like object; build_peppol_ubl_invoice_xml only needs a few attrs
|
||||
inv = SimpleNamespace(
|
||||
id=1,
|
||||
invoice_number="INV-X",
|
||||
@@ -122,6 +191,9 @@ def test_embed_zugferd_returns_original_pdf_on_embed_failure(app):
|
||||
buyer_reference=None,
|
||||
project=None,
|
||||
client=None,
|
||||
client_name="Test",
|
||||
client_email=None,
|
||||
client_address=None,
|
||||
items=[],
|
||||
expenses=[],
|
||||
extra_goods=[],
|
||||
|
||||
Reference in New Issue
Block a user