Changeset 17730 for main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/etranzact

Timestamp:

2 Apr 2024, 20:25:29 (6 months ago)

Author:

Henrik Bettermann

Message:

Implement Etranzact Credo platform payments.

Location:

main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/etranzact

Files:

• browser_templates/goto_credo.pt (added)
• credobrowser.py (added)
• helpers.py (modified) (3 diffs)
• tests.py (modified) (3 diffs)

main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/etranzact/helpers.py

@@ -21 +21 @@
 import re
 from datetime import datetime
+from ssl import SSLError
+from datetime import datetime
 from urllib import urlencode
 from urllib2 import urlopen
@@ -28 +30 @@
 import json
 from zope.event import notify
+from waeup.kofa.payments.interfaces import IPayer
 from waeup.kofa.utils.helpers import extract_formvars
 from kofacustom.nigeria.interfaces import MessageFactory as _
@@ -282 +285 @@
     payment.payment_date = datetime.utcnow()
     return True, msg, log
+
+
+# Etranzact Credo payments helper functions
+
+def get_JSON_response_initialize(payment, host, callbackUrl, public_api_key):
+    headers={
+        'Content-Type':'application/JSON',
+        'Authorization':public_api_key,
+    }
+    h = httplib.HTTPSConnection(host)
+
+    firstname = IPayer(payment).display_fullname.split()[0]
+    lastname = IPayer(payment).display_fullname.split()[-1]
+    email = IPayer(payment).email
+    phone = IPayer(payment).phone
+
+    args = {
+        'email': email,
+        'amount': 100 * payment.amount_auth,
+        'reference': payment.p_id,
+        'currency': 'NGN',
+        'callbackUrl': callbackUrl,
+        'customerFirstName': firstname,
+        'customerLastName': lastname,
+        'customerPhoneNumber': phone,
+        'bearer': '0',
+        }
+    try:
+        h.request('POST', '/transaction/initialize',
+                  body=json.dumps(args), headers=headers)
+    except SSLError:
+        return {'error': 'SSL handshake error'}
+    response = h.getresponse()
+    if response.status==400:
+        jsonout = response.read()
+        parsed_json = json.loads(jsonout)
+        return {'error': parsed_json['error']}
+    if response.status!=200:
+        return {'error': 'Connection error (%s, %s)' % (response.status, response.reason)}
+    jsonout = response.read()
+    parsed_json = json.loads(jsonout)
+
+    # A typical JSON result
+    
+    # {u'status': 200,
+    #  u'execTime': 5.109764,
+    #  u'message': u'Successfully processed',
+    #  u'data':
+    #     {u'credoReference': u'vsb200B5oM0521Mb00og',
+    #      u'reference': u'xyz',
+    #      u'authorizationUrl': u'https://pay.credodemo.com/vsb200B5oM0521Mb00og'
+    #      },
+    #  u'error': []
+    # }
+
+    return parsed_json
+
+
+def initiate_payment(payment, host, callbackUrl, public_api_key):
+    response = get_JSON_response_initialize(payment, host, callbackUrl, public_api_key)
+    if response['status'] == 200:
+        return True, response['data']['authorizationUrl']
+    return False, response['message']
+
+def get_JSON_response_verify(transref, host, secret_api_key):
+    headers={
+        'Content-Type':'text/xml; charset=utf-8',
+        'Authorization':secret_api_key,
+    }
+    h = httplib.HTTPSConnection(host)
+    url = '/transaction/%s/verify' % transref
+    try:
+        h.request("GET", url, headers=headers)
+    except SSLError:
+        return {'error': 'SSL handshake error'}
+    response = h.getresponse()
+    if response.status==404:
+        jsonout = response.read()
+        parsed_json = json.loads(jsonout)
+        return {'error': parsed_json['error']}
+    if response.status!=200:
+        return {'error': 'Connection error (%s, %s)' % (response.status, response.reason)}
+    jsonout = response.read()
+    parsed_json = json.loads(jsonout)
+    return parsed_json
+
+
+def query_credo_payment(payment, host, secret_api_key):
+
+    jr = get_JSON_response_verify(payment.p_id, host, secret_api_key)
+    error = jr.get('error')
+    if error:
+        msg = log = error
+        return False, msg, log
+
+    # A typical JSON result
+
+    #{u'status': 200,
+    # u'execTime': 35.20671,
+    # u'message': u'Pending bank credit notification',
+    # u'data': {
+    #   u'status': 12,
+    #   u'businessCode': u'700607003930001',
+    #   u'channelId': 0,
+    #   u'businessRef': u'xyz',
+    #   u'businessName': u'Igbinedion University Okada',
+    #   u'settlementAmount': 100000.0,
+    #   u'currencyCode': u'NGN',
+    #   u'transFeeAmount': 1600.0,
+    #   u'transRef': u'vsb200B5oM0521Mb00og',
+    #   u'transAmount': 100000.0,
+    #   u'debitedAmount': 101600.0,
+    #   u'customerId': u'aa@aa.ng',
+    #   u'statusMessage': u'Pending bank credit notification',
+    #   u'metadata': []
+    #  },
+    # u'error': []
+    #}
+
+    payment.r_code = unicode(jr['data']['status'])
+    payment.r_desc = jr['data']['statusMessage']
+    payment.r_amount_approved = jr['data']['debitedAmount']
+    payment.r_pay_reference = jr['data'].get('transRef', u'')
+    payment.r_payment_link = u'n/a'
+    payment.r_card_num = u'n/a'
+    if payment.r_code != '0':
+        msg = _('Unsuccessful callback: ${a}', mapping = {'a': payment.r_desc})
+        log = 'unsuccessful callback for %s payment %s: %s' % (
+            payment.p_category, payment.p_id, payment.r_desc)
+        payment.p_state = 'failed'
+        notify(grok.ObjectModifiedEvent(payment))
+        return False, msg, log
+    payment.p_state = 'paid'
+    payment.payment_date = datetime.utcnow()
+    msg = _('Successful callback received')
+    log = 'valid callback for %s payment %s: %s' % (
+        payment.p_category, payment.p_id, str(jr))
+    notify(grok.ObjectModifiedEvent(payment))
+    return True, msg, log
+

main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/etranzact/tests.py

@@ -33 +33 @@
 from kofacustom.nigeria.testing import FunctionalLayer
 from kofacustom.nigeria.etranzact.helpers import (
-    query_history, query_payoutlet, ERROR_PART1, ERROR_PART2)
+    query_history, query_payoutlet, ERROR_PART1, ERROR_PART2,
+    )
 
 #from kofacustom.nigeria.etranzact.helpers import (query_etranzact)
@@ -167 +168 @@
         return
 
+
+
 class EtranzactTestsApplicants(ApplicantsFullSetup):
     """Tests for the Etranzact payment gateway.
@@ -318 +321 @@
             % self.payment.p_id)
         self.assertEqual(self.browser.contents, ERROR_PART1 + 'PAYEE_ID already used' + ERROR_PART2)
+
+# Credo tests
+
+from kofacustom.nigeria.etranzact.helpers import (
+    query_credo_payment, get_JSON_response_initialize)
+
+SECRET_API_KEY = "0PRI0500LB11cBSSLcW0DcW1BcWgmf45"
+API_PUBLIC_KEY = "0PUB0500wuoOe9sdtSOLj5peqHKc8Q9W"
+EXTERNAL_TESTS_CREDO = True
+CREDO_HOST = "api.credodemo.com"
+
+def external_test_credo(func):
+    if not EXTERNAL_TESTS_CREDO:
+        myself = __file__
+        if myself.endswith('.pyc'):
+            myself = myself[:-1]
+        print "WARNING: external Credo tests are skipped!"
+        print "WARNING: edit %s to enable them." % myself
+        return
+    return func
+
+class CredoTestsStudents(StudentsFullSetup):
+
+    layer = FunctionalLayer
+
+    P_ID = 'p7120340727304' # works only if such a payment was made on the test platform
+
+    def setUp(self):
+        super(CredoTestsStudents, self).setUp()
+        self.app['configuration']['2004'].etranzact_credo_enabled = True
+        payment = createObject('waeup.StudentOnlinePayment')
+        payment.p_id = self.P_ID
+        payment.p_category = 'schoolfee'
+        payment.amount_auth = 100000.0
+        self.student['payments'][payment.p_id] = payment
+        self.payment2 = payment
+
+    @external_test_credo
+    def test_initialize(self):
+
+        public_api_key = API_PUBLIC_KEY
+        callbackUrl = 'aa.aa.aa'
+        response = get_JSON_response_initialize(self.payment2, CREDO_HOST, callbackUrl, public_api_key)
+
+        ## A typical response
+
+        #{u'status': 200,
+        # u'execTime': 5.109764,
+        # u'message': u'Successfully processed',
+        # u'data':
+        #    {u'credoReference': u'vsb200B5oM0521Mb00og',
+        #     u'reference': u'xyz',
+        #     u'authorizationUrl': u'https://pay.credodemo.com/vsb200B5oM0521Mb00og'
+        #     },
+        # u'error': []
+        #}
+
+        # Payment already exists
+        self.assertEqual(response, {'error': {u'reference': u'Reference must be unique'}})
+
+        #self.assertEqual(response['status'], 200)
+        #self.assertEqual(response['message'], 'Successfully processed')
+        #self.assertEqual(response['data']['reference'], self.P_ID)
+        #self.assertTrue('https://pay.credodemo.com/' in response['data']['authorizationUrl'])
+        return
+
+    @external_test_credo
+    def test_verify(self):
+        self.payment2.p_id = 'anything'
+        success, msg, log = query_credo_payment(self.payment2, CREDO_HOST, SECRET_API_KEY)
+        self.assertEqual(log, 'Transaction with ref# anything not found')
+        self.payment2.p_id = self.P_ID # manually paid
+        success, msg, log = query_credo_payment(self.payment2, CREDO_HOST, SECRET_API_KEY)
+        self.assertEqual(msg, 'Successful callback received')
+        self.assertEqual(self.payment2.r_code, '0')
+        self.assertEqual(self.payment2.r_desc, 'Successfully processed')
+        self.assertEqual(self.payment2.p_state, 'paid')
+        return
+
+    @external_test_credo
+    def test_student_credo_views(self):
+        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
+        self.browser.open(self.payments_path)
+        self.browser.getLink(self.P_ID).click()
+        self.browser.getLink("Requery Etranzact Credo").click()
+        self.assertTrue('<span>Successfully processed</span>' in self.browser.contents)
+        self.assertEqual(self.payment2.r_desc, 'Successfully processed')
+        return
+
+class CredoTestsApplicants(ApplicantsFullSetup):
+
+    layer = FunctionalLayer
+
+    P_ID = 'p7120808915674' # works only if such a payment was made on the test platform
+
+    def setUp(self):
+        super(CredoTestsApplicants, self).setUp()
+        # Add session configuration object
+        configuration = SessionConfiguration()
+        configuration.academic_session =  self.applicant.__parent__.year
+        configuration.etranzact_credo_enabled = True
+        self.app['configuration'].addSessionConfiguration(configuration)
+        payment = createObject('waeup.ApplicantOnlinePayment')
+        payment.p_id = self.P_ID
+        payment.p_category = 'application'
+        payment.amount_auth = 5000.0
+        self.applicant.email = 'xx@xx.xx'
+        self.applicant[payment.p_id] = payment
+        self.payment2 = payment
+
+    @external_test_credo
+    def test_initialize(self):
+
+        public_api_key = API_PUBLIC_KEY
+        callbackUrl = 'aa.aa.aa'
+        response = get_JSON_response_initialize(self.payment2, CREDO_HOST, callbackUrl, public_api_key)
+
+        # Payment already exists
+        self.assertEqual(response, {'error': {u'reference': u'Reference must be unique'}})
+
+        #self.assertEqual(response['status'], 200)
+        #self.assertEqual(response['message'], 'Successfully processed')
+        #self.assertEqual(response['data']['reference'], self.P_ID)
+        #self.assertTrue('https://pay.credodemo.com/' in response['data']['authorizationUrl'])
+        return
+
+    @external_test_credo
+    def test_verify(self):
+        self.payment2.p_id = 'anything'
+        success, msg, log = query_credo_payment(self.payment2, CREDO_HOST, SECRET_API_KEY)
+        self.assertEqual(log, 'Transaction with ref# anything not found')
+        self.payment2.p_id = self.P_ID # manually paid
+        success, msg, log = query_credo_payment(self.payment2, CREDO_HOST, SECRET_API_KEY)
+        self.assertEqual(msg, 'Successful callback received')
+        self.assertEqual(self.payment2.r_code, '0')
+        self.assertEqual(self.payment2.r_desc, 'Successfully processed')
+        self.assertEqual(self.payment2.p_state, 'paid')
+        return
+
+    @external_test_credo
+    def test_applicant_credo_views(self):
+        IWorkflowState(self.applicant).setState('started')
+        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
+        self.browser.open(self.view_path)
+        self.browser.getLink(self.P_ID).click()
+        self.browser.getLink("Requery Etranzact Credo").click()
+        self.assertTrue('<span>Successfully processed</span>' in self.browser.contents)
+        self.assertEqual(self.payment2.r_desc, 'Successfully processed')
+        self.assertEqual(self.applicant.state, 'paid')
+        return