Skip to content

Z
zmops-test
Commit 854115d9 by sanshi
Options

IP地址簿

parent c964d828
Showing with 318 additions and 139 deletions
WorkCase/APP/Experience/test_experience_create.py
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
from __future__ import division from __future__ import division
from WorkApi.APP.Api.api_login import ApiLogin from WorkApi.APP.Api.api_login import ApiLogin
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase from WorkCase import CaseBase
from WorkData.Argus.members import DataMembers from WorkData.Argus.members import DataMembers
from WorkUtils.UtilsLog import UtilsLog from WorkUtils.UtilsLog import UtilsLog
...@@ -31,7 +32,6 @@ from WorkData.Argus.hosts_items import DataHostsItems ...@@ -31,7 +32,6 @@ from WorkData.Argus.hosts_items import DataHostsItems
from WorkData.Argus.problem import DataProblem from WorkData.Argus.problem import DataProblem
from WorkData.Argus.tag import DataTag from WorkData.Argus.tag import DataTag
import allure import allure
...@@ -52,15 +52,16 @@ class TestExperienceCreate(object): ...@@ -52,15 +52,16 @@ class TestExperienceCreate(object):
base_type_agent = 126 base_type_agent = 126
base_type_snmp = 127 base_type_snmp = 127
base_business_id_1 = "12"
base_business_id_2 = "13"
base_ip = "172.16.3.170"
base_port = 161
base_title = "SS测试经验标题" base_title = "SS测试经验标题"
base_description = "SS测试经验描述"
base_content = "SS测试经验内容" base_content = "SS测试经验内容"
base_businessName = "SS测试经验业务名称"
base_monitorName = "SS测试经验监控名称"
base_hostid = 10266
base_name = "经验手册-测试数据" base_name = "经验手册-测试数据"
base_eventid = 91000000
@classmethod @classmethod
def setup_class(cls): def setup_class(cls):
...@@ -204,48 +205,108 @@ class TestExperienceCreate(object): ...@@ -204,48 +205,108 @@ class TestExperienceCreate(object):
return sql return sql
@allure.step("创建测试数据") @allure.step("创建测试数据")
def case_create(self,): def host_create(self, num):
# 删除第一个表 name = self.base_name + num
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_snmp, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None, port=None, businessTree=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.port = port
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.businessTree = businessTree
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session() session = self.db_session()
sql = DataHostsItems().select_all_from_allKeys(session=session, host_id=host_id)
return sql
@allure.step("创建测试数据")
def case_create(self, num):
self.title = self.base_title + num
self.content = self.base_content + num
self.name = self.base_name + num
self.eventid = self.base_eventid + int(num)
# 删除第一个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataAlarmRule().delete_like_name(session=session, name=self.base_name) DataAlarmRule().delete_like_name(session=session, name=self.base_name)
# 删除第二个表 # 删除第二个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataBusinesses().delete_like_name(session=session, name=self.base_name) DataBusinesses().delete_like_name(session=session, name=self.base_name)
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.base_name) sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.name)
for x, y in enumerate(sql): for x, y in enumerate(sql):
# 删除第三个表 # 删除第三个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid) DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表
session = self.db_session() # 删除第四个表
DataProblem().delete_hostid(session=session, hostid=y.hostid) session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_like_name(session=session, name=self.base_name)
# 删除第五个表 # 删除第五个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=self.base_name) DataHosts().delete_like_host_name(session=session, host_name=self.base_name)
# 删除第六个表 # 删除第六个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataTag().delete_like_tag(session=session, tag=self.base_name) DataTag().delete_like_tag(session=session, tag=self.base_name)
session = self.db_session() session = self.db_session()
self.alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.base_name, status=1, severity=1, self.alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.name, status=1, severity=1,
manual_close=0, enable_time_default=1, create_time=1) manual_close=0, enable_time_default=1, create_time=1)
session = self.db_session() session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.base_name, _type=1) self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
session = self.db_session()
self.hostid = DataHosts().insert_one(session=session, host_name=self.base_name, del_flag=0)
session = self.db_session() self.hostid = self.host_create(num=num)
self.itemid = DataHostsItems().insert_one(session=session, host_id=self.hostid, name=self.base_name, key_="key_")
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
for x, y in enumerate(sql):
itemids.append(y.id)
self.itemid = itemids[0]
session = self.db_session() session = self.db_session()
self.eventid = DataProblem().insert_one(session=session, eventid=999999, source=0, _object=0, objectid=0, clock=0, ns=0, DataProblem().insert_one(session=session, eventid=self.eventid, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0, r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.base_name) status=0, hostid=self.hostid, name=self.name)
@allure.step("校验数据") @allure.step("校验数据")
def check_select(self, response): def check_select(self, response):
...@@ -298,7 +359,7 @@ class TestExperienceCreate(object): ...@@ -298,7 +359,7 @@ class TestExperienceCreate(object):
self.title = self.base_title + "04" self.title = self.base_title + "04"
self.content = self.base_content + "04" self.content = self.base_content + "04"
response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_agent]) response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_snmp])
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
self.experienceid = UtilsResponse().get_data(response=response) self.experienceid = UtilsResponse().get_data(response=response)
sql = self.select_experience(experienceid=self.experienceid) sql = self.select_experience(experienceid=self.experienceid)
...@@ -310,12 +371,10 @@ class TestExperienceCreate(object): ...@@ -310,12 +371,10 @@ class TestExperienceCreate(object):
@allure.story("创建经验手册：所有字段") @allure.story("创建经验手册：所有字段")
@allure.severity("blocker") @allure.severity("blocker")
def test_case_05(self): def test_case_05(self):
self.case_create()
self.get_base_token() self.get_base_token()
self.case_create(num="05")
self.title = self.base_title + "05" response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_snmp],
self.content = self.base_content + "05"
response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_agent],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid], alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid], tagNameList=[self.base_name]) manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid], tagNameList=[self.base_name])
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
...@@ -344,7 +403,7 @@ class TestExperienceCreate(object): ...@@ -344,7 +403,7 @@ class TestExperienceCreate(object):
sql = self.select_experience_hosts_type(experienceid=self.experienceid) sql = self.select_experience_hosts_type(experienceid=self.experienceid)
assert sql != [] assert sql != []
for x, y in enumerate(sql): for x, y in enumerate(sql):
assert y.hosts_type == self.base_type_agent assert y.hosts_type == self.base_type_snmp
# 校验第五个表 # 校验第五个表
sql = self.select_experience_ips(experienceid=self.experienceid) sql = self.select_experience_ips(experienceid=self.experienceid)
assert sql != [] assert sql != []
......
WorkCase/APP/Experience/test_experience_delete.py
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
from __future__ import division from __future__ import division
from WorkApi.APP.Api.api_login import ApiLogin from WorkApi.APP.Api.api_login import ApiLogin
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase from WorkCase import CaseBase
from WorkData.Argus.members import DataMembers from WorkData.Argus.members import DataMembers
from WorkUtils.UtilsLog import UtilsLog from WorkUtils.UtilsLog import UtilsLog
...@@ -53,17 +54,17 @@ class TestExperienceDelete(object): ...@@ -53,17 +54,17 @@ class TestExperienceDelete(object):
base_type_agent = 126 base_type_agent = 126
base_type_snmp = 127 base_type_snmp = 127
base_business_id_1 = "12"
base_business_id_2 = "13"
base_ip = "172.16.3.170"
base_port = 161
base_title = "SS测试经验标题" base_title = "SS测试经验标题"
base_description = "SS测试经验描述"
base_content = "SS测试经验内容" base_content = "SS测试经验内容"
base_businessName = "SS测试经验业务名称" base_description = "SS测试经验描述"
base_monitorName = "SS测试经验监控名称"
base_hostid = 10266
base_name = "经验手册-测试数据" base_name = "经验手册-测试数据"
base_update_name = "经验手册-测试数据-删除" base_eventid = 91000000
base_ip = "172.99.1.100"
@classmethod @classmethod
def setup_class(cls): def setup_class(cls):
...@@ -203,71 +204,111 @@ class TestExperienceDelete(object): ...@@ -203,71 +204,111 @@ class TestExperienceDelete(object):
self.token = UtilsResponse().get_data(response=response) self.token = UtilsResponse().get_data(response=response)
@allure.step("创建测试数据") @allure.step("创建测试数据")
def host_create(self, num):
name = self.base_name + num
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_snmp, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None, port=None, businessTree=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.port = port
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.businessTree = businessTree
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session()
sql = DataHostsItems().select_all_from_allKeys(session=session, host_id=host_id)
return sql
@allure.step("创建测试数据")
def case_create(self, num): def case_create(self, num):
self.name = self.base_name + num
self.eventid = self.base_eventid + int(num)
# 删除第一个表 # 删除第一个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataAlarmRule().delete_like_name(session=session, name=self.base_name) DataAlarmRule().delete_like_name(session=session, name=self.base_name)
# 删除第二个表 # 删除第二个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataBusinesses().delete_like_name(session=session, name=self.base_name) DataBusinesses().delete_like_name(session=session, name=self.base_name)
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.base_name) sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.name)
for x, y in enumerate(sql): for x, y in enumerate(sql):
# 删除第三个表 # 删除第三个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid) DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表 # 删除第四个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_hostid(session=session, hostid=y.hostid) DataProblem().delete_like_name(session=session, name=self.base_name)
# 删除第五个表 # 删除第五个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=self.base_name) DataHosts().delete_like_host_name(session=session, host_name=self.base_name)
# 删除第六个表 # 删除第六个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataTag().delete_like_tag(session=session, tag=self.base_name) DataTag().delete_like_tag(session=session, tag=self.base_name)
session = self.db_session() session = self.db_session()
self.alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.base_name, status=1, severity=1, self.alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.name, status=1, severity=1,
manual_close=0, enable_time_default=1, create_time=1) manual_close=0, enable_time_default=1, create_time=1)
session = self.db_session() session = self.db_session()
self.update_alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.base_update_name, status=1, severity=1, self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
manual_close=0, enable_time_default=1, create_time=1)
session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.base_name, _type=1)
session = self.db_session()
self.update_businessid = DataBusinesses().insert_one(session=session, name=self.base_update_name, _type=1)
session = self.db_session()
self.hostid = DataHosts().insert_one(session=session, host_name=self.base_name, del_flag=0, monitor_status=1,
manage_ip=self.base_ip)
session = self.db_session()
self.update_hostid = DataHosts().insert_one(session=session, host_name=self.base_update_name, del_flag=0, monitor_status=1,
manage_ip=self.base_ip)
session = self.db_session() self.hostid = self.host_create(num=num)
self.itemid = DataHostsItems().insert_one(session=session, host_id=self.hostid, name=self.base_name, key_="key_")
session = self.db_session()
self.update_itemid = DataHostsItems().insert_one(session=session, host_id=self.hostid, name=self.base_update_name, key_="key_")
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
for x, y in enumerate(sql):
itemids.append(y.id)
self.itemid = itemids[0]
session = self.db_session() session = self.db_session()
self.eventid = DataProblem().insert_one(session=session, eventid=999999, source=0, _object=0, objectid=0, clock=0, ns=0, DataProblem().insert_one(session=session, eventid=self.eventid, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0, r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.base_name) status=0, hostid=self.hostid, name=self.name)
session = self.db_session()
self.update_eventid = DataProblem().insert_one(session=session, eventid=999998, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.base_update_name)
@allure.step("创建经验手册")
def case_create_experience(self, num):
self.title = self.base_title + num self.title = self.base_title + num
self.content = self.base_content + num self.content = self.base_content + num
self.description = self.base_description + num self.description = self.base_description + num
self.update_title = self.title + "修改"
self.update_content = self.content + "修改"
self.update_description = self.description + "修改"
response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_agent], response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_agent],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid], alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid], manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid],
...@@ -325,17 +366,19 @@ class TestExperienceDelete(object): ...@@ -325,17 +366,19 @@ class TestExperienceDelete(object):
def test_case_04(self): def test_case_04(self):
self.get_base_token() self.get_base_token()
self.case_create(num="04") self.case_create(num="04")
self.case_create_experience(num="04")
response = self.experience_delete(token=self.token, experienceids=[self.experienceid]) response = self.experience_delete(token=self.token, experienceids=[self.experienceid])
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
self.check_select(experienceid=self.experienceid) self.check_select(experienceid=self.experienceid)
@allure.title("experience.delete：删除多个") @allure.title("experience.delete：删除多个")
@allure.story("删除经验手册：删除多个") @allure.story("删除经验手册：删除多个")
def test_case_05(self): def test_case_06(self):
self.get_base_token() self.get_base_token()
experienceid_1 = self.case_create(num="051") self.case_create(num="06")
experienceid_2 = self.case_create(num="052") experienceid_1 = self.case_create_experience(num="061")
experienceid_3 = self.case_create(num="053") experienceid_2 = self.case_create_experience(num="062")
experienceid_3 = self.case_create_experience(num="063")
response = self.experience_delete(token=self.token, experienceids=[experienceid_1, experienceid_2, experienceid_3]) response = self.experience_delete(token=self.token, experienceids=[experienceid_1, experienceid_2, experienceid_3])
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
self.check_select(experienceid=experienceid_1) self.check_select(experienceid=experienceid_1)
...@@ -357,4 +400,4 @@ if __name__ == "__main__": ...@@ -357,4 +400,4 @@ if __name__ == "__main__":
# a = TestExperienceDelete() # a = TestExperienceDelete()
# a.setup_class() # a.setup_class()
# a.test_case_05() # a.test_case_06()
WorkCase/APP/Experience/test_experience_update.py
...@@ -7,6 +7,7 @@ ...@@ -7,6 +7,7 @@
from __future__ import division from __future__ import division
from WorkApi.APP.Api.api_login import ApiLogin from WorkApi.APP.Api.api_login import ApiLogin
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase from WorkCase import CaseBase
from WorkData.Argus.members import DataMembers from WorkData.Argus.members import DataMembers
from WorkUtils.UtilsLog import UtilsLog from WorkUtils.UtilsLog import UtilsLog
...@@ -53,17 +54,18 @@ class TestExperienceUpdate(object): ...@@ -53,17 +54,18 @@ class TestExperienceUpdate(object):
base_type_agent = 126 base_type_agent = 126
base_type_snmp = 127 base_type_snmp = 127
base_business_id_1 = "12"
base_business_id_2 = "13"
base_ip = "172.16.3.170"
base_port = 161
base_title = "SS测试经验标题" base_title = "SS测试经验标题"
base_description = "SS测试经验描述"
base_content = "SS测试经验内容" base_content = "SS测试经验内容"
base_businessName = "SS测试经验业务名称" base_description = "SS测试经验描述"
base_monitorName = "SS测试经验监控名称"
base_hostid = 10266
base_name = "经验手册-测试数据" base_name = "经验手册-测试数据"
base_update_name = "经验手册-测试数据-修改" base_update_name = "经验手册-测试数据修改"
base_ip = "172.99.1.99" base_eventid = 91000000
@classmethod @classmethod
def setup_class(cls): def setup_class(cls):
...@@ -217,95 +219,149 @@ class TestExperienceUpdate(object): ...@@ -217,95 +219,149 @@ class TestExperienceUpdate(object):
self.token = UtilsResponse().get_data(response=response) self.token = UtilsResponse().get_data(response=response)
@allure.step("创建测试数据") @allure.step("创建测试数据")
def host_create(self, num):
name = self.base_name + num
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_snmp, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None, port=None, businessTree=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.port = port
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.businessTree = businessTree
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session()
sql = DataHostsItems().select_all_from_allKeys(session=session, host_id=host_id)
return sql
@allure.step("创建测试数据")
def case_create(self, num): def case_create(self, num):
self.title = self.base_title + num
self.content = self.base_content + num
self.name = self.base_name + num
self.description = self.base_description + num
self.update_title = self.title + "修改"
self.update_content = self.content + "修改"
self.update_name = self.name + "修改"
self.update_description = self.description + "修改"
self.eventid = self.base_eventid + int(num)
self.update_eventid = self.eventid + 1
# 删除第一个表 # 删除第一个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataAlarmRule().delete_like_name(session=session, name=self.base_name) DataAlarmRule().delete_like_name(session=session, name=self.base_name)
session = self.db_session()
DataAlarmRule().delete_like_name(session=session, name=self.base_update_name)
# 删除第二个表 # 删除第二个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataBusinesses().delete_like_name(session=session, name=self.base_name) DataBusinesses().delete_like_name(session=session, name=self.base_name)
session = self.db_session()
DataBusinesses().delete_like_name(session=session, name=self.base_update_name)
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.base_name) sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.name)
for x, y in enumerate(sql): for x, y in enumerate(sql):
# 删除第三个表 # 删除第三个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid) DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表 # 删除第四个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_hostid(session=session, hostid=y.hostid) DataProblem().delete_like_name(session=session, name=self.base_name)
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.base_update_name) sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.update_name)
for x, y in enumerate(sql): for x, y in enumerate(sql):
# 删除第三个表 # 删除第三个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid) DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表 # 删除第四个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_hostid(session=session, hostid=y.hostid) DataProblem().delete_like_name(session=session, name=self.base_name)
# 删除第五个表 # 删除第五个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=self.base_name) DataHosts().delete_like_host_name(session=session, host_name=self.base_name)
session = self.db_session()
DataHosts().delete_like_host_name(session=session, host_name=self.base_update_name)
# 删除第六个表 # 删除第六个表
session = self.db_session() session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataTag().delete_like_tag(session=session, tag=self.base_name) DataTag().delete_like_tag(session=session, tag=self.base_name)
session = self.db_session()
DataTag().delete_like_tag(session=session, tag=self.base_update_name)
session = self.db_session() session = self.db_session()
self.alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.base_name, status=1, severity=1, self.alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.name, status=1, severity=1,
manual_close=0, enable_time_default=1, create_time=1) manual_close=0, enable_time_default=1, create_time=1)
session = self.db_session() session = self.db_session()
self.update_alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.base_update_name, status=1, severity=1, self.update_alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.name, status=1, severity=1,
manual_close=0, enable_time_default=1, create_time=1) manual_close=0, enable_time_default=1, create_time=1)
session = self.db_session() session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.base_name, _type=1) self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
session = self.db_session() session = self.db_session()
self.update_businessid = DataBusinesses().insert_one(session=session, name=self.base_update_name, _type=1) self.update_businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
session = self.db_session() self.hostid = self.host_create(num=num)
self.hostid = DataHosts().insert_one(session=session, host_name=self.base_name, del_flag=0, monitor_status=1,
manage_ip=self.base_ip)
session = self.db_session()
self.update_hostid = DataHosts().insert_one(session=session, host_name=self.base_update_name, del_flag=0, monitor_status=1,
manage_ip=self.base_ip)
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
for x, y in enumerate(sql):
itemids.append(y.id)
self.itemid = itemids[0]
session = self.db_session() session = self.db_session()
self.itemid = DataHostsItems().insert_one(session=session, host_id=self.hostid, name=self.base_name, key_="key_") DataProblem().insert_one(session=session, eventid=self.eventid, source=0, _object=0, objectid=0, clock=0, ns=0,
session = self.db_session() r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
self.update_itemid = DataHostsItems().insert_one(session=session, host_id=self.hostid, name=self.base_update_name, key_="key_") status=0, hostid=self.hostid, name=self.name)
self.update_hostid = self.host_create(num=num + "1")
sql = self.select_hosts_items(host_id=self.update_hostid)
update_itemid = []
for x, y in enumerate(sql):
update_itemid.append(y.id)
self.update_itemid = update_itemid[0]
session = self.db_session() session = self.db_session()
self.eventid = DataProblem().insert_one(session=session, eventid=999999, source=0, _object=0, objectid=0, clock=0, ns=0, DataProblem().insert_one(session=session, eventid=self.update_eventid, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0, r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.base_name) status=0, hostid=self.update_hostid, name=self.update_name)
session = self.db_session()
self.update_eventid = DataProblem().insert_one(session=session, eventid=999998, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.base_update_name)
self.title = self.base_title + num
self.content = self.base_content + num
self.description = self.base_description + num
self.update_title = self.title + "修改"
self.update_content = self.content + "修改"
self.update_description = self.description + "修改"
response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_agent], response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_agent],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid], alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid], manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid],
tagNameList=[self.base_name], description=self.description) tagNameList=[self.base_name], description=self.description)
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
self.experienceid = UtilsResponse().get_data(response=response) self.experienceid = UtilsResponse().get_data(response=response)
return self.experienceid
@allure.step("校验数据") @allure.step("校验数据")
def check_select(self, response): def check_select(self, response):
......
WorkData/Argus/problem.py
...@@ -98,6 +98,27 @@ class DataProblem(object): ...@@ -98,6 +98,27 @@ class DataProblem(object):
session.close() session.close()
return error return error
def delete_like_name(self, session, name):
"""
:param session: 指针
:param name:
:return:
"""
self.log.debug("查询数据库:")
try:
self.log.debug("删除相关测试数据")
session.query(self.table).filter(self.table.name.like(str(name) + "%")).delete(synchronize_session=False)
# session.delete(sql)
session.commit()
session.close()
self.log.debug("删除成功")
return True
except UtilsDataBase().errors as error:
self.log.error("异常:")
self.log.error(error)
session.close()
return error
def insert_one(self, session, eventid=None, source=None, _object=None, objectid=None, clock=None, ns=None, def insert_one(self, session, eventid=None, source=None, _object=None, objectid=None, clock=None, ns=None,
r_clock=None, r_ns=None, hostid=None, name=None, acknowledged=None, acknowledged_time=None, r_clock=None, r_ns=None, hostid=None, name=None, acknowledged=None, acknowledged_time=None,
severity=None, status=None,): severity=None, status=None,):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment