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‘, ‘[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 的操作