如有任何學習問題，可以添加作者微信：lockingfree

課程目錄

Python接口測試實戰1（上）- 接口測試理論
Python接口測試實戰1（下）- 接口測試工具的使用
Python接口測試實戰2 - 使用Python發送請求
Python接口測試實戰3（上）- Python操作數據庫
Python接口測試實戰3（下）- unittest測試框架
Python接口測試實戰4（上） - 接口測試框架實戰
Python接口測試實戰4（下） - 框架完善：用例基類，用例標簽，重新運行上次失敗用例
Python接口測試實戰5（上） - Git及Jenkins持續集成
Python接口測試實戰5（下） - RESTful、Web Service及Mock Server

更多學習資料請加QQ群: 822601020獲取

本節內容

• 數據庫操作
• 封裝數據庫操作

前言

在功能、接口測試中常常需要通過數據庫的操作，來準備數據、檢測環境及核對功能、接口的數據庫操作是否正確。
在自動化測試中，就需要我們用代碼連接數據庫自動完成數據準備、環境檢查及數據庫斷言的功能。
使用Python操作MySQL數據庫這裏我們需要用到三方庫PyMySQl

安裝方法：pip install pymysql

數據庫操作

1. 建立數據庫連接 conn = pymysql.connect()
2. 從連接建立操作遊標 cur = conn.cursor()
3. 使用遊標執行sql（讀/寫） cur.execute(sql)
4. 獲取結果（讀）/ 提交更改（寫） cur.fetchall() / conn.commit()
5. 關閉遊標及連接 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‘)   # 如果查詢有中文需要指定數據庫編碼
                    
# 2. 從連接建立遊標（有了遊標才能操作數據庫）
cur = conn.cursor()

# 3. 查詢數據庫（讀）
cur.execute("select * from user where name=‘張三‘")

# 4. 獲取查詢結果
result = cur.fetchall()
print(result)

# 3. 更改數據庫（寫）
cur.execute("delect from user where name=‘李四‘")

# 4. 提交更改
conn.commit()  # 註意是用的conn不是cur

# 5. 關閉遊標及連接
cur.close()
conn.close()
 

什麽是遊標？ 遊標類似文件句柄，可以逐條的訪問數據庫執行結果集。pymysql中只能通過遊標來執行sql和獲取結果

查詢操作
使用cur.execute(), 執行數據庫查詢後無返回的是影響的行數，而非查詢結果。我們要使用cur.fetchone()/cur.fetchmany()/cur.fetchall()來獲取查詢結果

• cur.fetchone(): 獲取一條數據（同時獲取的數據會從結果集刪除），返回元祖(‘張三‘,‘123456‘)
• cur.fetchmany(3): 獲取多條數據，返回嵌套元祖((‘張三‘,‘123456‘),(‘李四‘,‘123456‘),("王五","123456"))
• cur.fetchall(): 獲取所有數據，返回嵌套元祖，((‘張三‘,‘123456‘),)（只有一條數據時）

註意： 獲取完數據後，數據會從數據集中刪除，再次獲取獲取不到，如：

cur.execute(select * from user where name=‘張三‘)
print(cur.fetchone()) # 結果： (‘張三‘,‘123456‘)
print(cur.fetchone()) # 結果：None
print(cur.fetchall()) # 結果：()

所以我們需要重復使用查詢結果時，需要將查詢結果賦給某個變量

cur.execute(select * from user where name=‘張三‘)
result = cur.fetchall()
print(result)  # 結果： (‘張三‘,‘123456‘)
print(result)  # 結果： (‘張三‘,‘123456‘)

修改操作
執行修改數據庫的操作後不立即生效，使用連接conn.commit()提交後才生效，支持事物及回滾

try:
    cur.execute("insert into user (name,password) values (‘張三‘, ‘123456‘)")
    cur.execute("insert into user (name, passwd) values (‘李四‘), ‘123456‘") # 此處sql出錯
    conn.commit()  # 使用連接提交所有更改
except Exception as e:
    conn.rollback()  # 回滾所有更改（註意用的是conn)
    print(str(e))

封裝數據庫操作

由於經常要使用到數據庫操作，建議將所有數據庫操作封裝成公用的數據庫模塊

1. 新建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‘)  # 如果查詢有中文，需要指定測試集編碼

    return conn


# 封裝數據庫查詢操作
def query_db(sql):
    conn = get_db_conn()  # 獲取連接
    cur = conn.cursor()  # 建立遊標
    cur.execute(sql)  # 執行sql
    result = cur.fetchall()  # 獲取所有查詢結果
    cur.close()  # 關閉遊標
    conn.close()  # 關閉連接
    return result  # 返回結果


# 封裝更改數據庫操作
def change_db(sql):
    conn = get_db_conn()  # 獲取連接
    cur = conn.cursor()  # 建立遊標
    try:
        cur.execute(sql)  # 執行sql
        conn.commit()  # 提交更改
    except Exception as e:
        conn.rollback()  # 回滾
    finally:
        cur.close()  # 關閉遊標
        conn.close()  # 關閉連接


# 封裝常用數據庫操作
def check_user(name):
    # 註意sql中‘‘號嵌套的問題
    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操作數據庫，封裝常用的數據庫操作會更安全

1. 調用方法(其他模塊)

from db import *

if check_user("張三"):
    del_user("張三")

補充：另一種封裝方法

由於上面這種封裝方法，每一次查詢都會建立一次數據庫連接，效率較低，也可以采用下面面向對象的封裝方法

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): # 析構函數，實例刪除時觸發
        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("del from user where name=‘{}‘".format(name))
        

使用方法

from db2 import DB:

db = DB()  # 實例化一個數據庫操作對象
if db.check_user("張三"):
    db.del_user("張三")

後言

• 數據庫連接信息建議寫到配置文件中，從配置文件中讀取
• sql語句建議先在手工測試一下沒有語法問題再進行封裝
• 通過封裝各種sql可以完成各種業務操作
• 更改數據庫有風險，操作需謹慎！！！

此為北京龍騰育才 Python高級自動化（接口測試部分）授課筆記
課程介紹
想要參加現場(北京)/網絡課程的可以聯系作者微信：lockingfree

1. 高效學習，快速掌握Python自動化所有領域技能
2. 同步快速解決各種問題
3. 配套實戰項目練習

Python接口測試實戰3（上）- Python操作數據庫