RF库其实和用selenium自动化测试调用一些方法是类似的，在selenium中，我们会定义一些类和方法然后调用，但是RF框架是用关键字的，所以有时候需要我们用python进行方法和库的定义，然后把一些关键字传给RF框架。
我们举一个例子，首先，我这里新建了个userdata.txt文本，具体内容如下，然后我想通过定义方法，来获取这个文本中每一行的内容。具体代码：
# -*- coding:utf-8 -*-
def Load_Userdata(fname):
udata = []
with open(fname) as f:
for line in f:
r =line.strip().split(' ')
udata.append(r)
return udata
if __name__ == '__main__':
fpath = r'C:\Users\Administrator\Desktop\userdata.txt'
result= Load_Userdata(fpath)
#print(result)
for i in result:
print(i)
运行结果如下：
这样的话，我们现在就完成了读取txt文本的每行内容，现在如何RF框架来调用呢，继续上图：
step1：通过cmd运行ride.py打开RobotFrameWork，然后新建一个工程，在工程下新建一个suite，再在suite下新建一个case用例命名为test_library：然后鼠标选中Search，点击界面中的‘Library’，之后会弹出来个对话框，把data_read添加到该路径即可。
然后返回到主界面就可以看到已经添加的库了：
我们新建一个test_library来测试一下：具体代码如下
运行结果，没有报错表明我们自己新建的库可以正常使用。
实践：
最近我们项目要测试图库(neo4j)同步，而Robot Framework不支持图库API,所以我就写了一个neo4j的测试库，今天跟大家分享下。 (Neo4j是一个高性能的,NOSQL图形数据库，它完全支持ACID(原子性（Atomicity）、一致性（Consistency）、隔离性（Isolation）、持久性（Durability）)数据库事务属性，采用JAVA语音编写，由于其良好的图数据模型设计，Neo4j的速度非常快。对于连接的数据操作，Neo4j的速度要比传统的关系型数据库快1000倍。部署一个neo4j服务器便可以承载上亿级的节点和关系。当单节点无法承载我们的数据需求时，我们可以进行分布式集群部署.)
【需求】
neo4j图库从pg数据库同步数据，检查同步后数据源和目标库数据是否一致
【实现步骤】
1.测试设计
根据测试的需求我们设计如下的测试。
因为本次同步设计到多个节点和多个关系。基本的思路是：
1）编写sql脚本，Robot Framework调用Execute Sql Script关键字，增加如上的一个子图（包括节点和关系）到pg源数据库
2）通过远程ssh调用同步代码，进行neo4j图库同步
3）待同步完成，调用Library进行neo4j数据获取,调用自带的DatabaseLibrary进行源数据获取
4) 使用padas的dataframe进行数据比较，输出判断结果
5）编写sql脚本，进行子图更新
6）重复2-4步骤
7）编写sql脚本，进行子图删除
8）重复2-4步骤
2.编写Library
首先我们编写第3,4步骤的代码。编写完成后放到site-packages目录下。
主要提供如下功能：
主要方法1：get_node_and_relation
功能：封装neo4j的调用，实现查询图库的实体和关系的功能
核心代码：
def get_node_and_relation(self,url,username=None,password=None,queryType=None):
# 定义匹配,查询查询neo4j
logger.info('开始查询neo4j中%s的检查'%( queryType))
graph = Graph(url,username=username,password=password)
# print(graph)
if queryType.find('_') > -1:
rels = graph.match((), r_type=queryType)
else:
matcher = NodeMatcher(graph)
rels = matcher.match(queryType)
logger.info('数据长度为:%d'%(len(rels)))
# print(len(rels))
neo4jDf = pd.DataFrame(data=list(rels))
# print(queryType)
columns = self.getColumnsByType(queryType)[0]
# print(columns)
if columns != None:
neo4jDf.rename(columns=columns, inplace=True)
logger.info('neo4j表中%s的原始数据长度为:%d'%(queryType, neo4jDf.shape[0]))
return neo4jDf
主要方法2：getCompareAll
功能：对比源数据库pg取到的数据，与目标数据库Neo4j取到的数据库进行比较
1.首先确定是关系还是实体，如果是关系，去掉实体不存在的关系
2.比较neo4jDf，pgDf，计算出addset，diffset.并把不一致的结果输出成文件，并把不一致的条数返回
核心代码：
def getCompareAll(self, pgDf, neo4jDf, type,node1Result=None,node2Result=None):
logger.info('开始比较数据')
logger.info(type)
columnList = self.getColumnsByType(type)[1]
# 如果是关系，去掉实体不存在的关系
if type.find('_') > -1:
node1 = type.split("_", 2)[0]
node2 = type.split("_", 2)[2]
logger.info('开始检查关系实体数据是否存在')
nodedf1 = self.convertDbResultToDataframe(node1Result,node1)
node1list = list(set(list(nodedf1['code'])))
nodedf2 = self.convertDbResultToDataframe(node2Result,node2)
node2list = list(set(list(nodedf2['code'])))
pgDf = pgDf[pgDf['relation_code'].map(
lambda x: (x.split("_", 1)[0] in node1list) and (x.split("_", 1)[1] in node2list))]
logger.info('kettle表中%s的去掉无实体的关系后数据长度为:%d'%( type, pgDf.shape[0]))
# 去重
if type.find('_') > -1:
pgDf.drop_duplicates(['relation_code'], keep='first', inplace=True)
else:
pgDf.drop_duplicates(['code'], keep='first', inplace=True)
logger.info('kettle表中%s的去重后数据长度为:%d'%( type, pgDf.shape[0]))
returnCount = 0
if (pgDf is not None):
neo4jset = set(neo4jDf[columnList[0]])
pgDfset = set(pgDf[columnList[0]])
addset = neo4jset - pgDfset
diffset = pgDfset - neo4jset
if neo4jset != pgDfset:
logger.error('[' + type + ']数据重复的主键条数为: %d条' % len(addset))
# logger.error('[' + type + ']数据重复的主键数据为: %s条', addset)
logger.error('[' + type + ']数据缺失的主键条数为: %d条' % len(diffset))
# logger.error('[' + type + ']数据缺失的主键数据为: %s条', diffset)
if len(addset) > 0:
adddf = pd.DataFrame(list(addset))
returnCount = returnCount + adddf.shape[0]
adddf.to_csv('neo4j_' + type + 'addset.csv', encoding='UTF-8-SIG')
if len(diffset) > 0:
diffdf = pd.DataFrame(list(diffset))
returnCount = returnCount + diffdf.shape[0]
diffdf.to_csv('neo4j_' + type + 'diffset.csv', encoding='UTF-8-SIG')
else:
logger.info('数据条数一致，开始输出结果')
comparedf = pd.merge(neo4jDf, pgDf, on=columnList[0], how='outer')
comparedf['judge'] = 1
comparedf.fillna('', inplace=True)
i = 0
for column in columnList:
if i == 0:
i = i + 1
continue
comparedf[column + '_compare'] = comparedf.apply(
lambda x: True if x[column] == x[column + '_neo4j'] else False, axis=1)
comparedf[column + '_compare'] = comparedf[column + '_compare'].map(lambda x: 1 if x == True else 0)
comparedf['judge'] = comparedf['judge'] * comparedf[column + '_compare']
i = i + 1
comparedf['judge'] = comparedf['judge'].apply(lambda x: True if x == 1 else False)
deltadf = comparedf[comparedf['judge'] == False]
logger.info('完成比较neo4j数据')
logger.info('[' + type + ']数据不对应的条数为: %d条' % deltadf.shape[0])
# logger.info("数据不对应的条数为： %d条", deltadf.shape[0])
if deltadf.shape[0] > 0:
returnCount = returnCount + deltadf.shape[0]
deltadf.to_csv('neo4j_' + type + 'deltadf.csv', encoding='UTF-8-SIG')
return returnCount
3.编写测试脚本
如下脚本反映主要步骤：
脚本1：
*** Settings ***
Suite Setup Connect To Database Using Custom Params psycopg2 database=${pg_database},user= ${pg_username}, password=${pg_password} ,host=${pg_host}, port=${pg_port}
Suite Teardown Disconnect From Database
Library DatabaseLibrary
Library Neo4jLibrary
Library SSHLibrary
#Suite setup时候连接数据库，Suite teardown时候关闭数据库连接
脚本2：
*** Test Cases ***
dataInitial
[Tags] P1
Execute Sql Script ${rootDir}${sql_add_path
脚本3：
graphSync
[Tags] P0
Open Connection ${graph_ip} timeout=1 hour
Login ${graph_username} ${graph_password}
Write cd /app/graph
Write sh kg_job.sh
Read Until Program running status written to log library successfully!step_no:8510 INFO
Close All Connections
#调用SSHLibrary进行远程调用，执行数据同步，完成pg库到neo4j图库的数据同步过程
脚本4:
fundcompany
[Tags] P0
log ${neo4j_url}
log ${neo4j_username}
${pg_sql} getSqlByType fundcompany
${result_kettle} query ${pg_sql}
${df_pg} Convert Db Result To Dataframe ${result_kettle} fundcompany
${df_neo4j} Get Node And Relation ${neo4j_url} ${neo4j_username} ${neo4j_password} fundcompany
${returnCount} getCompareAll ${df_pg} ${df_neo4j} fundcompany
log ${returnCount}
Should Be Equal As Integers 0 ${returnCount} 'fundcompany compare failed'
脚本解析：
进行各个实体，关系的比较，这里以一个实体fundcompany为例
【pg源数据查询】
首先获取到实体查询语句
${pg_sql} getSqlByType fundcompany
进行查询
${result_kettle} query ${pg_sql}
把查询结果转换为dataframe
${df_pg} Convert Db Result To Dataframe ${result_kettle} fundcompany
【neo4j目标数据查询】
调用自定义library查询
${df_neo4j} Get Node And Relation ${neo4j_url} ${neo4j_username} ${neo4j_password} fundcompany
【源数据和目标数据比较】
${returnCount} getCompareAll ${df_pg} ${df_neo4j} fundcompany
log ${returnCount}
Should Be Equal As Integers 0 ${returnCount} 'fundcompany compare failed'
脚本5：更新子图数据，模拟update场景
data Update
Execute Sql Script ${rootDir}${sql_update_path}
脚本6：调用同步
graphSync
[Tags] P0
Open Connection ${graph_ip} timeout=1 hour
Login ${graph_username} ${graph_password}
Write cd /app/graph
Write sh kg_job.sh
Read Until Program running status written to log library successfully!step_no:8510 INFO
Close All Connections
脚本7：进行各个实体，关系的比较
fundcompany
[Tags] P0
[Setup] log hello liuqiong
log ${neo4j_url}
log ${neo4j_username}
${pg_sql} getSqlByType fundcompany
${result_kettle} query ${pg_sql}
${df_pg} Convert Db Result To Dataframe ${result_kettle} fundcompany
${df_neo4j} Get Node And Relation ${neo4j_url} ${neo4j_username} ${neo4j_password} fundcompany
${returnCount} getCompareAll ${df_pg} ${df_neo4j} fundcompany
log ${returnCount}
Should Be Equal As Integers 0 ${returnCount} 'fundcompany compare failed'
脚本8：更新子图数据，模拟delete场景
dataClear
[Tags] P1
Execute Sql Script ${rootDir}${sql_delete_path}
脚本9:调用同步
graphSync
[Tags] P0
Open Connection ${graph_ip} timeout=1 hour
Login ${graph_username} ${graph_password}
Write cd /app/graph
Write sh kg_job.sh
Read Until Program running status written to log library successfully!step_no:8510 INFO
Close All Connections
脚本10：进行各个实体，关系的比较
fundcompany
[Tags] P0
[Setup] log hello liuqiong
log ${neo4j_url}
log ${neo4j_username}
${pg_sql} getSqlByType fundcompany
${result_kettle} query ${pg_sql}
${df_pg} Convert Db Result To Dataframe ${result_kettle} fundcompany
${df_neo4j} Get Node And Relation ${neo4j_url} ${neo4j_username} ${neo4j_password} fundcompany
${returnCount} getCompareAll ${df_pg} ${df_neo4j} fundcompany
log ${returnCount}
Should Be Equal As Integers 0 ${returnCount} 'fundcompany compare failed'
4.对测试库进行测试
如果是开发环境/本地执行，直接使用Ride的Run界面。
如果是测试环境/生产环境，调用命令行执行，可编写如下的shell脚本定时执行：
export LANG=en_US.UTF-8
# up to env
log_path=/var/log/rbpaa/batch
ROBOT_HOME=/app/rbppTest/RobotFramework/RobotAdvisorProject/Graph
day=$(date +%Y%m%d)
log_file=${log_path}/cron_neo4j_job_${day}.log
err_file=${log_path}/error/rbpaa_${day}.err
echo $(date '+%Y-%m-%d %H:%M:%S') start execute job GraphTest
varabileFile=variables.py
case_list=( 'UpdateTest.txt' )
while [[ i -lt ${#case_list[@]} ]]; do
bh=${case_list[i]}
echo execute ${bh}
/var/python3.6.5/bin/robot -d $ROBOT_HOME -V $ROBOT_HOME/${varabileFile}:test $ROBOT_HOME/${bh} |tee -a ${log_file}
result=$?
echo $(date '+%Y-%m-%d %H:%M:%S') ${bh} result=${result}
if [ ${result} -ne 0 ]; then
echo error exit.
echo ${bh} error >> ${err_file}
exit
fi
let i++
done
echo $(date '+%Y-%m-%d %H:%M:%S') job GraphTest end.
核心语句为斜体部分，调用robot命令执行
-d设置代码目录
-V设置变量文件
网友评论