zoukankan      html  css  js  c++  java
  • python3 Mock开发

    直接举例

    请求报文:

     "request_headers" : "{"Content-Type":["application/x-www-form-urlencoded"]}", "request_info" : "uploadData={"data":"7174DBB5C979DE2827BA76ABBE318555D16C1355C79815","token":"752145eafdc023b"}" 

    对应接口mock:

      1 import datetime
      2 import json
      3 import re
      4 import pymongo
      5 import binascii
      6 from Crypto.Cipher import DES
      7 
      8 from flask import Flask, request, Response
      9 
     10 app = Flask(__name__)
     11 
     12 app.config['JSON_AS_ASCII'] = False
     13 
     14 
     15 def des_decode(data, key):
     16     """aes解密
     17     :param key:
     18     :param data:
     19     """
     20     cipher = DES.new(key)
     21     result2 = binascii.a2b_hex(data)  # 十六进制还原成二进制
     22     decrypted = cipher.decrypt(result2)
     23 
     24     return decrypted
     25 
     26 
     27 def mongodb_connect():
     28     """创建数据库及集合
     29     """
     30     myclient = pymongo.MongoClient("mongodb://1.1.1.1:20/")
     31     mydb = myclient["test"]
     32     mycol = mydb["test"]
     33 
     34     return mycol
     35 
     36 
     37 def mongodb_update_apply(*args):
     38     """根据申请信息更新表中的数据
     39     """
     40     mycol = mongodb_connect()
     41     mycol.update({"tran_no": args[0]},
     42                  {"$set": {"code": args[1], "data": args[2], "msg": args[3], "apply_status": args[4]}})
     43 
     44 
     45 def mongodb_update_verification(*args):
     46     """根据确认信息更新表中的数据
     47     """
     48     mycol = mongodb_connect()
     49     mycol.update({"tran_no": args[0]},
     50                  {"$set": {"code": args[1], "msg": args[2], "apply_status": args[3]}})
     51 
     52 
     53 def apply_info(tran_no, code, data, msg, apply_status):
     54     """申请响应结果
     55     :param tran_no:
     56     :param msg:
     57     :param data:
     58     :param code:
     59     :param apply_status:
     60     """
     61     result = {
     62         "code": code,
     63         "data": data,
     64         "msg": msg
     65     }
     66 
     67     mongodb_update_apply(tran_no, code, data, msg, apply_status)
     68 
     69     return result
     70 
     71 
     72 
     73 @app.route('/bxdb/xmbksq', methods=['POST', 'GET'])
     74 def bxdb_xmbksq():
     75     """签约接口"""
     76     json_data = request.form
     77     data = json.loads(json_data['uploadData'])['data']
     78     data1 = des_decode(data, '123456')
     79     data2 = re.findall('({.+})', data1.decode('utf-8'))
     80     request_data = json.loads(data2[0])
     81 
     82     tran_no = request_data['tranno']
     83     name = request_data['name']
     84     id_no = request_data['idNo']
     85     bank_card_no = request_data['bankCardNo']
     86     mobile = request_data['mobile']
     87 
     88     date_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
     89 
     90     mycol = mongodb_connect()
     91 
     92     mongodb_result = mycol.find_one(
     93         {"id_no": id_no, "name": name, "bank_card_no": bank_card_no, "mobile": mobile})
     94 
     95     if mongodb_result:
     96         apply_status = mongodb_result['apply_status']
     97         if apply_status != 'COMPLETED':
     98             mycol.update(
     99                 {{"id_no": id_no, "name": name, "bank_card_no": bank_card_no, "mobile": mobile}},
    100                 {"$set": {"tran_no": tran_no}})
    101             result = apply_info(tran_no, '0', '01', '交易成功', 'SUBMITTED')
    102         else:
    103             result = apply_info(tran_no, '0', '02', '该卡已签约绑定', 'COMPLETED')
    104     else:
    105         mydict = {"tran_no": tran_no, "id_no": id_no, "name": name, "bank_card_no": bank_card_no, "mobile": mobile,
    106                   "last_modified_time": datetime.datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S'),
    107                   "created_time": datetime.datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S')}
    108         mycol.insert_one(mydict)
    109 
    110         if len(mobile) > 11:
    111             result = apply_info(tran_no, '1', '', '银行预留手机号不正确!', 'FAILED')
    112         else:
    113             result = apply_info(tran_no, '0', '01', '交易成功', 'SUBMITTED')
    114 
    115     response = Response(json.dumps(result, ensure_ascii=False), mimetype='application/json')
    116 
    117     return response
    118 
    119 
    120 if __name__ == '__main__':
    121     app.run(host='0.0.0.0', port=9015)

    以上mock涉及MongoDB操作、DES解密(参考之前写的文章)、表单数据的接收及发送。其中表单接收时需要注意根据编码类型的不同,处理的方式也不一样:

    application/x-www-form-urlencoded对应request.form

    application/xml或text/xml对应request.data

    application/json对应request.json

    发送表单信息的时候,我们也可以设置编码类型,通过Response的mimetype参数设置:

    Response(json.dumps(result, ensure_ascii=False), mimetype='application/json')
  • 相关阅读:
    OSX中zsh新增环境变量
    新的开始 春光明媚
    tmux
    继承
    6
    Object类
    网页收藏
    画王八
    ES6 语法之import export
    ES6 语法 之 destructuring
  • 原文地址:https://www.cnblogs.com/sammy1989/p/9838220.html
Copyright © 2011-2022 走看看