1. 程式人生 > >Python接口測試之對MySQL/unittest框架/Requests 的操作

Python接口測試之對MySQL/unittest框架/Requests 的操作

l數據庫 dmi itl input ins 就是 user tool spa

單元測試支持測試自動化 共享安裝程序關閉代碼測試

聚合集合測試報告框架測試獨立性單元測試模塊提供可以容易支持這些素質測試。關於unittest

測試框架建議可以到官方查看詳細的說明以及演示的實例。unittest各個模塊的關系為:

技術分享圖片

#!/usr/bin/env python
#coding:utf-8

import  unittest

class TestDiv(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_001(self):
        self.assertEqual(div(1,1),1)

    def test_002(self):
        self.assertRaises(ZeroDivisionError,div,1,0)

if __name__==‘__main__‘:
    unittest.main(verbosity=2)

最常使用的分別有GET,PUT,DELETE,POST,通過request庫也是可以實現

的,見如下的代碼:

技術分享圖片

在python的命令行環境下

看是否可以導入MySQLdb,如果可以導入並且無任何的錯誤提示,表示已經安裝成功了,見截圖:

技術分享圖片

已經很成功的安裝了python操作mysql的數據庫,在這裏,我們詳細的介紹對python對mysql的增加,刪除,修改

和查詢的基本操作,這裏使用的數據庫名稱是“day2017”,我們對數據庫的操作,首先是創建數據庫,然後是在數據庫中

創建表,在這裏,表的名稱為:userInfo,見創建好的表字段信息:

技術分享圖片

OK,創建好數據庫以及創建好了數據庫中的表以後,下來開始操作數據庫,操作數據庫的第一步當然是連接數據庫,然後是

創建遊標,接下來是對數據庫的各種操作,這裏我們先來操作Insert數據的操作,見實現的代碼:

技術分享圖片
#!/usr/bin/env python 
#coding:utf-8

import MySQLdb

def insert_One():
    ‘‘‘插入一條數據‘‘‘
    try:
        conn=MySQLdb.connect(host=‘127.0.0.1‘,user=‘root‘,passwd=‘server‘,db=‘day2017‘)
    except:
        print u‘連接mysql數據庫失敗‘
    else:
        cur = conn.cursor()
        sql = ‘INSERT INTO userInfo VALUES (%s,%s,%s,%s)‘
        params = (1, ‘admin‘, ‘admin‘, ‘[email protected]‘,)
        cur.execute(sql, params)
        conn.commit()
    finally:
        cur.close()
        conn.close()

if __name__==‘__main__‘:
    insert_One()
技術分享圖片

技術分享圖片

查看數據庫,可以看到,數據已經插入到數據庫中,見查詢的結果:

技術分享圖片

在上面的案例中,只是插入了單條數據,實際上,某些時候,會插入多條數據,也就是批量插入,批量插入實現的代碼為:

技術分享圖片
def insert_Many():
    ‘‘‘批量插入數據‘‘‘
    try:
        conn=MySQLdb.connect(host=‘127.0.0.1‘,user=‘root‘,passwd=‘server‘,db=‘day2017‘)
    except:
        print u‘連接mysql數據庫失敗‘
    else:
        cur = conn.cursor()
        sql = ‘INSERT INTO userInfo VALUES (%s,%s,%s,%s)‘
        params = [
            (2,‘wuya‘,‘admin‘,‘[email protected]‘),
            (3,‘weke‘,‘admin‘,‘[email protected]‘)
        ]
        cur.executemany(sql, params)
        conn.commit()
    finally:
        cur.close()
        conn.close()

if __name__==‘__main__‘:
    insert_Many()
技術分享圖片

技術分享圖片

接下來,我們來查看數據庫的查詢,數據查詢分為二種,一種是查詢的結果是一條語句,使用的是fetchone()方法,另外一種是查詢的數據

結果是多條,使用的方法是fetchmany(),我們分別來看這二個方法的使用,我們先來看單條數據的查詢,見實現的代碼:

技術分享圖片
import  MySQLdb

def select_one():
    ‘‘‘單條數據的查詢‘‘‘
    try:
        conn=MySQLdb.connect(host=‘127.0.0.1‘,user=‘root‘,passwd=‘server‘,db=‘day2017‘)
    except:
        print u‘連接mysql數據庫失敗‘
    else:
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        sql = ‘select * from userInfo where id=%s‘
        params = (1,)
        data=cur.execute(sql, params)
        print cur.fetchone()
        conn.commit()
    finally:
        cur.close()
        conn.close()

if __name__==‘__main__‘:
    select_one()
技術分享圖片

技術分享圖片

多條數據的查詢,見實現的代碼:

技術分享圖片
import MySQLdb

def select_Many():
    ‘‘‘多條數據的查詢‘‘‘
    try:
        conn=MySQLdb.connect(host=‘127.0.0.1‘,user=‘root‘,passwd=‘server‘,db=‘day2017‘)
    except:
        print u‘連接mysql數據庫失敗‘
    else:
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        sql = ‘select * from userInfo‘
        ret=cur.execute(sql)
        data=cur.fetchall()
        for item in data:
            print item
    finally:
        cur.close()
        conn.close()

if __name__==‘__main__‘:
    select_Many()
技術分享圖片

技術分享圖片

下面我們來看更新語句的測試,見實現的代碼:

技術分享圖片
import  MySQLdb

def update_Test():
    ‘‘‘更新語句測試‘‘‘
    try:
        conn=MySQLdb.connect(host=‘127.0.0.1‘,user=‘root‘,passwd=‘server‘,db=‘day2017‘)
    except:
        print u‘連接mysql數據庫失敗‘
    else:
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        sql = ‘update userInfo set username=%s where id=%s‘
        params=(‘system‘,1,)
        ret=cur.execute(sql,params)
        conn.commit()
    finally:
        cur.close()
        conn.close()

if __name__==‘__main__‘:
    update_Test()
技術分享圖片

技術分享圖片

最後一步,也就是刪除數據了,直接看如下的實現代碼:

技術分享圖片
import  MySQLdb

def delete_Test():
    ‘‘‘刪除語句測試‘‘‘
    try:
        conn=MySQLdb.connect(host=‘127.0.0.1‘,user=‘root‘,passwd=‘server‘,db=‘day2017‘)
    except:
        print u‘連接mysql數據庫失敗‘
    else:
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        sql = ‘delete from userInfo where id=%s‘
        params=(3,)
        ret=cur.execute(sql,params)
        conn.commit()
    finally:
        cur.close()
        conn.close()

if __name__==‘__main__‘:
    delete_Test()
技術分享圖片

技術分享圖片

事實上,對於如上操作數據庫的方式,有很多的代碼是可以重夠的,比如連接數據庫的方式,另外,我們可以把操作數據庫的

方式寫在一個類裏面,在業務調用的時候直接調用我們的數據庫方法進行操作,見下面操作mysql數據庫的方法,見源碼:

技術分享圖片
#!/usr/bin/env python 
#coding:utf-8

import  MySQLdb

class MySQLHelper(object):
    def __init__(self):
        pass

    def get_one(self,sql,params):
        conn = MySQLdb.connect(host=‘127.0.0.1‘,user=‘root‘,passwd=‘server‘,db=‘day2017‘)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        retCount = cur.execute(sql,params)
        data = cur.fetchone()
        cur.close()
        conn.close()
        return data

    def get_many(self,sql,params):
        conn = MySQLdb.connect(host=‘127.0.0.1‘,user=‘root‘,passwd=‘server‘,db=‘day2017‘)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        retCount = cur.execute(sql,params)
        data = cur.fetchall()
        cur.close()
        conn.close()
        return data

    def insert_one(self,sql,params):
        conn = MySQLdb.connect(host=‘127.0.0.1‘, user=‘root‘, passwd=‘server‘, db=‘day2017‘)
        cur = conn.cursor()
        cur.execute(sql, params)
        conn.commit()
        cur.close()
        return u‘插入數據庫成功‘

    def insert_many(self,sql,params):
        conn = MySQLdb.connect(host=‘127.0.0.1‘, user=‘root‘, passwd=‘server‘, db=‘day2017‘)
        cur = conn.cursor()
        cur.executemany(sql, params)
        conn.commit()
        cur.close()
        return u‘批量插入數據庫成功‘

    def update_one(self,sql,params):
        conn = MySQLdb.connect(host=‘127.0.0.1‘, user=‘root‘, passwd=‘server‘, db=‘day2017‘)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        ret = cur.execute(sql, params)
        conn.commit()
        cur.close()
        conn.close()
        return u‘更新數據庫成功‘

    def delete_one(self,sql,params):
        conn = MySQLdb.connect(host=‘127.0.0.1‘, user=‘root‘, passwd=‘server‘, db=‘day2017‘)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        ret = cur.execute(sql, params)
        conn.commit()
        cur.close()
        conn.close()
        return u‘刪除數據庫成功‘
技術分享圖片

把連接數據庫部分進行重構,放到一個config.py的文件中,這樣我們連接數據庫的方法就只需要在config.py文件維護了,而不

需要在如上代碼中每個都得看的修改,這實在是很糟糕,見重構後的config.py文件源碼:

#!/usr/bin/env python 
#coding:utf-8

conn_dict=dict(host=‘127.0.0.1‘, user=‘root‘, passwd=‘server‘, db=‘day2017‘)

見重構後操作mysql的數據庫方法,見源碼:

技術分享圖片
#!/usr/bin/env python 
#coding:utf-8

import  MySQLdb
import  config

class MySQLHelper(object):
    def __init__(self):
        self.conn=config.conn_dict

    def get_one(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        retCount = cur.execute(sql,params)
        data = cur.fetchone()
        cur.close()
        conn.close()
        return data

    def get_many(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        retCount = cur.execute(sql,params)
        data = cur.fetchall()
        cur.close()
        conn.close()
        return data

    def insert_one(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor()
        cur.execute(sql, params)
        conn.commit()
        cur.close()
        return u‘插入數據庫成功‘

    def insert_many(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor()
        cur.executemany(sql, params)
        conn.commit()
        cur.close()
        return u‘批量插入數據庫成功‘

    def update_one(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        ret = cur.execute(sql, params)
        conn.commit()
        cur.close()
        conn.close()
        return u‘更新數據庫成功‘

    def delete_one(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        ret = cur.execute(sql, params)
        conn.commit()
        cur.close()
        conn.close()
        return u‘刪除數據庫成功‘
技術分享圖片

寫數據庫的操作方法,是為了進行對業務的操作,要不僅僅寫這些沒什麽實際的意義,如我們實現輸入用戶名和密碼,在

在數據庫中驗證,如果用戶名和密碼都是admin,那麽通過,如果有其中一個不是admin,就提示用戶,請提示用戶用戶名

或者密碼錯誤,下面來實現這樣的一個過程,見實現的源碼:

技術分享圖片
#!/usr/bin/env python 
#coding:utf-8

import  MySQLdb
import  config

class MySQLHelper(object):
    def __init__(self):
        self.conn=config.conn_dict

    def get_one(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        retCount = cur.execute(sql,params)
        data = cur.fetchone()
        cur.close()
        conn.close()
        return data

    def get_many(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        retCount = cur.execute(sql,params)
        data = cur.fetchall()
        cur.close()
        conn.close()
        return data

    def insert_one(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor()
        cur.execute(sql, params)
        conn.commit()
        cur.close()
        return u‘插入數據庫成功‘

    def insert_many(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor()
        cur.executemany(sql, params)
        conn.commit()
        cur.close()
        return u‘批量插入數據庫成功‘

    def update_one(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        ret = cur.execute(sql, params)
        conn.commit()
        cur.close()
        conn.close()
        return u‘更新數據庫成功‘

    def delete_one(self,sql,params):
        conn = MySQLdb.connect(**self.conn)
        cur = conn.cursor(cursorclass=MySQLdb.cursors.DictCursor)
        ret = cur.execute(sql, params)
        conn.commit()
        cur.close()
        conn.close()
        return u‘刪除數據庫成功‘

class CheckUserInfo(object):
    def __init__(self):
        self.__helper=MySQLHelper()

    def checkValid(self,username,password):
        sql=‘select * from userInfo where  username=%s and password=%s‘
        params=(username,password)
        return self.__helper.get_one(sql,params)

def info():
    username=raw_input(u‘請輸入你的用戶名:\n‘)
    password=raw_input(u‘請輸入你的密碼:\n‘)
    userInfo=CheckUserInfo()
    result=userInfo.checkValid(username,password)
    if not result:
        print u‘用戶名或者密碼錯誤,請聯系管理員‘
    else:
        print u‘恭喜您,輸入正確!‘

if __name__==‘__main__‘:
    info()

Python接口測試之對MySQL/unittest框架/Requests 的操作