需求背景：

在测试过程中，遇到电商审核项目重构，此时包含以下测试点：

1、进审对比json数据的一致性（包括各个方面，对比数据量大）

2、空跑数据对比数据库

3、普通case覆盖

那么问题来了，其实我们本身可以人工diff json，但是对于这种送审数据量比较大的情况，对比起来就十分的不方便。而且同一个字段可能存在多个位置。就很容易出错。此时，我们需要写一个脚本来对比json，为了后期的测试数据好整理并且有迹可循，我们记录在log日志中。

那么我们开始喽~~~~~~~~~

具体实现逻辑：

首先，逻辑如下：

1、去数据库中读取自己要对比的每个进审任务的id

2、拼接成url，版本1的url和版本2的url

3、对两个url进行请求，获取返回的json，写入文件。

4、对这两个json进行diff

5、引入logger，打印日志

不知道大家对审核业务是否有了解，审核存在着不同的队列，每个队列存在不同的任务id，业务上存在两个队列（就是代码中的project），一个是python，一个是go，现想用go的队列代替python的队列，拿着任务id和队列id拼接出来的url可以直接拿到该任务的详情信息等。也就是进审的数据。这是我们需要diff的。

代码实现：

1、目录：

根目录为jsondiff

jsonfile文件夹：主要是放不同版本返回的数据

logs文件夹：放打印出来的log

get_tcs_request.py：这个就是主要的逻辑实现

loger.py：日志方法的封装，可以循环利用。

2、具体代码：

get_tcs_request.py

import requests
import json
import json_tools
from urllib.parse import urlencode
from leecode.jsondiff.loger import Logger
import pymysql
import jsonpathbase_url = "https://tcs-bos.bytedance.net/api/v2/task_info/?"
cookie = {
'initLang':'zh-CN',
'initLang':'zh-CN',
'initLang':'zh-CN',   #此处省略，用自己的cookie奥～}class TCSGetResp_Diff():def __init__(self):self.logger = Logger(__name__)def  make_url(self,project,base_url,task_id):param = {"project_id": "project_id","work_mode": "scan","task_id": task_id,}if project == "GO":param["project_id"] = 6966108025115427000result = urlencode(param)self.logger.get_log().info('go_url:%s' % base_url+result)return base_url+resultelif project == "PYTHON":param["project_id"] = 6921223155105137000result = urlencode(param)self.logger.get_log().info('python_url:%s' % base_url + result)return base_url + resultelse:return "not go and python"def get_response(self,url):response = requests.get(url=url,cookies=cookie,verify = False)json_str = json.dumps(response.json(), indent=2, sort_keys=False, ensure_ascii=False)return json_strdef write_file(self,json_str,project):jsonobj = json.loads(json_str)data = jsonpath.jsonpath(jsonobj, '$..cell_list')if data is None:self.logger.get_log().info('project:%s返回值为空' % project)else:if project == "GO":self.logger.get_log().info('json_str_go:%s' % json_str)with open('/Users/dongyue/Desktop/code/leecode/jsondiff/json_file/json_go.txt', 'w') as f:f.write(json_str)if project == "PYTHON":self.logger.get_log().info('json_str_python:%s' % json_str)with open('/Users/dongyue/Desktop/code/leecode/jsondiff/json_file/json_python.txt', 'w') as f:f.write(json_str)def diff_json(self):go = open('/Users/dongyue/Desktop/code/leecode/jsondiff/json_file/json_go.txt', encoding='UTF-8')python = open('/Users/dongyue/Desktop/code/leecode/jsondiff/json_file/json_python.txt', encoding='UTF-8')#以python为基准，对比go的操作。后面的为python的result = json_tools.diff(json.load(python), json.load(go))self.logger.get_log().info('diff_result:%s' %result)class DbFunc():def __init__(self,host,port,user,pwd,name):self.host = hostself.port  = portself.user = userself.pwd = pwdself.name = nameself.sql = ""self.conn = Noneself.logger = Logger(__name__)def connectsql(self):# 链接数据库try:self.conn = pymysql.connect(self.host, user=self.user, passwd=self.pwd, db=self.name)cursor = self.conn.cursor()print("数据库链接成功")return cursorexcept pymysql.Error as e:self.logger.get_log().info(e)def getall(self,sql):try:cur = self.connectsql()cur.execute(sql)return cur.fetchall()except pymysql.Error as e:self.logger.get_log().info('sql为:%s,execute failed.reason:%s' % (sql,e))if __name__ == '__main__':host = "10.231.11.111"port = "3307"user = "ecom"password = "*************_kunfnNTw8QtoBeJK"db = "ecom_global_audit_boe"#最新的任务在最上面，早的在最下面sql_go ="select tcs_task_id from t_live_slice_task where room_id = 6975015238768872197 order by start_time_stamp desc;"sql_python ="select task_id from t_room_shift_task where room_id=6975015238768872197 order by start_time_stamp desc;"cursor = DbFunc(host,port,user,password,db)list_go = cursor.getall(sql_go)print("go:",list_go)list_python = cursor.getall(sql_python)print("python:", list_python)#遍历list找到两个队列对应的tcs_task_id进行difffor index in range(len(list_go)):tcs_obj = TCSGetResp_Diff()go_url = tcs_obj.make_url("GO",base_url,list_go[index][0])python_url = tcs_obj.make_url("PYTHON",base_url,list_python[index][0])json_str_go = tcs_obj.get_response(go_url)json_str_python = tcs_obj.get_response(python_url)tcs_obj.write_file(json_str_go,"GO")tcs_obj.write_file(json_str_python, "PYTHON")tcs_obj.diff_json()

loger.py

# -*- coding:utf-8 -*-
import logging
import oslog_path = './jsondiff/logs/log'class Logger:def __init__(self, loggername):# 创建一个loggerself.logger = logging.getLogger(loggername)self.logger.setLevel(logging.DEBUG)#每次清除下已经有的handler。 不然会重复打印之前handler的日志self.logger.handlers.clear()# 创建一个handler，用于写入日志文件# log_path = os.path.dirname(getcwd.get_cwd())+"/logs/" # 指定文件输出路径，注意logs是个文件夹，一定要加上/，不然会导致输出路径错误，把logs变成文件名的一部分了logname = log_path + '.log'  # 指定输出的日志文件名fh = logging.FileHandler(logname, encoding='utf-8')  # 指定utf-8格式编码，避免输出的日志文本乱码fh.setLevel(logging.DEBUG)# 创建一个handler，用于将日志输出到控制台ch = logging.StreamHandler()ch.setLevel(logging.DEBUG)# 定义handler的输出格式formatter = logging.Formatter('%(asctime)s-%(name)s-%(levelname)s-%(message)s')fh.setFormatter(formatter)ch.setFormatter(formatter)# 给logger添加handlerself.logger.addHandler(fh)self.logger.addHandler(ch)# def get_log(self):#   """定义一个函数，回调logger实例"""#     return self.loggerdef get_log(self):return self.loggerif __name__ == '__main__':t = Logger("hmk").get_log().debug("User %s is loging" % 'jeck')

log.log

执行后得到的log文件，就是下面的格式。

2021-06-20 21:16:34,285-__main__-INFO-go_url:https://tcs-bos.bytedance.net/api/v2/task_info/?project_id=6966108025115427000&work_mode=scan&task_id=0
2021-06-20 21:16:34,286-__main__-INFO-python_url:https://tcs-bos.bytedance.net/api/v2/task_info/?project_id=6921223155105137000&work_mode=scan&task_id=6975114769346888000
2021-06-20 21:16:49,161-__main__-INFO-json_str_go:{"code": 0,"message": "success","data": []
}
2021-06-20 21:16:49,165-__main__-INFO-json_str_python:{"code": 0,"message": "success","data": [2]
}
2021-06-20 21:16:49,167-__main__-INFO-diff_result:[{'remove': '/data/0', 'prev': {'id': '6975114769346888198', 'task_id': '6975114769346888198', 'project_id': '6921223155105137157', 'object_id': '6975015238768872197_1624020544', 'batch_id': None, 'priority': 1586151632, 'status': 4, 'object_data': {'video': {'start_time': 1624020544, 'video_slice': 'https://pull-hls-l1-mus-admin.pstatp.com/stage-dev-6975039157215169286.m3u8?wsStart=1624020544&wsEnd=1624020664&session_id=2021061812481401023102816498404', 'end_time': 1624020664}, 'audit_type': 1, 'object_id': '6975015238768872197_1624020544', 'record_id': '6975108508756738054', 'room_status': 2, 'stream_id': '6975039157215169286', 'user': {'authentication': False, 'bio': 'javascript:alert(1)', 'follows': 17, 'user_id': '6932401500031697925', 'user_name': 'weilinyy'}, 'user_name': '', 'room_et': None, 'room_id': '6975015238768872197', 'room_st': 1624018984, 'room_start_time': '2021-06-18 12:23:04', 'language': 'en', 'live_id': 12, 'products': {'product_items': [{'pictures': ['http://p0.sgpstatp.com/origin/tos-alisg-i-qeln3uh9lc-sg/0ab9563ef2c941298782e4863b575012'], 'pin_time': 0, 'product_id': '1729382477524797039', 'product_type': 2, 'add_time': 1624018984, 'pin': False, 'price': 'Rp3.443 - Rp3.443', 'product_url': 'https://magellan-boe.bytedance.net/view/fe_tiktok_ecommerce_upgrade/index.html?enter_from=live', 'name': 'weilin-test-012'}, {'product_id': '6931962391827579142', 'product_url': 'https://chen-sync-ad-02.myshopify.com/products/test110?sub_id=TTEC&utm_source=TTEC', 'add_time': 1624018984, 'pictures': [], 'pin': False, 'price': '$13.00', 'name': 'test110_rename', 'pin_time': 0, 'product_type': 1}]}, 'user_count': 0, 'app_id': 1233, 'audit_log': [], 'config_key_type': 41, 'region': 'ID', 'user_id': '6932401500031697925'}, 'origin_object_data': None, 'sample_type': None, 'object_version': 0, 'should_verify_count': 1, 'verify_count': 0, 'verifiers': [], 'verify_data': {}, 'standard_data': None, 'abandon_reason': 'call by key:yangshizhao, reason:live is closed', 'challengers': None, 'audit_status': 0, 'mode': 0, 'open_time': '2021-06-18 20:48:21', 'closed_time': '2021-06-18 20:49:20', 'resolve_time': None, 'second_verify': {}, 'first_verify': {}, 'third_verify': {}, 'task_verifies': None, 'task_verify_id': None, 'verify_others': [], 'postpone_reason': [], 'location': None, 'project_title': '[oec] live test', 'tags': ['tcs'], 'prev_task_verifies': [], 'final_verify_data': None, 'create_time': '2021-06-18 20:48:21', 'modify_time': '2021-06-18 20:49:20', 'verifies': None, 'label_map': {'fr_idc': 'boei18n', 'to_idc': 'boei18n'}, 'entity': None}, 'details': 'array-item'}]

好啦～结束战斗～

python实现diff json 并且打印出log日志相关推荐

python批量执行linux命令并写入log日志
linux下使用python运行make命令并把日志有错的路径写入.log日志中 #!/usr/bin/python # -*- coding: UTF-8 -*- import subprocess ...
Python + logging 输出到屏幕，将log日志写入文件(亲测)
日志日志是跟踪软件运行时所发生的事件的一种方法.软件开发者在代码中调用日志函数,表明发生了特定的事件.事件由描述性消息描述,该描述性消息可以可选地包含可变数据(即,对于事件的每次出现都潜在地不同的数 ...
python多进程log日志问题_Python 如何安全地实现实现多进程日志以及日志正常的分割...
在Python中我们经常需要使用到多进程来提高我们程序性能,但是多进程的编程中经常有各种各样的问题来困扰我们,比如多进程和多线程的公用导致的子进程的卡死,进程间的通信等问题.还有一个问题我们也许不经常 ...
Python语言+pytest框架+allure报告+log日志+yaml文件+mysql断言实现接口自动化框架
目录前言实现功能目录结构依赖库安装教程接口文档如何创建用例创建用例步骤用例中相关字段的介绍如何发送get请求如何发送post请求如何测试上传文件接口上传文件接口,即需要上传文 ...
python中的json.loads_Python中使用json.loads解码字符串时出错：
1.报错信息: File "C:\Users\lenovo\Desktop\client.py", line 90, in callback ret = json.loads(re ...
在python中使用json格式存储数据
在python中使用json格式存储数据代码如下: import jsonlist1 = [{'A': [1, 2, 3, 4, 5, 6], 'B': [3, 4, 5, 6, 7]},{'C': ...
python log函数怎么打_Python的log日志功能及设置方法
python log函数怎么打_Python的log日志功能及设置方法_Elaine要当律师的博客-CSDN博客
Python 生成 JWT(json web token) 及解析方式
一.关于 jwt 的原理及概念可以自行在网络上搜索了解一下,这里推荐一篇写的比较好的博客深入了解Json Web Token之概念篇另附 JWT 的官方文档: https://jwt.io/int ...
python json.loads()中文问题-Python实现的json文件读取及中文乱码显示问题解决方法...
本文实例讲述了Python实现的json文件读取及中文乱码显示问题解决方法.分享给大家供大家参考,具体如下: city.json文件的内容如下: { "cities": [ { & ...
python中文编码-python中文编码与json中文输出问题详解
前言 python2.x版本的字符编码有时让人很头疼,遇到问题,网上方法可以解决错误,但对原理还是一知半解,本文主要介绍 python 中字符串处理的原理,附带解决 json 文件输出时,显示中文而非 ...

python实现diff json 并且打印出log日志