python3.6+requests实现接口自动化3

逐步完善中……

 本篇记录连接mysql，并执行sql语句，可以参考https://www.runoob.com/python3/python3-mysql.html

目录

1、安装连接mysql的模块

2、了解模块中的常用方法以及使用

1、安装连接mysql模块

使用pip完成模块的安装，执行pip3 install PyMySQL：

安装安装后，可以看到模块是可以正常导入的

 

2、了解模块中的常用方法以及使用

第一步：使用该模块连接mysql

#!/usr/bin/python3
 
import pymysql
 
# 打开数据库连接
db = pymysql.connect("localhost","testuser","test123","TESTDB" )

封装：

def  Connect(mysql_info):        
        try:
            conn = MySQLdb.connect(host=db_info['host'],
            port=db_info['port'],
            user=db_info['user'],
            passwd=db_info['passwd'],
            db=db_info['db'],
            charset=db_info['charset'])
            return conn
        except Exception as a:
            print("数据库连接异常：%s"%a)　　

传入数据库信息：

mysql_info = {"host": 'localhost',
                "port": 3306,
                "user": 'root',
                "passwd": '123456',
                "db": 'aiopms',
                "charset": 'utf8'}

　　

第二步：数据库的查、增、删

查：

    def mysql_getrows(self, sql):
        ''' 返回查询结果'''
        cur = self.conn.cursor()
        try:
            cur.execute(sql)
        except Exception as a:
            print("执行 SQL 语句出现异常：%s"%a)
        else:
            rows = cur.fetchall()
            cur.close()
            return rows

增、删：使用execute()方法执行语句

def mysql_execute(self, sql):
        #通过cursor()方法来创建游标
        cur = self.conn.cursor()
        try:
            #execute()方法执行语句
            a=cur.execute(sql)
        except Exception as a:
            self.conn.rollback() # sql 执行异常后回滚
            print("执行 SQL 语句出现异常：%s"%a)
        else:
            cur.close()
            self.conn.commit() # sql 无异常时提交

应用：

比如现在我要测一个新增接口，那么就先调用数据库查一下，如果存在就先执行sql删除；或者是我现在测新增接口，用sql来处理脏数据，比如现在setup中查一下，不存在就执行新增接口，存在就是执行删除sql，在tearDown中再删除调接口新增的数据：

新增接口：

import requests
class add_user():
	def add_user01(self,username):
		add_user_header={"":""}
		add_user_url="http://xx:8089/user/add"
		add_body={"username":username,
				 "password":"123456")
		p=requests.post(add_user_url,headers=add_user_header,data=add_body)
		return p.json()['message']　

test新增接口的用例：

import  unittest
class add(unittest.TestCase):
    def setUp(self):
        addresult1=self.add()
        sql1="select***"
        sql2="delete***"
        m1=sql.getrows(sql)
        if m1==None:
            pass
        else
            excute(sql2)
        
    def test_add_01(self,username):
        m=self.add("qq1")
        self.assertEqual(m,"新增成功")
        
    def test_add_02(self,username): 
        ……
        
    def tearDown(self):
        excute(sql2)