集成数据库操作
在做接口测试时，有时我们需要去查询业务数据库，做一些数据相关的校验，所以aomaker集成了一些数据库的相关操作，目前支持 sqlite和mysql两种数据库的一些常规操作。
如何使用？（以mysql为例）
先在配置文件conf/config.yaml中配置对应环境的数据库连接信息：
config.yaml
env: test
test:
account:
pwd: test01
user: test01
host: http://test.aomaker.com
# 配置数据库信息
mysql:
host: localhost
user: admin
password: 123456
port: 3306
一般以pytest本地插件的方式去封装获取数据库连接对象的方法，在conftest.py下，封装一个mysql连接插件：
conftest.py
import pytest
from aomaker.database.mysql import Mysql
from aomaker.cache import config
@pytest.fixture(scope="class")
def connect_mysql():
# 读取数据库配置信息
mysql_conf = config.get('mysql')
# 创建mysql连接对象
db = Mysql(**mysql_conf)
yield db
# 测试结束后，关闭游标对象和数据库连接
db.close()
本地插件做好后，在用例层直接传入数据库连接对象：
test_demo.py
class TestDemo(BaseTestcase):
# 以参数形式传入数据库连接插件
def test_demo(self, connect_mysql):
# 获取数据库连接对象
db = connect_mysql
# 通过数据连接对象操作数据库
sql_res = db.get_all("select * from xxx")
aomaker内置的mysql：
class Mysql:
def __init__(self, **kwargs):
# 连接到数据库
try:
self.con = pymysql.connect(charset="utf8", **kwargs)
except Exception as e:
logger.error(f'数据库连接失败，连接参数：{kwargs}')
raise e
else:
# 创建一个游标
self.cur = self.con.cursor()
def get_one(self, sql):
"""获取查询到的第一条数据"""
self.con.commit()
self.cur.execute(sql)
return self.cur.fetchone()
def get_all(self, sql):
"""获取sql语句查询到的所有数据"""
self.con.commit()
self.cur.execute(sql)
return self.cur.fetchall()
def count(self, sql):
"""获取sql语句查询到的数量"""
self.con.commit()
res = self.cur.execute(sql)
return res
def close(self):
# 关闭游标对象
self.cur.close()
# 断开连接
self.con.close()
如果有其它数据库需求，可以自行扩展，使用方式同上。