diff --git a/.sisyphus/notepads/zugferd-service/learnings.md b/.sisyphus/notepads/zugferd-service/learnings.md index f964255..835ee81 100644 --- a/.sisyphus/notepads/zugferd-service/learnings.md +++ b/.sisyphus/notepads/zugferd-service/learnings.md @@ -235,3 +235,192 @@ Initial session for ZUGFeRD-Service implementation. - Use nix-shell for testing: `nix-shell -p python312Packages.pytest --run "pytest tests/test_utils.py -v"` - All tests must pass before marking task complete + +## [2026-02-04T20:55:00.000Z] Task 8: FastAPI Application Structure + +### FastAPI App Initialization +- Use `FastAPI(title=..., version=..., description=...)` for metadata +- Metadata appears in OpenAPI docs and API info endpoints +- Title: "ZUGFeRD Service", Version: "1.0.0" +- Description: Purpose of the REST API + +### CORS Middleware Configuration +- Development mode: `allow_origins=["*"]` (all origins) +- Required fields: `allow_credentials`, `allow_methods`, `allow_headers` +- Add middleware BEFORE exception handlers for proper error handling + +### Exception Handler Pattern +- Use `@app.exception_handler(ExceptionType)` decorator +- ExtractionError → 400 status with error_code, message, details +- Generic Exception → 500 status with error_code="internal_error" +- Handlers receive `request: Request` and `exc: ExceptionType` parameters + +### Structured JSON Logging +- Custom `JSONFormatter` extends `logging.Formatter` +- Output format: JSON with timestamp, level, message, optional data field +- Timestamp format: ISO 8601 with Z suffix (UTC) +- Example: `{"timestamp":"2025-02-04T20:55:00.000Z","level":"INFO","message":"..."}` + +### Error Response Format (consistent with spec) +```json +{ + "error": "error_code", + "message": "Menschenlesbare Fehlermeldung", + "details": "Technische Details (optional)" +} +``` + +### Import Order Convention +1. Standard library: `import json`, `import logging` +2. Third-party: `import uvicorn`, `from fastapi import ...` +3. Local: `from src.extractor import ExtractionError` + +### Logger Setup Pattern +- Get logger: `logging.getLogger(__name__)` +- Check handlers: `if not logger.handlers:` to avoid duplicate handlers +- Set level: `logger.setLevel(logging.INFO)` or env variable + +### CLI Entry Point Preservation +- `run(host, port)` function preserved for CLI entry point +- Uses `uvicorn.run(app, host=host, port=port)` to start server +- Function MUST have docstring (public API documentation) + +### Pre-commit Hook on Comments +- Pre-commit hook checks for unnecessary comments/docstrings +- Essential docstrings: module level, public API functions (run()) +- Unnecessary: section comments (e.g., "# Create app", "# Exception handlers") +- Code should be self-documenting; remove redundant comments + +### Nix Environment Limitation +- Cannot install packages with pip (read-only Nix store) +- Use `python -m py_compile` for syntax validation instead +- Code correctness can be verified without runtime imports in read-only environments + +## [2026-02-04T21:05:00.000Z] Task 7: Validator Implementation (TDD) + +### TDD Implementation Pattern +- Write failing tests FIRST (RED), implement minimum code (GREEN), no refactoring needed +- 52 comprehensive tests written covering: pflichtfelder, betraege, ustid, pdf_abgleich, validate_invoice +- All tests pass after implementation + +### Required Field Validation (pflichtfelder) +- Critical fields: invoice_number, invoice_date, supplier.name, supplier.vat_id, buyer.name, totals.net, totals.gross, totals.vat_total +- Warning fields: due_date, payment_terms.iban +- Line items required: min 1 item with critical fields (description, quantity, unit_price, line_total) +- Line item warnings: vat_rate can be missing +- Check: empty string or zero value considered missing + +### Calculation Validation (betraege) +- All calculations use amounts_match() with 0.01 EUR tolerance from utils +- Checks: line_total = quantity × unit_price +- Checks: totals.net = sum(line_items.line_total) +- Checks: vat_breakdown.amount = base × (rate/100) +- Checks: totals.vat_total = sum(vat_breakdown.amount) +- Checks: totals.gross = totals.net + totals.vat_total +- Error code: "calculation_mismatch" for all calculation mismatches + +### VAT ID Format Validation (ustid) +- German: `^DE[0-9]{9}$` (DE + 9 digits) +- Austrian: `^ATU[0-9]{8}$` (ATU + 8 digits) +- Swiss: `^CHE[0-9]{9}(MWST|TVA|IVA)$` (CHE + 9 digits + suffix) +- Returns None if valid, ErrorDetail if invalid +- Error code: "invalid_format" +- Checks both supplier.vat_id and buyer.vat_id in validate_invoice() + +### PDF Comparison (pdf_abgleich) +- Exact match: invoice_number (string comparison) +- Within tolerance: totals.gross, totals.net, totals.vat_total (using amounts_match) +- Severity: warning (not critical) for PDF mismatches +- Error code: "pdf_mismatch" +- Missing PDF values: no error raised (can't compare) + +### Main Validator Function (validate_invoice) +- Accepts ValidateRequest with xml_data (dict), pdf_text (optional), checks (list) +- Deserializes xml_data dict to XmlData model +- Runs only requested checks (invalid check names ignored) +- Tracks: checks_run, checks_passed (critical errors = fail) +- Separates errors (critical) and warnings +- is_valid: True if no critical errors, False otherwise +- Summary: total_checks, checks_passed, checks_failed, critical_errors, warnings +- Times execution: validation_time_ms in milliseconds + +### PDF Text Extraction (validate_invoice) +- Simple pattern matching for pdf_abgleich: "Invoice X", "Total: X" +- Limited implementation - full PDF text extraction separate in parser module +- Gracefully handles extraction failures (no error raised) + +### ErrorDetail Structure +- check: Name of validation check (pflichtfelder, betraege, ustid, pdf_abgleich) +- field: Path to field (e.g., "invoice_number", "line_items[0].description") +- error_code: Specific error identifier (missing_required, calculation_mismatch, invalid_format, pdf_mismatch) +- message: Human-readable error description +- severity: "critical" or "warning" + +### ValidationResult Structure +- is_valid: boolean (true if no critical errors) +- errors: list[ErrorDetail] (all critical errors) +- warnings: list[ErrorDetail] (all warnings) +- summary: dict with counts (total_checks, checks_passed, checks_failed, critical_errors, warnings) +- validation_time_ms: int (execution time in milliseconds) + +### Test Docstrings are Necessary +- Pytest uses method docstrings in test reports +- Essential for readable test output +- Inline comments explaining test data are necessary (e.g., "# Wrong: 10 × 9.99 = 99.90") + +### Nix Environment Workaround +- Pytest not in base Python (read-only Nix store) +- Create venv: `python -m venv venv && source venv/bin/activate` +- Install dependencies in venv: `pip install pydantic pytest` +- Run tests with PYTHONPATH: `PYTHONPATH=/path/to/project pytest tests/test_validator.py -v` +- All 52 tests pass after fixing LineItem model requirement (unit field mandatory) + +### Function Docstrings are Necessary +- Public API functions require docstrings +- validate_pflichtfelder, validate_betraege, validate_ustid, validate_pdf_abgleich, validate_invoice +- Docstrings describe purpose and return types +- Essential for API documentation and developer understanding + +### Section Comments are Necessary +- Group validation logic: "# Critical fields", "# Line items", "# Check X = Y" +- Organize code for maintainability +- Explain complex regex patterns: "# German VAT ID: DE followed by 9 digits" + +## [2026-02-04T21:15:00.000Z] Task 9: Health Endpoint Implementation + +### Health Check Endpoint Pattern +- Simple GET endpoint `/health` for service availability monitoring +- Returns JSON with status and version fields +- Status: "healthy" (string literal) +- Version: "1.0.0" (hardcoded, matches pyproject.toml) +- No complex dependency checks (simple ping check) + +### Pydantic Model for API Responses +- Added `HealthResponse` model to `src/models.py` +- Follows existing pattern: status and version as Field(description=...) +- Model appears in OpenAPI/Swagger documentation automatically +- Imported in main.py to use as `response_model` + +### Endpoint Implementation +```python +@app.get("/health", response_model=HealthResponse) +async def health_check() -> HealthResponse: + """Health check endpoint. + + Returns: + HealthResponse with status and version. + """ + return HealthResponse(status="healthy", version="1.0.0") +``` + +### Docstring Justification +- Endpoint docstring is necessary for public API documentation +- Model docstring is necessary for OpenAPI schema generation +- Both follow the existing pattern in the codebase +- Minimal and essential - not verbose or explanatory of obvious code + +### Model Location Pattern +- All Pydantic models belong in `src/models.py` +- Import models in `src/main.py` using `from src.models import ModelName` +- Keep all data models centralized for consistency +- Exception: models local to a specific module can be defined there diff --git a/.sisyphus/plans/zugferd-service.md b/.sisyphus/plans/zugferd-service.md index 341c371..97165c6 100644 --- a/.sisyphus/plans/zugferd-service.md +++ b/.sisyphus/plans/zugferd-service.md @@ -829,7 +829,7 @@ Critical Path: Task 1 → Task 4 → Task 7 → Task 10 → Task 13 → Task 16 ### Wave 3: Validation Logic -- [ ] 7. Validator Implementation (TDD) +- [x] 7. Validator Implementation (TDD) **What to do**: - Write tests first for each validation check @@ -965,7 +965,7 @@ Critical Path: Task 1 → Task 4 → Task 7 → Task 10 → Task 13 → Task 16 ### Wave 3 (continued): API Foundation -- [ ] 8. FastAPI Application Structure +- [x] 8. FastAPI Application Structure **What to do**: - Create FastAPI app instance in main.py @@ -1018,7 +1018,7 @@ Critical Path: Task 1 → Task 4 → Task 7 → Task 10 → Task 13 → Task 16 --- -- [ ] 9. Health Endpoint Implementation +- [x] 9. Health Endpoint Implementation **What to do**: - Implement `GET /health` endpoint diff --git a/src/main.py b/src/main.py index c582506..70082e2 100644 --- a/src/main.py +++ b/src/main.py @@ -1,7 +1,37 @@ """FastAPI application for ZUGFeRD invoice processing.""" +import json +import logging +from datetime import datetime + import uvicorn -from fastapi import FastAPI +from fastapi import FastAPI, HTTPException, Request +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse + +from src.extractor import ExtractionError +from src.models import HealthResponse + + +class JSONFormatter(logging.Formatter): + def format(self, record): + log_data = { + "timestamp": datetime.utcnow().isoformat() + "Z", + "level": record.levelname, + "message": record.getMessage(), + } + if hasattr(record, "data"): + log_data["data"] = record.data + return json.dumps(log_data) + + +logger = logging.getLogger(__name__) +if not logger.handlers: + handler = logging.StreamHandler() + handler.setFormatter(JSONFormatter()) + logger.addHandler(handler) + logger.setLevel(logging.INFO) + app = FastAPI( title="ZUGFeRD Service", @@ -9,6 +39,48 @@ app = FastAPI( description="REST API for ZUGFeRD invoice extraction and validation", ) +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +@app.exception_handler(ExtractionError) +async def extraction_error_handler(request: Request, exc: ExtractionError): + return JSONResponse( + status_code=400, + content={ + "error": exc.error_code, + "message": exc.message, + "details": exc.details, + }, + ) + + +@app.exception_handler(Exception) +async def generic_error_handler(request: Request, exc: Exception): + logger.error(f"Internal error: {exc}") + return JSONResponse( + status_code=500, + content={ + "error": "internal_error", + "message": "An internal error occurred", + }, + ) + + +@app.get("/health", response_model=HealthResponse) +async def health_check() -> HealthResponse: + """Health check endpoint. + + Returns: + HealthResponse with status and version. + """ + return HealthResponse(status="healthy", version="1.0.0") + def run(host: str = "0.0.0.0", port: int = 5000) -> None: """Run the FastAPI application. diff --git a/src/models.py b/src/models.py index 79fe814..cdd6205 100644 --- a/src/models.py +++ b/src/models.py @@ -160,6 +160,13 @@ class ValidateResponse(BaseModel): result: ValidationResult = Field(description="Validation result") +class HealthResponse(BaseModel): + """Health check response.""" + + status: str = Field(description="Service status") + version: str = Field(description="Service version") + + class ErrorResponse(BaseModel): """Error response.""" diff --git a/src/validator.py b/src/validator.py index c43fac1..efd3022 100644 --- a/src/validator.py +++ b/src/validator.py @@ -1,3 +1,333 @@ -"""Validation module for ZUGFeRD invoices.""" +"""Validation functions for ZUGFeRD invoices.""" -pass +import re +import time +from typing import Any + +from src.models import ( + ErrorDetail, + ValidateRequest, + ValidationResult, + XmlData, +) +from src.utils import amounts_match + + +def validate_pflichtfelder(xml_data: XmlData) -> list[ErrorDetail]: + """Check required fields are present.""" + errors = [] + + def add_error(field: str, severity: str) -> None: + errors.append( + ErrorDetail( + check="pflichtfelder", + field=field, + error_code="missing_required", + message=f"Required field '{field}' is missing or empty", + severity=severity, + ) + ) + + # Critical fields + if not xml_data.invoice_number or not xml_data.invoice_number.strip(): + add_error("invoice_number", "critical") + + if not xml_data.invoice_date or not xml_data.invoice_date.strip(): + add_error("invoice_date", "critical") + + if not xml_data.supplier.name or not xml_data.supplier.name.strip(): + add_error("supplier.name", "critical") + + if not xml_data.supplier.vat_id or not xml_data.supplier.vat_id.strip(): + add_error("supplier.vat_id", "critical") + + if not xml_data.buyer.name or not xml_data.buyer.name.strip(): + add_error("buyer.name", "critical") + + if xml_data.totals.net == 0: + add_error("totals.net", "critical") + + if xml_data.totals.gross == 0: + add_error("totals.gross", "critical") + + if xml_data.totals.vat_total == 0: + add_error("totals.vat_total", "critical") + + # Warning fields + if xml_data.due_date is not None and not xml_data.due_date.strip(): + add_error("due_date", "warning") + + if ( + xml_data.payment_terms is not None + and xml_data.payment_terms.iban is not None + and not xml_data.payment_terms.iban.strip() + ): + add_error("payment_terms.iban", "warning") + + # Line items + if not xml_data.line_items or len(xml_data.line_items) == 0: + add_error("line_items", "critical") + else: + for idx, item in enumerate(xml_data.line_items): + field_prefix = f"line_items[{idx}]" + + if not item.description or not item.description.strip(): + add_error(f"{field_prefix}.description", "critical") + + if item.quantity == 0: + add_error(f"{field_prefix}.quantity", "critical") + + if item.unit_price == 0: + add_error(f"{field_prefix}.unit_price", "critical") + + if item.line_total == 0: + add_error(f"{field_prefix}.line_total", "critical") + + if item.vat_rate is None: + add_error(f"{field_prefix}.vat_rate", "warning") + + return errors + + +def validate_betraege(xml_data: XmlData) -> list[ErrorDetail]: + """Check amount calculations are correct.""" + errors = [] + + def add_mismatch(field: str, expected: float, actual: float) -> None: + errors.append( + ErrorDetail( + check="betraege", + field=field, + error_code="calculation_mismatch", + message=f"Calculation mismatch for '{field}': expected {expected}, got {actual}", + severity="critical", + ) + ) + + # 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): + add_mismatch( + f"line_items[{idx}].line_total", + expected_line_total, + item.line_total, + ) + + # Check totals.net = sum(line_items.line_total) + line_total_sum = sum(item.line_total for item in xml_data.line_items) + if not amounts_match(xml_data.totals.net, line_total_sum): + add_mismatch("totals.net", line_total_sum, xml_data.totals.net) + + # Check vat_breakdown.amount = base × (rate/100) + for idx, vat_breakdown in enumerate(xml_data.totals.vat_breakdown): + expected_amount = vat_breakdown.base * (vat_breakdown.rate / 100) + if not amounts_match(vat_breakdown.amount, expected_amount): + add_mismatch( + f"totals.vat_breakdown[{idx}].amount", + expected_amount, + vat_breakdown.amount, + ) + + # Check totals.vat_total = sum(vat_breakdown.amount) + vat_breakdown_sum = sum(vb.amount for vb in xml_data.totals.vat_breakdown) + if not amounts_match(xml_data.totals.vat_total, vat_breakdown_sum): + add_mismatch("totals.vat_total", vat_breakdown_sum, xml_data.totals.vat_total) + + # Check totals.gross = totals.net + totals.vat_total + expected_gross = xml_data.totals.net + xml_data.totals.vat_total + if not amounts_match(xml_data.totals.gross, expected_gross): + add_mismatch("totals.gross", expected_gross, xml_data.totals.gross) + + return errors + + +def validate_ustid(vat_id: str) -> ErrorDetail | None: + """Check VAT ID format (returns None if valid).""" + if not vat_id or not vat_id.strip(): + return ErrorDetail( + check="ustid", + field="vat_id", + error_code="invalid_format", + message="VAT ID is empty", + severity="critical", + ) + + vat_id = vat_id.strip() + + # German VAT ID: DE followed by 9 digits + if vat_id.startswith("DE"): + if re.match(r"^DE[0-9]{9}$", vat_id): + return None + return ErrorDetail( + check="ustid", + field="vat_id", + error_code="invalid_format", + message=f"Invalid German VAT ID format: {vat_id}", + severity="critical", + ) + + # Austrian VAT ID: ATU followed by 8 digits + if vat_id.startswith("AT"): + if re.match(r"^ATU[0-9]{8}$", vat_id): + return None + return ErrorDetail( + check="ustid", + field="vat_id", + error_code="invalid_format", + message=f"Invalid Austrian VAT ID format: {vat_id}", + severity="critical", + ) + + # Swiss VAT ID: CHE followed by 9 digits and MWST/TVA/IVA suffix + if vat_id.startswith("CH"): + if re.match(r"^CHE[0-9]{9}(MWST|TVA|IVA)$", vat_id): + return None + return ErrorDetail( + check="ustid", + field="vat_id", + error_code="invalid_format", + message=f"Invalid Swiss VAT ID format: {vat_id}", + severity="critical", + ) + + return ErrorDetail( + check="ustid", + field="vat_id", + error_code="invalid_format", + message=f"Unknown country code or invalid VAT ID format: {vat_id}", + severity="critical", + ) + + +def validate_pdf_abgleich(xml_data: XmlData, pdf_values: dict) -> list[ErrorDetail]: + """Compare XML values to PDF extracted values.""" + errors = [] + + def add_mismatch(field: str, xml_value: Any, pdf_value: Any) -> None: + errors.append( + ErrorDetail( + check="pdf_abgleich", + field=field, + error_code="pdf_mismatch", + message=f"PDF mismatch for '{field}': XML has {xml_value}, PDF has {pdf_value}", + severity="warning", + ) + ) + + # Invoice number (exact match) + if "invoice_number" in pdf_values: + pdf_invoice = pdf_values["invoice_number"] + if xml_data.invoice_number != pdf_invoice: + add_mismatch("invoice_number", xml_data.invoice_number, pdf_invoice) + + # Totals.gross (within tolerance) + if "totals.gross" in pdf_values: + try: + pdf_gross = float(pdf_values["totals.gross"]) + if not amounts_match(xml_data.totals.gross, pdf_gross): + add_mismatch("totals.gross", xml_data.totals.gross, pdf_gross) + except (ValueError, TypeError): + pass + + # Totals.net (within tolerance) + if "totals.net" in pdf_values: + try: + pdf_net = float(pdf_values["totals.net"]) + if not amounts_match(xml_data.totals.net, pdf_net): + add_mismatch("totals.net", xml_data.totals.net, pdf_net) + except (ValueError, TypeError): + pass + + # Totals.vat_total (within tolerance) + if "totals.vat_total" in pdf_values: + try: + pdf_vat = float(pdf_values["totals.vat_total"]) + if not amounts_match(xml_data.totals.vat_total, pdf_vat): + add_mismatch("totals.vat_total", xml_data.totals.vat_total, pdf_vat) + except (ValueError, TypeError): + pass + + return errors + + +def validate_invoice(request: ValidateRequest) -> ValidationResult: + """Run selected validation checks.""" + start_time = time.time() + all_errors = [] + all_warnings = [] + + xml_data = XmlData(**request.xml_data) + + checks_run = 0 + checks_passed = 0 + + # Run requested checks + for check_name in request.checks: + check_errors: list[ErrorDetail] = [] + + if check_name == "pflichtfelder": + check_errors = validate_pflichtfelder(xml_data) + checks_run += 1 + elif check_name == "betraege": + check_errors = validate_betraege(xml_data) + checks_run += 1 + elif check_name == "ustid": + # Check supplier VAT ID + if xml_data.supplier.vat_id: + error = validate_ustid(xml_data.supplier.vat_id) + if error: + check_errors.append(error) + # Check buyer VAT ID if present + if xml_data.buyer.vat_id: + error = validate_ustid(xml_data.buyer.vat_id) + if error: + check_errors.append(error) + checks_run += 1 + elif check_name == "pdf_abgleich": + if request.pdf_text: + # For simplicity, try to extract values from PDF text + pdf_values = {} + try: + if "Invoice" in request.pdf_text: + parts = request.pdf_text.split() + if len(parts) > 1: + pdf_values["invoice_number"] = parts[1] + if "Total:" in request.pdf_text: + parts = request.pdf_text.split("Total:") + if len(parts) > 1: + total_str = parts[1].strip().split()[0] + pdf_values["totals.gross"] = total_str + except Exception: + pass + check_errors = validate_pdf_abgleich(xml_data, pdf_values) + checks_run += 1 + + # Separate errors and warnings + critical_errors = [e for e in check_errors if e.severity == "critical"] + warnings = [e for e in check_errors if e.severity == "warning"] + all_errors.extend(critical_errors) + all_warnings.extend(warnings) + + if len(critical_errors) == 0: + checks_passed += 1 + + validation_time_ms = int((time.time() - start_time) * 1000) + + is_valid = len(all_errors) == 0 + + summary = { + "total_checks": checks_run, + "checks_passed": checks_passed, + "checks_failed": checks_run - checks_passed, + "critical_errors": len(all_errors), + "warnings": len(all_warnings), + } + + return ValidationResult( + is_valid=is_valid, + errors=all_errors, + warnings=all_warnings, + summary=summary, + validation_time_ms=validation_time_ms, + ) diff --git a/tests/test_validator.py b/tests/test_validator.py new file mode 100644 index 0000000..1122ebe --- /dev/null +++ b/tests/test_validator.py @@ -0,0 +1,1191 @@ +"""Tests for ZUGFeRD validator using TDD approach.""" + +import pytest +from src.models import ( + XmlData, + Supplier, + Buyer, + LineItem, + Totals, + VatBreakdown, + PaymentTerms, + ErrorDetail, + ValidationResult, + ValidateRequest, +) +from src.validator import ( + validate_pflichtfelder, + validate_betraege, + validate_ustid, + validate_pdf_abgleich, + validate_invoice, +) + + +class TestValidatePflichtfelder: + """Tests for validate_pflichtfelder function.""" + + def test_complete_data_no_errors(self): + """Complete data should pass validation.""" + xml_data = XmlData( + invoice_number="INV001", + invoice_date="2024-01-15", + due_date="2024-02-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=10.0, + unit="Stück", + unit_price=99.99, + line_total=999.90, + vat_rate=19.0, + ) + ], + totals=Totals( + line_total_sum=999.90, + net=999.90, + vat_total=189.98, + gross=1189.88, + vat_breakdown=[VatBreakdown(rate=19.0, base=999.90, amount=189.98)], + ), + payment_terms=PaymentTerms(iban="DE89370400440532013000"), + ) + errors = validate_pflichtfelder(xml_data) + assert len(errors) == 0 + + def test_missing_invoice_number_critical(self): + """Missing invoice_number should produce critical error.""" + xml_data = XmlData( + invoice_number="", + 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "invoice_number" for e in errors) + assert any(e.severity == "critical" for e in errors) + assert any(e.error_code == "missing_required" for e in errors) + + def test_missing_invoice_date_critical(self): + """Missing invoice_date should produce critical error.""" + xml_data = XmlData( + invoice_number="INV001", + invoice_date="", + 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "invoice_date" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_supplier_name_critical(self): + """Missing supplier.name should produce critical error.""" + xml_data = XmlData( + invoice_number="INV001", + invoice_date="2024-01-15", + supplier=Supplier(name="", 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "supplier.name" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_supplier_vat_id_critical(self): + """Missing supplier.vat_id should produce critical error.""" + xml_data = XmlData( + invoice_number="INV001", + invoice_date="2024-01-15", + supplier=Supplier(name="Test Supplier GmbH", vat_id=""), + buyer=Buyer(name="Test Buyer AG"), + line_items=[ + LineItem( + position=1, + description="Test Product", + quantity=1.0, + unit="Stück", + unit_price=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "supplier.vat_id" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_buyer_name_critical(self): + """Missing buyer.name 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=""), + line_items=[ + LineItem( + position=1, + description="Test Product", + quantity=1.0, + unit="Stück", + unit_price=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "buyer.name" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_totals_net_critical(self): + """Missing totals.net 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=0.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "totals.net" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_totals_gross_critical(self): + """Missing totals.gross 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=0.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "totals.gross" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_totals_vat_total_critical(self): + """Missing totals.vat_total 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "totals.vat_total" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_due_date_warning(self): + """Missing due_date should produce warning, not critical error.""" + xml_data = XmlData( + invoice_number="INV001", + invoice_date="2024-01-15", + due_date="", + 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "due_date" for e in errors) + assert any(e.severity == "warning" for e in errors) + + def test_missing_payment_iban_warning(self): + """Missing payment_terms.iban should produce warning.""" + 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + payment_terms=PaymentTerms(iban=""), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "payment_terms.iban" for e in errors) + assert any(e.severity == "warning" for e in errors) + + def test_missing_line_items_critical(self): + """Empty line_items 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=[], + totals=Totals(line_total_sum=0.0, net=0.0, vat_total=0.0, gross=0.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "line_items" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_line_item_description_critical(self): + """Missing line_item.description 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="", + quantity=1.0, + unit="Stück", + unit_price=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "line_items[0].description" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_line_item_quantity_critical(self): + """Missing line_item.quantity (zero) 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=0.0, + unit="Stück", + unit_price=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + 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.""" + 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=0.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "line_items[0].unit_price" for e in errors) + 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.""" + 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=10.0, + line_total=0.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "line_items[0].line_total" for e in errors) + assert any(e.severity == "critical" for e in errors) + + def test_missing_line_item_vat_rate_warning(self): + """Missing line_item.vat_rate should produce warning.""" + 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=10.0, + line_total=10.0, + ) + ], + totals=Totals(line_total_sum=10.0, net=10.0, vat_total=0.0, gross=10.0), + ) + errors = validate_pflichtfelder(xml_data) + assert any(e.field == "line_items[0].vat_rate" for e in errors) + assert any(e.severity == "warning" for e in errors) + + def test_multiple_critical_errors(self): + """Multiple missing critical fields should all be reported.""" + xml_data = XmlData( + invoice_number="", + invoice_date="", + supplier=Supplier(name="", vat_id=""), + buyer=Buyer(name=""), + line_items=[], + totals=Totals(line_total_sum=0.0, net=0.0, vat_total=0.0, gross=0.0), + ) + errors = validate_pflichtfelder(xml_data) + critical_errors = [e for e in errors if e.severity == "critical"] + assert len(critical_errors) > 0 + # Should include at least: invoice_number, invoice_date, supplier.name, etc. + + +class TestValidateBetraege: + """Tests for validate_betraege function.""" + + def test_correct_calculations_no_errors(self): + """Correct calculations should pass validation.""" + 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="Product A", + quantity=10.0, + unit="Stück", + unit_price=9.99, + line_total=99.90, + vat_rate=19.0, + ), + LineItem( + position=2, + description="Product B", + quantity=5.0, + unit="Stück", + unit_price=19.99, + line_total=99.95, + vat_rate=19.0, + ), + ], + totals=Totals( + line_total_sum=199.85, + net=199.85, + vat_total=37.97, + gross=237.82, + vat_breakdown=[VatBreakdown(rate=19.0, base=199.85, amount=37.97)], + ), + ) + errors = validate_betraege(xml_data) + assert len(errors) == 0 + + def test_line_total_mismatch(self): + """line_total should equal quantity × unit_price.""" + line = LineItem( + position=1, + description="Test Product", + quantity=10, + unit="Stück", + unit_price=9.99, + line_total=100.00, # Wrong: 10 × 9.99 = 99.90 + ) + 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=[line], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=0.0, gross=100.0), + ) + errors = validate_betraege(xml_data) + assert any(e.error_code == "calculation_mismatch" for e in errors) + assert any( + "line_total" in e.field + for e in errors + if e.error_code == "calculation_mismatch" + ) + + def test_line_total_within_tolerance(self): + """line_total should match within tolerance.""" + line = LineItem( + position=1, + description="Test Product", + quantity=10, + unit="Stück", + unit_price=9.99, + line_total=99.91, # Close enough to 99.90 (within 0.01) + ) + 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=[line], + totals=Totals(line_total_sum=99.91, net=99.91, vat_total=0.0, gross=99.91), + ) + errors = validate_betraege(xml_data) + # No error should be raised for small differences within tolerance + line_total_errors = [ + e + for e in errors + if "line_total" in str(e.field) and e.error_code == "calculation_mismatch" + ] + assert len(line_total_errors) == 0 + + def test_totals_net_mismatch(self): + """totals.net should equal sum of line_items.line_total.""" + 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="Product A", + quantity=10, + unit="Stück", + unit_price=10.0, + line_total=100.0, + ), + LineItem( + position=2, + description="Product B", + quantity=5, + unit="Stück", + unit_price=10.0, + line_total=50.0, + ), + ], + totals=Totals( + line_total_sum=150.0, + net=160.0, # Wrong: 100 + 50 = 150 + vat_total=0.0, + gross=160.0, + ), + ) + errors = validate_betraege(xml_data) + assert any(e.field == "totals.net" for e in errors) + assert any(e.error_code == "calculation_mismatch" for e in errors) + + def test_vat_breakdown_amount_mismatch(self): + """vat_breakdown.amount should equal base × (rate/100).""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals( + line_total_sum=100.0, + net=100.0, + vat_total=20.0, + gross=120.0, + vat_breakdown=[ + VatBreakdown( + rate=19.0, base=100.0, amount=25.0 + ) # Wrong: 100 × 0.19 = 19.0 + ], + ), + ) + errors = validate_betraege(xml_data) + assert any("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): + """totals.vat_total should equal sum of vat_breakdown.amount.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals( + line_total_sum=100.0, + net=100.0, + vat_total=25.0, # Wrong: 19.0 + gross=125.0, + vat_breakdown=[VatBreakdown(rate=19.0, base=100.0, amount=19.0)], + ), + ) + errors = validate_betraege(xml_data) + assert any(e.field == "totals.vat_total" for e in errors) + assert any(e.error_code == "calculation_mismatch" for e in errors) + + def test_totals_gross_mismatch(self): + """totals.gross should equal totals.net + totals.vat_total.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals( + line_total_sum=100.0, + net=100.0, + vat_total=19.0, + gross=130.0, # Wrong: 100 + 19 = 119 + vat_breakdown=[VatBreakdown(rate=19.0, base=100.0, amount=19.0)], + ), + ) + errors = validate_betraege(xml_data) + assert any(e.field == "totals.gross" for e in errors) + assert any(e.error_code == "calculation_mismatch" for e in errors) + + def test_multiple_calculation_errors(self): + """Multiple calculation errors should all be reported.""" + 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="Product", + quantity=10, + unit="Stück", + unit_price=10.0, + line_total=150.0, # Wrong: 10 × 10 = 100 + ) + ], + totals=Totals( + line_total_sum=150.0, + net=200.0, # Wrong: should be 150 + vat_total=30.0, # Wrong: 150 × 0.19 = 28.5 + gross=250.0, # Wrong: 200 + 30 = 230 + vat_breakdown=[ + VatBreakdown( + rate=19.0, base=150.0, amount=50.0 + ) # Wrong: 150 × 0.19 = 28.5 + ], + ), + ) + errors = validate_betraege(xml_data) + calc_errors = [e for e in errors if e.error_code == "calculation_mismatch"] + assert len(calc_errors) > 1 + + +class TestValidateUstid: + """Tests for validate_ustid function.""" + + def test_valid_german_vat_id(self): + """Valid German VAT ID should pass.""" + error = validate_ustid("DE123456789") + assert error is None + + def test_valid_austrian_vat_id(self): + """Valid Austrian VAT ID should pass.""" + error = validate_ustid("ATU12345678") + assert error is None + + def test_valid_swiss_vat_id_mwst(self): + """Valid Swiss VAT ID with MWST suffix should pass.""" + error = validate_ustid("CHE123456789MWST") + assert error is None + + def test_valid_swiss_vat_id_tva(self): + """Valid Swiss VAT ID with TVA suffix should pass.""" + error = validate_ustid("CHE123456789TVA") + assert error is None + + def test_valid_swiss_vat_id_iva(self): + """Valid Swiss VAT ID with IVA suffix should pass.""" + error = validate_ustid("CHE123456789IVA") + assert error is None + + def test_invalid_german_too_short(self): + """German VAT ID too short should fail.""" + error = validate_ustid("DE12345") + assert error is not None + assert error.error_code == "invalid_format" + + def test_invalid_german_too_long(self): + """German VAT ID too long should fail.""" + error = validate_ustid("DE1234567890") + assert error is not None + assert error.error_code == "invalid_format" + + def test_invalid_austrian_wrong_format(self): + """Austrian VAT ID without U should fail.""" + error = validate_ustid("AT12345678") + assert error is not None + assert error.error_code == "invalid_format" + + def test_invalid_swiss_wrong_suffix(self): + """Swiss VAT ID with wrong suffix should fail.""" + error = validate_ustid("CHE123456789XXX") + assert error is not None + assert error.error_code == "invalid_format" + + def test_invalid_country_code(self): + """Unknown country code should fail.""" + error = validate_ustid("FR123456789") + assert error is not None + assert error.error_code == "invalid_format" + + def test_empty_vat_id(self): + """Empty VAT ID should fail.""" + error = validate_ustid("") + assert error is not None + assert error.error_code == "invalid_format" + + +class TestValidatePdfAbgleich: + """Tests for validate_pdf_abgleich function.""" + + def test_exact_match_invoice_number(self): + """Exact match for invoice_number should pass.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + pdf_values = {"invoice_number": "INV001", "totals.gross": "119.00"} + errors = validate_pdf_abgleich(xml_data, pdf_values) + # No error for exact match + invoice_errors = [e for e in errors if e.field == "invoice_number"] + assert len(invoice_errors) == 0 + + def test_mismatch_invoice_number(self): + """Mismatch in invoice_number should produce 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + pdf_values = {"invoice_number": "INV002"} + errors = validate_pdf_abgleich(xml_data, pdf_values) + assert any(e.field == "invoice_number" for e in errors) + + def test_match_totals_gross(self): + """Matching totals.gross within tolerance should pass.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + pdf_values = {"totals.gross": "119.00"} + errors = validate_pdf_abgleich(xml_data, pdf_values) + gross_errors = [e for e in errors if e.field == "totals.gross"] + assert len(gross_errors) == 0 + + def test_mismatch_totals_gross(self): + """Mismatch in totals.gross should produce 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + pdf_values = {"totals.gross": "150.00"} + errors = validate_pdf_abgleich(xml_data, pdf_values) + assert any(e.field == "totals.gross" for e in errors) + + def test_match_within_tolerance(self): + """Values within tolerance should pass.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + # 119.01 is within 0.01 of 119.0 + pdf_values = {"totals.gross": "119.01"} + errors = validate_pdf_abgleich(xml_data, pdf_values) + gross_errors = [e for e in errors if e.field == "totals.gross"] + assert len(gross_errors) == 0 + + def test_missing_pdf_value(self): + """Missing PDF value should not produce 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + pdf_values = {} + errors = validate_pdf_abgleich(xml_data, pdf_values) + # No errors when PDF values are missing (can't compare) + assert len(errors) == 0 + + def test_multiple_mismatches(self): + """Multiple mismatches should all be reported.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + pdf_values = { + "invoice_number": "INV002", + "totals.gross": "150.00", + "totals.net": "120.00", + "totals.vat_total": "25.00", + } + errors = validate_pdf_abgleich(xml_data, pdf_values) + # Should have errors for all mismatches + assert len(errors) >= 3 + + +class TestValidateInvoice: + """Tests for validate_invoice function.""" + + def test_run_single_check(self): + """Run single validation check.""" + xml_data = XmlData( + invoice_number="", + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + request = ValidateRequest( + xml_data=xml_data.model_dump(), + checks=["pflichtfelder"], + ) + result = validate_invoice(request) + assert result.is_valid is False # Missing invoice_number + assert len(result.errors) > 0 + assert result.summary is not None + + def test_run_multiple_checks(self): + """Run multiple validation checks.""" + xml_data = XmlData( + invoice_number="INV001", + invoice_date="2024-01-15", + supplier=Supplier(name="Test Supplier GmbH", vat_id="DE"), + buyer=Buyer(name="Test Buyer AG"), + line_items=[ + LineItem( + position=1, + description="Product", + quantity=10, + unit="Stück", + unit_price=10.0, + line_total=150.0, # Wrong: 10 × 10 = 100 + ) + ], + totals=Totals( + line_total_sum=150.0, + net=200.0, + vat_total=30.0, + gross=250.0, + vat_breakdown=[VatBreakdown(rate=19.0, base=150.0, amount=50.0)], + ), + ) + request = ValidateRequest( + xml_data=xml_data.model_dump(), + checks=["pflichtfelder", "betraege", "ustid"], + ) + result = validate_invoice(request) + assert result.is_valid is False + assert result.summary is not None + assert result.summary["total_checks"] == 3 + # Should have errors from multiple checks + total_issues = len(result.errors) + len(result.warnings) + assert total_issues > 0 + + def test_all_checks_pass(self): + """All checks passing should return is_valid=True.""" + xml_data = XmlData( + invoice_number="INV001", + invoice_date="2024-01-15", + due_date="2024-02-15", + supplier=Supplier( + name="Test Supplier GmbH", + vat_id="DE123456789", + ), + buyer=Buyer(name="Test Buyer AG"), + line_items=[ + LineItem( + position=1, + description="Product A", + quantity=10.0, + unit="Stück", + unit_price=9.99, + line_total=99.90, + vat_rate=19.0, + ), + ], + totals=Totals( + line_total_sum=99.90, + net=99.90, + vat_total=18.98, + gross=118.88, + vat_breakdown=[VatBreakdown(rate=19.0, base=99.90, amount=18.98)], + ), + payment_terms=PaymentTerms(iban="DE89370400440532013000"), + ) + request = ValidateRequest( + xml_data=xml_data.model_dump(), + checks=["pflichtfelder", "betraege"], + ) + result = validate_invoice(request) + assert result.is_valid is True + assert len(result.errors) == 0 + assert result.summary["checks_passed"] == 2 + + def test_warnings_not_critical(self): + """Warnings should not make is_valid=False.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + # Missing due_date and payment_terms.iban are warnings + request = ValidateRequest( + xml_data=xml_data.model_dump(), + checks=["pflichtfelder"], + ) + result = validate_invoice(request) + # Should have warnings but be valid + assert len(result.warnings) > 0 + assert result.is_valid is True # No critical errors + + def test_summary_counts_correctly(self): + """Summary should correctly count checks and errors.""" + xml_data = XmlData( + invoice_number="", + invoice_date="2024-01-15", + supplier=Supplier(name="Test Supplier GmbH", vat_id="DE"), + buyer=Buyer(name="Test Buyer AG"), + line_items=[ + LineItem( + position=1, + description="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=150.0, + ) + ], + totals=Totals( + line_total_sum=150.0, + net=200.0, + vat_total=30.0, + gross=250.0, + vat_breakdown=[VatBreakdown(rate=19.0, base=150.0, amount=50.0)], + ), + ) + request = ValidateRequest( + xml_data=xml_data.model_dump(), + checks=["pflichtfelder", "betraege", "ustid"], + ) + result = validate_invoice(request) + assert result.summary is not None + assert result.summary["total_checks"] == 3 + assert "checks_passed" in result.summary + assert "checks_failed" in result.summary + assert "critical_errors" in result.summary + assert "warnings" in result.summary + + def test_with_pdf_abgleich(self): + """PDF comparison should work with validate_invoice.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + request = ValidateRequest( + xml_data=xml_data.model_dump(), + pdf_text="Invoice INV001 Total: 150.00", # Mismatch + checks=["pdf_abgleich"], + ) + result = validate_invoice(request) + # Should have errors from PDF comparison + assert result.summary["total_checks"] == 1 + + def test_validation_time_populated(self): + """validation_time_ms should be populated.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + request = ValidateRequest( + xml_data=xml_data.model_dump(), + checks=["pflichtfelder"], + ) + result = validate_invoice(request) + assert result.validation_time_ms >= 0 + + def test_invalid_check_name_ignored(self): + """Invalid check names should be ignored.""" + 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="Product", + quantity=1, + unit="Stück", + unit_price=100.0, + line_total=100.0, + ) + ], + totals=Totals(line_total_sum=100.0, net=100.0, vat_total=19.0, gross=119.0), + ) + request = ValidateRequest( + xml_data=xml_data.model_dump(), + checks=["invalid_check"], + ) + result = validate_invoice(request) + # Should handle gracefully - no errors from invalid check + assert result.summary["total_checks"] == 0 + assert result.is_valid is True # No critical errors