提交代码
This commit is contained in:
parent
e93d5bf046
commit
e7da81ca7a
|
|
@ -4,6 +4,7 @@ Resource Element.robot
|
|||
Resource Common.robot
|
||||
Variables ../config/myConfig.py
|
||||
Library DateTime
|
||||
Library ../Script/mssqlTest.py
|
||||
|
||||
*** Keywords ***
|
||||
登录黑方
|
||||
|
|
@ -148,4 +149,37 @@ Library DateTime
|
|||
Log ${recoveryname}
|
||||
作业校验 ${recoveryname}
|
||||
|
||||
初始化数据库
|
||||
# 创建数据库
|
||||
mssqlTest.Create Database
|
||||
# 创建表
|
||||
mssqlTest.Create Table
|
||||
# 插入数据
|
||||
mssqlTest.Insert Data Table
|
||||
# 查询插入的结果
|
||||
${select_result}= mssqlTest.Get Insert Res
|
||||
Log 查询到的结果是：${select_result}
|
||||
Set Global Variable ${first_result} ${select_result}
|
||||
|
||||
删除表确认该表不存在
|
||||
# 先确认表是否存在
|
||||
${table_status1}= mssqlTest.Check Table Exists
|
||||
Should Be True ${table_status1} 要删除的表不存在
|
||||
Log TABLE_TEST表存在,开始做删除表操作
|
||||
mssqlTest.Drop Table
|
||||
# 删除后检查
|
||||
${table_status2}= mssqlTest.Check Table Exists
|
||||
Should Not Be True ${table_status2} 删除失败，删除的表还能查询到
|
||||
Log TABLE_TEST已删除
|
||||
|
||||
校验恢复结果
|
||||
# 先确认表是否存在
|
||||
${table_status}= mssqlTest.Check Table Exists
|
||||
Should Be True ${table_status} 要检查的表不存在，请确认恢复结果
|
||||
Log 表已经恢复成功，开始检查表结果
|
||||
# 查询插入的结果
|
||||
${select_result}= mssqlTest.Get Insert Res
|
||||
Log 查询到的结果是：${select_result}
|
||||
# 将其与之前的结果做对比
|
||||
Should Be Equal As Strings ${first_result} ${select_result} 恢复前后的结果不匹配
|
||||
Log 恢复成功，恢复结果与预期一致
|
||||
|
|
|
|||
|
|
@ -0,0 +1,115 @@
|
|||
# encoding=utf-8
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import jaydebeapi
|
||||
|
||||
|
||||
class mssqlTest(object):
|
||||
def __init__(self):
|
||||
# 公共的
|
||||
path = Path(__file__).parent
|
||||
# 数据库连接相关的参数
|
||||
driver_name = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
|
||||
jdbc_url = "jdbc:sqlserver://10.10.13.249:1433"
|
||||
OAuth = ["sa", "unary@2008"]
|
||||
driver_jar_path = path / "jdbc-driver" / "mssql-jdbc-9.2.0.jre8.jar"
|
||||
|
||||
# 连接数据库
|
||||
self.conn = jaydebeapi.connect(driver_name,
|
||||
jdbc_url,
|
||||
OAuth,
|
||||
str(driver_jar_path))
|
||||
|
||||
|
||||
# 执行sql
|
||||
def execute_sql(self, sql):
|
||||
with self.conn.cursor() as curs:
|
||||
curs.execute(sql)
|
||||
|
||||
|
||||
# 创建库
|
||||
def create_database(self):
|
||||
print("start create databases...")
|
||||
sql_string = "CREATE DATABASE UNATEST"
|
||||
try:
|
||||
self.execute_sql(sql_string)
|
||||
except:
|
||||
self.drop_database()
|
||||
print("start try create databases again")
|
||||
self.execute_sql(sql_string)
|
||||
print("CREATE DATABASE UNATEST sucess!")
|
||||
|
||||
#创建表
|
||||
def create_table(self):
|
||||
print("start create table...")
|
||||
sql_string = "create table UNATEST.dbo.TABLE_TEST(test char(15))"
|
||||
try:
|
||||
self.execute_sql(sql_string)
|
||||
except:
|
||||
self.drop_table()
|
||||
print("start try create table TABLE_TEST again")
|
||||
self.execute_sql(sql_string)
|
||||
print("CREATE create TABLE_TEST sucess!")
|
||||
|
||||
|
||||
# 插入数据
|
||||
def insert_data_table(self):
|
||||
print("start insert into UNATEST.dbo.TABLE_TEST")
|
||||
# 插入当前的时间
|
||||
actime_time = datetime.now().strftime('%Y%m%d%H%M%S')
|
||||
sql_string ="INSERT INTO UNATEST.dbo.TABLE_TEST (test) VALUES(N'"+ actime_time +"')"
|
||||
print("insert data is : " + actime_time)
|
||||
self.execute_sql(sql_string)
|
||||
|
||||
|
||||
|
||||
# 删除表
|
||||
def drop_table(self):
|
||||
print("start drop table TABLE_TEST...")
|
||||
sql_string = "DROP table UNATEST.dbo.TABLE_TEST"
|
||||
try:
|
||||
self.execute_sql(sql_string)
|
||||
except:
|
||||
print("TABLE_TEST already drop.")
|
||||
|
||||
|
||||
# 删除库
|
||||
def drop_database(self):
|
||||
print("start drop databases...")
|
||||
sql_string = "DROP DATABASE UNATEST"
|
||||
try:
|
||||
self.execute_sql(sql_string)
|
||||
except:
|
||||
print("databases already drop.")
|
||||
|
||||
|
||||
# 获取插入的结果
|
||||
def get_insert_res(self):
|
||||
sql_string = "select * from UNATEST.dbo.TABLE_TEST"
|
||||
with self.conn.cursor() as select:
|
||||
select.execute(sql_string)
|
||||
res = select.fetchall()
|
||||
print(res)
|
||||
# 将第一个结果返回
|
||||
return res[0]
|
||||
|
||||
|
||||
# 检查表是否存在
|
||||
def check_table_exists(self):
|
||||
sql_string = "select * from UNATEST.dbo.sysObjects WHERE name='TABLE_TEST'"
|
||||
with self.conn.cursor() as select:
|
||||
select.execute(sql_string)
|
||||
res = select.fetchall()
|
||||
# 根据结果数据量来判断表是否存在
|
||||
if len(res) == 1:
|
||||
exists = True
|
||||
else:
|
||||
exists = False
|
||||
return exists
|
||||
|
||||
|
||||
# 关闭连接
|
||||
def close_conn(self):
|
||||
self.conn.close()
|
||||
|
||||
|
|
@ -7,10 +7,10 @@ Resource ../Resources/Common.robot
|
|||
*** Test Cases ***
|
||||
Sqlserver副本备份恢复
|
||||
[Setup] 登录黑方 # 登录黑方
|
||||
# 初始化数据库
|
||||
初始化数据库
|
||||
创建Sqlserver副本备份策略
|
||||
执行备份任务并校验
|
||||
# 删除表确认该表不存在
|
||||
删除表确认该表不存在
|
||||
创建恢复任务并校验
|
||||
# 校验恢复结果
|
||||
校验恢复结果
|
||||
[Teardown] Close Browser # Close Browser
|
||||
|
|
|
|||
|
|
@ -18,5 +18,5 @@ PolicyName = "AutoTest-SQLServerCDM-Backup-" + actime_time
|
|||
# 代理名称
|
||||
hostname = "SqlserverCDM"
|
||||
# 过滤关键词
|
||||
keyFilter = "test1"
|
||||
keyFilter = "test"
|
||||
recTargetPath='C:\\data'
|
||||
|
|
|
|||
Loading…
Reference in New Issue