pythonmysql单元测试_Python接口测试之对MySQL/unittest框架/Requests的操作

单元测试支持测试自动化、共享的安装程序和关闭代码测试、

聚合成集合&＃xff0c;测试和报告框架从测试的独立性。单元测试模块提供可以很容易地支持这些素质的一组测试的类。关于unittest

测试框架建议可以到官方查看详细的说明以及演示的实例。unittest各个模块的关系为&＃xff1a;

#!/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__&＃61;&＃61;&＃39;__main__&＃39;:

unittest.main(verbosity&＃61;2)

最常使用的分别有GET,PUT,DELETE,POST&＃xff0c;通过request库也是可以实现

的&＃xff0c;见如下的代码&＃xff1a;

在python的命令行环境下

看是否可以导入MySQLdb&＃xff0c;如果可以导入并且无任何的错误提示&＃xff0c;表示已经安装成功了&＃xff0c;见截图&＃xff1a;

已经很成功的安装了python操作mysql的数据库&＃xff0c;在这里&＃xff0c;我们详细的介绍对python对mysql的增加&＃xff0c;删除&＃xff0c;修改

和查询的基本操作&＃xff0c;这里使用的数据库名称是“day2017”&＃xff0c;我们对数据库的操作&＃xff0c;首先是创建数据库&＃xff0c;然后是在数据库中

创建表&＃xff0c;在这里&＃xff0c;表的名称为&＃xff1a;userInfo&＃xff0c;见创建好的表字段信息&＃xff1a;

OK&＃xff0c;创建好数据库以及创建好了数据库中的表以后&＃xff0c;下来开始操作数据库&＃xff0c;操作数据库的第一步当然是连接数据库&＃xff0c;然后是

创建游标&＃xff0c;接下来是对数据库的各种操作&＃xff0c;这里我们先来操作Insert数据的操作&＃xff0c;见实现的代码&＃xff1a;

#!/usr/bin/env python

#coding:utf-8

import MySQLdb

def insert_One():

&＃39;&＃39;&＃39;插入一条数据&＃39;&＃39;&＃39;

try:

conn&＃61;MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;,user&＃61;&＃39;root&＃39;,passwd&＃61;&＃39;server&＃39;,db&＃61;&＃39;day2017&＃39;)

except:

print u&＃39;连接mysql数据库失败&＃39;

else:

cur &＃61; conn.cursor()

sql &＃61; &＃39;INSERT INTO userInfo VALUES (%s,%s,%s,%s)&＃39;

params &＃61; (1, &＃39;admin&＃39;, &＃39;admin&＃39;, &＃39;wuya&＃64;outlook.com&＃39;,)

cur.execute(sql, params)

conn.commit()

finally:

cur.close()

conn.close()

if __name__&＃61;&＃61;&＃39;__main__&＃39;:

insert_One()

查看数据库&＃xff0c;可以看到&＃xff0c;数据已经插入到数据库中&＃xff0c;见查询的结果:

在上面的案例中&＃xff0c;只是插入了单条数据&＃xff0c;实际上&＃xff0c;某些时候&＃xff0c;会插入多条数据&＃xff0c;也就是批量插入&＃xff0c;批量插入实现的代码为&＃xff1a;

def insert_Many():

&＃39;&＃39;&＃39;批量插入数据&＃39;&＃39;&＃39;

try:

conn&＃61;MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;,user&＃61;&＃39;root&＃39;,passwd&＃61;&＃39;server&＃39;,db&＃61;&＃39;day2017&＃39;)

except:

print u&＃39;连接mysql数据库失败&＃39;

else:

cur &＃61; conn.cursor()

sql &＃61; &＃39;INSERT INTO userInfo VALUES (%s,%s,%s,%s)&＃39;

params &＃61; [

(2,&＃39;wuya&＃39;,&＃39;admin&＃39;,&＃39;wuya&＃64;outlook.com&＃39;),

(3,&＃39;weke&＃39;,&＃39;admin&＃39;,&＃39;weke&＃64;outlook.com&＃39;)

]

cur.executemany(sql, params)

conn.commit()

finally:

cur.close()

conn.close()

if __name__&＃61;&＃61;&＃39;__main__&＃39;:

insert_Many()

接下来&＃xff0c;我们来查看数据库的查询&＃xff0c;数据查询分为二种&＃xff0c;一种是查询的结果是一条语句&＃xff0c;使用的是fetchone()方法&＃xff0c;另外一种是查询的数据

结果是多条&＃xff0c;使用的方法是fetchmany()&＃xff0c;我们分别来看这二个方法的使用&＃xff0c;我们先来看单条数据的查询&＃xff0c;见实现的代码:

import MySQLdb

def select_one():

&＃39;&＃39;&＃39;单条数据的查询&＃39;&＃39;&＃39;

try:

conn&＃61;MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;,user&＃61;&＃39;root&＃39;,passwd&＃61;&＃39;server&＃39;,db&＃61;&＃39;day2017&＃39;)

except:

print u&＃39;连接mysql数据库失败&＃39;

else:

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

sql &＃61; &＃39;select * from userInfo where id&＃61;%s&＃39;

params &＃61; (1,)

data&＃61;cur.execute(sql, params)

print cur.fetchone()

conn.commit()

finally:

cur.close()

conn.close()

if __name__&＃61;&＃61;&＃39;__main__&＃39;:

select_one()

多条数据的查询&＃xff0c;见实现的代码&＃xff1a;

import MySQLdb

def select_Many():

&＃39;&＃39;&＃39;多条数据的查询&＃39;&＃39;&＃39;

try:

conn&＃61;MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;,user&＃61;&＃39;root&＃39;,passwd&＃61;&＃39;server&＃39;,db&＃61;&＃39;day2017&＃39;)

except:

print u&＃39;连接mysql数据库失败&＃39;

else:

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

sql &＃61; &＃39;select * from userInfo&＃39;

ret&＃61;cur.execute(sql)

data&＃61;cur.fetchall()

for item in data:

print item

finally:

cur.close()

conn.close()

if __name__&＃61;&＃61;&＃39;__main__&＃39;:

select_Many()

下面我们来看更新语句的测试&＃xff0c;见实现的代码&＃xff1a;

import MySQLdb

def update_Test():

&＃39;&＃39;&＃39;更新语句测试&＃39;&＃39;&＃39;

try:

conn&＃61;MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;,user&＃61;&＃39;root&＃39;,passwd&＃61;&＃39;server&＃39;,db&＃61;&＃39;day2017&＃39;)

except:

print u&＃39;连接mysql数据库失败&＃39;

else:

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

sql &＃61; &＃39;update userInfo set username&＃61;%s where id&＃61;%s&＃39;

params&＃61;(&＃39;system&＃39;,1,)

ret&＃61;cur.execute(sql,params)

conn.commit()

finally:

cur.close()

conn.close()

if __name__&＃61;&＃61;&＃39;__main__&＃39;:

update_Test()

最后一步&＃xff0c;也就是删除数据了&＃xff0c;直接看如下的实现代码&＃xff1a;

import MySQLdb

def delete_Test():

&＃39;&＃39;&＃39;删除语句测试&＃39;&＃39;&＃39;

try:

conn&＃61;MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;,user&＃61;&＃39;root&＃39;,passwd&＃61;&＃39;server&＃39;,db&＃61;&＃39;day2017&＃39;)

except:

print u&＃39;连接mysql数据库失败&＃39;

else:

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

sql &＃61; &＃39;delete from userInfo where id&＃61;%s&＃39;

params&＃61;(3,)

ret&＃61;cur.execute(sql,params)

conn.commit()

finally:

cur.close()

conn.close()

if __name__&＃61;&＃61;&＃39;__main__&＃39;:

delete_Test()

事实上&＃xff0c;对于如上操作数据库的方式&＃xff0c;有很多的代码是可以重够的&＃xff0c;比如连接数据库的方式&＃xff0c;另外&＃xff0c;我们可以把操作数据库的

方式写在一个类里面&＃xff0c;在业务调用的时候直接调用我们的数据库方法进行操作&＃xff0c;见下面操作mysql数据库的方法&＃xff0c;见源码&＃xff1a;

#!/usr/bin/env python

#coding:utf-8

import MySQLdb

class MySQLHelper(object):

def __init__(self):

pass

def get_one(self,sql,params):

conn &＃61; MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;,user&＃61;&＃39;root&＃39;,passwd&＃61;&＃39;server&＃39;,db&＃61;&＃39;day2017&＃39;)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

retCount &＃61; cur.execute(sql,params)

data &＃61; cur.fetchone()

cur.close()

conn.close()

return data

def get_many(self,sql,params):

conn &＃61; MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;,user&＃61;&＃39;root&＃39;,passwd&＃61;&＃39;server&＃39;,db&＃61;&＃39;day2017&＃39;)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

retCount &＃61; cur.execute(sql,params)

data &＃61; cur.fetchall()

cur.close()

conn.close()

return data

def insert_one(self,sql,params):

conn &＃61; MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;, user&＃61;&＃39;root&＃39;, passwd&＃61;&＃39;server&＃39;, db&＃61;&＃39;day2017&＃39;)

cur &＃61; conn.cursor()

cur.execute(sql, params)

conn.commit()

cur.close()

return u&＃39;插入数据库成功&＃39;

def insert_many(self,sql,params):

conn &＃61; MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;, user&＃61;&＃39;root&＃39;, passwd&＃61;&＃39;server&＃39;, db&＃61;&＃39;day2017&＃39;)

cur &＃61; conn.cursor()

cur.executemany(sql, params)

conn.commit()

cur.close()

return u&＃39;批量插入数据库成功&＃39;

def update_one(self,sql,params):

conn &＃61; MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;, user&＃61;&＃39;root&＃39;, passwd&＃61;&＃39;server&＃39;, db&＃61;&＃39;day2017&＃39;)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

ret &＃61; cur.execute(sql, params)

conn.commit()

cur.close()

conn.close()

return u&＃39;更新数据库成功&＃39;

def delete_one(self,sql,params):

conn &＃61; MySQLdb.connect(host&＃61;&＃39;127.0.0.1&＃39;, user&＃61;&＃39;root&＃39;, passwd&＃61;&＃39;server&＃39;, db&＃61;&＃39;day2017&＃39;)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

ret &＃61; cur.execute(sql, params)

conn.commit()

cur.close()

conn.close()

return u&＃39;删除数据库成功&＃39;

把连接数据库部分进行重构&＃xff0c;放到一个config.py的文件中&＃xff0c;这样我们连接数据库的方法就只需要在config.py文件维护了&＃xff0c;而不

需要在如上代码中每个都得看的修改&＃xff0c;这实在是很糟糕&＃xff0c;见重构后的config.py文件源码&＃xff1a;

#!/usr/bin/env python

#coding:utf-8

conn_dict&＃61;dict(host&＃61;&＃39;127.0.0.1&＃39;, user&＃61;&＃39;root&＃39;, passwd&＃61;&＃39;server&＃39;, db&＃61;&＃39;day2017&＃39;)

见重构后操作mysql的数据库方法&＃xff0c;见源码&＃xff1a;

#!/usr/bin/env python

#coding:utf-8

import MySQLdb

import config

class MySQLHelper(object):

def __init__(self):

self.conn&＃61;config.conn_dict

def get_one(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

retCount &＃61; cur.execute(sql,params)

data &＃61; cur.fetchone()

cur.close()

conn.close()

return data

def get_many(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

retCount &＃61; cur.execute(sql,params)

data &＃61; cur.fetchall()

cur.close()

conn.close()

return data

def insert_one(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor()

cur.execute(sql, params)

conn.commit()

cur.close()

return u&＃39;插入数据库成功&＃39;

def insert_many(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor()

cur.executemany(sql, params)

conn.commit()

cur.close()

return u&＃39;批量插入数据库成功&＃39;

def update_one(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

ret &＃61; cur.execute(sql, params)

conn.commit()

cur.close()

conn.close()

return u&＃39;更新数据库成功&＃39;

def delete_one(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

ret &＃61; cur.execute(sql, params)

conn.commit()

cur.close()

conn.close()

return u&＃39;删除数据库成功&＃39;

写数据库的操作方法&＃xff0c;是为了进行对业务的操作&＃xff0c;要不仅仅写这些没什么实际的意义&＃xff0c;如我们实现输入用户名和密码&＃xff0c;在

在数据库中验证&＃xff0c;如果用户名和密码都是admin,那么通过&＃xff0c;如果有其中一个不是admin,就提示用户&＃xff0c;请提示用户用户名

或者密码错误&＃xff0c;下面来实现这样的一个过程&＃xff0c;见实现的源码&＃xff1a;

#!/usr/bin/env python

#coding:utf-8

import MySQLdb

import config

class MySQLHelper(object):

def __init__(self):

self.conn&＃61;config.conn_dict

def get_one(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

retCount &＃61; cur.execute(sql,params)

data &＃61; cur.fetchone()

cur.close()

conn.close()

return data

def get_many(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

retCount &＃61; cur.execute(sql,params)

data &＃61; cur.fetchall()

cur.close()

conn.close()

return data

def insert_one(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor()

cur.execute(sql, params)

conn.commit()

cur.close()

return u&＃39;插入数据库成功&＃39;

def insert_many(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor()

cur.executemany(sql, params)

conn.commit()

cur.close()

return u&＃39;批量插入数据库成功&＃39;

def update_one(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

ret &＃61; cur.execute(sql, params)

conn.commit()

cur.close()

conn.close()

return u&＃39;更新数据库成功&＃39;

def delete_one(self,sql,params):

conn &＃61; MySQLdb.connect(**self.conn)

cur &＃61; conn.cursor(cursorclass&＃61;MySQLdb.cursors.DictCursor)

ret &＃61; cur.execute(sql, params)

conn.commit()

cur.close()

conn.close()

return u&＃39;删除数据库成功&＃39;

class CheckUserInfo(object):

def __init__(self):

self.__helper&＃61;MySQLHelper()

def checkValid(self,username,password):

sql&＃61;&＃39;select * from userInfo where username&＃61;%s and password&＃61;%s&＃39;

params&＃61;(username,password)

return self.__helper.get_one(sql,params)

def info():

username&＃61;raw_input(u&＃39;请输入你的用户名:\n&＃39;)

password&＃61;raw_input(u&＃39;请输入你的密码:\n&＃39;)

userInfo&＃61;CheckUserInfo()

result&＃61;userInfo.checkValid(username,password)

if not result:

print u&＃39;用户名或者密码错误&＃xff0c;请联系管理员&＃39;

else:

print u&＃39;恭喜您&＃xff0c;输入正确&＃xff01;&＃39;

if __name__&＃61;&＃61;&＃39;__main__&＃39;:

info()