Skip to content

Kunto / backoffice

Go to a project
Commit d0ce9351 by Solo Group
Options

test_odeo

1 parent ebf37c00
Inline Side-by-side
Showing 4 changed files with 423 additions and 130 deletions
src/agratek/api/merchant/views/odeo.py
......@@ -11,9 +11,9 @@ import urllib3
urllib3.disable_warnings()
class Vendor(VendorClass):
def __init__(self, vendor_produk, bill_no, customer_id=None,
cust_trx_id=None, row=None):
cust_trx=None, row=None):
VendorClass.__init__(self, vendor_produk, bill_no, customer_id,
cust_trx_id, row)
cust_trx, row)
settings = get_settings()
self.mid = 'odeo_mid' in settings and settings['odeo_mid'] or None
self.key = 'odeo_key' in settings and settings['odeo_key'] or None
......@@ -25,32 +25,33 @@ class Vendor(VendorClass):
def request_post(self, params, url=None):
headers = {'Authorization': 'Bearer {key}'.format(key=self.auth)}
data = json.dumps(params)
url = url and self.url+url or self.url
self.url = url and self.url + url or self.url
resp = None
with requests.session():
resp = requests.post(url, data=data,
resp = requests.post(self.url, data=data,
verify=False, headers=headers)
return resp
def request_get(self, params, url=None):
headers = {'Authorization': 'Bearer {key}'.format(key=self.auth)}
url = url and self.url+url or self.url
self.url = url and self.url + url or self.url
with requests.session():
resp = requests.get(url, params=params,
resp = requests.get(self.url, params=params,
verify=False, headers=headers)
return resp
def inquiry(self):
# PLN ONLY
if not self.v_produk_kd or not self.bill_no:
return
b
return dict(code=9999,
message='Parameter tidak lengkap')
if self.v_produk_kd =='PLN' and self.v_produk_kd != 'PLNPASCA':
params = dict(
data=dict(
number=self.bill_no
)
)
self.request = params
self.request = json.dumps(params)
self.save_log("inquiry")
resp = self.request_get(params, '/prepaid/pln/inquiry')
else:
......@@ -58,118 +59,340 @@ class Vendor(VendorClass):
denom=self.v_produk_kd,
number=self.bill_no
)
self.request = params
self.request = json.dumps(params)
self.save_log("inquiry")
resp = self.request_get(params, '/postpaid/inquiry')
content = None
if resp and resp.text:
content = json.loads(resp.text)
content = resp.text
if resp.ok:
# todo mapping content
"""
{
"status": "SUCCESS",
"data": {
"number": "XXX",
"subscriber_id": "XXX",
"name": "XXX",
"tariff": "XXX",
"power": "XXX"
},
"message": ""
}"""
result = content
result = json.loads(content)
else:
result = dict(code=resp.status_code,
message=resp.text)
self.response = content
self.save_log('inquiry')
return result
return self.pars_data(result)
def payment(self):
if not self.v_produk_kd or not self.bill_no:
return
return dict(code=9999,
message='Parameter tidak lengkap')
params = dict(
data=dict(
denom=self.v_produk_kd,
number=self.bill_no
)
)
self.request = params
self.request = json.dumps(params)
self.save_log("payment")
resp = self.request_post(params, '/prepaid/purchase')
content = None
if resp and resp.text:
content = json.loads(resp.text)
content = resp.text
if resp.ok:
#todo mapping content
"""
{
"status": "SUCCESS",
"data": {
"order_id": 1,
"price": 5675
},
"message": ""
}
{
"status": "BAD_REQUEST",
"data": {
"errors": [
"ERROR_MESSAGE_1",
"ERROR_MESSAGE_2",
"ERROR_MESSAGE_ETC"
]
},
"message": ""
}
{
"status": "UNAUTHORIZED",
"data": {
"errors": [
"Unauthorized"
]
},
"message": ""
}
"""
result = content
result = json.loads(content)
else:
content = json.loads(resp.text)
result = dict(code=resp.status_code,
message=resp.text)
self.response = content
self.save_log('payment')
return result
return self.pars_data(result)
def advice(self):
if not self.v_produk_kd or not self.bill_no:
return
if not self.v_produk_kd or not self.bill_no or not self.partner_log_id:
return dict(code=9999,
message='Parameter tidak lengkap')
#todo: product category e-payment '/order/{order_id}'
if self.kategori == 'e-payment':
params = dict(
data=dict(
denom=self.v_produk_kd,
number=self.bill_no
order_id = self.partner_log.vend_trx
url = '/order/{order_id}'.format(order_id=order_id)
params = None
self.save_log("advice")
resp = self.request_get(params, url)
else:
params = dict(
data=dict(
denom=self.v_produk_kd,
number=self.bill_no
)
)
)
self.request = params
self.save_log("advice")
resp = self.request_get(params, '/prepaid/purchase-get')
content = None
if resp and resp.text:
content = json.loads(resp.text)
self.request = params
self.save_log("advice")
resp = self.request_get(params, '/prepaid/purchase-get')
content = None
content = resp.text
if resp.ok:
#todo mapping content
result = content
result = json.loads(content)
else:
result = dict(code=resp.status_code,
message=resp.text)
data = 'data' in result and result['data'] or None
if data:
self.vend_trx = 'order_id' in data and data['order_id'] or self.vend_trx
self.response = content
self.save_log('payment')
return result
self.save_log('advice')
return self.pars_data(result)
def pars_data(self, data):
# https://api.odeo.co.id/v1/affiliate/prepaid/purchase
"""
{
"data": {
"denom": "TSEL5",
"number": "0812121212"
}
}
Response
{
"status": "SUCCESS",
"data": {
"order_id": 1,
"price": 5675
},
"message": ""
}
{
"status": "BAD_REQUEST",
"data": {
"errors": [
"ERROR_MESSAGE_1",
"ERROR_MESSAGE_2",
"ERROR_MESSAGE_ETC"
]
},
"message": ""
}
{
"status": "UNAUTHORIZED",
"data": {
"errors": [
"Unauthorized"
]
},
"message": ""
}
"""
# GET https://api.odeo.co.id/v1/affiliate/prepaid/pln/inquiry
"""
Request
{
"data": {
"number": "0812121212"
}
}
Response
{
"status": "SUCCESS",
"data": {
"number": "XXX",
"subscriber_id": "XXX",
"name": "XXX",
"tariff": "XXX",
"power": "XXX"
},
"message": ""
}
"""
# GET https://api.odeo.co.id/v1/affiliate/postpaid/inquiry?denom=BPJSKES1&number=XXXX
"""
{
"status": "SUCCESS",
"data": {
"subscriber_id": "XXX",
"subscriber_name": "XXX",
"ref_id": "XXX",
"inquiries": {
"branch_code": "-",
"branch_name": "XXX",
"month_counts": 1,
"participant_counts": 3,
"base_price": 154200,
"bill_rest": 0,
"admin": 2500
},
"subtotal": 156700,
"discount": 600,
"total": 156100
},
"message": ""
}
"""
# GET https://api.odeo.co.id/v1/affiliate/postpaid/inquiry?denom=PLNPASCA&number=XXXX
"""
{
"status": "SUCCESS",
"data": {
"subscriber_id": "XXX",
"subscriber_name": "XXX",
"ref_id": "XXX",
"tariff": "R1/1300VA",
"inquiries": [
{
"period": "2016-08",
"base_price": 300000,
"fine": 0,
"admin_fee": 2500
}
],
"subtotal": 302500,
"discount": 900,
"total": 301600
},
"message": ""
}
"""
# GET https://api.odeo.co.id/v1/affiliate/postpaid/inquiry?denom=PAM...&number=XXXX
"""
{
"status": "SUCCESS",
"data": {
"subscriber_id": "XXX",
"subscriber_name": "XXX",
"ref_id": "XXX",
"inquiries": [
{
"period": "XXX",
"base_price": 300000,
"fine": 0,
"admin_fee": 2500,
"meter_changes": "XXX"
}
],
"subtotal": 302500,
"discount": 900,
"total": 301600
},
"message": ""
}
"""
# GET https://api.odeo.co.id/v1/affiliate/postpaid/inquiry?denom=PGN&number=XXXX
"""
{
"status": "SUCCESS",
"data": {
"subscriber_id": "XXX",
"subscriber_name": "XXX",
"ref_id": "XXX",
"inquiries": {
"period": "XXX",
"meter_changes": "XXX",
"usages": "XXX",
"base_price": 300000,
"admin_fee": 2500
},
"subtotal": 302500,
"discount": 900,
"total": 301600
},
"message": ""
}
"""
# Multi Finance
# GET https://api.odeo.co.id/v1/affiliate/postpaid/inquiry?denom=XXX&number=XXXX
"""
{
"status": "SUCCESS",
"data": {
"subscriber_id": "XXX",
"subscriber_name": "XXX",
"ref_id": "XXX",
"inquiries": {
"installment": "XXX",
"due_date": "XXX",
"platform": "XXX",
"base_price": 300000,
"coll_fee": 0,
"fine": 75000,
"admin": 2000
},
"subtotal": 377000,
"discount": 900,
"total": 376100
},
"message": ""
}
"""
# PLN Pasca
# GET https://api.odeo.co.id/v1/affiliate/postpaid/inquiry?denom=XXXX&number=XXXX
"""
{
"status": "SUCCESS",
"data": {
"subscriber_id": "XXX",
"subscriber_name": "XXX",
"ref_id": "XXX",
"inquiries": {
"period": "XXX",
"base_price": 300000,
"admin_fee": 1500
},
"subtotal": 301500,
"discount": 900,
"total": 300600
},
"message": ""
}
"""
"""
POST https://api.odeo.co.id/affiliate_url
{
"order_id": 1,
"status": "COMPLETED",
"message": "XXX"
"signature": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
}
sha256({order_id} + {MID} + {SECRET_KEY} + {status}) [COMPLETED/REFUNDED]
sha256({MID} + {SECRET_KEY} + {status}) [BROADCAST_NEW_PRICE]
{
"new_prices": {
"TSEL5": 5000
},
"status": "BROADCAST_NEW_PRICE",
"message": "XXX",
"signature": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
}
"""
# GET https://api.odeo.co.id/v1/affiliate/order/order_id (Cek status)
"""
{
"order_id": 123,
"name": "Telkomsel 5",
"msisdn": "08xx",
"serial_number": "XXX",
"status": "COMPLETED",
"price": 5000
}
"""
return {
"trx_id": self.partner_log_id,
"bill_no": self.bill_no,
"cust_trx": self.cust_trx,
"vend_trx": "order_id" in data and data["order_id"] or self.vend_trx,
"nama": "nama" in data and data["nama"] or "NO-NAME",
"pokok": "tagihan" in data and data["tagihan"] or 0,
"denda": "denda" in data and data["denda"] or 0,
"admin": "admin" in data and data["admin"] or 0,
"total": "total" in data and data["total"] or 0,
"ppn": "ppn" in data and data["ppn"] or 0,
"ref_no": "refnum" in data and data["refnum"] or "",
"jml_bulan": "bulan" in data and data["bulan"] or "",
"tgh_bulan": "blth" in data and data["blth"] or "",
"tarif": "tarif" in data and data["tarif"] or "",
"tgl_bayar" : "date" in data and data["date"] or "",
"info" : "info1" in data and data["info1"] or "",
"stand_meter" : "standmeter" in data and data["standmeter"] or "",
}
src/agratek/api/merchant/views/vendor.py
......@@ -52,7 +52,7 @@ def build_request(typ, values):
.filter(PartnerProduk.partner_id == vendor_id,
PartnerProduk.produk_id == produk_id).first()
if not vendor_product:
return dict(error='Data Tidak Ditemukan')
return dict(data = dict(error='Data Tidak Ditemukan'))
row = None
if "trx_id" in values and values['trx_id']:
......@@ -63,36 +63,48 @@ def build_request(typ, values):
cls_module = modules.Vendor(vendor_product, bill_no, customer_id,
cust_trx_id, row)
if typ=='inquiry':
result = hasattr(cls_module, typ) and cls_module.inquiry() or None
if result:
result["f_request"] = cls_module.request
result["f_response"] = cls_module.response
return result
elif typ=='payment':
result = cls_module.payment()
if result:
result["f_request"] = cls_module.request
result["f_response"] = cls_module.response
return result
elif typ == 'advice':
result = hasattr(cls_module, typ) and cls_module.advice() or None
if result :
result["f_request"] = cls_module.request
result["f_response"] = cls_module.response
return result
elif typ == 'info':
result = hasattr(cls_module, typ) and cls_module.info() or None
if result:
result["f_request"] = cls_module.request
result["f_response"] = cls_module.reesponse
return result
cls = hasattr(cls_module, typ) and getattr(cls_module, typ) or None
if cls:
data = cls()
result = dict(data=data)
result["f_request"] = cls_module.request
result["f_response"] = cls_module.response
result["url"] = cls_module.url
else:
return dict(error='Function %s tidak ditemukan' % typ)
result = dict(data=dict(error='Fungsi %s tidak ada' % typ,
code=9999))
return result
# if typ=='inquiry':
# data = cls_module.inquiry()
# result = dict(data=data)
# result["f_request"] = cls_module.request
# result["f_response"] = cls_module.response
# return result
#
# elif typ=='payment':
# data = cls_module.payment()
# result = dict(data=data)
# result["f_request"] = cls_module.request
# result["f_response"] = cls_module.response
# return result
#
# elif typ == 'advice':
# data = hasattr(cls_module, typ) and cls_module.advice() or None
# result = dict(data=data)
# result["f_request"] = cls_module.request
# result["f_response"] = cls_module.response
# return result
#
# elif typ == 'info':
# data = hasattr(cls_module, typ) and cls_module.info() or None
# result = dict(data=data)
# result["f_request"] = cls_module.request
# result["f_response"] = cls_module.reesponse
# return result
# else:
# return dict(error='Function %s tidak ditemukan' % typ)
@colander.deferred
......@@ -284,7 +296,9 @@ class ViewHome(BaseView):
values = dict(c.items())
# if 'trx_id' not in values or not values['trx_id']:
# values['trx_id'] = get_random_number(10)
result = []
# result = build_request('inquiry', values)
#
result = None
if 'inquiry' in request.POST:
result = build_request('inquiry', values)
elif 'payment' in request.POST:
......@@ -296,16 +310,23 @@ class ViewHome(BaseView):
elif 'info' in request.POST:
result = build_request('info', values)
if not result:
session.flash('Data Tidak Ditemukan', 'error')
# if not result:
# session.flash('Data Tidak Ditemukan', 'error')
if result and 'error' in result:
session.flash(result['error'], 'error')
if result and 'data' in result:
data = result['data']
values['f_request'] = 'f_request' in result \
and json.dumps(result['f_request'], indent=4) or ''
values['f_response'] = 'f_response' in result \
and json.dumps(result['f_response'], indent=4) or ''
else:
values['result'] = json.dumps(result, indent=4)
values['ref_no'] = 'ref_no' in result and result["ref_no"] or None
values['trx_id'] = 'trx_id' in result and result['trx_id'] or None
values['f_result'] = 'data' in result \
and json.dumps(result['data'], indent=4) or ''
values['ref_no'] = 'ref_no' in data and data["ref_no"] or ""
values['trx_id'] = 'trx_id' in data and data['trx_id'] or ""
values['vend_trx'] = 'vend_trx' in data and data['vend_trx'] or ""
if 'error' in data and data['error']:
session.flash(data['error'], 'error')
form.render(values)
return dict(form=form, params=form_params_view)
......@@ -314,7 +335,7 @@ class ViewHome(BaseView):
class VendorClass(object):
def __init__(self, vendor_produk, bill_no, customer_id=None,
cust_trx_id=None, row=None):
cust_trx=None, row=None):
"""
:param cid:
merupakan customer id seperti MSN dan atu id pelanggaan
......@@ -325,11 +346,12 @@ class VendorClass(object):
self.produk_id = vendor_produk.produk_id
self.bill_no = bill_no
self.customer_id = customer_id
self.cust_trx_id = cust_trx_id
self.cust_trx = cust_trx
self.vend_trx = None
self.response = None
self.request = None
self.partner_log = row
self.partner_log_id = None
self.partner_log_id = row and row.id or None
self.v_produk_kd = vendor_produk.kode
self.kategori = vendor_produk.produk.kategori.nama
......@@ -340,7 +362,8 @@ class VendorClass(object):
row.produk_id = self.produk_id
row.bill_no = self.bill_no
row.customer_id = self.customer_id
row.cust_trx_id = self.cust_trx_id
row.cust_trx = self.cust_trx
row.vend_trx = self.vend_trx
if typ == 'inquiry':
row.inquiry = dict(request=self.request,
response=self.response)
......
test_odeo.py 0 → 100644
import unittest
import transaction
from agratek.api.merchant.views.vendor import build_request
from opensipkd.base.models import Partner
from opensipkd.pasar.models import PartnerProduk, Produk
from pyramid.paster import get_appsettings
from pyramid_rpc.jsonrpc import JsonRpcError
from pyramid import testing
class TestOdeo(unittest.TestCase):
def setUp(self):
settings = get_appsettings('test.ini', name='main')
self.config = testing.setUp(settings=settings)
# self.config.include('..models')
# self.config.include('..routes')
#session_factory = self.config.registry['dbsession_factory']
#self.session = get_tm_session(session_factory, transaction.manager)
#
# self.config = testing.setUp()
# from pyramid.paster import get_appsettings
from opensipkd.base import main
app = main({}, **settings)
def test_vendor(self):
"""
Test that it can sum a list of integers
"""
produk_kd = 'TS-5'
bill_no = '081311045668'
partner_kd = 'ODEO'
vendor = Partner.query_kode(partner_kd).first()
produk = Produk.query_kode(produk_kd).first()
if not vendor or not produk:
raise JsonRpcError(message="Vendor Not Found")
values = dict(vendor_id = vendor.id,
produk_id = produk.id,
destination=bill_no)
result = build_request('payment', values)
for r in result:
print(r,":", result[r])
if __name__ == '__main__':
unittest.main()
\ No newline at end of file
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!