added webhook support (#3)

Author: kdpisda

src/pay_ccavenue/ccavenue.py

@@ -8,6 +8,7 @@
 from Crypto.Cipher import AES
 from pay_ccavenue.models.config import CCavenueConfig
 from pay_ccavenue.models.form import CCavenueFormData
+from pay_ccavenue.models.webhook import CCavenueWebhookData
 
 
 class CCAvenue:
@@ -191,3 +192,16 @@ def decrypt(self, data: Dict[str, str]) -> Dict[str, str]:
         decrypted = cipher.decrypt(encrypted_text)
         self._unflatten_decrypted_data(decrypted)
         return self.decrypted_data
+
+    def process_webhook(self, response_body: Dict[str, str]) -> CCavenueWebhookData:
+        """
+        Process and validate the webhook/notification data from CCAvenue.
+
+        Args:
+            response_body (Dict[str, str]): The response body containing 'encResp'.
+
+        Returns:
+            CCavenueWebhookData: The decrypted and parsed data object.
+        """
+        decrypted_dict = self.decrypt(response_body)
+        return CCavenueWebhookData.from_dict(decrypted_dict)

src/pay_ccavenue/models/__init__.py

@@ -0,0 +1,5 @@
+from pay_ccavenue.models.config import CCavenueConfig
+from pay_ccavenue.models.form import CCavenueFormData
+from pay_ccavenue.models.webhook import CCavenueWebhookData
+
+__all__ = ["CCavenueConfig", "CCavenueFormData", "CCavenueWebhookData"]

src/pay_ccavenue/models/webhook.py

@@ -0,0 +1,82 @@
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class CCavenueWebhookData:
+    """
+    Represents the data received from a CCAvenue webhook/notification.
+
+    Covers fields for:
+    - Order Status (including Reconcilation and Echo)
+    - Order Risk Status
+    - Payment Type Status
+    """
+
+    # Core Fields (Order Status / generic)
+    order_id: Optional[str] = None
+    tracking_id: Optional[str] = None
+    bank_ref_no: Optional[str] = None
+    order_status: Optional[str] = None  # Success, Failure, Aborted, Invalid
+    failure_message: Optional[str] = None
+    payment_mode: Optional[str] = None
+
+    # Card & Bank Details
+    card_name: Optional[str] = None
+    status_code: Optional[str] = None
+    status_message: Optional[str] = None
+
+    # Transaction Details
+    currency: Optional[str] = None
+    amount: Optional[str] = None
+
+    # Billing Details
+    billing_name: Optional[str] = None
+    billing_address: Optional[str] = None
+    billing_city: Optional[str] = None
+    billing_state: Optional[str] = None
+    billing_zip: Optional[str] = None
+    billing_country: Optional[str] = None
+    billing_tel: Optional[str] = None
+    billing_email: Optional[str] = None
+
+    # Delivery Details
+    delivery_name: Optional[str] = None
+    delivery_address: Optional[str] = None
+    delivery_city: Optional[str] = None
+    delivery_state: Optional[str] = None
+    delivery_zip: Optional[str] = None
+    delivery_country: Optional[str] = None
+    delivery_tel: Optional[str] = None
+
+    # Merchant Custom Fields
+    merchant_param1: Optional[str] = None
+    merchant_param2: Optional[str] = None
+    merchant_param3: Optional[str] = None
+    merchant_param4: Optional[str] = None
+    merchant_param5: Optional[str] = None
+
+    # Offers & Vault
+    vault: Optional[str] = None
+    offer_type: Optional[str] = None
+    offer_code: Optional[str] = None
+    discount_value: Optional[str] = None
+
+    # Risk Status Specific
+    risk_status: Optional[str] = None  # High, Low, NR, GA
+    risk_reason: Optional[str] = None
+
+    # Payment Type Status Specific
+    payment_option: Optional[str] = None
+    current_status: Optional[str] = None  # ACTI, FLCT, INAC, DOWN, NEW
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "CCavenueWebhookData":
+        """
+        Create a CCavenueWebhookData instance from a dictionary.
+        Keys in the dictionary that don't match fields are ignored.
+        """
+        # Filter data to only include keys that match field names
+        known_fields = set(cls.__annotations__.keys())
+        filtered_data = {k: v for k, v in data.items() if k in known_fields}
+        return cls(**filtered_data)

tests/test_core.py

@@ -108,3 +108,13 @@ def mock_get_cipher():
     decrypted_data = {k: v.rstrip("\r") for k, v in decrypted_data.items()}
 
     assert decrypted_data == {"key1": "value1", "key2": "value2", "key3": "value3"}
+
+
+def test_get_cipher(ccavenue_instance):
+    """Test _get_cipher logic directly without mocking."""
+    cipher = ccavenue_instance._get_cipher()
+    assert cipher is not None
+    # We can't easily check the key inside the cipher object in PyCryptodome,
+    # but successful creation is what we want to test.
+    # The block size should be AES.block_size (16)
+    assert cipher.block_size == 16

tests/test_webhook.py

@@ -0,0 +1,60 @@
+from pay_ccavenue.models.webhook import CCavenueWebhookData
+
+
+def test_webhook_data_from_dict():
+    """Test that CCavenueWebhookData correctly parses a dictionary."""
+    data = {
+        "order_id": "123",
+        "order_status": "Success",
+        "amount": "100.00",
+        "unknown_field": "should_be_ignored",
+        "merchant_param1": "custom_data",
+    }
+
+    webhook_data = CCavenueWebhookData.from_dict(data)
+
+    assert webhook_data.order_id == "123"
+    assert webhook_data.order_status == "Success"
+    assert webhook_data.amount == "100.00"
+    assert webhook_data.merchant_param1 == "custom_data"
+
+    # Check that unknown fields are ignored (not present as attributes)
+    # Dataclasses only have defined fields
+    assert not hasattr(webhook_data, "unknown_field")
+
+
+def test_process_webhook(ccavenue_instance, monkeypatch):
+    """Test the process_webhook method of CCAvenue class."""
+
+    # Mock decrypt to return a specific dictionary
+    expected_dict = {
+        "order_id": "ORDER123",
+        "tracking_id": "TRACK123",
+        "order_status": "Success",
+        "amount": "500.50",
+        "currency": "INR",
+        "payment_mode": "Net Banking",
+        "card_name": "HDFC",
+    }
+
+    def mock_decrypt(data):
+        return expected_dict
+
+    monkeypatch.setattr(ccavenue_instance, "decrypt", mock_decrypt)
+
+    # The input to process_webhook is a dict with 'encResp'
+    # Since we mocked decrypt, the content doesn't matter much, but structure does
+    dummy_input = {"encResp": "dummy_encrypted_string"}
+
+    result = ccavenue_instance.process_webhook(dummy_input)
+
+    assert isinstance(result, CCavenueWebhookData)
+    assert result.order_id == "ORDER123"
+    assert result.tracking_id == "TRACK123"
+    assert result.order_status == "Success"
+    assert result.amount == "500.50"
+    assert result.currency == "INR"
+    assert result.payment_mode == "Net Banking"
+    assert result.card_name == "HDFC"
+    # Ensure optional fields are None
+    assert result.failure_message is None