当前位置：首页 > news >正文

金融项目实战 07|Python实现接口自动化——连接数据库和数据清洗、测试报告、持续集成

news 2025/7/22 6:07:34

一、投资模块（投资接口+投资业务）

二、连接数据库封装和清洗数据

1、连接数据库

2、数据清洗

4、调用

三、批量执行测试用例并生成测试报告

四、持续集成

1、代码上传gitee

2、Jenkin持续集成

一、投资模块（投资接口+投资业务）

略，之后有机会再补充

二、连接数据库封装和清洗数据

1、连接数据库

在util.py文件中添加连接数据库工具类

# 连接库工具  # 数据库连接不上
def conn_mysql(sql):conn = Nonecursor = Nonetry:# 1、获取连接对象conn = pymysql.connect(host="121.43.169.97", user="root", password="Itcast_p2p_20191228", database="czbk_member", port=3306,charset="utf8", autocommit=True)# 2、获取游标对象cursor = conn.cursor()# 3、执行sql语句cursor.execute(sql)# 判断sql语句是否为查询if sql.split()[0].lower() == "select":# 返回所有结果return cursor.fetchall()# 否则else:# 返回受影响的行数return "受影响的行数：{}".format(cursor.rowcount)except Exception as e:GetLog.get_log().error(e)raisefinally:# 4、关闭游标cursor.close()# 5、关闭连接conn.close()

2、数据清洗

在util.py文件中添加数据清洗工具类

该工具类实现的目的是删掉注册手机号及其关联表的关联数据，保证每次执行注册接口都能通过

# 清除方法
def clear_data():sql1 = """delete i.* from mb_member_info i INNER JOIN mb_member m on i.member_id=m.id where m.phone in ("13600001111","13600001112","13600001113","13600001114")"""conn_mysql(sql1)sql2 = """delete l.* from mb_member_login_log l INNER JOIN mb_member m on l.member_id=m.id where m.phone in ("13600001111","13600001112","13600001113","13600001114")"""conn_mysql(sql2)sql3 = """delete from mb_member_register_log where phone in ("13600001111","13600001112","13600001113","13600001114")"""conn_mysql(sql3)sql4 = """delete from mb_member where phone in ("13600001111","13600001112","13600001113","13600001114")"""conn_mysql(sql4)

4、调用

建议下面的代码放在类TestRegisterLogin中的开头

@classmethoddef setUpClass(cls) -> None:# 清除测试数据clear_data()# 提示：必须在test01_xxxx.py中类⽅法中调⽤

三、批量执行测试用例并生成测试报告

import os
import time
import unittest
from HTMLTestRunner import HTMLTestRunner
from config import DIR_PATH# 自动发现当前目录下的所有测试模块并运行
suite = unittest.TestSuite()
case = unittest.defaultTestLoader.discover(start_dir="./script",pattern="test*.py")suite.addTest(case)# 设置生成文件的路径和名字
now = time.strftime("%Y-%m-%d-%H-%M-%S")
report_path = DIR_PATH + os.sep + "report" + os.sep + f"测试报告-{now}.html" with open(report_path, "wb") as f:runner = HTMLTestRunner(stream=f,title="python自动化测试报告",description="共15+9条测试用例，1条不通过（因为无法执行数据清洗）")runner.run(suite)