Skip to content

Z
zmops-test
Commit 6c0830fb by sanshi
Options

告警规则模块完成

parent 10145084
Showing with 483 additions and 94 deletions
.idea/dictionaries/Administrator.xml
......@@ -55,6 +55,7 @@
<w>nodata</w>
<w>objectid</w>
<w>objectids</w>
<w>percpu</w>
<w>preservekeys</w>
<w>privatekey</w>
<w>privpassphrase</w>
......
WorkApi/APP/Device/device_create.py
......@@ -29,6 +29,7 @@ class DeviceCreate(object):
self.token = None
self.hostName = None
self.hostType = None
self.manageLevel = None
self.iplist = None
self.dns = None
self.monitorInterface = None
......@@ -58,6 +59,7 @@ class DeviceCreate(object):
base = ApiBase()
base.dict_add_key(_key="hostName", value=self.hostName)
base.dict_add_key(_key="hostType", value=self.hostType)
base.dict_add_key(_key="manageLevel", value=self.manageLevel)
base.dict_add_key(_key="iplist", value=self.iplist)
base.dict_add_key(_key="dns", value=self.dns)
base.dict_add_key(_key="monitorInterface", value=self.monitorInterface)
......
WorkApi/APP/Rule/rule_update.py
......@@ -27,7 +27,7 @@ class RuleUpdate(object):
self.response = ""
self.token = None
self.alarmRuleId = None
self.id = None
self.name = None
self.severity = None
self.url = None
......@@ -75,7 +75,7 @@ class RuleUpdate(object):
base.dict_add_key(_key="hostName", value=self.triggersList_hostName, num=x)
base.dict_add_key(_key="itemid", value=self.triggersList_itemid, num=x)
base.dict_add_key(_key="itemName", value=self.triggersList_itemName, num=x)
base.dict_add_key(_key="parameters", value=self.triggersList_parameters, num=x)
base.dict_add_key(_key="parameters", value=self.triggersList_parameters)
base.dict_add_key(_key="operator", value=self.triggersList_operator, num=x)
base.dict_add_key(_key="constant", value=self.triggersList_constant, num=x)
base.dict_add_key(_key="logicalSymbol", value=self.triggersList_logicalSymbol, num=x)
......@@ -85,7 +85,7 @@ class RuleUpdate(object):
pass
base = ApiBase()
base.dict_add_key(_key="alarmRuleId", value=self.alarmRuleId)
base.dict_add_key(_key="id", value=self.id)
base.dict_add_key(_key="name", value=self.name)
base.dict_add_key(_key="severity", value=self.severity)
base.dict_add_key(_key="url", value=self.url)
......
WorkCase/APP/Device/test_device_create.py
......@@ -82,7 +82,7 @@ class TestDeviceCreate(object):
return sql
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, iplist=None, dns=None,
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
......@@ -92,6 +92,7 @@ class TestDeviceCreate(object):
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.dns = dns
api.monitorInterface = monitorInterface
......
WorkCase/APP/Rule/test_rule_create.py
......@@ -207,23 +207,44 @@ class TestRuleCreate(object):
api.get_response()
return api.response
@allure.title("创建测试数据")
@allure.step("创建测试数据")
def case_create(self, num, item_key):
name = self.base_host_name + num
# 创建资产设备
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_agent,
iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1)
self.check_code(response=response, code=0)
# 得到host_id
self.host_id = UtilsResponse().get_data(response=response)
sql = self.select_hosts(hostid=self.host_id)
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到zabbix的host_name
self.host_name = y.asset_code
sql = self.select_hosts_items(host_id=self.host_id, key_=item_key)
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到item_name
self.item_name = y.name
self.item_id = y.id
@allure.step("校验数据库")
def check_sql(self, rule_id, name, severity, url, status, description, manualClose, enableTimeDefault,
period, periodType):
sql = self.select_alarm_rule(alarm_ruleid=rule_id)
for x, y in enumerate(sql):
assert x == 0
assert y.name == name
assert y.severity == severity
assert y.url == url
assert y.status == status
assert y.description == description
assert y.manual_close == manualClose
assert y.enable_time_default == enableTimeDefault
assert y.period == period
assert y.period_unit == periodType
@allure.title("rule.create：token未传")
@allure.story("创建告警规则：token未传")
......@@ -263,12 +284,15 @@ class TestRuleCreate(object):
item_key = "system.cpu.load[percpu,avg15]"
self.case_create(num="04", item_key=item_key)
name = self.base_name + "04"
response = self.rule_create(token=self.token, name=name, severity=0, status=1, manualClose=0, triggersList_hostid=[self.host_id],
response = self.rule_create(token=self.token, name=name, severity=2, status=0, manualClose=0, triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.base_template_id_agent],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5])
self.check_code(response=response, code=0)
rule_id = UtilsResponse().get_data(response=response)
self.check_sql(rule_id=rule_id, name=name, severity=2, url="url", description="description", status=0, manualClose=0,
enableTimeDefault=0, period=None, periodType=None)
@allure.title("rule.create：创建agent设备在线规则")
@allure.story("创建告警规则：创建agent设备在线规则")
......@@ -278,7 +302,7 @@ class TestRuleCreate(object):
self.case_create(num="05", item_key=item_key)
name = self.base_name + "05"
response = self.rule_create(token=self.token, name=name, severity=0, status=1, manualClose=1, triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.base_template_id_agent],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5], triggersList_onlineFlag=[True])
......@@ -292,7 +316,7 @@ class TestRuleCreate(object):
self.case_create(num="06", item_key=item_key)
name = self.base_name + "06"
response = self.rule_create(token=self.token, name=name, severity=0, status=1, manualClose=1, triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.base_template_id_agent],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5], triggersList_onlineFlag=[False])
......@@ -306,7 +330,7 @@ class TestRuleCreate(object):
self.case_create(num="07", item_key=item_key)
name = self.base_name + "07"
response = self.rule_create(token=self.token, name=name, severity=0, status=1, manualClose=1, triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.base_template_id_snmp],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5], triggersList_onlineFlag=[True])
......@@ -320,12 +344,30 @@ class TestRuleCreate(object):
self.case_create(num="08", item_key=item_key)
name = self.base_name + "08"
response = self.rule_create(token=self.token, name=name, severity=0, status=1, manualClose=1, triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.base_template_id_snmp],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5], triggersList_onlineFlag=[False])
self.check_code(response=response, code=0)
@allure.title("rule.create：创建普通告警规则-所有字段校验")
@allure.story("创建告警规则：创建普通告警规则-所有字段校验")
def test_case_09(self):
self.get_base_token()
item_key = "system.cpu.load[percpu,avg15]"
self.case_create(num="09", item_key=item_key)
name = self.base_name + "09"
response = self.rule_create(token=self.token, name=name, severity=3, url="url", description="description",
status=1, manualClose=1, triggersList_hostid=[self.host_id], enableTimeDefault=1,
period=10, periodType="m",
triggersList_hostName=[self.host_name], triggersList_itemid=[self.item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5])
self.check_code(response=response, code=0)
rule_id = UtilsResponse().get_data(response=response)
self.check_sql(rule_id=rule_id, name=name, severity=3, url="url", description="description", status=1, manualClose=1,
enableTimeDefault=1, period=10, periodType="m")
if __name__ == "__main__":
......@@ -342,4 +384,4 @@ if __name__ == "__main__":
# a = TestRuleCreate()
# a.setup_class()
# a.test_case_08()
# a.test_case_09()
WorkCase/APP/Rule/test_rule_delete.py
......@@ -6,6 +6,7 @@
from __future__ import division
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
......@@ -16,6 +17,9 @@ from WorkApi.APP.Rule.rule_create import RuleCreate
from WorkApi.APP.Rule.rule_delete import RuleDelete
from WorkData.Argus.members import DataMembers
from WorkData.Argus.hosts_items import DataHostsItems
from WorkData.Argus.alarm_rule import DataAlarmRule
from WorkData.Argus.hosts import DataHosts
import allure
......@@ -24,6 +28,7 @@ import allure
class TestRuleDelete(object):
log = UtilsLog()
env = CaseBase().app_environment
host_create = "http://172.16.3.78:3334"
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
......@@ -35,11 +40,22 @@ class TestRuleDelete(object):
base_id = 10
base_password = "123456"
base_type_agent = 126
base_type_snmp = 127
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 999
base_host_name = "SS测试主机"
base_name = "SS测试告警规则名称"
base_template_id_agent = 10001
base_template_id_snmp = 10190
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=cls.base_host_name)
@classmethod
def teardown_class(cls):
......@@ -127,24 +143,28 @@ class TestRuleDelete(object):
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)
return session
# @allure.step("查询表：businesses")
# def select_businesses(self, businessid=None, name=None):
# session = self.db_session()
# sql = DataBusinesses().select_all_from_allKeys(session=session, businessid=businessid, name=name)
# return sql
@allure.step("查询表：alarm_rule")
def select_alarm_rule(self, alarm_ruleid=None, name=None, triggerid=None):
session = self.db_session()
sql = DataAlarmRule().select_all_from_allKeys(session=session, alarm_ruleid=alarm_ruleid, name=name, triggerid=triggerid)
return sql
@allure.step("查询表：hosts")
def select_hosts(self, hostid=None, host_name=None, available=None, monitor_status=None):
session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid, host_name=host_name, available=available, monitor_status=monitor_status)
return sql
@allure.step("查询表：hosts_items")
def select_hosts_items(self, _id=None, host_id=None, name=None, key_=None):
session = self.db_session()
sql = DataHostsItems().select_all_from_allKeys(session=session, _id=_id, host_id=host_id, name=name, key_=key_)
return sql
@allure.step("校验查询结果")
def check_select(self, response, name, description, _type, principal_name, principal_phone, memberid):
data = UtilsResponse().get_data(response=response)
sql = self.select_businesses(businessid=data)
for x, y in enumerate(sql):
assert y.name == name
assert y.description == description
assert y.type == _type
assert y.principal_name == principal_name
assert y.principal_phone == principal_phone
assert y.memberid == memberid
assert x == 0
def check_select(self, alarm_ruleid):
sql = self.select_alarm_rule(alarm_ruleid=alarm_ruleid)
assert sql == []
@allure.step("断言返回结果")
def check_code(self, response, code):
......@@ -156,8 +176,78 @@ class TestRuleDelete(object):
_msg = UtilsResponse().get_msg(response=response)
assert _msg == msg
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factory=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host_create)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.iplist = iplist
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factory = factory
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self, num, item_key):
name = self.base_host_name + num
# 创建资产设备
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_agent,
iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1)
self.check_code(response=response, code=0)
# 得到host_id
self.host_id = UtilsResponse().get_data(response=response)
sql = self.select_hosts(hostid=self.host_id)
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到zabbix的host_name
self.host_name = y.asset_code
sql = self.select_hosts_items(host_id=self.host_id, key_=item_key)
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到item_name
self.item_name = y.name
self.item_id = y.id
@allure.step("创建测试数据-告警规则")
def case_create_rule(self, item_key, num):
self.name = self.base_name + num
# 告警规则创建
response = self.rule_create(token=self.token, name=self.name, severity=0, status=1, manualClose=0, triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5], triggersList_onlineFlag=[True])
self.check_code(response=response, code=0)
# 得到告警规则编号
self.rule_id = UtilsResponse().get_data(response=response)
return self.rule_id
@allure.title("rule.delete：token未传")
@allure.story("查询告警规则：token未传")
@allure.story("删除告警规则：token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
......@@ -167,7 +257,7 @@ class TestRuleDelete(object):
self.check_msg(response=response, msg="无token，请重新登录")
@allure.title("rule.delete：token的用户已删除")
@allure.story("查询告警规则：token的用户已删除")
@allure.story("删除告警规则：token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.get_base_token()
......@@ -177,7 +267,7 @@ class TestRuleDelete(object):
self.check_msg(response=response, msg="用户不存在")
@allure.title("rule.delete：token的用户已禁用")
@allure.story("查询告警规则：token的用户已禁用")
@allure.story("删除告警规则：token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.get_base_token()
......@@ -186,15 +276,35 @@ class TestRuleDelete(object):
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
@allure.title("rule.delete：无参数调用")
@allure.story("查询告警规则：无参数调用")
@allure.title("rule.delete：删除一个")
@allure.story("删除告警规则：删除一个")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
item_key = "agent.ping"
self.case_create(num="04", item_key=item_key)
self.case_create_rule(item_key=item_key, num="04")
name = self.base_name + "04"
response = self.rule_delete(token=self.token)
response = self.rule_delete(token=self.token, alarmRuleidList=[self.rule_id])
self.check_code(response=response, code=0)
self.check_select(alarm_ruleid=self.rule_id)
@allure.title("rule.delete：删除多个")
@allure.story("删除告警规则：删除多个")
@allure.severity("blocker")
def test_case_05(self):
self.get_base_token()
item_key = "agent.ping"
self.case_create(num="04", item_key=item_key)
rule_id_1 = self.case_create_rule(item_key=item_key, num="041")
rule_id_2 = self.case_create_rule(item_key=item_key, num="042")
rule_id_3 = self.case_create_rule(item_key=item_key, num="043")
response = self.rule_delete(token=self.token, alarmRuleidList=[rule_id_1, rule_id_2, rule_id_3])
self.check_code(response=response, code=0)
self.check_select(alarm_ruleid=rule_id_1)
self.check_select(alarm_ruleid=rule_id_2)
self.check_select(alarm_ruleid=rule_id_3)
if __name__ == "__main__":
......@@ -204,11 +314,11 @@ if __name__ == "__main__":
import os
# 执行自动化测试用例
# case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case, case_info=case_info)
# r.run_main()
a = TestRuleDelete()
a.setup_class()
a.test_case_04()
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestRuleDelete()
# a.setup_class()
# a.test_case_05()
WorkCase/APP/Rule/test_rule_get.py
......@@ -6,6 +6,7 @@
from __future__ import division
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
......@@ -16,6 +17,9 @@ from WorkApi.APP.Rule.rule_create import RuleCreate
from WorkApi.APP.Rule.rule_get import RuleGet
from WorkData.Argus.members import DataMembers
from WorkData.Argus.hosts_items import DataHostsItems
from WorkData.Argus.alarm_rule import DataAlarmRule
from WorkData.Argus.hosts import DataHosts
import allure
......@@ -24,6 +28,7 @@ import allure
class TestRuleGet(object):
log = UtilsLog()
env = CaseBase().app_environment
host_create = "http://172.16.3.78:3334"
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
......@@ -35,11 +40,22 @@ class TestRuleGet(object):
base_id = 10
base_password = "123456"
base_type_agent = 126
base_type_snmp = 127
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 999
base_host_name = "SS测试主机"
base_name = "SS测试告警规则名称"
base_template_id_agent = 10001
base_template_id_snmp = 10190
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=cls.base_host_name)
@classmethod
def teardown_class(cls):
......@@ -133,24 +149,23 @@ class TestRuleGet(object):
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)
return session
# @allure.step("查询表：businesses")
# def select_businesses(self, businessid=None, name=None):
# session = self.db_session()
# sql = DataBusinesses().select_all_from_allKeys(session=session, businessid=businessid, name=name)
# return sql
@allure.step("查询表：alarm_rule")
def select_alarm_rule(self, alarm_ruleid=None, name=None, triggerid=None):
session = self.db_session()
sql = DataAlarmRule().select_all_from_allKeys(session=session, alarm_ruleid=alarm_ruleid, name=name, triggerid=triggerid)
return sql
@allure.step("校验查询结果")
def check_select(self, response, name, description, _type, principal_name, principal_phone, memberid):
data = UtilsResponse().get_data(response=response)
sql = self.select_businesses(businessid=data)
for x, y in enumerate(sql):
assert y.name == name
assert y.description == description
assert y.type == _type
assert y.principal_name == principal_name
assert y.principal_phone == principal_phone
assert y.memberid == memberid
assert x == 0
@allure.step("查询表：hosts")
def select_hosts(self, hostid=None, host_name=None, available=None, monitor_status=None):
session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid, host_name=host_name, available=available, monitor_status=monitor_status)
return sql
@allure.step("查询表：hosts_items")
def select_hosts_items(self, _id=None, host_id=None, name=None, key_=None):
session = self.db_session()
sql = DataHostsItems().select_all_from_allKeys(session=session, _id=_id, host_id=host_id, name=name, key_=key_)
return sql
@allure.step("断言返回结果")
def check_code(self, response, code):
......@@ -162,6 +177,74 @@ class TestRuleGet(object):
_msg = UtilsResponse().get_msg(response=response)
assert _msg == msg
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factory=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host_create)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factory = factory
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self, num, item_key):
name = self.base_host_name + num
# 创建资产设备
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_agent,
iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=3)
self.check_code(response=response, code=0)
# 得到host_id
self.host_id = UtilsResponse().get_data(response=response)
sql = self.select_hosts(hostid=self.host_id)
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到zabbix的host_name
self.host_name = y.asset_code
sql = self.select_hosts_items(host_id=self.host_id, key_=item_key)
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到item_name
self.item_name = y.name
self.item_id = y.id
self.name = self.base_name + num
# 告警规则创建
response = self.rule_create(token=self.token, name=self.name, severity=0, status=1, manualClose=0, triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5], triggersList_onlineFlag=[True])
self.check_code(response=response, code=0)
# 得到告警规则编号
self.rule_id = UtilsResponse().get_data(response=response)
@allure.title("rule.get：token未传")
@allure.story("查询告警规则：token未传")
@allure.severity("blocker")
......@@ -197,11 +280,44 @@ class TestRuleGet(object):
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
name = self.base_name + "04"
response = self.rule_get(token=self.token)
self.check_code(response=response, code=0)
@allure.title("rule.get：severity查询")
@allure.story("查询告警规则：severity查询")
def test_case_05(self):
self.get_base_token()
response = self.rule_get(token=self.token, severity=2)
self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
for x, y in enumerate(data["list"]):
assert y["severity"] == 2
@allure.title("rule.get：item_id查询")
@allure.story("查询告警规则：item_id查询")
def test_case_06(self):
self.get_base_token()
item_key = "agent.ping"
self.case_create(num="06", item_key=item_key)
response = self.rule_get(token=self.token, itemid=self.item_id)
self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
for x, y in enumerate(data["list"]):
for q, w in enumerate(y["triggers"]):
assert w["itemid"] == self.item_id
@allure.title("rule.get：manageLevel查询")
@allure.story("查询告警规则：manageLevel查询")
def test_case_07(self):
self.get_base_token()
item_key = "agent.ping"
self.case_create(num="07", item_key=item_key)
response = self.rule_get(token=self.token, manageLevel=3)
self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
for x, y in enumerate(data["list"]):
assert y["status"] == 0
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
......@@ -210,11 +326,11 @@ if __name__ == "__main__":
import os
# 执行自动化测试用例
# case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case, case_info=case_info)
# r.run_main()
a = TestRuleGet()
a.setup_class()
a.test_case_04()
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestRuleGet()
# a.setup_class()
# a.test_case_06()
WorkCase/APP/Rule/test_rule_update.py
......@@ -6,6 +6,7 @@
from __future__ import division
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
......@@ -16,6 +17,9 @@ from WorkApi.APP.Rule.rule_create import RuleCreate
from WorkApi.APP.Rule.rule_update import RuleUpdate
from WorkData.Argus.members import DataMembers
from WorkData.Argus.hosts_items import DataHostsItems
from WorkData.Argus.alarm_rule import DataAlarmRule
from WorkData.Argus.hosts import DataHosts
import allure
......@@ -24,6 +28,7 @@ import allure
class TestRuleUpdate(object):
log = UtilsLog()
env = CaseBase().app_environment
host_create = "http://172.16.3.78:3334"
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
......@@ -35,11 +40,22 @@ class TestRuleUpdate(object):
base_id = 10
base_password = "123456"
base_type_agent = 126
base_type_snmp = 127
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 999
base_host_name = "SS测试主机"
base_name = "SS测试告警规则名称"
base_template_id_agent = 10001
base_template_id_snmp = 10190
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=cls.base_host_name)
@classmethod
def teardown_class(cls):
......@@ -110,13 +126,14 @@ class TestRuleUpdate(object):
return api.response
@allure.step("调用接口：rule.update")
def rule_update(self, token=None, name=None, severity=None, url=None, description=None, status=None,
def rule_update(self, token=None, _id=None, name=None, severity=None, url=None, description=None, status=None,
manualClose=None, enableTimeDefault=None, period=None, periodType=None, triggersList=None,
triggersList_hostid=None, triggersList_hostName=None, triggersList_itemid=None,
triggersList_itemName=None, triggersList_function=None, triggersList_parameters=None, triggersList_parameters_value=None,
triggersList_parameters_unit=None, triggersList_operator=None, triggersList_constant=None,
triggersList_logicalSymbol=None, triggersList_onlineFlag=None):
api = RuleUpdate(_host=self.host)
api.id = _id
api.token = token
api.name = name
api.severity = severity
......@@ -153,24 +170,106 @@ class TestRuleUpdate(object):
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)
return session
# @allure.step("查询表：businesses")
# def select_businesses(self, businessid=None, name=None):
# session = self.db_session()
# sql = DataBusinesses().select_all_from_allKeys(session=session, businessid=businessid, name=name)
# return sql
@allure.step("查询表：alarm_rule")
def select_alarm_rule(self, alarm_ruleid=None, name=None, triggerid=None):
session = self.db_session()
sql = DataAlarmRule().select_all_from_allKeys(session=session, alarm_ruleid=alarm_ruleid, name=name, triggerid=triggerid)
return sql
@allure.step("查询表：hosts")
def select_hosts(self, hostid=None, host_name=None, available=None, monitor_status=None):
session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid, host_name=host_name, available=available, monitor_status=monitor_status)
return sql
@allure.step("查询表：hosts_items")
def select_hosts_items(self, _id=None, host_id=None, name=None, key_=None):
session = self.db_session()
sql = DataHostsItems().select_all_from_allKeys(session=session, _id=_id, host_id=host_id, name=name, key_=key_)
return sql
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factory=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host_create)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.iplist = iplist
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factory = factory
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self, num, item_key):
name = self.base_host_name + num
# 创建资产设备
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_agent,
iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1)
self.check_code(response=response, code=0)
# 得到host_id
self.host_id = UtilsResponse().get_data(response=response)
sql = self.select_hosts(hostid=self.host_id)
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到zabbix的host_name
self.host_name = y.asset_code
sql = self.select_hosts_items(host_id=self.host_id, key_=item_key)
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到item_name
self.item_name = y.name
self.item_id = y.id
self.name = self.base_name + num
# 告警规则创建
response = self.rule_create(token=self.token, name=self.name, severity=0, status=1, manualClose=0, triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[self.item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5], triggersList_onlineFlag=[True])
self.check_code(response=response, code=0)
# 得到告警规则编号
self.rule_id = UtilsResponse().get_data(response=response)
@allure.step("校验查询结果")
def check_select(self, response, name, description, _type, principal_name, principal_phone, memberid):
data = UtilsResponse().get_data(response=response)
sql = self.select_businesses(businessid=data)
@allure.step("校验数据库")
def check_sql(self, rule_id, name, severity, url, status, description, manualClose, enableTimeDefault,
period, periodType):
sql = self.select_alarm_rule(alarm_ruleid=rule_id)
for x, y in enumerate(sql):
assert x == 0
assert y.name == name
assert y.severity == severity
assert y.url == url
assert y.status == status
assert y.description == description
assert y.type == _type
assert y.principal_name == principal_name
assert y.principal_phone == principal_phone
assert y.memberid == memberid
assert x == 0
assert y.manual_close == manualClose
assert y.enable_time_default == enableTimeDefault
assert y.period == period
assert y.period_unit == periodType
@allure.step("断言返回结果")
def check_code(self, response, code):
......@@ -217,10 +316,28 @@ class TestRuleUpdate(object):
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
item_key = "agent.ping"
self.case_create(num="04", item_key=item_key)
name = self.base_name + "04"
response = self.rule_update(token=self.token)
item_key_update = "system.cpu.load[percpu,avg15]"
sql = self.select_hosts_items(key_=item_key_update, host_id=self.host_id)
item_id = 0
for x, y in enumerate(sql):
assert x == 0
item_id = y.id
response = self.rule_update(token=self.token, _id=self.rule_id, name=self.name + "修改", severity=5,
url="URL", description="DESCRIPTION", status=0, manualClose=0,
enableTimeDefault=0, period=1, periodType="h",
triggersList_hostid=[self.host_id],
triggersList_hostName=[self.host_name], triggersList_itemid=[item_id],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5],
triggersList_onlineFlag=[True])
self.check_code(response=response, code=0)
rule_id = UtilsResponse().get_data(response=response)
self.check_sql(rule_id=rule_id, name=self.name + "修改", severity=5, url="URL", description="DESCRIPTION", status=0, manualClose=0,
enableTimeDefault=0, period=1, periodType="h")
if __name__ == "__main__":
......@@ -230,11 +347,11 @@ if __name__ == "__main__":
import os
# 执行自动化测试用例
# case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case, case_info=case_info)
# r.run_main()
a = TestRuleUpdate()
a.setup_class()
a.test_case_04()
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestRuleUpdate()
# a.setup_class()
# a.test_case_04()
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