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

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

聚合成集合，測試和報告框架從測試的獨立性。單元測試模塊提供可以很容易地支持這些素質的一組測試的類。關於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', '[e-mail]',)
        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','[e-mail]'),
            (3,'weke','admin','[e-mail]')
        ]
        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()