Ignore:
Timestamp:
2 Apr 2024, 20:25:29 (9 months ago)
Author:
Henrik Bettermann
Message:

Implement Etranzact Credo platform payments.

Location:
main/kofacustom.nigeria/trunk/src/kofacustom/nigeria
Files:
2 added
5 edited

Legend:

Unmodified
Added
Removed
  • main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/etranzact/helpers.py

    r16167 r17730  
    2121import re
    2222from datetime import datetime
     23from ssl import SSLError
     24from datetime import datetime
    2325from urllib import urlencode
    2426from urllib2 import urlopen
     
    2830import json
    2931from zope.event import notify
     32from waeup.kofa.payments.interfaces import IPayer
    3033from waeup.kofa.utils.helpers import extract_formvars
    3134from kofacustom.nigeria.interfaces import MessageFactory as _
     
    282285    payment.payment_date = datetime.utcnow()
    283286    return True, msg, log
     287
     288
     289# Etranzact Credo payments helper functions
     290
     291def get_JSON_response_initialize(payment, host, callbackUrl, public_api_key):
     292    headers={
     293        'Content-Type':'application/JSON',
     294        'Authorization':public_api_key,
     295    }
     296    h = httplib.HTTPSConnection(host)
     297
     298    firstname = IPayer(payment).display_fullname.split()[0]
     299    lastname = IPayer(payment).display_fullname.split()[-1]
     300    email = IPayer(payment).email
     301    phone = IPayer(payment).phone
     302
     303    args = {
     304        'email': email,
     305        'amount': 100 * payment.amount_auth,
     306        'reference': payment.p_id,
     307        'currency': 'NGN',
     308        'callbackUrl': callbackUrl,
     309        'customerFirstName': firstname,
     310        'customerLastName': lastname,
     311        'customerPhoneNumber': phone,
     312        'bearer': '0',
     313        }
     314    try:
     315        h.request('POST', '/transaction/initialize',
     316                  body=json.dumps(args), headers=headers)
     317    except SSLError:
     318        return {'error': 'SSL handshake error'}
     319    response = h.getresponse()
     320    if response.status==400:
     321        jsonout = response.read()
     322        parsed_json = json.loads(jsonout)
     323        return {'error': parsed_json['error']}
     324    if response.status!=200:
     325        return {'error': 'Connection error (%s, %s)' % (response.status, response.reason)}
     326    jsonout = response.read()
     327    parsed_json = json.loads(jsonout)
     328
     329    # A typical JSON result
     330   
     331    # {u'status': 200,
     332    #  u'execTime': 5.109764,
     333    #  u'message': u'Successfully processed',
     334    #  u'data':
     335    #     {u'credoReference': u'vsb200B5oM0521Mb00og',
     336    #      u'reference': u'xyz',
     337    #      u'authorizationUrl': u'https://pay.credodemo.com/vsb200B5oM0521Mb00og'
     338    #      },
     339    #  u'error': []
     340    # }
     341
     342    return parsed_json
     343
     344
     345def initiate_payment(payment, host, callbackUrl, public_api_key):
     346    response = get_JSON_response_initialize(payment, host, callbackUrl, public_api_key)
     347    if response['status'] == 200:
     348        return True, response['data']['authorizationUrl']
     349    return False, response['message']
     350
     351def get_JSON_response_verify(transref, host, secret_api_key):
     352    headers={
     353        'Content-Type':'text/xml; charset=utf-8',
     354        'Authorization':secret_api_key,
     355    }
     356    h = httplib.HTTPSConnection(host)
     357    url = '/transaction/%s/verify' % transref
     358    try:
     359        h.request("GET", url, headers=headers)
     360    except SSLError:
     361        return {'error': 'SSL handshake error'}
     362    response = h.getresponse()
     363    if response.status==404:
     364        jsonout = response.read()
     365        parsed_json = json.loads(jsonout)
     366        return {'error': parsed_json['error']}
     367    if response.status!=200:
     368        return {'error': 'Connection error (%s, %s)' % (response.status, response.reason)}
     369    jsonout = response.read()
     370    parsed_json = json.loads(jsonout)
     371    return parsed_json
     372
     373
     374def query_credo_payment(payment, host, secret_api_key):
     375
     376    jr = get_JSON_response_verify(payment.p_id, host, secret_api_key)
     377    error = jr.get('error')
     378    if error:
     379        msg = log = error
     380        return False, msg, log
     381
     382    # A typical JSON result
     383
     384    #{u'status': 200,
     385    # u'execTime': 35.20671,
     386    # u'message': u'Pending bank credit notification',
     387    # u'data': {
     388    #   u'status': 12,
     389    #   u'businessCode': u'700607003930001',
     390    #   u'channelId': 0,
     391    #   u'businessRef': u'xyz',
     392    #   u'businessName': u'Igbinedion University Okada',
     393    #   u'settlementAmount': 100000.0,
     394    #   u'currencyCode': u'NGN',
     395    #   u'transFeeAmount': 1600.0,
     396    #   u'transRef': u'vsb200B5oM0521Mb00og',
     397    #   u'transAmount': 100000.0,
     398    #   u'debitedAmount': 101600.0,
     399    #   u'customerId': u'aa@aa.ng',
     400    #   u'statusMessage': u'Pending bank credit notification',
     401    #   u'metadata': []
     402    #  },
     403    # u'error': []
     404    #}
     405
     406    payment.r_code = unicode(jr['data']['status'])
     407    payment.r_desc = jr['data']['statusMessage']
     408    payment.r_amount_approved = jr['data']['debitedAmount']
     409    payment.r_pay_reference = jr['data'].get('transRef', u'')
     410    payment.r_payment_link = u'n/a'
     411    payment.r_card_num = u'n/a'
     412    if payment.r_code != '0':
     413        msg = _('Unsuccessful callback: ${a}', mapping = {'a': payment.r_desc})
     414        log = 'unsuccessful callback for %s payment %s: %s' % (
     415            payment.p_category, payment.p_id, payment.r_desc)
     416        payment.p_state = 'failed'
     417        notify(grok.ObjectModifiedEvent(payment))
     418        return False, msg, log
     419    payment.p_state = 'paid'
     420    payment.payment_date = datetime.utcnow()
     421    msg = _('Successful callback received')
     422    log = 'valid callback for %s payment %s: %s' % (
     423        payment.p_category, payment.p_id, str(jr))
     424    notify(grok.ObjectModifiedEvent(payment))
     425    return True, msg, log
     426
  • main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/etranzact/tests.py

    r16784 r17730  
    3333from kofacustom.nigeria.testing import FunctionalLayer
    3434from kofacustom.nigeria.etranzact.helpers import (
    35     query_history, query_payoutlet, ERROR_PART1, ERROR_PART2)
     35    query_history, query_payoutlet, ERROR_PART1, ERROR_PART2,
     36    )
    3637
    3738#from kofacustom.nigeria.etranzact.helpers import (query_etranzact)
     
    167168        return
    168169
     170
     171
    169172class EtranzactTestsApplicants(ApplicantsFullSetup):
    170173    """Tests for the Etranzact payment gateway.
     
    318321            % self.payment.p_id)
    319322        self.assertEqual(self.browser.contents, ERROR_PART1 + 'PAYEE_ID already used' + ERROR_PART2)
     323
     324# Credo tests
     325
     326from kofacustom.nigeria.etranzact.helpers import (
     327    query_credo_payment, get_JSON_response_initialize)
     328
     329SECRET_API_KEY = "0PRI0500LB11cBSSLcW0DcW1BcWgmf45"
     330API_PUBLIC_KEY = "0PUB0500wuoOe9sdtSOLj5peqHKc8Q9W"
     331EXTERNAL_TESTS_CREDO = True
     332CREDO_HOST = "api.credodemo.com"
     333
     334def external_test_credo(func):
     335    if not EXTERNAL_TESTS_CREDO:
     336        myself = __file__
     337        if myself.endswith('.pyc'):
     338            myself = myself[:-1]
     339        print "WARNING: external Credo tests are skipped!"
     340        print "WARNING: edit %s to enable them." % myself
     341        return
     342    return func
     343
     344class CredoTestsStudents(StudentsFullSetup):
     345
     346    layer = FunctionalLayer
     347
     348    P_ID = 'p7120340727304' # works only if such a payment was made on the test platform
     349
     350    def setUp(self):
     351        super(CredoTestsStudents, self).setUp()
     352        self.app['configuration']['2004'].etranzact_credo_enabled = True
     353        payment = createObject('waeup.StudentOnlinePayment')
     354        payment.p_id = self.P_ID
     355        payment.p_category = 'schoolfee'
     356        payment.amount_auth = 100000.0
     357        self.student['payments'][payment.p_id] = payment
     358        self.payment2 = payment
     359
     360    @external_test_credo
     361    def test_initialize(self):
     362
     363        public_api_key = API_PUBLIC_KEY
     364        callbackUrl = 'aa.aa.aa'
     365        response = get_JSON_response_initialize(self.payment2, CREDO_HOST, callbackUrl, public_api_key)
     366
     367        ## A typical response
     368
     369        #{u'status': 200,
     370        # u'execTime': 5.109764,
     371        # u'message': u'Successfully processed',
     372        # u'data':
     373        #    {u'credoReference': u'vsb200B5oM0521Mb00og',
     374        #     u'reference': u'xyz',
     375        #     u'authorizationUrl': u'https://pay.credodemo.com/vsb200B5oM0521Mb00og'
     376        #     },
     377        # u'error': []
     378        #}
     379
     380        # Payment already exists
     381        self.assertEqual(response, {'error': {u'reference': u'Reference must be unique'}})
     382
     383        #self.assertEqual(response['status'], 200)
     384        #self.assertEqual(response['message'], 'Successfully processed')
     385        #self.assertEqual(response['data']['reference'], self.P_ID)
     386        #self.assertTrue('https://pay.credodemo.com/' in response['data']['authorizationUrl'])
     387        return
     388
     389    @external_test_credo
     390    def test_verify(self):
     391        self.payment2.p_id = 'anything'
     392        success, msg, log = query_credo_payment(self.payment2, CREDO_HOST, SECRET_API_KEY)
     393        self.assertEqual(log, 'Transaction with ref# anything not found')
     394        self.payment2.p_id = self.P_ID # manually paid
     395        success, msg, log = query_credo_payment(self.payment2, CREDO_HOST, SECRET_API_KEY)
     396        self.assertEqual(msg, 'Successful callback received')
     397        self.assertEqual(self.payment2.r_code, '0')
     398        self.assertEqual(self.payment2.r_desc, 'Successfully processed')
     399        self.assertEqual(self.payment2.p_state, 'paid')
     400        return
     401
     402    @external_test_credo
     403    def test_student_credo_views(self):
     404        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
     405        self.browser.open(self.payments_path)
     406        self.browser.getLink(self.P_ID).click()
     407        self.browser.getLink("Requery Etranzact Credo").click()
     408        self.assertTrue('<span>Successfully processed</span>' in self.browser.contents)
     409        self.assertEqual(self.payment2.r_desc, 'Successfully processed')
     410        return
     411
     412class CredoTestsApplicants(ApplicantsFullSetup):
     413
     414    layer = FunctionalLayer
     415
     416    P_ID = 'p7120808915674' # works only if such a payment was made on the test platform
     417
     418    def setUp(self):
     419        super(CredoTestsApplicants, self).setUp()
     420        # Add session configuration object
     421        configuration = SessionConfiguration()
     422        configuration.academic_session =  self.applicant.__parent__.year
     423        configuration.etranzact_credo_enabled = True
     424        self.app['configuration'].addSessionConfiguration(configuration)
     425        payment = createObject('waeup.ApplicantOnlinePayment')
     426        payment.p_id = self.P_ID
     427        payment.p_category = 'application'
     428        payment.amount_auth = 5000.0
     429        self.applicant.email = 'xx@xx.xx'
     430        self.applicant[payment.p_id] = payment
     431        self.payment2 = payment
     432
     433    @external_test_credo
     434    def test_initialize(self):
     435
     436        public_api_key = API_PUBLIC_KEY
     437        callbackUrl = 'aa.aa.aa'
     438        response = get_JSON_response_initialize(self.payment2, CREDO_HOST, callbackUrl, public_api_key)
     439
     440        # Payment already exists
     441        self.assertEqual(response, {'error': {u'reference': u'Reference must be unique'}})
     442
     443        #self.assertEqual(response['status'], 200)
     444        #self.assertEqual(response['message'], 'Successfully processed')
     445        #self.assertEqual(response['data']['reference'], self.P_ID)
     446        #self.assertTrue('https://pay.credodemo.com/' in response['data']['authorizationUrl'])
     447        return
     448
     449    @external_test_credo
     450    def test_verify(self):
     451        self.payment2.p_id = 'anything'
     452        success, msg, log = query_credo_payment(self.payment2, CREDO_HOST, SECRET_API_KEY)
     453        self.assertEqual(log, 'Transaction with ref# anything not found')
     454        self.payment2.p_id = self.P_ID # manually paid
     455        success, msg, log = query_credo_payment(self.payment2, CREDO_HOST, SECRET_API_KEY)
     456        self.assertEqual(msg, 'Successful callback received')
     457        self.assertEqual(self.payment2.r_code, '0')
     458        self.assertEqual(self.payment2.r_desc, 'Successfully processed')
     459        self.assertEqual(self.payment2.p_state, 'paid')
     460        return
     461
     462    @external_test_credo
     463    def test_applicant_credo_views(self):
     464        IWorkflowState(self.applicant).setState('started')
     465        self.browser.addHeader('Authorization', 'Basic mgr:mgrpw')
     466        self.browser.open(self.view_path)
     467        self.browser.getLink(self.P_ID).click()
     468        self.browser.getLink("Requery Etranzact Credo").click()
     469        self.assertTrue('<span>Successfully processed</span>' in self.browser.contents)
     470        self.assertEqual(self.payment2.r_desc, 'Successfully processed')
     471        self.assertEqual(self.applicant.state, 'paid')
     472        return
  • main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/interfaces.py

    r17588 r17730  
    154154        )
    155155
     156    etranzact_credo_enabled = schema.Bool(
     157        title = _(u'Etranzact Credo integration enabled'),
     158        default = False,
     159        )
     160
    156161    etranzact_payoutlet_enabled = schema.Bool(
    157162        title = _(u'Etranzact Payoutlet integration enabled'),
  • main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/payments/interfaces.py

    r17246 r17730  
    4242
    4343    r_amount_approved = schema.Float(
    44         title = _(u'Response Amount Approved'),
     44        title = _(u'Response Amount Debited'),
    4545        default = 0.0,
    4646        required = False,
  • main/kofacustom.nigeria/trunk/src/kofacustom/nigeria/students/payments.py

    r15755 r17730  
    4848    @property
    4949    def student(self):
    50         return self.__parent__.__parent__
     50        try:
     51            return self.__parent__.__parent__
     52        except AttributeError:
     53            return None
    5154
    5255
Note: See TracChangeset for help on using the changeset viewer.