簡(jiǎn)書已停更，歡迎轉(zhuǎn)到個(gè)人博客查看對(duì)應(yīng)教程：https://www.cnblogs.com/superhin/p/10338985.html
課程目錄
Python接口測(cè)試實(shí)戰(zhàn)1（上）- 接口測(cè)試?yán)碚?/a>
Python接口測(cè)試實(shí)戰(zhàn)1（下）- 接口測(cè)試工具的使用
Python接口測(cè)試實(shí)戰(zhàn)2 - 使用Python發(fā)送請(qǐng)求
Python接口測(cè)試實(shí)戰(zhàn)3（上）- Python操作數(shù)據(jù)庫(kù)
Python接口測(cè)試實(shí)戰(zhàn)3（下）- unittest測(cè)試框架
Python接口測(cè)試實(shí)戰(zhàn)4（上） - 接口測(cè)試框架實(shí)戰(zhàn)
Python接口測(cè)試實(shí)戰(zhàn)4（下） - 框架完善：用例基類，用例標(biāo)簽，重新運(yùn)行上次失敗用例
Python接口測(cè)試實(shí)戰(zhàn)5（上） - Git及Jenkins持續(xù)集成
Python接口測(cè)試實(shí)戰(zhàn)5（下） - RESTful、Web Service及Mock Server
更多學(xué)習(xí)資料請(qǐng)加添加作者微信：lockingfree獲取
本節(jié)內(nèi)容
- 數(shù)據(jù)庫(kù)操作
- 封裝數(shù)據(jù)庫(kù)操作
前言
在功能、接口測(cè)試中常常需要通過(guò)數(shù)據(jù)庫(kù)的操作，來(lái)準(zhǔn)備數(shù)據(jù)、檢測(cè)環(huán)境及核對(duì)功能、接口的數(shù)據(jù)庫(kù)操作是否正確。
在自動(dòng)化測(cè)試中，就需要我們用代碼連接數(shù)據(jù)庫(kù)自動(dòng)完成數(shù)據(jù)準(zhǔn)備、環(huán)境檢查及數(shù)據(jù)庫(kù)斷言的功能。
使用Python操作MySQL數(shù)據(jù)庫(kù)這里我們需要用到三方庫(kù)PyMySQl
安裝方法：
pip install pymysql
數(shù)據(jù)庫(kù)操作
- 建立數(shù)據(jù)庫(kù)連接
conn = pymysql.connect() - 從連接建立操作游標(biāo)
cur = conn.cursor() - 使用游標(biāo)執(zhí)行sql（讀/寫）
cur.execute(sql) - 獲取結(jié)果（讀）/ 提交更改（寫）
cur.fetchall()/conn.commit() - 關(guān)閉游標(biāo)及連接
cur.close();conn.close()
import pymysql
# 1. 建立連接
conn = pymysql.connect(host='127.0.0.1',
port=3306,
user='root',
passwd='123456', # password也可以
db='api_test',
charset='utf8') # 如果查詢有中文需要指定數(shù)據(jù)庫(kù)編碼
# 2. 從連接建立游標(biāo)（有了游標(biāo)才能操作數(shù)據(jù)庫(kù)）
cur = conn.cursor()
# 3. 查詢數(shù)據(jù)庫(kù)（讀）
cur.execute("select * from user where name='張三'")
# 4. 獲取查詢結(jié)果
result = cur.fetchall()
print(result)
# 3. 更改數(shù)據(jù)庫(kù)（寫）
cur.execute("delete from user where name='李四'")
# 4. 提交更改
conn.commit() # 注意是用的conn不是cur
# 5. 關(guān)閉游標(biāo)及連接
cur.close()
conn.close()
什么是游標(biāo)？ 游標(biāo)類似文件句柄，可以逐條的訪問(wèn)數(shù)據(jù)庫(kù)執(zhí)行結(jié)果集。pymysql中只能通過(guò)游標(biāo)來(lái)執(zhí)行sql和獲取結(jié)果
查詢操作
使用cur.execute(), 執(zhí)行數(shù)據(jù)庫(kù)查詢后無(wú)返回的是影響的行數(shù)，而非查詢結(jié)果。我們要使用cur.fetchone()/cur.fetchmany()/cur.fetchall()來(lái)獲取查詢結(jié)果
- cur.fetchone(): 獲取一條數(shù)據(jù)（同時(shí)獲取的數(shù)據(jù)會(huì)從結(jié)果集刪除），返回元祖
('張三','123456') - cur.fetchmany(3): 獲取多條數(shù)據(jù)，返回嵌套元祖
(('張三','123456'),('李四','123456'),("王五","123456")) - cur.fetchall(): 獲取所有數(shù)據(jù)，返回嵌套元祖，
(('張三','123456'),)（只有一條數(shù)據(jù)時(shí)）
注意： 獲取完數(shù)據(jù)后，數(shù)據(jù)會(huì)從數(shù)據(jù)集中刪除，再次獲取獲取不到，如：
cur.execute(select * from user where name='張三')
print(cur.fetchone()) # 結(jié)果： ('張三','123456')
print(cur.fetchone()) # 結(jié)果：None
print(cur.fetchall()) # 結(jié)果：()
所以我們需要重復(fù)使用查詢結(jié)果時(shí)，需要將查詢結(jié)果賦給某個(gè)變量
cur.execute(select * from user where name='張三')
result = cur.fetchall()
print(result) # 結(jié)果： ('張三','123456')
print(result) # 結(jié)果： ('張三','123456')
修改操作
執(zhí)行修改數(shù)據(jù)庫(kù)的操作后不立即生效，使用連接conn.commit()提交后才生效，支持事物及回滾
try:
cur.execute("insert into user (name,password) values ('張三', '123456')")
cur.execute("insert into user (name, passwd) values ('李四'), '123456'") # 此處sql出錯(cuò)
conn.commit() # 使用連接提交所有更改
except Exception as e:
conn.rollback() # 回滾所有更改（注意用的是conn)
print(str(e))
封裝數(shù)據(jù)庫(kù)操作
由于經(jīng)常要使用到數(shù)據(jù)庫(kù)操作，建議將所有數(shù)據(jù)庫(kù)操作封裝成公用的數(shù)據(jù)庫(kù)模塊
- 新建db.py, 代碼如下：
import pymysql
# 獲取連接方法
import pymysql
# 獲取連接方法
def get_db_conn():
conn = pymysql.connect(host='127.0.0.1',
port=3306,
user='test',
passwd='123456',
db='api_test',
charset='utf8') # 如果查詢有中文，需要指定測(cè)試集編碼
return conn
# 封裝數(shù)據(jù)庫(kù)查詢操作
def query_db(sql):
conn = get_db_conn() # 獲取連接
cur = conn.cursor() # 建立游標(biāo)
cur.execute(sql) # 執(zhí)行sql
result = cur.fetchall() # 獲取所有查詢結(jié)果
cur.close() # 關(guān)閉游標(biāo)
conn.close() # 關(guān)閉連接
return result # 返回結(jié)果
# 封裝更改數(shù)據(jù)庫(kù)操作
def change_db(sql):
conn = get_db_conn() # 獲取連接
cur = conn.cursor() # 建立游標(biāo)
try:
cur.execute(sql) # 執(zhí)行sql
conn.commit() # 提交更改
except Exception as e:
conn.rollback() # 回滾
finally:
cur.close() # 關(guān)閉游標(biāo)
conn.close() # 關(guān)閉連接
# 封裝常用數(shù)據(jù)庫(kù)操作
def check_user(name):
# 注意sql中''號(hào)嵌套的問(wèn)題
sql = "select * from user where name = '{}'".format(name)
result = query_db(sql)
return True if result else False
def add_user(name, password):
sql = "insert into user (name, passwd) values ('{}','{}')".format(name, password)
change_db(sql)
def del_user(name):
sql = "delete from user where name='{}'".format(name)
change_db(sql)
相比用例中直接使用sql操作數(shù)據(jù)庫(kù)，封裝常用的數(shù)據(jù)庫(kù)操作會(huì)更安全
- 調(diào)用方法(其他模塊)
from db import *
if check_user("張三"):
del_user("張三")
補(bǔ)充：另一種封裝方法
由于上面這種封裝方法，每一次查詢都會(huì)建立一次數(shù)據(jù)庫(kù)連接，效率較低，也可以采用下面面向?qū)ο蟮姆庋b方法
db2.py
import pymysql
class DB:
def __init__(self):
self.conn = pymysql.connect(host='127.0.0.1',
port=3306,
user='root',
passwd='123456', # passwd 不是 password
db='api_test')
self.cur = self.conn.cursor()
def __del__(self): # 析構(gòu)函數(shù)，實(shí)例刪除時(shí)觸發(fā)
self.cur.close()
self.conn.close()
def query(self, sql):
self.cur.execute(sql)
return self.cur.fetchall()
def exec(self, sql):
try:
self.cur.execute(sql)
self.conn.commit()
except Exception as e:
self.conn.rollback()
print(str(e))
def check_user(self,name):
result = self.query("select * from user where name='{}'".format(name))
return True if result else False
def del_user(self, name)
self.exec("delete from user where name='{}'".format(name))
使用方法
from db2 import DB:
db = DB() # 實(shí)例化一個(gè)數(shù)據(jù)庫(kù)操作對(duì)象
if db.check_user("張三"):
db.del_user("張三")
后言
- 數(shù)據(jù)庫(kù)連接信息建議寫到配置文件中，從配置文件中讀取
- sql語(yǔ)句建議先在手工測(cè)試一下沒(méi)有語(yǔ)法問(wèn)題再進(jìn)行封裝
- 通過(guò)封裝各種sql可以完成各種業(yè)務(wù)操作
- 更改數(shù)據(jù)庫(kù)有風(fēng)險(xiǎn)，操作需謹(jǐn)慎！??！
