test: exercise all gateway services with mocked HTTP

2026-09-03 00:40:51 +05:45
1 parent c2de8c02c0
commit f7a88497d7
@@ -0,0 +1,384 @@
+"""Gateway service integration tests with mocked HTTP (constitution II)."""
+
+import base64
+import json
+
+import pytest
+import requests
+
+from nepali_payment.enums import PaymentMethod, PaymentMode
+from nepali_payment.exceptions import PaymentError
+from nepali_payment.helpers import convert_to
+from nepali_payment.http import ApiService
+from nepali_payment.manager import PaymentManager
+from nepali_payment.models import PaymentResult
+from nepali_payment.models.esewa import PaymentRequest as EsewaRequest
+from nepali_payment.models.fonepay import QrRequest, QrResponse, StaticQrRequest, TaxRefundRequest
+from nepali_payment.models.khalti import PaymentRequest as KhaltiRequest
+from nepali_payment.service_factory import get_payment_service
+
+SECRET_KEY = "a7e3512f5032480a83137793cb2021dc"
+
+
+class FakeResponse:
+    def __init__(self, payload, status=200, content_type="application/json", url="https://example.test/x"):
+        self._payload = payload
+        self.status_code = status
+        self.headers = {"Content-Type": content_type}
+        self.url = url
+        self.text = json.dumps(payload) if not isinstance(payload, str) else payload
+
+    def json(self):
+        if isinstance(self._payload, str):
+            raise TypeError("not json")
+        return self._payload
+
+    @property
+    def ok(self):
+        return 200 <= self.status_code < 300
+
+
+class FakeSession:
+    def __init__(self, responses):
+        self.responses = responses
+        self.last_kwargs = None
+
+    def request(self, method, url, headers=None, data=None, json=None, timeout=None):
+        self.last_kwargs = {"method": method, "url": url, "headers": headers, "data": data, "json": json}
+        return self.responses.pop(0)
+
+
+def test_esewa_v2_initiate_form_encoded_signature_and_endpoint():
+    session = FakeSession([FakeResponse("https://epay/redirect", content_type="text/html")])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.ESEWA, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+
+    request = EsewaRequest(
+        amount="100",
+        total_amount="113",
+        transaction_uuid="uuid-1",
+        product_code="EPAYTEST",
+        signed_field_names="total_amount,transaction_uuid,product_code",
+    )
+    result = manager.initiate_payment(PaymentResult, request)
+
+    assert result.success is True
+    assert result.data is not None
+    assert session.last_kwargs["url"] == "https://rc-epay.esewa.com.np/api/epay/main/v2/form"
+    assert session.last_kwargs["method"] == "POST"
+    assert session.last_kwargs["data"]["transaction_uuid"] == "uuid-1"
+    assert request.signature  # signature was generated
+
+
+def test_esewa_verify_maps_status_to_success():
+    manager = PaymentManager(PaymentMethod.ESEWA, PaymentMode.SANDBOX, SECRET_KEY)
+    base64_content = lambda status: base64.b64encode(
+        json.dumps({"status": status, "total_amount": 100.0}).encode()
+    ).decode()
+
+    ok = manager.verify_payment(PaymentResult, base64_content("COMPLETE"))
+    assert ok.success is True
+    assert ok.data.status == "COMPLETE"
+
+    cancelled = manager.verify_payment(PaymentResult, base64_content("CANCELLED"))
+    assert cancelled.success is False
+    assert cancelled.data.status == "CANCELLED"
+
+    failed = manager.verify_payment(PaymentResult, base64_content("FAILED"))
+    assert failed.success is False
+    assert failed.data.status == "FAILED"
+
+
+def test_esewa_service_direct_empty_and_malformed_content():
+    """Cover eSewa's own guards reached by calling the service directly."""
+    from nepali_payment.services.esewa import PaymentService as EsewaService
+
+    svc = EsewaService("secret", PaymentMode.SANDBOX)
+    with pytest.raises(ValueError):
+        svc.verify_payment(PaymentResult, "")
+    with pytest.raises(PaymentError):
+        # JSON decodes to a list, so PaymentResponse(**list) fails -> wrapped in PaymentError.
+        svc.verify_payment(PaymentResult, base64.b64encode(b"[1,2,3]").decode())
+
+
+def test_esewa_initiate_with_wrong_content_type_raises_type_error():
+    manager = PaymentManager(PaymentMethod.ESEWA, PaymentMode.SANDBOX, SECRET_KEY)
+    req = KhaltiRequest(return_url="https://return", website_url="https://site", amount=100)
+    with pytest.raises(TypeError):
+        manager.initiate_payment(PaymentResult, req)
+
+
+def test_esewa_raises_payment_error_not_failed_result_on_failure():
+    """eSewa's distinct contract: a network failure raises, it does not fold."""
+
+    class _BadSession:
+        def request(self, **kwargs):
+            raise requests.ConnectionError("offline")
+
+    api = ApiService(session=_BadSession())
+    manager = PaymentManager(PaymentMethod.ESEWA, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+    req = EsewaRequest(
+        amount="100",
+        total_amount="113",
+        transaction_uuid="uuid-1",
+        product_code="EPAYTEST",
+        signed_field_names="total_amount,transaction_uuid,product_code",
+    )
+    with pytest.raises(PaymentError):
+        manager.initiate_payment(PaymentResult, req)
+
+
+def test_esewa_passthrough_preserves_payment_error_type():
+    """A ProviderError (a PaymentError) from the HTTP layer must be re-raised."""
+
+    class _Resp:
+        ok = False
+        status_code = 500
+        text = "Internal error"
+        url = "https://epay/x"
+
+        def __init__(self):
+            self.headers = {"Content-Type": "application/json"}
+
+    class _BadSession:
+        def request(self, **kwargs):
+            return _Resp()
+
+    api = ApiService(session=_BadSession())
+    manager = PaymentManager(PaymentMethod.ESEWA, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+    req = EsewaRequest(
+        amount="100",
+        total_amount="113",
+        transaction_uuid="uuid-1",
+        product_code="EPAYTEST",
+        signed_field_names="total_amount,transaction_uuid,product_code",
+    )
+    with pytest.raises(PaymentError):
+        manager.initiate_payment(PaymentResult, req)
+
+
+def test_esewa_verify_with_empty_content_raises_value_error():
+    from nepali_payment.exceptions import ValidationError
+
+    manager = PaymentManager(PaymentMethod.ESEWA, PaymentMode.SANDBOX, SECRET_KEY)
+    with pytest.raises(ValidationError):
+        manager.verify_payment(PaymentResult, "")
+
+
+def test_service_factory_raises_on_unsupported_gateway():
+    from nepali_payment.exceptions import UnsupportedGatewayError
+
+    with pytest.raises(UnsupportedGatewayError):
+        get_payment_service("NotAGateway", SECRET_KEY, PaymentMode.SANDBOX)
+
+
+def test_service_factory_requires_secret_key():
+    with pytest.raises(ValueError):
+        get_payment_service(PaymentMethod.KHALTI, "", PaymentMode.SANDBOX)
+
+
+def test_endpoint_for_raises_on_unsupported_gateway_or_action():
+    from nepali_payment.enums import PaymentAction
+    from nepali_payment.exceptions import UnsupportedGatewayError
+    from nepali_payment.factory import _endpoint_for
+
+    with pytest.raises(UnsupportedGatewayError):
+        _endpoint_for("NotAGateway", PaymentAction.PROCESS_PAYMENT, PaymentMode.SANDBOX)
+    with pytest.raises(UnsupportedGatewayError):
+        # Fonepay has no CHECK_PAYMENT mapping in the endpoint registry.
+        _endpoint_for(PaymentMethod.FONEPAY, PaymentAction.CHECK_PAYMENT, PaymentMode.SANDBOX)
+
+
+def test_khalti_service_requires_secret_key():
+    from nepali_payment.services.khalti import PaymentService as KhaltiService
+
+    with pytest.raises(ValueError):
+        KhaltiService("", PaymentMode.SANDBOX)
+
+
+def test_fonepay_service_requires_secret_key():
+    from nepali_payment.services.fonepay import PaymentService as FonepayService
+
+    with pytest.raises(ValueError):
+        FonepayService("", PaymentMode.SANDBOX)
+
+
+def test_esewa_service_requires_secret_key():
+    from nepali_payment.services.esewa import PaymentService as EsewaService
+
+    with pytest.raises(ValueError):
+        EsewaService("", PaymentMode.SANDBOX)
+
+
+def test_manager_requires_secret_key():
+    from nepali_payment.manager import PaymentManager
+
+    with pytest.raises(ValueError):
+        PaymentManager(PaymentMethod.KHALTI, PaymentMode.SANDBOX, "")
+
+
+def test_khalti_v2_initiate_authorization_header_and_json_body():
+    khalti_response = {"pidx": "abc123", "payment_url": "https://pay.khalti/...", "expires_in": 30}
+    session = FakeSession([FakeResponse(khalti_response)])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.KHALTI, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+
+    request = KhaltiRequest(return_url="https://return", website_url="https://site", amount=100)
+    result = manager.initiate_payment(PaymentResult, request)
+
+    assert result.success is True
+    assert result.data.pidx == "abc123"
+    assert session.last_kwargs["url"] == "https://a.khalti.com/api/epayment/initiate/"
+    assert session.last_kwargs["headers"]["Authorization"].startswith("key ")
+    assert session.last_kwargs["json"]["return_url"] == "https://return"
+
+
+def test_khalti_v2_verify_form_encoding_with_pidx():
+    verify_response = {
+        "pidx": "abc123",
+        "total_amount": 100,
+        "status": "Completed",
+        "transaction_id": "txn",
+        "fee": 3,
+        "refunded": False,
+    }
+    session = FakeSession([FakeResponse(verify_response)])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.KHALTI, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+
+    result = manager.verify_payment(PaymentResult, "abc123")
+
+    assert result.success is True
+    assert result.data.status == "Completed"
+    assert session.last_kwargs["url"] == "https://a.khalti.com/api/epayment/lookup/"
+    assert session.last_kwargs["data"] == {"pidx": "abc123"}
+
+
+def test_fonepay_initiate_hmac_sha512_hex_data_validation():
+    qr_response = {
+        "message": "ok",
+        "qrMessage": "QRDATA",
+        "status": "success",
+        "success": True,
+        "thirdpartyQrWebSocketUrl": "wss://foo",
+    }
+    session = FakeSession([FakeResponse(qr_response)])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.FONEPAY, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+
+    request = QrRequest(
+        amount="14",
+        remarks1="test1",
+        remarks2="test2",
+        prn="prn-1",
+        merchant_code="NBQM",
+        username="u",
+        password="p",
+    )
+    result = manager.initiate_payment(PaymentResult, request)
+
+    assert result.success is True
+    assert result.data.qr_message == "QRDATA"
+    assert session.last_kwargs["url"].endswith("thirdPartyDynamicQrDownload")
+    body = session.last_kwargs["json"]
+    assert len(body["dataValidation"]) == 128
+    assert body["merchantCode"] == "NBQM"
+
+
+def test_fonepay_verify_hmac_sha512_status():
+    status_response = {"fonepayTraceId": 123, "paymentStatus": "Paid", "prn": "prn-1"}
+    session = FakeSession([FakeResponse(status_response)])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.FONEPAY, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+
+    content = json.dumps({"prn": "prn-1", "merchantCode": "NBQM", "username": "u", "password": "p"})
+    result = manager.verify_payment(PaymentResult, content)
+
+    assert result.success is True
+    assert result.data.payment_status == "Paid"
+    assert len(session.last_kwargs["json"]["dataValidation"]) == 128
+
+
+def test_fonepay_tax_refund():
+    refund_response = {"fonepayTraceId": 35132, "message": "ok", "success": True}
+    session = FakeSession([FakeResponse(refund_response)])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.FONEPAY, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+
+    request = TaxRefundRequest(
+        fonepay_trace_id=35132,
+        transaction_amount="100",
+        merchant_prn="PRN",
+        invoice_number="INV",
+        invoice_date="2076.09.29",
+        merchant_code="NBQM",
+    )
+    result = manager.process_tax_refund(PaymentResult, request)
+
+    assert result.success is True
+    assert len(session.last_kwargs["json"]["dataValidation"]) == 128
+
+
+def test_fonepay_static_qr():
+    static_response = {"qrMessage": "STATICQRDATA"}
+    session = FakeSession([FakeResponse(static_response)])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.FONEPAY, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+
+    request = StaticQrRequest(
+        prn="prn-static",
+        merchant_code="NBQM",
+        username="u",
+        password="p",
+    )
+    result = manager.process_static_qr(PaymentResult, request)
+
+    assert result.success is True
+    assert result.data.qr_message == "STATICQRDATA"
+    assert session.last_kwargs["url"].endswith("thirdPartyStaticQrDownload")
+    body = session.last_kwargs["json"]
+    assert len(body["dataValidation"]) == 128
+    assert body["merchantCode"] == "NBQM"
+    assert body["prn"] == "prn-static"
+
+
+def test_fonepay_static_qr_rejects_wrong_type():
+    session = FakeSession([])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.FONEPAY, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+    with pytest.raises(TypeError):
+        manager.process_static_qr(PaymentResult, {})
+
+
+def test_static_qr_only_supported_for_fonepay():
+    session = FakeSession([])
+    api = ApiService(session=session)
+    manager = PaymentManager(PaymentMethod.KHALTI, PaymentMode.SANDBOX, SECRET_KEY, api=api)
+    with pytest.raises(Exception) as excinfo:
+        manager.process_static_qr(PaymentResult, StaticQrRequest(merchant_code="NBQM"))
+    assert "only supported for Fonepay" in str(excinfo.value)
+
+
+def test_convert_to_builds_dataclass_from_dict():
+
+    obj = convert_to(QrResponse, {"message": "hi", "success": True})
+    assert obj.message == "hi"
+    assert obj.success is True
+
+
+def test_manager_reuses_single_session_across_operations():
+    """The manager must not open a fresh connection per call (N+1 sessions).
+
+    When no ApiService is injected, PaymentManager lazily creates one shared
+    ApiService and reuses it (and its underlying requests.Session) for every
+    initiate/verify/tax-refund operation instead of building a new session per
+    call.
+    """
+    manager = PaymentManager(PaymentMethod.FONEPAY, PaymentMode.SANDBOX, SECRET_KEY)
+
+    api1 = manager._api_service()
+    api2 = manager._api_service()
+    assert api1 is api2, "manager must cache a single ApiService across calls"
+    # Both operations flow through the same shared session object.
+    service = get_payment_service(PaymentMethod.FONEPAY, SECRET_KEY, PaymentMode.SANDBOX, api=api1)
+    assert service._api_service is api1