AZ-Intec-GmbH/zugferd-service
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Allow zero-euro line item amounts

Browse Source
This commit is contained in:
m3ta-chiron
2026-06-03 07:52:52 +02:00
parent d497a4cb5b
commit e2e1d8c046
5 changed files with 71 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/extractor.py
+4 -4
View File
@@ -8,7 +8,7 @@ import io
import time
from facturx import get_flavor, get_level, get_xml_from_pdf
from lxml import etree
import lxml.etree as etree
from pypdf import PdfReader
from pypdf.errors import PdfReadError, PyPdfError
@@ -63,7 +63,7 @@ def extract_text_from_pdf(pdf_bytes: bytes) -> str:
except (PdfReadError, PyPdfError) as e:
raise ExtractionError(
error_code="corrupt_pdf", message="Failed to read PDF", details=str(e)
)
) from e
def parse_supplier(xml_root: etree._Element) -> Supplier:
@@ -235,8 +235,8 @@ def parse_line_items(xml_root: etree._Element) -> list[LineItem]:
description=description[0] if description else "",
quantity=float(quantity[0]) if quantity else 0.0,
unit=unit,
unit_price=float(unit_price[0]) if unit_price else 0.0,
line_total=float(line_total[0]) if line_total else 0.0,
unit_price=float(unit_price[0]) if unit_price else None,
line_total=float(line_total[0]) if line_total else None,
vat_rate=float(vat_rate[0]) if vat_rate else None,
vat_amount=float(vat_amount[0]) if vat_amount else None,
)
src/models.py
+3 -2
View File
@@ -1,6 +1,7 @@
"""Pydantic models for ZUGFeRD service."""
from typing import Any
from pydantic import BaseModel, Field
@@ -76,8 +77,8 @@ class LineItem(BaseModel):
description: str = Field(description="Item description")
quantity: float = Field(description="Quantity")
unit: str = Field(description="Unit (human-readable)")
unit_price: float = Field(description="Unit price")
line_total: float = Field(description="Line total amount")
unit_price: float | None = Field(default=None, description="Unit price")
line_total: float | None = Field(default=None, description="Line total amount")
vat_rate: float | None = Field(default=None, description="VAT rate percentage")
vat_amount: float | None = Field(
default=None, description="VAT amount for this line"
src/validator.py
+20 -9
View File
@@ -5,6 +5,7 @@ import time
from typing import Any
from pydantic import ValidationError
from src.models import (
ErrorDetail,
ValidateRequest,
@@ -78,10 +79,10 @@ def validate_pflichtfelder(xml_data: XmlData) -> list[ErrorDetail]:
if item.quantity == 0:
add_error(f"{field_prefix}.quantity", "critical")
if item.unit_price == 0:
if item.unit_price is None:
add_error(f"{field_prefix}.unit_price", "critical")
if item.line_total == 0:
if item.line_total is None:
add_error(f"{field_prefix}.line_total", "critical")
if item.vat_rate is None:
@@ -94,7 +95,7 @@ def validate_betraege(xml_data: XmlData) -> list[ErrorDetail]:
"""Check amount calculations are correct."""
errors = []
def add_mismatch(field: str, expected: float, actual: float) -> None:
def add_mismatch(field: str, expected: float, actual: float | None) -> None:
errors.append(
ErrorDetail(
check="betraege",
@@ -107,16 +108,24 @@ def validate_betraege(xml_data: XmlData) -> list[ErrorDetail]:
# Check line_total = quantity × unit_price
for idx, item in enumerate(xml_data.line_items):
expected_line_total = item.quantity * item.unit_price
if not amounts_match(item.line_total, expected_line_total):
unit_price = item.unit_price
line_total = item.line_total
if unit_price is None or line_total is None:
continue
expected_line_total = item.quantity * unit_price
if not amounts_match(line_total, expected_line_total):
add_mismatch(
f"line_items[{idx}].line_total",
expected_line_total,
item.line_total,
line_total,
)
# Check totals.net = sum(line_items.line_total)
line_total_sum = sum(item.line_total for item in xml_data.line_items)
line_total_sum = 0.0
for item in xml_data.line_items:
if item.line_total is not None:
line_total_sum += item.line_total
if not amounts_match(xml_data.totals.net, line_total_sum):
add_mismatch("totals.net", line_total_sum, xml_data.totals.net)
@@ -280,12 +289,14 @@ def validate_invoice(request: ValidateRequest) -> ValidationResult:
xml_data = XmlData(**request.xml_data)
except ValidationError as e:
# Convert Pydantic validation errors to ValidationResult
validation_errors = []
validation_errors: list[ErrorDetail] = []
for error in e.errors():
loc = error["loc"]
field = str(loc[0]) if loc else None
validation_errors.append(
ErrorDetail(
check="schema_validation",
field=error["loc"][0] if error["loc"] else None,
field=field,
error_code=error["type"],
message=error["msg"],
severity="critical",
tests/test_extractor.py
+12 -12
View File
@@ -4,8 +4,10 @@ Tests are written following TDD: FAILING TESTS FIRST (RED phase),
then implementation makes them pass (GREEN phase).
"""
import pytest
import base64
import binascii
import pytest
from pypdf import PdfReader, PdfWriter
@@ -49,7 +51,7 @@ class TestFileSizeValidation:
def test_file_size_limit_exactly_10mb(self):
"""Test PDF exactly at 10MB limit passes size check but fails PDF parsing."""
from src.extractor import extract_zugferd, ExtractionError
from src.extractor import ExtractionError, extract_zugferd
# 10MB = 10 * 1024 * 1024 bytes
large_pdf = b"X" * (10 * 1024 * 1024)
@@ -63,7 +65,7 @@ class TestFileSizeValidation:
def test_file_size_limit_10mb_plus_one_byte(self):
"""Test PDF one byte over 10MB limit is rejected."""
from src.extractor import extract_zugferd, ExtractionError
from src.extractor import ExtractionError, extract_zugferd
# 10MB + 1 byte
too_large = b"X" * (10 * 1024 * 1024 + 1)
@@ -75,7 +77,7 @@ class TestFileSizeValidation:
def test_file_size_under_10mb_accepted(self):
"""Test PDF under 10MB is accepted for processing."""
from src.extractor import extract_zugferd, ExtractionError
from src.extractor import ExtractionError, extract_zugferd
# Small PDF (9MB)
small_pdf = b"X" * (9 * 1024 * 1024)
@@ -166,8 +168,8 @@ class TestEN16931Extraction:
assert first_item.description is not None and len(first_item.description) > 0
assert first_item.quantity > 0
assert first_item.unit is not None and len(first_item.unit) > 0
assert first_item.unit_price > 0
assert first_item.line_total > 0
assert first_item.unit_price is not None and first_item.unit_price > 0
assert first_item.line_total is not None and first_item.line_total > 0
# Totals
assert xml_data.totals.line_total_sum > 0
@@ -181,7 +183,7 @@ class TestErrorHandling:
def test_corrupt_pdf_raises_error(self):
"""Test corrupt PDF raises ExtractionError with correct code."""
from src.extractor import extract_zugferd, ExtractionError
from src.extractor import ExtractionError, extract_zugferd
# Invalid PDF data
corrupt_pdf = b"NOT A PDF FILE AT ALL"
@@ -194,14 +196,14 @@ class TestErrorHandling:
def test_empty_pdf_raises_error(self):
"""Test empty PDF raises ExtractionError."""
from src.extractor import extract_zugferd, ExtractionError
from src.extractor import ExtractionError, extract_zugferd
with pytest.raises(ExtractionError):
extract_zugferd(b"")
def test_invalid_base64(self):
"""Test invalid base64 raises ExtractionError."""
from src.extractor import extract_zugferd, ExtractionError
from src.extractor import ExtractionError, extract_zugferd
# This would be called by API layer, but we can test the concept
# Invalid PDF that's not valid base64-encoded
@@ -210,7 +212,7 @@ class TestErrorHandling:
# If API layer decodes invalid base64, it gets error
decoded = base64.b64decode(invalid_base64, validate=True)
extract_zugferd(decoded)
except (base64.binascii.Error, ValueError):
except (binascii.Error, ValueError):
# base64 error is expected
pass
except ExtractionError as e:
@@ -232,8 +234,6 @@ class TestPDFTextExtraction:
assert result.pdf_text is not None
assert len(result.pdf_text) > 0
# Should contain some common German invoice terms
text_lower = result.pdf_text.lower()
# PDF text may contain invoice-related terms in German or English
tests/test_validator.py
+32 -9
View File
@@ -1,6 +1,5 @@
"""Tests for ZUGFeRD validator using TDD approach."""
import pytest
from src.models import (
XmlData,
Supplier,
@@ -9,8 +8,6 @@ from src.models import (
Totals,
VatBreakdown,
PaymentTerms,
ErrorDetail,
ValidationResult,
ValidateRequest,
)
from src.validator import (
@@ -352,8 +349,8 @@ class TestValidatePflichtfelder:
assert any(e.field == "line_items[0].quantity" for e in errors)
assert any(e.severity == "critical" for e in errors)
def test_missing_line_item_unit_price_critical(self):
"""Missing line_item.unit_price (zero) should produce critical error."""
def test_zero_euro_line_item_amounts_allowed(self):
"""Zero-valued line item prices should count as present."""
xml_data = XmlData(
invoice_number="INV001",
invoice_date="2024-01-15",
@@ -366,6 +363,29 @@ class TestValidatePflichtfelder:
quantity=1.0,
unit="Stück",
unit_price=0.0,
line_total=0.0,
)
],
totals=Totals(line_total_sum=10.0, net=10.0, vat_total=19.0, gross=29.0),
)
errors = validate_pflichtfelder(xml_data)
assert not any(e.field == "line_items[0].unit_price" for e in errors)
assert not any(e.field == "line_items[0].line_total" for e in errors)
def test_missing_line_item_unit_price_critical(self):
"""Missing line_item.unit_price should produce critical error."""
xml_data = XmlData(
invoice_number="INV001",
invoice_date="2024-01-15",
supplier=Supplier(name="Test Supplier GmbH", vat_id="DE123456789"),
buyer=Buyer(name="Test Buyer AG"),
line_items=[
LineItem(
position=1,
description="Test Product",
quantity=1.0,
unit="Stück",
unit_price=None,
line_total=10.0,
)
],
@@ -376,7 +396,7 @@ class TestValidatePflichtfelder:
assert any(e.severity == "critical" for e in errors)
def test_missing_line_item_line_total_critical(self):
"""Missing line_item.line_total (zero) should produce critical error."""
"""Missing line_item.line_total should produce critical error."""
xml_data = XmlData(
invoice_number="INV001",
invoice_date="2024-01-15",
@@ -389,7 +409,7 @@ class TestValidatePflichtfelder:
quantity=1.0,
unit="Stück",
unit_price=10.0,
line_total=0.0,
line_total=None,
)
],
totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0),
@@ -499,7 +519,7 @@ class TestValidateBetraege:
errors = validate_betraege(xml_data)
assert any(e.error_code == "calculation_mismatch" for e in errors)
assert any(
"line_total" in e.field
e.field is not None and "line_total" in e.field
for e in errors
if e.error_code == "calculation_mismatch"
)
@@ -597,7 +617,7 @@ class TestValidateBetraege:
),
)
errors = validate_betraege(xml_data)
assert any("vat_breakdown" in e.field for e in errors)
assert any(e.field is not None and "vat_breakdown" in e.field for e in errors)
assert any(e.error_code == "calculation_mismatch" for e in errors)
def test_totals_vat_total_mismatch(self):
@@ -1040,6 +1060,7 @@ class TestValidateInvoice:
result = validate_invoice(request)
assert result.is_valid is True
assert len(result.errors) == 0
assert result.summary is not None
assert result.summary["checks_passed"] == 2
def test_warnings_not_critical(self):
@@ -1134,6 +1155,7 @@ class TestValidateInvoice:
)
result = validate_invoice(request)
# Should have errors from PDF comparison
assert result.summary is not None
assert result.summary["total_checks"] == 1
def test_validation_time_populated(self):
@@ -1187,5 +1209,6 @@ class TestValidateInvoice:
)
result = validate_invoice(request)
# Should handle gracefully - no errors from invalid check
assert result.summary is not None
assert result.summary["total_checks"] == 0
assert result.is_valid is True # No critical errors