赞
踩
是存储数据的仓库,程序中数据的载体
数据库和变量都可以存储数据,二者的区别是
持久性不同:数据库可以持久性能够存储数据(数据被写入磁盘中),变量不能(运行在内存中)
python(等不同语言)本身不具备直连数据库的功能,必须导入第三方包
数据库驱动:
# 连接pymysql # 1. 导包 import pymysql # 2. 创建游标 con = pymysql.connect(host='127.0.0.1', port=3306, database='test', user='root', password='123456', charset='utf8') # 3. 创建链接 cur = con.cursor() # 4. 执行sql # 编写sql # sql = "select * from t_area" sql = "insert into t_area(area_name, priority) values('西安', '1')" # 执行SQL cur.execute(sql) # 逐行获取数据 # result = cur.fetchone() # print(result) # 获取素有数据 result = cur.fetchall() for row in result: print(row[3]) # 影响的行数 print("影响行数:", cur.rowcount) # 提交事务 con.commit() # 5. 释放资源 cur.close() con.close()
注意:
# 1. 创建连接connection # 2. 获取游标cursor # 3. 执行SQL语句 # 1. 执行查询语句 # 2. 执行增删改语句 # 1. 判断是否出现异常 # - 否(没有出现异常)-->提交事务 # - 是(出现异常)-->回滚事务 # 4. 关闭游标cursor # 5. 关闭连接connection import pymysql con = pymysql.connect(host='127.0.0.1', port=3306, database='test', user='root', passwd='123456', charset='utf8') cur = con.cursor() try: sql = "select * from t_area" cur.execute(sql) result = cur.fetchall() for row in result: print(row) # 在添加和修改的时候需要提交事务 # con.commit() except Exception as e: # 回滚事务 cur.rollback() cur.close() con.close()
import pymysql class DBUtil: # 获取连接 @classmethod def get_connect(cls): # 创建连接 return pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd=123456, charset='utf8') # 获取游标 @classmethod def get_cursor(cls, con): # 创建连接 return con.cursor() # 释放资源 @classmethod def close(cls, cur, con): if cur: cur.close() if con: con.close()
案例
# 获取连接
from demo.utils.DBUtil import DBUtil
con = DBUtil.get_connect()
# 创建游标
cur = DBUtil.get_cursor(con)
# 编写和执行sql
sql = "select * from t_aera"
# 执行sql
result = cur.execute(sql)
# 查看结果
for row in result:
print(row[3])
# 关闭资源
DBUtil.close(con, cur)
银行转账:
事务:是一套完整的业务逻辑,在业务逻辑中,困难包含多条sql语句,在这些sql执行的时候,要么都成功,要么都失败
requests库是使用python编写的,可以调用该库的函数直接向服务器发送请求,并接收响应
类似于Jmeter中http请求
pip install requests
在命令行输入pip list
命令查找requests的名称和对应的版本号
import requests as requests
# 请求
response = requests.get("https://api-v2.xdclass.net/api/rank/v1/hot_product")
print("状态码:", response.status_code)
print("响应体:", response.text)
import requests as requests
data = {"page": 1, "size": 4}
response = requests.post("https://api-v2.xdclass.net/api/play_record/v1/page", data=data)
print("状态码:", response.status_code)
print("响应体:", response.text)
import requests as requests
# 请求
myJson = {
"areaId": 55,
"areaName": "上海",
"priority": "111"
}
response = requests.put("http://localhost:8080/sa/modifyarea", json=myJson)
print("状态码:", response.status_code)
print("响应体:", response.text)
import requests as requests
# 请求
response = requests.delete("http://localhost:8080/sa/removearea", params={"areaId": "40"})
print("状态码:", response.status_code)
print("响应体:", response.text)
import requests response = requests.get("https://www.baidu.com") # 响应行 print("url", response.url) print("状态码", response.status_code) print("-"*100) # 响应头 print("获取所有响应体头", response.headers) print("获取所有cookie", response.cookies) print("获取所有编码集", response.encoding) print("-"*100) # 响应体 print("以文本的方式获取响应体", response.text) # print("以二进制的方式获取响应体", response.content) # print("以JSON的方式获取响应体", response.json()) print("-"*100)
需求案例:先登录,登陆成成功后获取“订单页面”
cookie:
# 核心知识点:关联
# 获取登录接口响应的 cookie,提取出来作为查询订单接口要提交的参数
import requests
# 访问接口1:访问登录接口
response1 = requests.post("login接囗", data={"username": “xxx", "password":"yyy"})
# 获取 cookie,再获取 cookie 中的 xXID 形式类似于{“xxID":“zzz"}
id= response1.cookies.get("xxID")
# 访问接口2:订单查询接口
requests.get("查询订单接口",cookies={“xxID":id})
Session:
import requests
# 获取 session 对象
mySession =requests.session()
# 请求1:使用 session 登录
#requests.post(...)
response1 = mysession.post("1ogin接囗",data={"username":"xxx","password": "yyy"})
# 请求2:使用 session 获取订单
response2 = mysession.get("查询订单接口")
需求:使用 requests库调用 tpshop登录功能的相关接口,完成登录操作,登录成功后获取“我的订单"页面(访问订单列表接口)
相关接口:
1.获取验证码:http://localhost/index.php?m=Home&c=User&a=verify GET
(此接口返回验证码 和 cookie)
2.登录:http://localhost/index.php?m=Home&c=User&a=do_login POST
参数: {"username":"xxxxx","password":"yyyy","verify_code":"zzzz"},非 JSON 提交
3.我的订单:http://localhost/Home/Order/order_list.html GET
Cookie实现
import requests # 请求 cookieId_response = requests.get("http://192.168.157.130/index.php?m=Home&c=User&a=verify") print("状态码:", cookieId_response.status_code) print("Cookie对象", cookieId_response.cookies) # 获取cookies,PHPSESSID id = cookieId_response.cookies.get("PHPSESSID") print("Cookie=", id) print("-"*100) # 请求登录接口 data = { "username": "13012345678", "password": "123456", "verify_code": "8888" } # 获取cookie的ID的值 cookie = {"PHPSESSID": id} # 登录接口 res_login = requests.post("http://192.168.157.130/index.php?m=Home&c=User&a=do_login", data=data, cookies=cookie) print("登录状态码:", res_login.status_code) print("登录响应体:", res_login.text) print("-"*100) # 订单接口 order_login = requests.get("http://192.168.157.130/Home/Order/order_list.html", cookies=cookie) print("订单状态码:", order_login.status_code) print("订单响应体:", order_login.text)
import requests # 创建session对象 sesion = requests.session() print("-"*100) # 请求登录接口 data = { "username": "13012345678", "password": "123456", "verify_code": "8888" } # 登录接口 res_login = sesion.post("http://192.168.157.130/index.php?m=Home&c=User&a=do_login", data=data) print("登录状态码:", res_login.status_code) print("登录响应体:", res_login.text) print("-"*100) # 订单接口 order_login = sesion.get("http://192.168.157.130/Home/Order/order_list.html") print("订单状态码:", order_login.status_code) print("订单响应体:", order_login.text)
pytest是python第三方的单元测试框架
概述:
概念
运行于测试方法的始末, 运行一次测试函数会执行一次 setup 和 teardown
有多少个测试函数就会运行多少次的 setup 和 teardown方法
代码
test_xxx.py
class TestLogin: # 函数级初始化方法 def setup(self): print("---setup---") # 函数级结束 def teardown(self): print("---teardown---") def test_a(self): print("test_a") assert 1 # 断言成功 def test_b(self): print("test_b") assert 0 # 断言失败
结果
test_setup和teardown[39].py ---setup--- # 第1次运行 setup
test_a
.---teardown--- # 第1次运行 teardown
---setup--- # 第2次运行 setup
test_b
F---teardown--- # 第2次运行 teardown
应用场景
使用配置文件, 可以通过配置项来选择执行哪些目录下的哪些测试模块
使用方式
示例
[pytest]
addopts = -s
testpaths = ./scripts
python_files = test_*.py
python_classes = Test*
python_functions = test_*
你写的时候可以直接拿来复制粘贴
参数解释:
addopts = -s 表示命令行参数
testpaths, python_files, python_classes, python_functions
表示执行哪一个包下面的哪些.py结尾的文件, 以及哪些前缀开头的类, 以及哪些前缀开头的测试函数
注意点
方法
@pytest.mark.parametrize("参数名", 参数值)
参数对应的值: 类型必须为可迭代的类型, 一般使用 list
示例
import pytest
class TestLogin:
@pytest.mark.parametrize("params", [{"username": "zhangsan", "password": "111"}, {"username": "lisi", "password": "222"}])
def test_a(self, params):
print(params)
print(params["username"])
print(params["password"])
结果
test_login[43].py {'username': 'zhangsan', 'password': '111'}
zhangsan
111
.{'username': 'lisi', 'password': '222'}
lisi
222
.
参数化后, 有几组参数, 测试函数就会执行几次
安装
# [推荐安装1.21.1的版本]命令行输入
pip install pytest-html==1.21.1
校验方式 pip list
使用
在配置文件中的命令行参数增加 --html=用户路径/xxx.html
伪代码
在 scripts 包下的文件 test_xxx.py
# 测试类 class TestDemo: # 初始化函数 def setup(self): self.session = requests.Session() # 资源销毁函数 def teardown(self): self.session.close() # 测试函数1: 登录 def test_login(self): 写登录相关代码 # 测试函数2: 我的订单 def test_order(self): # 1.登录 # 2.获取订单 ...如, # 3.断言
存在的问题
[pytest]
addopts = -s --html=report/report.html # 将测试报告放入到report/report.htm的文件
testpath = ./scripts # 放入测试脚本
python_files = test_*.py # 以test_开头的脚本进行读取
python_classes = Test* # 将Test开头的类文件读取
python_function = test_* # 将test_开头的方法读取3.
BASE_URL = "127.0.0.1:8080" # 基础地址:IP:端口号
AREA_URL = "/sa" # 接口地址
创建test_area.py文件
import requests from api.AreaAPI import AreaAPI from api.AreaDBAPI import AreaDBAPI class TestArea: def setup(self): self.session = requests.Session() self.area_api = AreaAPI(self.session) def teardown(self): self.session.close() # 测试查询列表接口 def test_list_area(self): response = self.area_api.area_list_url() print("状态码 = ", response.status_code) print("响应体 = ", response.text) # 测试新增接口 def test_add_area(self): data = { "areaName": "12", "priority": "321" } response = self.area_api.add_area(data=data) print("状态码 = ", response.status_code) print("响应体 = ", response.text) # 测试修改接口 def test_update_area(self): id = AreaDBAPI.select_id_by_name("12") json = { "areaId": id, "areaName": "123" } response = self.area_api.update_area(json=json) print("状态码 = ", response.status_code) print("响应体 = ", response.text) # 测试删除接口 def test_remove_area(self): id = AreaDBAPI.select_id_by_name("123 ") # 需要删除的变量 params = { "areaId": id } # 响应结果 response = self.area_api.removeArea_url(params=params) print("状态码 = ", response.status_code) print("响应体 = ", response.text)
创建AreaAPI.py
from app import BASE_URL, AREA_URL class AreaAPI: def __init__(self, session): self.session = session self.area_list_url = BASE_URL + AREA_URL + "/listarea" self.addArea_url = BASE_URL + AREA_URL + "/addArea" self.updateArea_url = BASE_URL + AREA_URL + "/updateArea" self.removeArea_url = BASE_URL + AREA_URL + "/removeArea" # 1查询area def area_list(self): response = self.session.get(self.area_list_url) return response # 2新增area def add_area(self, data): response = self.session.post(self.addArea_url, data=data) return response # 3修改area def update_area(self, json): response = self.session.put(self.updateArea_url, json=json) return response # 4删除area def remove_area(self, params): response = self.session.delete(self.removeArea_url, params=params) return response
创建AreaDBAPI.py文件【用于数据库查询】
from utils.DBUtils import DBUtils class AreaDBAPI: @classmethod def select_id_by_name(cls, area_name): # 获取连接 con = DBUtils.get_connect() # 获取游标 cur = DBUtils.get_cursor(con) # 编写sql sql = "select area_id from area where area_name = `%s`" % (area_name) # 执行sql cur.execute(sql) # 获取所有数据 row = cur.fechall() # 释放资源 DBUtils.close_res(con, cur) # 返回第一行第一列 return row[0][0]
创建DBUtils.py
import pymysql class DBUtils: @classmethod def get_connect(cls): # 获取连接 return pymysql.Connect(host="127.0.0.1", user="root", password="123456", database="test", port=3306, charset="utf8") @classmethod def get_cursor(cls, con): # 获取游标 return con.cursor() @classmethod def close_res(cls, con, cursor): # 释放资源 if cursor: cursor.close() if con: con.close()
结果截图
总结接口自动化各个目录之间的互相调用
以脚本为基础【test_area.py】
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。