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

課程目錄

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

更多學習資料請加添加做者微信：lockingfree獲取

本節內容

• 數據庫操做
• 封裝數據庫操做

前言

在功能、接口測試中經常須要經過數據庫的操做，來準備數據、檢測環境及覈對功能、接口的數據庫操做是否正確。 在自動化測試中，就須要咱們用代碼鏈接數據庫自動完成數據準備、環境檢查及數據庫斷言的功能。 使用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("delete 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("delete 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. 配套實戰項目練習