Preserve payment receipt extensions

.changelog/receipt-extensions.md

@@ -0,0 +1,5 @@
+---
+pympp: patch
+---
+
+Preserve method-specific top-level fields when parsing and formatting payment receipts.

src/mpp/__init__.py

@@ -365,6 +365,9 @@ def to_authorization(self) -> str:
 class Receipt:
     """Payment receipt returned after verification.
 
+    Method-specific top-level fields are stored in ``extensions``. The ``extra``
+    field represents the distinct, nested ``extra`` wire field.
+
     Example:
         from datetime import datetime, UTC
 
@@ -382,6 +385,7 @@ class Receipt:
     external_id: str | None = None
     subscription_id: str | None = None
     extra: dict[str, Any] | None = None
+    extensions: dict[str, Any] | None = None
 
     @classmethod
     def from_payment_receipt(cls, header: str) -> Receipt:

src/mpp/_parsing.py

@@ -29,6 +29,9 @@
 _AUTH_PARAM_RE = re.compile(r'([a-zA-Z_][\w-]*)\s*=\s*(?:"((?:[^"\\]|\\.)*)"|([^\s,]+))')
 # Syntax-level Payment Auth grammar. Supported-method dispatch is handled after parsing.
 _PAYMENT_METHOD_ID_RE = re.compile(r"^[a-z]+$")
+_RECEIPT_RESERVED_FIELDS = frozenset(
+    {"status", "timestamp", "reference", "method", "externalId", "subscriptionId", "extra"}
+)
 
 
 class ParseError(Exception):
@@ -318,6 +321,7 @@ def parse_payment_receipt(header: str) -> Receipt:
     _validate_payment_method_id(method)
 
     extra = data.get("extra")
+    extensions = {key: value for key, value in data.items() if key not in _RECEIPT_RESERVED_FIELDS}
 
     return Receipt(
         status=status,
@@ -327,6 +331,7 @@ def parse_payment_receipt(header: str) -> Receipt:
         external_id=str(data["externalId"]) if data.get("externalId") else None,
         subscription_id=str(data["subscriptionId"]) if data.get("subscriptionId") else None,
         extra=extra if isinstance(extra, dict) else None,
+        extensions=extensions or None,
     )
 
 
@@ -350,4 +355,9 @@ def format_payment_receipt(receipt: Receipt) -> str:
         payload["subscriptionId"] = receipt.subscription_id
     if receipt.extra:
         payload["extra"] = receipt.extra
+    if receipt.extensions:
+        reserved = receipt.extensions.keys() & _RECEIPT_RESERVED_FIELDS
+        if reserved:
+            raise ValueError(f"Receipt extensions contain reserved fields: {sorted(reserved)}")
+        payload.update(receipt.extensions)
     return _b64_encode(payload)

tests/test_parsing.py

@@ -365,6 +365,41 @@ def test_parse_preserves_foreign_subscription_id(self) -> None:
         roundtripped = Receipt.from_payment_receipt(parsed.to_payment_receipt())
         assert roundtripped.subscription_id == "sub_123"
 
+    def test_roundtrip_preserves_top_level_extensions(self) -> None:
+        payload = {
+            "status": "success",
+            "method": "tempo",
+            "timestamp": "2024-01-20T12:00:00Z",
+            "reference": "session-123",
+            "extra": {"plan": "pro"},
+            "intent": "session",
+            "channelId": "0xabc123",
+            "acceptedCumulative": "1000",
+            "units": 3,
+        }
+
+        parsed = Receipt.from_payment_receipt(_b64_json(payload))
+
+        assert parsed.extra == {"plan": "pro"}
+        assert parsed.extensions == {
+            "intent": "session",
+            "channelId": "0xabc123",
+            "acceptedCumulative": "1000",
+            "units": 3,
+        }
+        assert Receipt.from_payment_receipt(parsed.to_payment_receipt()) == parsed
+
+    def test_extensions_cannot_override_receipt_fields(self) -> None:
+        receipt = Receipt(
+            status="success",
+            timestamp=datetime(2024, 1, 20, 12, 0, 0, tzinfo=UTC),
+            reference="0xabc123",
+            extensions={"status": "failed"},
+        )
+
+        with pytest.raises(ValueError, match=r"reserved fields: \['status'\]"):
+            receipt.to_payment_receipt()
+
     def test_parse_invalid_timestamp(self) -> None:
         payload = {
             "status": "success",