本接口自动化框架采用 python + unittest + request + openpyxl + myddt + pymysql 来实现接口自动化。
1、总体框架
2、单元测试框架 unittest
unittest 是 Python 自带的一个单元测试框架
2.1 作用
-
管理用例
-
批量执行用例
-
组织运行结果/报告
-
让代码更稳健
-
可拓展
2.2 unittest 框架中，有以下几个组件：
TestCase：即测试用例，Unittest提供testCase类来编写测试用例，一个TestCase的实例就是一个测试用例。一条测试用例就是一个完整的测试流程，包括测试前准备环境的搭建(setUp)，执行测试代码(run)，以及测试后环境的还原(tearDown)，通过运行一条测试用例，可以对某一个问题进行验证。
Fixture：即测试固件，用于测试用例环境的搭建和销毁。在测试步骤执行前需要为该测试用例准备环境（SetUp），如启动app或打开浏览器，测试步骤执行后需要恢复环境（TearDown），如关闭app或浏览器，这时候就需要用到Fixture，使代码更简洁。
TestSuite：即测试套件，把需要执行的测试用例集合在一起就是TestSuite。使用TestLoader来加载TestCase到TestSuite中
TextTestRunner：即测试执行器，用于执行测试用例。该模块中提供run方法执行TestSuite中的测试用例，并返回测试用例的执行结果，如运行的用例总数、用例通过数、用例失败数。
report：即测试报告。unittest框架没有自带的用于生成测试报告的模块或接口，需要使用第三方的扩展模块HTMLTestRunner。
2.3 跳过执行测试用例共有四种写法
- @unittest.skip(reason) ：跳过测试用例，reason 为测试被跳过的原因。
- @unittest.skipIf(condition, reason) ：当 condition 为真时，跳过测试用例。
- @unittest.skipUnless(condition, reason) ：跳过测试用例，除非 condition 为真。
2.4 断言
2.5 报告
from BeautifulReport import BeautifulReportfrom common.HTMLTestRunnerNew import HTMLTestRunner# 4种测试报告"""1、生成 HTML 类型2、生成 Br 类型3、生成 txt 类型"""# ts0 = unittest.TestLoader().discover('test_cases')# with open('reports/html_do接口自动化.html','wb') as f:# runner = HTMLTestRunner(f)# runner.run(ts0)## ts1 = unittest.TestLoader().discover('test_cases')# br = BeautifulReport(ts1)# br.report(description='DO',filename='br_do接口自动化',report_dir='reports',theme='theme_memories')### ts2 = unittest.TestLoader().discover('test_cases')# with open('reports/txt_do接口自动化.txt','w+') as f:# unittest.TextTestRunner(f,2).run(ts2)if __name__ == "__main__": unittest.main()3、基础框架搭建
在项目根目录下新建 common 文件夹，用来存储公用方法。
在项目根目录下新建 reports 文件夹，用来存储项目报告。
在项目根目录下新建 logs 文件夹，用来存储结果日志。
在项目根目录下新建 test_data 文件夹，用来存储用例数据。
在项目根目录下新建 test_cases 文件夹，用例存储测试用例模块。
在项目根目录下新建 main.py 文件，作为入口函数，方便项目调试。
3.1 common公用方法文件
3.1.1 init.py
# /usr/bin/env python# __*__ coding: utf-8 __*__# @Time : 2021/9/9 22:22# @Author: 夜华import settingsfrom common.log_handler import get_loggerfrom common.db_handler import DB# 日志logger = get_logger(**settings.LOG_CONFIG)# 数据库db = DB(settings.DB_CONFIG)3.1.2 http_requests.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 3:38 PM@File : cliend_http_requests.py@Project : PyCharm"""import requestsdef cliend_http_requests(url,method,**kwargs): method = method.lower() return getattr(requests,method)(url,**kwargs)if __name__ == "__main__": case = { 'url' : 'http://10.21.5.74:33140/api/v1/login', 'method' : 'post', 'requests':{ 'json' : {"email": "name", "password": "password"}, 'headers' : {"Content-Type": "application/json;charset=UTF-8"} } } response = cliend_http_requests(url=case['url'],method=case['method'],**case['requests']) print(response.json())3.1.3 data_handler.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 4:10 PM@File : data_handler.py@Project : PyCharm"""import jsonfrom openpyxl import load_workbookdef get_test_data(filename,sheet_name): wb = load_workbook(filename=filename) sh = wb[sheet_name] row = sh.max_row column = sh.max_column data = [] keys = [] for i in range(1,column+1): keys.append(sh.cell(1,i).value) for i in range(2,row+1): temp = {} for j in range(1,column+1): temp[keys[j-1]] = sh.cell(i,j).value try: temp['request'] = json.loads(temp['request']) temp['exportx_code'] = json.loads(temp['exportx_code']) except json.decoder.JSONDecodeError: raise ValueError('json数据转换错误') data.append(temp) return dataif __name__ == "__main__": res = get_test_data(filename='../test_data/test_cases.xlsx',sheet_name='login') print(res[0])3.1.4 db_handler.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 2:58 PM@File : db_config_handler.py@Project : PyCharm"""import settingsimport pymysqlclass DB: def __init__(self,db_config): self.conn = pymysql.connect(**db_config) def sql_one(self,sql): with self.conn.cursor() as cursor: cursor.execute(sql) return cursor.fetchone() def sql_many(self,sql,size=int): with self.conn.cursor() as cursor: cursor.execute(sql) return cursor.fetchmany(size) def sql_all(self,sql): with self.conn.cursor() as cursor: cursor.execute(sql) return cursor.fetchall() def exisx(self,sql): with self.conn.cursor() as cursor: cursor.execute(sql) if cursor.fetchone(): return True else: return False def sql_update(self,sql): with self.conn.cursor() as cursor: try: cursor.execute(sql) self.conn.commit() except: self.conn.rollback() return cursor.fetchone() def __del__(self): self.conn.close()if __name__ == "__main__": db = DB(db_config=settings.DB_CONFIG) print(db.sql_one("select * from help_category;")) print(db.sql_many("select * from help_category;",2)) print(db.sql_all("select name from help_category;")) print(db.exisx("select * from help_category where name = 'Contents';")) print(db.sql_update("update help_category set url='' where name = 'Contents';"))3.1.5 fixtrue
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 8:27 PM@File : fixtrue.py@Project : PyCharm"""import requestsimport settingsfrom common import loggerdef login(email,password): data = { 'email': email, 'password': password } headers = {"Content-Type":"application/json;charset=UTF-8"} url = settings.PROJECT_URL + settings.INTERFACE['login'] res = requests.post(url=url,json=data,headers=headers) if res.status_code == 200: logger.info('用户登录成功') return res.json() else: logger.warning('用户登录失败')if __name__ == "__main__": res = login(email='name',password='password')3.1.6 logs_handler.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 3:28 PM@File : log_handler.py@Project : PyCharm"""import loggingdef get_logger(name,filename,debug=False,fmt=None,mode='w',encoding='utf-8'): logger = logging.getLogger(name=name) logger.setLevel(level=logging.DEBUG) if debug: file_level = logging.DEBUG console_level = logging.DEBUG else: file_level = logging.WARNING console_level = logging.INFO if fmt is None: #fmt = '%(asctime)s %(levelname)s [%(name)s] [%(filename)s (%(funcName)s:%(lineno)d] - %(message)s' fmt = '%(asctime)s-[%(filename)s-->line:%(lineno)d]-%(levelname)s:%(message)s' format = logging.Formatter(fmt) file_handler = logging.FileHandler(filename=filename,mode=mode,encoding=encoding) file_handler.setLevel(level=file_level) console_handler = logging.StreamHandler() console_handler.setLevel(level=console_level) file_handler.setFormatter(format) console_handler.setFormatter(format) logger.addHandler(file_handler) logger.addHandler(console_handler) return loggerif __name__ == "__main__": logger = get_logger(name='do',filename='../logs/do.txt',debug=False,mode='a') logger.debug(10) logger.info(20) logger.warning(30) logger.error(40) logger.critical(50)3.1.7 reports_handler.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 3:42 PM@File : reports_handler.py@Project : PyCharm"""import osfrom BeautifulReport import BeautifulReportfrom common.HTMLTestRunnerNew import HTMLTestRunnerfrom datetime import datetimedef reports(ts,filename,report_dir,theme='theme_default',title=None,description=None,tester=None,_type='br'): time_prefix=datetime.now().strftime('%Y-%m-%d_%H:%M') filename = '{}_{}'.format(time_prefix,filename) if _type == 'br': br = BeautifulReport(ts) br.report(description=description,filename=filename,report_dir=report_dir,theme=theme) else: with open(os.path.join(report_dir,filename),'wb') as f: runner = HTMLTestRunner(f,title=title,description=description,tester=tester) runner.run(ts)4、config 配置文件夹
4.1 config_dev.ini
[URL]api_url = http://10.21.5.74:331404.2 config_handler.py
# /usr/bin/env python# __*__ coding: utf-8 __*__# @Time : 2021/9/10 21:00# @Author: 夜华"""封装配置文件"""import yamlfrom configparser import ConfigParserdef get_config(filename,encoding='utf-8'): # 根据 . 获取文件后缀，并获取后面的内容 suffix = filename.split('.')[-1] if suffix in ['ini','cfg','cng']: # 判断文件后缀是否存在列表内 # 就是ini 配置 config = ConfigParser() # 实例 config.read(filename,encoding=encoding) # 读取文件 data = {} # for section in config.sections(): #获取 文件里面的所有段名 data[section] = dict(config.items(section)) elif suffix in ['yaml','yml']: # 就是 yaml 配置 with open(filename,'r',encoding=encoding) as f: data = yaml.load(f,Loader=yaml.FullLoader) else: raise ValueError('不能识别的配置后缀') return dataif __name__ == '__main__': get = get_config('../config.ini') print(get) res = get_config('../config.yaml') print(res)4.3 init.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/6/1 2:18 PM@File : __init__.py@Project : PyCharm"""import osimport sysfrom config.config_handler import get_configBASE_DIR = os.path.dirname(os.path.abspath(__file__))if sys.argv[1] == "DEV": Config = get_config(os.path.join(BASE_DIR, './config_dev.ini'))else: Config = get_config(os.path.join(BASE_DIR, './config_test.ini'))5、 logs 文件夹
保存接口测试过程中输出的日志
6、reports 文件夹
保存接口测试报告
7、test_cases 文件夹
7.1 base_case.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 4:39 PM@File : base_case.py@Project : PyCharm"""import unittestfrom common import logger,dbimport requestsimport settingsclass Basic_test_case(unittest.TestCase): name = '基类' logger = logger requests = requests session = requests.session() db = db settings = settings @classmethod def setUpClass(cls) -> None: cls.logger.info('---------------【{}】开始测试---------------'.format(cls.name)) @classmethod def tearDownClass(cls) -> None: cls.logger.info('---------------【{}】结束测试---------------'.format(cls.name)) def check(self,case): self.logger.info('---------------【{}】开始测试---------------'.format(self.name)) self.case = case # 测试用例 self.step() # 测试步骤 self.assert_status_code() # 断言状态码 self.assert_json() # 断言响应信息 self.assert_db() # 断言数据库是否存在数据 self.logger.info('---------------【{}】结束测试---------------'.format(self.name)) def step(self): self.case['url'] = self.settings.PROJECT_URL + self.settings.INTERFACE[self.case['url']] try: self.resposen = self.http_requests(url=self.case['url'],method=self.case['method'],**self.case['request']) except Exception as e: self.logger.warning('用例【{}】发送请求错误'.format(self.case['title'])) self.logger.debug('url:【{}】'.format(self.case['url'])) self.logger.debug('method:【{}】'.format(self.case['method'])) raise e else: self.logger.info('用例【{}】发送请求成功'.format(self.case['title'])) def assert_status_code(self): try: self.assertEqual(self.resposen.status_code,self.case['status_code']) except AssertionError as e: self.logger.warning('用例【{}】状态码断言错误'.format(self.case['title'])) self.logger.debug('预期状态码：【{}】'.format(self.case['status_code'])) self.logger.debug('实际状态码：【{}】'.format(self.resposen.status_code)) raise e else: self.logger.info('用例【{}】状态码断言成功'.format(self.case['title'])) def assert_json(self): res = self.resposen.json() res_data = { 'phone':res.get('phone',None), 'roleType':res.get('roleType',None) } try: self.assertEqual(res_data,self.case['exportx_code']) except AssertionError as e: self.logger.warning('用例【{}】响应信息断言错误'.format(self.case['title'])) self.logger.debug('预期内容:【{}】'.format(self.case['exportx_code'])) self.logger.debug('实际内容:【{}】'.format(res_data)) self.logger.debug('响应内容：【{}】'.format(res)) raise e else: self.logger.info('用例【{}】响应信息断言成功'.format(self.case['title'])) def assert_db(self): if self.case.get('sql'): try: db_res = self.db.exisx(self.case['sql']) self.assertTrue(db_res) except Exception as e: self.logger.warning('用例【{}】数据库查询失败'.format(self.case['title'])) self.logger.debug('sql:【{}】'.format(self.case['sql'])) raise e else: self.logger.info('用例【{}】数据库查询成功'.format(self.case['title'])) def http_requests(self,url,method,**kwargs)->requests.Response: method = method.lower() return getattr(self.session,method)(url=url,**kwargs)if __name__ == "__main__": unittest.main()7.2 test_login.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 3:43 PM@File : test_login.py@Project : PyCharm"""import unittestimport settingsfrom test_cases.base_case import Basic_test_casefrom common.data_handler import get_test_datafrom common.myddt import data,ddtcases = get_test_data(settings.TEST_DATA,sheet_name='login')@ddtclass Test(Basic_test_case): name = '登录' @data(*cases) def test_01(self,case): self.check(case)if __name__ == "__main__": unittest.main()8、test_data 文件夹
8.1 使用Excel表格维护测试用例
用例：id、title、url、method、requests、status_code、exportx_code、sql
9、main.py
main.py 为 测试入口。
from BeautifulReport import BeautifulReportfrom common.HTMLTestRunnerNew import HTMLTestRunnerimport settingsimport unittestfrom common.reports_handler import reportsts = unittest.TestLoader().discover('test_cases')runner = reports(ts,**settings.REPORTS_CONFIG)if __name__ == "__main__": unittest.main()10、settings.py
"""-*- coding: utf-8 -*-@Author : 夜华@Time : 2023/5/14 2:53 PM@File : settings.py@Project : PyCharm"""import osimport project_apiBASE_DIR = os.path.dirname(os.path.abspath(__file__))TEST_DATA = os.path.join(BASE_DIR,'test_data/test_cases.xlsx')PROJECT_URL = 'http://10.00.5.74:00000'INTERFACE = { 'login' : '/api/v1/login', 'query' : '/api/v1/approve/query?page=1&size=4294967295'}# 数据库配置DB_CONFIG = { 'user': 'root', 'password': '123456', 'host': '127.0.0.1', 'database': 'mysql', 'port': 3306, 'autocommit': False}# 输出日志LOG_CONFIG = { 'name' : 'DPO', 'filename' : os.path.join(BASE_DIR,'logs/dpo.txt'), 'debug' : True, 'mode' : 'w', 'fmt' : None, 'encoding': 'utf-8'}# 报告REPORTS_CONFIG = { 'filename':'do接口自动化', 'report_dir' : os.path.join(BASE_DIR,'reports'), 'theme' : 'theme_default', 'description' : 'DO', 'title': 'DO1期', '_type': 'br'}USER_LOGIN = {"email":"name","password":"password"}11、终端内执行
注意：DEV表示开发环境，如果想在非开发环境进行测试，就输入TEST。也可以在4.3init.py 修改
来源地址：https://blog.csdn.net/weixin_53846408/article/details/130951795
