diff --git a/README.md b/README.md index d5c7fbd..ab3e15b 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ ## EDocument Integration -Integration app for sending and receiving PEPPOL e-documents via service providers (B2B Router, Recommand). +Integration app for sending and receiving PEPPOL e-documents via service providers (B2B Router, Recommand, Peppyrus). This app extends the [edocument](https://github.com/prilk-consulting/edocument) app by providing integration capabilities with PEPPOL service providers for: - **Sending e-documents**: Transmit PEPPOL invoices and credit notes via API to service providers @@ -52,12 +52,26 @@ Configure integration with PEPPOL service providers using the **EDocument Integr - **EDocument Integrator**: Choose your service provider: - **B2B Router**: For B2B Router integration - **Recommand**: For Recommand integration + - **Peppyrus**: For Peppyrus integration - **API Configuration**: Enter your API credentials: - **API Key**: Your service provider API key - - **API Secret**: Your service provider API secret (for Recommand) - - **Base URL**: Your service provider's API base URL - - **Account ID**: Your account identifier (for B2B Router, also used as team_id for Recommand) - - **Company ID**: Your company identifier (for Recommand) + - **API Secret**: Your service provider API secret (Recommand only) + - **Base URL**: Your service provider's API base URL. For Peppyrus use `https://api.test.peppyrus.be/v1` for test or `https://api.peppyrus.be/v1` for production. + - **Account ID**: Your account identifier for providers that require it, such as B2B Router and Recommand + - **Company ID**: Your provider company identifier where required, such as Recommand + +### Peppyrus Setup + +To use Peppyrus integration: + +1. Create or obtain a Peppyrus API key +2. Set **EDocument Integrator** to **Peppyrus** +3. Set **API Key** to your Peppyrus key +4. Set **Base URL** to the Peppyrus environment you want to target: + - Test: `https://api.test.peppyrus.be/v1` + - Production: `https://api.peppyrus.be/v1` + +Peppyrus message transmission uses the participant identifiers embedded in the generated UBL XML, so supplier and customer EndpointID values must be present in the document. ### Recommand Setup @@ -106,7 +120,7 @@ For receiving incoming documents via polling: 3. Each document will be automatically processed and create an **EDocument** record 4. The system will automatically detect the profile and validate the XML -**Automatic Polling**: The app includes a scheduled task that automatically polls for incoming documents every hour. This runs for all active integration settings with Recommand integrator. +**Automatic Polling**: The app includes a scheduled task that automatically polls for incoming documents every hour. This runs for all active integration settings. ## Usage @@ -166,6 +180,15 @@ After receiving an incoming document: - Document verification - Transparent pricing +### Peppyrus + +- **API Documentation**: [Peppyrus API](https://api.test.peppyrus.be/v1) +- **Features**: + - Send documents via the documented `/message` endpoint + - Poll inbox messages via the documented message list/detail endpoints + - Validate connectivity using organization information endpoints + - Look up recipients in the Peppol directory + ## API Endpoints ### Webhook Endpoint diff --git a/edocument_integration/api.py b/edocument_integration/api.py index d97f2a8..e274ad9 100644 --- a/edocument_integration/api.py +++ b/edocument_integration/api.py @@ -168,6 +168,12 @@ def transmit_edocument(edocument_name: str): elif integrator == "Recommand": from .recommand_api import transmit_invoice + transmission_result = transmit_invoice( + xml_content, invoice_doc=edocument_doc, integration_settings=integration_settings + ) + elif integrator == "Peppyrus": + from .peppyrus_api import transmit_invoice + transmission_result = transmit_invoice( xml_content, invoice_doc=edocument_doc, integration_settings=integration_settings ) @@ -376,14 +382,18 @@ def poll_incoming_invoices(profile: str | None = None, company: str | None = Non frappe.throw(_("No integration settings found for profile: {0}").format(profile)) integrator = integration_settings.get("edocument_integrator") - if integrator != "Recommand": - frappe.throw(_("Polling incoming invoices is only supported for Recommand integrator.")) + if integrator == "Recommand": + from .recommand_api import poll_inbox - # Import poll_inbox function - from .recommand_api import poll_inbox + poll_result = poll_inbox(integration_settings=integration_settings, company_id=company) + elif integrator == "Peppyrus": + from .peppyrus_api import poll_inbox - # Poll inbox for incoming invoices - poll_result = poll_inbox(integration_settings=integration_settings, company_id=company) + poll_result = poll_inbox(integration_settings=integration_settings, company_id=company) + else: + frappe.throw( + _("Polling incoming invoices is only supported for Recommand and Peppyrus integrators.") + ) if not poll_result.get("invoices"): return { diff --git a/edocument_integration/edocument_integration/doctype/edocument_integration_settings/edocument_integration_settings.json b/edocument_integration/edocument_integration/doctype/edocument_integration_settings/edocument_integration_settings.json index 50296ef..4bf7757 100644 --- a/edocument_integration/edocument_integration/doctype/edocument_integration_settings/edocument_integration_settings.json +++ b/edocument_integration/edocument_integration/doctype/edocument_integration_settings/edocument_integration_settings.json @@ -35,7 +35,7 @@ "fieldname": "edocument_integrator", "fieldtype": "Select", "label": "EDocument Integrator", - "options": "B2B Router\nRecommand" + "options": "B2B Router\nRecommand\nPeppyrus" }, { "fieldname": "company", diff --git a/edocument_integration/edocument_integration/doctype/edocument_integration_settings/edocument_integration_settings.py b/edocument_integration/edocument_integration/doctype/edocument_integration_settings/edocument_integration_settings.py index 2508fd7..e600e8b 100644 --- a/edocument_integration/edocument_integration/doctype/edocument_integration_settings/edocument_integration_settings.py +++ b/edocument_integration/edocument_integration/doctype/edocument_integration_settings/edocument_integration_settings.py @@ -21,7 +21,7 @@ class EDocumentIntegrationSettings(Document): base_url: DF.Data | None company: DF.Link | None company_id: DF.Data | None - edocument_integrator: DF.Literal["B2B Router", "Recommand"] + edocument_integrator: DF.Literal["B2B Router", "Recommand", "Peppyrus"] edocument_profile: DF.Link | None # end: auto-generated types @@ -129,6 +129,10 @@ def poll_incoming_documents(self): elif self.edocument_integrator == "B2B Router": from edocument_integration.b2brouter_api import poll_inbox + poll_result = poll_inbox(integration_settings=integration_settings, company_id=self.company_id) + elif self.edocument_integrator == "Peppyrus": + from edocument_integration.peppyrus_api import poll_inbox + poll_result = poll_inbox(integration_settings=integration_settings, company_id=self.company_id) else: frappe.throw(_("Unsupported E-document integrator: {0}").format(self.edocument_integrator)) diff --git a/edocument_integration/peppyrus_api.py b/edocument_integration/peppyrus_api.py new file mode 100644 index 0000000..8188fbe --- /dev/null +++ b/edocument_integration/peppyrus_api.py @@ -0,0 +1,422 @@ +# Copyright (c) 2026, Prilk Consulting BV and contributors +# For license information, please see license.txt + +""" +Peppyrus PEPPOL API Client + +This module provides a client for interacting with the Peppyrus PEPPOL API. +""" + +import base64 +import json +from typing import Any + +import frappe +import requests + +BASE_URL = "https://api.peppyrus.be/v1" +TEST_BASE_URL = "https://api.test.peppyrus.be/v1" +TIMEOUT_SECONDS = 30 + + +class PeppyrusAPIError(Exception): + """Custom exception for Peppyrus API errors.""" + + ... + + +def _normalize_participant_id(endpoint_value: str | None, scheme_id: str | None) -> str: + if not endpoint_value: + raise PeppyrusAPIError("EndpointID value is missing in XML") + + participant_scheme = scheme_id or "0088" + return f"{participant_scheme}:{endpoint_value.strip()}" + + +def _is_participant_id(value: str | None) -> bool: + return bool(value and ":" in value) + + +def _extract_message_fields(xml_content: str | bytes) -> dict[str, str]: + from lxml import etree as ET + + try: + from edocument.edocument.profiles.peppol import UBL_NAMESPACES + + xml_bytes = xml_content.encode("utf-8") if isinstance(xml_content, str) else xml_content + root = ET.fromstring(xml_bytes) + + supplier_endpoint = root.find( + ".//cac:AccountingSupplierParty/cac:Party/cbc:EndpointID", UBL_NAMESPACES + ) + customer_endpoint = root.find( + ".//cac:AccountingCustomerParty/cac:Party/cbc:EndpointID", UBL_NAMESPACES + ) + + if supplier_endpoint is None: + raise PeppyrusAPIError( + "Sender EndpointID not found in XML. Please configure electronic address for supplier." + ) + + if customer_endpoint is None: + raise PeppyrusAPIError( + "Recipient EndpointID not found in XML. Please configure electronic address for customer." + ) + + customization_id = root.findtext(".//cbc:CustomizationID", namespaces=UBL_NAMESPACES) + profile_id = root.findtext(".//cbc:ProfileID", namespaces=UBL_NAMESPACES) + ubl_version = root.findtext(".//cbc:UBLVersionID", namespaces=UBL_NAMESPACES) or "2.1" + + if not customization_id: + raise PeppyrusAPIError("CustomizationID not found in XML; cannot determine document type") + + if not profile_id: + raise PeppyrusAPIError("ProfileID not found in XML; cannot determine process type") + + if root.tag.startswith("{"): + root_namespace, root_name = root.tag[1:].split("}", 1) + else: + root_namespace = "" + root_name = root.tag + + return { + "sender": _normalize_participant_id(supplier_endpoint.text, supplier_endpoint.get("schemeID")), + "recipient": _normalize_participant_id(customer_endpoint.text, customer_endpoint.get("schemeID")), + "process_type": f"cenbii-procid-ubl::{profile_id.strip()}", + "document_type": ( + f"busdox-docid-qns::{root_namespace}::{root_name}##{customization_id.strip()}::{ubl_version.strip()}" + ), + } + except ET.ParseError as exc: + raise PeppyrusAPIError(f"Failed to parse XML to extract message metadata: {exc!s}") from exc + except PeppyrusAPIError: + raise + except Exception as exc: + raise PeppyrusAPIError(f"Failed to extract message metadata from XML: {exc!s}") from exc + + +class PeppyrusAPIClient: + """Client for interacting with the Peppyrus PEPPOL API using API key header auth.""" + + def __init__(self, api_key: str, base_url: str = BASE_URL): + self.api_key = api_key + self.base_url = base_url.rstrip("/") + self.session = requests.Session() + + # Use API key header like B2B Router + self.session.headers.update( + { + "X-Api-Key": api_key, + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "Frappe-EDocument/1.0", + } + ) + + def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response: + url = f"{self.base_url}{endpoint}" + + try: + response = self.session.request(method, url, timeout=TIMEOUT_SECONDS, **kwargs) + return response + except requests.exceptions.RequestException as exc: + error_msg = f"Peppyrus API request failed: {exc!s}" + frappe.log_error(error_msg, "Peppyrus API Error") + raise PeppyrusAPIError(error_msg) from exc + + def _parse_json_response(self, response: requests.Response) -> dict[str, Any] | list[Any] | bool: + if not response.ok: + try: + error_data = response.json() + error_text = json.dumps(error_data) + except json.JSONDecodeError: + error_text = response.text + + error_msg = f"Peppyrus API error ({response.status_code}): {error_text}" + frappe.log_error(error_msg, "Peppyrus API Error") + raise PeppyrusAPIError(error_msg) + + if not response.content: + return {} + + try: + return response.json() + except json.JSONDecodeError as exc: + error_msg = f"Peppyrus API returned invalid JSON: {response.text}" + frappe.log_error(error_msg, "Peppyrus API Error") + raise PeppyrusAPIError(error_msg) from exc + + # DOCUMENT TRANSMISSION METHODS + + def send_document(self, xml_content: str | bytes) -> dict[str, Any]: + """Send a PEPPOL document via Peppyrus. + + The payload follows the documented MessageBody schema. + """ + endpoint = "/message" + + if isinstance(xml_content, bytes): + xml_bytes = xml_content + else: + xml_bytes = xml_content.encode("utf-8") + + message_fields = _extract_message_fields(xml_bytes) + + payload = { + "sender": message_fields["sender"], + "recipient": message_fields["recipient"], + "processType": message_fields["process_type"], + "documentType": message_fields["document_type"], + "fileContent": base64.b64encode(xml_bytes).decode("utf-8"), + } + + try: + response = self._make_request("POST", endpoint, json=payload) + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when posting message") + document_id = result.get("id", "unknown") + + return { + "status": "success", + "document_id": document_id, + "sender": message_fields["sender"], + "recipient": message_fields["recipient"], + "response": result, + } + + except PeppyrusAPIError: + raise + except Exception as e: + error_msg = f"Peppyrus document transmission failed: {e!s}" + frappe.log_error(error_msg, "Peppyrus Transmission Error") + raise PeppyrusAPIError(error_msg) + + def list_messages( + self, + folder: str | None = None, + sender: str | None = None, + receiver: str | None = None, + confirmed: bool | None = None, + page: int = 1, + per_page: int = 50, + ) -> dict[str, Any]: + endpoint = "/message/list" + params = {"page": page, "perPage": min(per_page, 100)} + if folder: + params["folder"] = folder + if sender: + params["sender"] = sender + if receiver: + params["receiver"] = receiver + if confirmed is not None: + params["confirmed"] = str(confirmed).lower() + try: + response = self._make_request("GET", endpoint, params=params) + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when listing messages") + return result + except requests.exceptions.RequestException as e: + error_msg = f"Peppyrus list messages failed: {e!s}" + frappe.log_error(error_msg, "Peppyrus Message List Error") + raise PeppyrusAPIError(error_msg) + + def get_message(self, message_id: str) -> dict[str, Any]: + endpoint = f"/message/{message_id}" + try: + response = self._make_request("GET", endpoint) + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when fetching message") + return result + except requests.exceptions.RequestException as e: + error_msg = f"Peppyrus get message failed: {e!s}" + frappe.log_error(error_msg, "Peppyrus Get Message Error") + raise PeppyrusAPIError(error_msg) + + def confirm_message(self, message_id: str) -> bool: + endpoint = f"/message/{message_id}/confirm" + response = self._make_request("PATCH", endpoint) + result = self._parse_json_response(response) + if not isinstance(result, bool): + raise PeppyrusAPIError("Unexpected Peppyrus response when confirming message") + return result + + def render_message(self, message_id: str, output_format: str = "PDF") -> dict[str, Any]: + endpoint = f"/message/{message_id}/render" + response = self._make_request("GET", endpoint, params={"format": output_format}) + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when rendering message") + return result + + def get_message_report(self, message_id: str) -> dict[str, Any]: + endpoint = f"/message/{message_id}/report" + response = self._make_request("GET", endpoint) + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when fetching message report") + return result + + def get_organization_info(self) -> dict[str, Any]: + response = self._make_request("GET", "/organization/info") + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when fetching organization info") + return result + + def get_organization_peppol_info(self) -> dict[str, Any]: + response = self._make_request("GET", "/organization/peppol") + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when fetching organization peppol info") + return result + + def lookup_participant(self, participant_id: str) -> dict[str, Any]: + response = self._make_request("GET", "/peppol/lookup", params={"participantId": participant_id}) + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when looking up participant") + return result + + def best_match_participant(self, vat_number: str, country_code: str) -> dict[str, Any]: + response = self._make_request( + "GET", "/peppol/bestMatch", params={"vatNumber": vat_number, "countryCode": country_code} + ) + result = self._parse_json_response(response) + if not isinstance(result, dict): + raise PeppyrusAPIError("Unexpected Peppyrus response when finding participant best match") + return result + + def search_peppol_directory(self, **search_params) -> list[dict[str, Any]]: + response = self._make_request("GET", "/peppol/search", params=search_params) + result = self._parse_json_response(response) + if not isinstance(result, list): + raise PeppyrusAPIError("Unexpected Peppyrus response when searching directory") + return result + + def get_documents(self, team_id: str | None = None, limit: int = 50, offset: int = 0) -> dict[str, Any]: + page = (offset // limit) + 1 if limit else 1 + return self.list_messages(page=page, per_page=limit) + + def get_inbox(self, team_id: str | None = None, company_id: str | None = None) -> dict[str, Any]: + params: dict[str, Any] = {"folder": "INBOX"} + if _is_participant_id(company_id): + params["receiver"] = company_id + return self.list_messages(**params) + + def get_document_status(self, team_id: str | None, document_id: str) -> dict[str, Any]: + try: + return self.get_message(document_id) + except requests.exceptions.RequestException as e: + error_msg = f"Peppyrus get document status failed: {e!s}" + frappe.log_error(error_msg, "Peppyrus Document Status Error") + raise PeppyrusAPIError(error_msg) + + +# HELPER FUNCTIONS + + +def get_peppyrus_client(integration_settings: dict[str, Any]) -> PeppyrusAPIClient: + if not integration_settings or not isinstance(integration_settings, dict): + raise PeppyrusAPIError("Peppyrus integration settings must be provided as a dict") + + api_key = integration_settings.get("api_key") + base_url = integration_settings.get("base_url", BASE_URL) + + if not api_key: + raise PeppyrusAPIError("Peppyrus API key not configured") + + return PeppyrusAPIClient(api_key, base_url) + + +def transmit_invoice(xml_content: str, invoice_doc=None, integration_settings=None) -> dict[str, Any]: + if not integration_settings or not isinstance(integration_settings, dict): + raise PeppyrusAPIError( + "Peppyrus integration settings are required for transmission and must be a dict" + ) + + try: + client = get_peppyrus_client(integration_settings) + transmission_result = client.send_document(xml_content=xml_content) + + return { + "status": "success", + "document_id": transmission_result.get("document_id"), + "recipient": transmission_result.get("recipient"), + "sender": transmission_result.get("sender"), + "response": transmission_result, + } + + except Exception as e: + error_msg = f"Peppyrus transmission failed: {e!s}" + frappe.log_error(error_msg, "Peppyrus Transmission Error") + raise PeppyrusAPIError(error_msg) + + +def validate_peppyrus_connection(api_key: str, base_url: str = BASE_URL) -> dict[str, Any]: + try: + client = PeppyrusAPIClient(api_key, base_url) + organization_info = client.get_organization_info() + organization_name = organization_info.get("name", "Peppyrus organization") + return {"status": "success", "message": f"Peppyrus API connection successful for {organization_name}"} + except Exception as e: + return {"status": "error", "message": f"Peppyrus API connection failed: {e!s}"} + + +def verify_recipient(participant_id: str, integration_settings: dict[str, Any]) -> dict[str, Any]: + client = get_peppyrus_client(integration_settings) + lookup_result = client.lookup_participant(participant_id) + services = lookup_result.get("services") or [] + return { + "status": "success", + "participant_id": lookup_result.get("participantId", participant_id), + "scheme": lookup_result.get("scheme"), + "services": services, + "can_receive": bool(services), + } + + +def poll_inbox( + integration_settings: dict[str, Any] | None = None, company_id: str | None = None +) -> dict[str, Any]: + if not integration_settings or not isinstance(integration_settings, dict): + raise PeppyrusAPIError( + "Peppyrus integration settings are required for inbox polling and must be a dict" + ) + + client = get_peppyrus_client(integration_settings) + inbox_result = client.get_inbox(company_id=company_id or integration_settings.get("company_id")) + documents = inbox_result.get("items", []) or [] + + if not documents: + return {"status": "success", "message": "No new invoices found", "invoices": []} + + invoices = [] + for doc in documents: + try: + document_id = doc.get("id") + if not document_id: + frappe.log_error("Message entry is missing an ID", "Peppyrus Inbox Polling Error") + continue + + doc_details = client.get_message(document_id) + + file_content = doc_details.get("fileContent") + if not file_content: + frappe.log_error( + f"No 'fileContent' key found in response for {document_id}. Response keys: {list(doc_details.keys())}", + "Peppyrus Inbox Polling Error", + ) + continue + + xml_bytes = base64.b64decode(file_content) + invoices.append({"xml_bytes": xml_bytes, "document_id": document_id, "metadata": doc_details}) + except Exception as e: + frappe.log_error( + f"Failed to fetch document {doc.get('id')}: {e!s}\nTraceback: {frappe.get_traceback()}", + "Peppyrus Inbox Polling Error", + ) + + return {"status": "success", "message": f"Found {len(invoices)} invoice(s)", "invoices": invoices} diff --git a/edocument_integration/test_peppyrus_api.py b/edocument_integration/test_peppyrus_api.py new file mode 100644 index 0000000..2c3b079 --- /dev/null +++ b/edocument_integration/test_peppyrus_api.py @@ -0,0 +1,140 @@ +import base64 +import importlib +import json +import sys +import types +import unittest +import xml.etree.ElementTree as stdlib_etree +from unittest.mock import Mock, patch + + +def _install_test_stubs(): + fake_frappe = types.ModuleType("frappe") + fake_frappe.log_error = Mock() + fake_frappe.get_traceback = lambda: "traceback" + sys.modules["frappe"] = fake_frappe + + fake_lxml = types.ModuleType("lxml") + fake_lxml_etree = types.ModuleType("lxml.etree") + fake_lxml_etree.ParseError = stdlib_etree.ParseError + fake_lxml_etree.QName = stdlib_etree.QName + fake_lxml_etree.fromstring = stdlib_etree.fromstring + fake_lxml.etree = fake_lxml_etree + sys.modules["lxml"] = fake_lxml + sys.modules["lxml.etree"] = fake_lxml_etree + + edocument_module = types.ModuleType("edocument") + edocument_inner_module = types.ModuleType("edocument.edocument") + profiles_module = types.ModuleType("edocument.edocument.profiles") + peppol_module = types.ModuleType("edocument.edocument.profiles.peppol") + peppol_module.UBL_NAMESPACES = { + "cbc": "urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2", + "cac": "urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2", + } + sys.modules["edocument"] = edocument_module + sys.modules["edocument.edocument"] = edocument_inner_module + sys.modules["edocument.edocument.profiles"] = profiles_module + sys.modules["edocument.edocument.profiles.peppol"] = peppol_module + + +class MockResponse: + def __init__(self, payload, status_code=200): + self._payload = payload + self.status_code = status_code + self.ok = status_code < 400 + self.text = payload if isinstance(payload, str) else json.dumps(payload) + self.content = self.text.encode("utf-8") + + def json(self): + return self._payload + + +class PeppyrusAPITestCase(unittest.TestCase): + @classmethod + def setUpClass(cls): + _install_test_stubs() + cls.peppyrus_api = importlib.import_module("edocument_integration.peppyrus_api") + + def setUp(self): + self.module = importlib.reload(self.peppyrus_api) + self.sample_xml = """ + + 2.1 + urn:cen.eu:en16931:2017#compliant#urn:fdc:peppol.eu:2017:poacc:billing:3.0 + urn:fdc:peppol.eu:2017:poacc:billing:01:1.0 + + + BE0123456789 + + + + + 1234567890123 + + + +""" + + def test_client_uses_documented_auth_header(self): + client = self.module.PeppyrusAPIClient("secret-key") + self.assertEqual(client.session.headers["X-Api-Key"], "secret-key") + self.assertNotIn("X-PEPPYRUS-API-Key", client.session.headers) + + def test_extract_message_fields_from_ubl_invoice(self): + fields = self.module._extract_message_fields(self.sample_xml) + + self.assertEqual(fields["sender"], "0208:BE0123456789") + self.assertEqual(fields["recipient"], "0088:1234567890123") + self.assertEqual( + fields["process_type"], + "cenbii-procid-ubl::urn:fdc:peppol.eu:2017:poacc:billing:01:1.0", + ) + self.assertEqual( + fields["document_type"], + "busdox-docid-qns::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", + ) + + def test_send_document_posts_message_body(self): + client = self.module.PeppyrusAPIClient("secret-key") + client._make_request = Mock(return_value=MockResponse({"id": "message-123"})) + + result = client.send_document(self.sample_xml) + + self.assertEqual(result["document_id"], "message-123") + call_args = client._make_request.call_args.kwargs + payload = call_args["json"] + self.assertEqual(payload["sender"], "0208:BE0123456789") + self.assertEqual(payload["recipient"], "0088:1234567890123") + self.assertEqual(base64.b64decode(payload["fileContent"]).decode("utf-8"), self.sample_xml) + + def test_poll_inbox_decodes_message_content(self): + fake_client = Mock() + fake_client.get_inbox.return_value = {"items": [{"id": "message-1"}]} + fake_client.get_message.return_value = { + "id": "message-1", + "fileContent": base64.b64encode(b"payload").decode("utf-8"), + } + + with patch.object(self.module, "get_peppyrus_client", return_value=fake_client): + result = self.module.poll_inbox({"api_key": "secret-key"}) + + self.assertEqual(result["status"], "success") + self.assertEqual(result["invoices"][0]["xml_bytes"], b"payload") + self.assertEqual(result["invoices"][0]["document_id"], "message-1") + + def test_validate_connection_uses_organization_info(self): + with patch.object( + self.module.PeppyrusAPIClient, + "get_organization_info", + return_value={"name": "Example Org"}, + ): + result = self.module.validate_peppyrus_connection("secret-key") + + self.assertEqual(result["status"], "success") + self.assertIn("Example Org", result["message"]) + + +if __name__ == "__main__": + unittest.main()