Skip to content

Z
zmops-test
Commit 1f9801e1 by sanshi
Options

告警规则优化

parent efd531db
Showing with 596 additions and 949 deletions
WorkCase/APP/Rule/test_rule_create.py
......@@ -8,6 +8,7 @@ from __future__ import division
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -19,6 +20,7 @@ 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
from WorkData.Zabbix.triggers import DataTriggers
import allure
......@@ -33,6 +35,7 @@ class TestRuleCreate(object):
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
db_base_zabbix = env["db_base_zabbix"]
login_name = "SS测试用户名"
base_id = 10
......@@ -40,12 +43,14 @@ class TestRuleCreate(object):
base_type_agent = 126
base_type_snmp = 127
base_business_id_1 = "12"
base_business_id_2 = "13"
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 161
base_host_name = "SS自动化专用主机"
base_name = "SS测试告警规则名称"
base_host_name = "SS 自动化专用"
base_name = "SS 自动化告警规则"
base_template_id_agent = 10001
base_template_id_snmp = 10186
......@@ -55,90 +60,49 @@ class TestRuleCreate(object):
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)
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)
DataAlarmRule().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：rule.create")
def rule_create(self, token=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 = RuleCreate(_host=self.host)
api.token = token
api.name = name
api.severity = severity
api.url = url
api.description = description
api.status = status
api.manualClose = manualClose
api.enableTimeDefault = enableTimeDefault
api.period = period
api.periodType = periodType
api.triggersList = triggersList
api.triggersList_hostid = triggersList_hostid
api.triggersList_hostName = triggersList_hostName
api.triggersList_itemid = triggersList_itemid
api.triggersList_itemName = triggersList_itemName
api.triggersList_function = triggersList_function
api.triggersList_parameters = triggersList_parameters
api.triggersList_parameters_value = triggersList_parameters_value
api.triggersList_parameters_unit = triggersList_parameters_unit
api.triggersList_operator = triggersList_operator
api.triggersList_constant = triggersList_constant
api.triggersList_logicalSymbol = triggersList_logicalSymbol
api.triggersList_onlineFlag = triggersList_onlineFlag
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
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("连接数据库：Zabbix")
def db_session_zabbix(self):
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base_zabbix, db_user=self.db_user, db_pw=self.db_pw)
return session
@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("查询表：triggers")
def select_triggers(self, description=None, triggerid=None):
session = self.db_session_zabbix()
sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid, description=description)
return sql
@allure.step("查询表：hosts")
def select_hosts(self, hostid=None, host_name=None, available=None, monitor_status=None):
session = self.db_session()
......@@ -151,78 +115,17 @@ class TestRuleCreate(object):
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
@allure.step("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.step("断言返回结果数据")
def check_msg(self, response, msg):
_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,
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.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.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("创建测试数据")
def case_create(self, num, hostType, item_key, snmpCommunity=None):
name = self.base_host_name + num
# 创建资产设备
response = self.device_create(token=self.token, hostName=name, hostType=hostType,
iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, snmpCommunity=snmpCommunity)
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)
@allure.step("获取查询测试数据")
def get_case(self, host_id, item_key):
sql = self.select_hosts(hostid=host_id)
assert sql != []
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)
sql = self.select_hosts_items(host_id=host_id, key_=item_key)
assert sql != []
for x, y in enumerate(sql):
assert x == 0
# 查询数据库得到item_name
......@@ -230,11 +133,13 @@ class TestRuleCreate(object):
self.item_id = y.id
@allure.step("校验数据库")
def check_sql(self, rule_id, name, severity, url, status, description, manualClose, enableTimeDefault,
def check_sql(self, response, name, severity, url, status, description, manualClose, enableTimeDefault,
period, periodType):
rule_id = UtilsResponse().get_data(response=response)
sql = self.select_alarm_rule(alarm_ruleid=rule_id)
assert sql != []
trigger_id = None
for x, y in enumerate(sql):
assert x == 0
assert y.name == name
assert y.severity == severity
assert y.url == url
......@@ -244,128 +149,145 @@ class TestRuleCreate(object):
assert y.enable_time_default == enableTimeDefault
assert y.period == period
assert y.period_unit == periodType
assert x == 0
trigger_id = y.triggerid
sql = self.select_triggers(triggerid=trigger_id)
assert sql != []
for x, y in enumerate(sql):
assert y.description == name
assert x == 0
@allure.title("rule.create：token未传")
@allure.story("创建告警规则：token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.rule_create()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.rule_create()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("rule.create：token的用户已删除")
@allure.story("创建告警规则：token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.rule_create(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.rule_create(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("rule.create：token的用户已禁用")
@allure.story("创建告警规则：token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.rule_create(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.rule_create(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("rule.create：创建普通告警规则")
@allure.story("创建告警规则：创建普通告警规则")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
item_key = "system.cpu.load[percpu,avg15]"
self.case_create(num="04", item_key=item_key, hostType=self.base_type_agent)
name = self.base_name + "04"
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.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=2, url=None, description=None, status=0, manualClose=0,
host_id = self.app.case_create_host(name=name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
response = self.app.rule_create(token=self.token, name=name, severity=2, status=0, manualClose=0, triggersList_hostid=[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])
self.app.check_code(response=response, code=0)
self.check_sql(response=response, name=name, severity=2, url=None, description=None, status=0, manualClose=0,
enableTimeDefault=0, period=None, periodType=None)
@allure.title("rule.create：创建agent设备在线规则")
@allure.story("创建告警规则：创建agent设备在线规则")
def test_case_05(self):
self.get_base_token()
item_key = "agent.ping"
self.case_create(num="05", item_key=item_key, hostType=self.base_type_agent)
name = self.base_name + "05"
response = self.rule_create(token=self.token, name=name, severity=0, status=0, manualClose=1, 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)
host_id = self.app.case_create_host(name=name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "agent.ping"
self.get_case(host_id=host_id, item_key=item_key)
response = self.app.rule_create(token=self.token, name=name, severity=0, status=0, manualClose=1, triggersList_hostid=[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.app.check_code(response=response, code=0)
@allure.title("rule.create：创建agent设备离线规则")
@allure.story("创建告警规则：创建agent设备离线规则")
def test_case_06(self):
self.get_base_token()
item_key = "agent.ping"
self.case_create(num="06", item_key=item_key, hostType=self.base_type_agent)
name = self.base_name + "06"
response = self.rule_create(token=self.token, name=name, severity=0, status=0, manualClose=1, 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=[False])
self.check_code(response=response, code=0)
host_id = self.app.case_create_host(name=name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "agent.ping"
self.get_case(host_id=host_id, item_key=item_key)
response = self.app.rule_create(token=self.token, name=name, severity=0, status=0, manualClose=1, triggersList_hostid=[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=[False])
self.app.check_code(response=response, code=0)
@allure.title("rule.create：创建icmp设备在线规则")
@allure.story("创建告警规则：创建icmp设备在线规则")
def test_case_07(self):
self.get_base_token()
item_key = "icmpping"
self.case_create(num="07", item_key=item_key, hostType=self.base_type_snmp, snmpCommunity="public")
name = self.base_name + "07"
response = self.rule_create(token=self.token, name=name, severity=0, status=0, manualClose=1, 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)
host_id = self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "icmpping"
self.get_case(host_id=host_id, item_key=item_key)
response = self.app.rule_create(token=self.token, name=name, severity=0, status=0, manualClose=1, triggersList_hostid=[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.app.check_code(response=response, code=0)
@allure.title("rule.create：创建icmp设备离线规则")
@allure.story("创建告警规则：创建icmp设备离线规则")
def test_case_08(self):
self.get_base_token()
item_key = "icmpping"
self.case_create(num="08", item_key=item_key, hostType=self.base_type_snmp, snmpCommunity="public")
name = self.base_name + "08"
response = self.rule_create(token=self.token, name=name, severity=0, status=0, manualClose=1, 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=[False])
self.check_code(response=response, code=0)
host_id = self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "icmpping"
self.get_case(host_id=host_id, item_key=item_key)
response = self.app.rule_create(token=self.token, name=name, severity=0, status=0, manualClose=1, triggersList_hostid=[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=[False])
self.app.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, hostType=self.base_type_agent)
name = self.base_name + "09"
response = self.rule_create(token=self.token, name=name, severity=3, url="url", description="description",
status=0, 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,
host_id = self.app.case_create_host(name=name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
response = self.app.rule_create(token=self.token, name=name, severity=3, url="url", description="description",
status=1, manualClose=1, triggersList_hostid=[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.app.check_code(response=response, code=0)
self.check_sql(response=response, name=name, severity=3, url="url", description="description", status=1, manualClose=1,
enableTimeDefault=1, period=10, periodType="m")
......
WorkCase/APP/Rule/test_rule_delete.py
......@@ -8,6 +8,8 @@ from __future__ import division
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkData.Zabbix.triggers import DataTriggers
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -35,6 +37,7 @@ class TestRuleDelete(object):
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
db_base_zabbix = env["db_base_zabbix"]
login_name = "SS测试用户名"
base_id = 10
......@@ -42,14 +45,16 @@ class TestRuleDelete(object):
base_type_agent = 126
base_type_snmp = 127
base_business_id_1 = "12"
base_business_id_2 = "13"
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 161
base_host_name = "SS自动化专用主机"
base_name = "SS测试告警规则名称"
base_host_name = "SS 自动化专用"
base_name = "SS 自动化告警规则"
base_template_id_agent = 10001
base_template_id_snmp = 10190
base_template_id_snmp = 10186
@classmethod
def setup_class(cls):
......@@ -57,98 +62,49 @@ class TestRuleDelete(object):
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)
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)
DataAlarmRule().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：rule.create")
def rule_create(self, token=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 = RuleCreate(_host=self.host)
api.token = token
api.name = name
api.severity = severity
api.url = url
api.description = description
api.status = status
api.manualClose = manualClose
api.enableTimeDefault = enableTimeDefault
api.period = period
api.periodType = periodType
api.triggersList = triggersList
api.triggersList_hostid = triggersList_hostid
api.triggersList_hostName = triggersList_hostName
api.triggersList_itemid = triggersList_itemid
api.triggersList_itemName = triggersList_itemName
api.triggersList_function = triggersList_function
api.triggersList_parameters = triggersList_parameters
api.triggersList_parameters_value = triggersList_parameters_value
api.triggersList_parameters_unit = triggersList_parameters_unit
api.triggersList_operator = triggersList_operator
api.triggersList_constant = triggersList_constant
api.triggersList_logicalSymbol = triggersList_logicalSymbol
api.triggersList_onlineFlag = triggersList_onlineFlag
api.get_response()
return api.response
@allure.step("调用接口：rule.delete")
def rule_delete(self, token=None, alarmRuleidList=None):
api = RuleDelete(_host=self.host)
api.token = token
api.alarmRuleidList = alarmRuleidList
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
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("连接数据库：Zabbix")
def db_session_zabbix(self):
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base_zabbix, db_user=self.db_user, db_pw=self.db_pw)
return session
@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("查询表：triggers")
def select_triggers(self, description=None, triggerid=None):
session = self.db_session_zabbix()
sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid, description=description)
return sql
@allure.step("查询表：hosts")
def select_hosts(self, hostid=None, host_name=None, available=None, monitor_status=None):
session = self.db_session()
......@@ -162,149 +118,122 @@ class TestRuleDelete(object):
return sql
@allure.step("校验查询结果")
def check_select(self, alarm_ruleid):
def check_select(self, alarm_ruleid, trigger_id):
sql = self.select_alarm_rule(alarm_ruleid=alarm_ruleid)
assert sql == []
@allure.step("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.step("断言返回结果数据")
def check_msg(self, response, msg):
_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,
factoryId=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.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("创建测试数据")
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)
sql = self.select_triggers(triggerid=trigger_id)
assert sql == []
@allure.step("获取查询测试数据")
def get_case(self, host_id, item_key):
sql = self.select_hosts(hostid=host_id)
assert sql != []
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)
sql = self.select_hosts_items(host_id=host_id, key_=item_key)
assert sql != []
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.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.rule_delete()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.rule_delete()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("rule.delete：token的用户已删除")
@allure.story("删除告警规则：token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.rule_delete(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.rule_delete(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("rule.delete：token的用户已禁用")
@allure.story("删除告警规则：token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.rule_delete(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.rule_delete(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@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")
host_name = self.base_host_name + "04"
name = self.base_name + "04"
host_id = self.app.case_create_host(name=host_name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
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)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
rule_id = self.app.case_create_rule(name=name, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=0)
sql = self.select_alarm_rule(alarm_ruleid=rule_id)
trigger_id = None
for x, y in enumerate(sql):
trigger_id = y.triggerid
response = self.app.rule_delete(token=self.token, alarmRuleidList=[rule_id])
self.app.check_code(response=response, code=0)
self.check_select(alarm_ruleid=rule_id, trigger_id=trigger_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)
host_name = self.base_host_name + "05"
name_1 = self.base_name + "051"
name_2 = self.base_name + "052"
name_3 = self.base_name + "053"
host_id = self.app.case_create_host(name=host_name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
rule_id_1 = self.app.case_create_rule(name=name_1, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=0)
rule_id_2 = self.app.case_create_rule(name=name_2, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=0)
rule_id_3 = self.app.case_create_rule(name=name_3, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=0)
sql = self.select_alarm_rule(alarm_ruleid=rule_id_1)
trigger_id_1 = None
for x, y in enumerate(sql):
trigger_id_1 = y.triggerid
sql = self.select_alarm_rule(alarm_ruleid=rule_id_2)
trigger_id_2 = None
for x, y in enumerate(sql):
trigger_id_2 = y.triggerid
sql = self.select_alarm_rule(alarm_ruleid=rule_id_3)
trigger_id_3 = None
for x, y in enumerate(sql):
trigger_id_3 = y.triggerid
response = self.app.rule_delete(token=self.token, alarmRuleidList=[rule_id_1, rule_id_2, rule_id_3])
self.app.check_code(response=response, code=0)
self.check_select(alarm_ruleid=rule_id_1, trigger_id=trigger_id_1)
self.check_select(alarm_ruleid=rule_id_2, trigger_id=trigger_id_2)
self.check_select(alarm_ruleid=rule_id_3, trigger_id=trigger_id_3)
if __name__ == "__main__":
......
WorkCase/APP/Rule/test_rule_get.py
......@@ -8,6 +8,7 @@ from __future__ import division
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -42,14 +43,16 @@ class TestRuleGet(object):
base_type_agent = 126
base_type_snmp = 127
base_business_id_1 = "12"
base_business_id_2 = "13"
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 161
base_host_name = "SS自动化专用主机"
base_name = "SS测试告警规则名称"
base_host_name = "SS 自动化专用"
base_name = "SS 自动化告警规则"
base_template_id_agent = 10001
base_template_id_snmp = 10190
base_template_id_snmp = 10186
@classmethod
def setup_class(cls):
......@@ -57,92 +60,26 @@ class TestRuleGet(object):
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)
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)
DataAlarmRule().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：rule.create")
def rule_create(self, token=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 = RuleCreate(_host=self.host)
api.token = token
api.name = name
api.severity = severity
api.url = url
api.description = description
api.status = status
api.manualClose = manualClose
api.enableTimeDefault = enableTimeDefault
api.period = period
api.periodType = periodType
api.triggersList = triggersList
api.triggersList_hostid = triggersList_hostid
api.triggersList_hostName = triggersList_hostName
api.triggersList_itemid = triggersList_itemid
api.triggersList_itemName = triggersList_itemName
api.triggersList_function = triggersList_function
api.triggersList_parameters = triggersList_parameters
api.triggersList_parameters_value = triggersList_parameters_value
api.triggersList_parameters_unit = triggersList_parameters_unit
api.triggersList_operator = triggersList_operator
api.triggersList_constant = triggersList_constant
api.triggersList_logicalSymbol = triggersList_logicalSymbol
api.triggersList_onlineFlag = triggersList_onlineFlag
api.get_response()
return api.response
@allure.step("调用接口：rule.get")
def rule_get(self, token=None, severity=None, itemid=None, status=None,
manageLevel=None, manageIp=None, hostName=None):
api = RuleGet(_host=self.host)
api.token = token
api.severity = severity
api.status = status
api.itemid = itemid
api.manageLevel = manageLevel
api.manageIp = manageIp
api.hostName = hostName
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
......@@ -167,155 +104,120 @@ class TestRuleGet(object):
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):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.step("断言返回结果数据")
def check_msg(self, response, msg):
_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,
factoryId=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.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("创建测试数据")
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)
@allure.step("获取查询测试数据")
def get_case(self, host_id, item_key):
sql = self.select_hosts(hostid=host_id)
assert sql != []
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)
sql = self.select_hosts_items(host_id=host_id, key_=item_key)
assert sql != []
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")
def test_case_01(self):
self.get_base_token()
response = self.rule_get()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.rule_get()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("rule.get：token的用户已删除")
@allure.story("查询告警规则：token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.rule_get(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.rule_get(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("rule.get：token的用户已禁用")
@allure.story("查询告警规则：token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.rule_get(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.rule_get(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("rule.get：无参数调用")
@allure.story("查询告警规则：无参数调用")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
response = self.rule_get(token=self.token)
self.check_code(response=response, code=0)
response = self.app.rule_get(token=self.token)
self.app.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
host_name = self.base_host_name + "05"
name = self.base_name + "05"
host_id = self.app.case_create_host(name=host_name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
self.app.case_create_rule(name=name, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=2)
response = self.app.rule_get(token=self.token, severity=2)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="severity", _value=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"]):
host_name = self.base_host_name + "06"
name = self.base_name + "06"
host_id = self.app.case_create_host(name=host_name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
self.app.case_create_rule(name=name, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=2)
response = self.app.rule_get(token=self.token, itemid=self.item_id)
self.app.check_code(response=response, code=0)
for x, y in enumerate(UtilsResponse().get_data(response=response)["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)
assert len(data["list"]) > 0
host_name = self.base_host_name + "07"
name = self.base_name + "07"
host_id = self.app.case_create_host(name=host_name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
self.app.case_create_rule(name=name, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=2)
response = self.app.rule_get(token=self.token, manageLevel=1)
self.app.check_code(response=response, code=0)
for x, y in enumerate(UtilsResponse().get_data(response=response)["list"]):
for q, w in enumerate(y["triggers"]):
sql = self.select_hosts(hostid=w["hostid"])
assert sql != []
for e, t in enumerate(sql):
assert t.manage_level == 1
if __name__ == "__main__":
......
WorkCase/APP/Rule/test_rule_list.py
......@@ -10,6 +10,7 @@ import pytest
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -44,14 +45,16 @@ class TestRuleList(object):
base_type_agent = 126
base_type_snmp = 127
base_business_id_1 = "12"
base_business_id_2 = "13"
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 161
base_host_name = "SS自动化专用主机"
base_name = "SS测试告警规则名称"
base_host_name = "SS 自动化专用"
base_name = "SS 自动化告警规则"
base_template_id_agent = 10001
base_template_id_snmp = 10190
base_template_id_snmp = 10186
@classmethod
def setup_class(cls):
......@@ -59,86 +62,26 @@ class TestRuleList(object):
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)
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)
DataAlarmRule().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：rule.create")
def rule_create(self, token=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 = RuleCreate(_host=self.host)
api.token = token
api.name = name
api.severity = severity
api.url = url
api.description = description
api.status = status
api.manualClose = manualClose
api.enableTimeDefault = enableTimeDefault
api.period = period
api.periodType = periodType
api.triggersList = triggersList
api.triggersList_hostid = triggersList_hostid
api.triggersList_hostName = triggersList_hostName
api.triggersList_itemid = triggersList_itemid
api.triggersList_itemName = triggersList_itemName
api.triggersList_function = triggersList_function
api.triggersList_parameters = triggersList_parameters
api.triggersList_parameters_value = triggersList_parameters_value
api.triggersList_parameters_unit = triggersList_parameters_unit
api.triggersList_operator = triggersList_operator
api.triggersList_constant = triggersList_constant
api.triggersList_logicalSymbol = triggersList_logicalSymbol
api.triggersList_onlineFlag = triggersList_onlineFlag
api.get_response()
return api.response
@allure.step("调用接口：rule.list")
def rule_list(self, token=None, ruleName=None):
api = RuleList(_host=self.host)
api.token = token
api.ruleName = ruleName
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
......@@ -163,149 +106,104 @@ class TestRuleList(object):
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):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.step("断言返回结果数据")
def check_msg(self, response, msg):
_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,
factoryId=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.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("创建测试数据")
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)
@allure.step("获取查询测试数据")
def get_case(self, host_id, item_key):
sql = self.select_hosts(hostid=host_id)
assert sql != []
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)
sql = self.select_hosts_items(host_id=host_id, key_=item_key)
assert sql != []
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.list：token未传")
@allure.story("查询告警规则下拉框：token未传")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_01(self):
self.get_base_token()
response = self.rule_list()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.rule_list()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("rule.list：token的用户已删除")
@allure.story("查询告警规则下拉框：token的用户已删除")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.rule_list(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.rule_list(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("rule.list：token的用户已禁用")
@allure.story("查询告警规则下拉框：token的用户已禁用")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.rule_list(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.rule_list(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("rule.list：无参数调用")
@allure.story("查询告警规则下拉框：无参数调用")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_04(self):
self.get_base_token()
response = self.rule_list(token=self.token)
self.check_code(response=response, code=0)
response = self.app.rule_list(token=self.token)
self.app.check_code(response=response, code=0)
@allure.title("rule.list：ruleName模糊查询")
@allure.story("查询告警规则下拉框：ruleName模糊查询")
def test_case_05(self):
self.get_base_token()
item_key = "agent.ping"
self.case_create(num="05", item_key=item_key)
response = self.rule_list(token=self.token, ruleName=self.base_name)
self.check_code(response=response, code=0)
host_name = self.base_host_name + "05"
name = self.base_name + "05"
host_id = self.app.case_create_host(name=host_name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
self.app.case_create_rule(name=name, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=0)
response = self.app.rule_list(token=self.token, ruleName=self.base_name)
self.app.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
assert data != []
for x, y in enumerate(data):
assert self.base_name in y["name"]
@allure.title("rule.list：ruleName精确查询")
@allure.story("查询告警规则下拉框：ruleName精确查询")
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_list(token=self.token, ruleName=self.name)
self.check_code(response=response, code=0)
host_name = self.base_host_name + "06"
name = self.base_name + "06"
host_id = self.app.case_create_host(name=host_name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
self.app.case_create_rule(name=name, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=0)
response = self.app.rule_list(token=self.token, ruleName=name)
self.app.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
assert data != []
for x, y in enumerate(data):
assert self.name == y["name"]
assert y["name"] == name
if __name__ == "__main__":
......
WorkCase/APP/Rule/test_rule_update.py
......@@ -8,6 +8,8 @@ from __future__ import division
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkData.Zabbix.triggers import DataTriggers
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -42,14 +44,16 @@ class TestRuleUpdate(object):
base_type_agent = 126
base_type_snmp = 127
base_business_id_1 = "12"
base_business_id_2 = "13"
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 161
base_host_name = "SS自动化专用主机"
base_name = "SS测试告警规则名称"
base_host_name = "SS 自动化专用"
base_name = "SS 自动化告警规则"
base_template_id_agent = 10001
base_template_id_snmp = 10190
base_template_id_snmp = 10186
@classmethod
def setup_class(cls):
......@@ -57,125 +61,49 @@ class TestRuleUpdate(object):
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)
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)
DataAlarmRule().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：rule.create")
def rule_create(self, token=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 = RuleCreate(_host=self.host)
api.token = token
api.name = name
api.severity = severity
api.url = url
api.description = description
api.status = status
api.manualClose = manualClose
api.enableTimeDefault = enableTimeDefault
api.period = period
api.periodType = periodType
api.triggersList = triggersList
api.triggersList_hostid = triggersList_hostid
api.triggersList_hostName = triggersList_hostName
api.triggersList_itemid = triggersList_itemid
api.triggersList_itemName = triggersList_itemName
api.triggersList_function = triggersList_function
api.triggersList_parameters = triggersList_parameters
api.triggersList_parameters_value = triggersList_parameters_value
api.triggersList_parameters_unit = triggersList_parameters_unit
api.triggersList_operator = triggersList_operator
api.triggersList_constant = triggersList_constant
api.triggersList_logicalSymbol = triggersList_logicalSymbol
api.triggersList_onlineFlag = triggersList_onlineFlag
api.get_response()
return api.response
@allure.step("调用接口：rule.update")
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
api.url = url
api.description = description
api.status = status
api.manualClose = manualClose
api.enableTimeDefault = enableTimeDefault
api.period = period
api.periodType = periodType
api.triggersList = triggersList
api.triggersList_hostid = triggersList_hostid
api.triggersList_hostName = triggersList_hostName
api.triggersList_itemid = triggersList_itemid
api.triggersList_itemName = triggersList_itemName
api.triggersList_function = triggersList_function
api.triggersList_parameters = triggersList_parameters
api.triggersList_parameters_value = triggersList_parameters_value
api.triggersList_parameters_unit = triggersList_parameters_unit
api.triggersList_operator = triggersList_operator
api.triggersList_constant = triggersList_constant
api.triggersList_logicalSymbol = triggersList_logicalSymbol
api.triggersList_onlineFlag = triggersList_onlineFlag
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
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("连接数据库：Zabbix")
def db_session_zabbix(self):
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base_zabbix, db_user=self.db_user, db_pw=self.db_pw)
return session
@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("查询表：triggers")
def select_triggers(self, description=None, triggerid=None):
session = self.db_session_zabbix()
sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid, description=description)
return sql
@allure.step("查询表：hosts")
def select_hosts(self, hostid=None, host_name=None, available=None, monitor_status=None):
session = self.db_session()
......@@ -188,77 +116,28 @@ class TestRuleUpdate(object):
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,
factoryId=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.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("创建测试数据")
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)
@allure.step("获取查询测试数据")
def get_case(self, host_id, item_key):
sql = self.select_hosts(hostid=host_id)
assert sql != []
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)
sql = self.select_hosts_items(host_id=host_id, key_=item_key)
assert sql != []
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_sql(self, rule_id, name, severity, url, status, description, manualClose, enableTimeDefault,
period, periodType):
sql = self.select_alarm_rule(alarm_ruleid=rule_id)
assert sql != []
for x, y in enumerate(sql):
assert x == 0
assert y.name == name
......@@ -271,72 +150,64 @@ class TestRuleUpdate(object):
assert y.period == period
assert y.period_unit == periodType
@allure.step("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.step("断言返回结果数据")
def check_msg(self, response, msg):
_msg = UtilsResponse().get_msg(response=response)
assert _msg == msg
@allure.title("rule.update：token未传")
@allure.story("更新告警规则：token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.rule_update()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.rule_update()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("rule.update：token的用户已删除")
@allure.story("更新告警规则：token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.rule_update(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.rule_update(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("rule.update：token的用户已禁用")
@allure.story("更新告警规则：token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.rule_update(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.rule_update(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("rule.update：更新普通告警规则")
@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)
host_name = self.base_host_name + "04"
name = self.base_name + "04"
host_id = self.app.case_create_host(name=host_name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
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,
item_key = "system.cpu.load[percpu,avg15]"
self.get_case(host_id=host_id, item_key=item_key)
rule_id = self.app.case_create_rule(name=name, status=CaseBase().status["ON"], host_id=host_id, host_name=self.host_name,
item_key=item_key, itemid=self.item_id, severity=0)
item_key = "agent.ping"
self.get_case(host_id=host_id, item_key=item_key)
response = self.app.rule_update(token=self.token, _id=rule_id, name=name + "修改", severity=5,
url="URL", description="DESCRIPTION", status=CaseBase().status["OFF"],
manualClose=0,
enableTimeDefault=0, period=1, periodType="h",
triggersList_hostid=[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.app.check_code(response=response, code=0)
self.check_sql(rule_id=rule_id, name=name + "修改", severity=5, url="URL", description="DESCRIPTION",
status=CaseBase().status["OFF"], manualClose=0,
enableTimeDefault=0, period=1, periodType="h")
......
WorkCase/APP/__init__.py
......@@ -19,6 +19,7 @@ import allure
from WorkApi.APP.Device.device_delete import DeviceDelete
from WorkApi.APP.Device.device_get import DeviceGet
from WorkApi.APP.File.file_upload import FileUpload
from WorkApi.APP.Members.Get.members_get_role import MembersGetRole
from WorkApi.APP.Members.Update.members_update_info import MembersUpdateInfo
from WorkApi.APP.Members.Update.members_update_status import MembersUpdateStatus
......@@ -26,6 +27,11 @@ from WorkApi.APP.Members.members_create import MembersCreate
from WorkApi.APP.Members.members_delete import MembersDelete
from WorkApi.APP.Members.members_get import MembersGet
from WorkApi.APP.Members.members_list import MembersList
from WorkApi.APP.Rule.rule_create import RuleCreate
from WorkApi.APP.Rule.rule_delete import RuleDelete
from WorkApi.APP.Rule.rule_get import RuleGet
from WorkApi.APP.Rule.rule_list import RuleList
from WorkApi.APP.Rule.rule_update import RuleUpdate
from WorkApi.APP.Settings.Get.settings_get_member import SettingsGetMember
from WorkApi.APP.Settings.settings_changePassword import SettingsChangePassword
from WorkData.Argus.members import DataMembers
......@@ -36,7 +42,6 @@ class AppBase(object):
def __init__(self, host):
self.host = host
self.token = None
self.host_id = None
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
......@@ -257,9 +262,8 @@ class AppBase(object):
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create_host(self, base_name, num, host_type, iplist, port, businessIds):
name = base_name + num
@allure.step("创建测试数据-资产")
def case_create_host(self, name, host_type, iplist, port, businessIds):
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, monitorInterface=port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=businessIds,
......@@ -271,20 +275,20 @@ class AppBase(object):
host_id = UtilsResponse().get_data(response=response)
return host_id
@allure.step("创建测试数据")
@allure.step("创建测试数据-业务类型")
def case_create_businesses_type(self, name):
response = self.businesses_type_create(token=self.token, typeName=name)
self.check_code(response=response, code=0)
type_id = UtilsResponse().get_data(response=response)
return type_id
@allure.step("创建测试数据")
@allure.step("创建测试数据-业务")
def case_create_businesses(self, name, _type, principalPhone=None, principalName=None):
response = self.businesses_create(token=self.token, name=name, _type=_type, principalPhone=principalPhone, principalName=principalName)
business_id = UtilsResponse().get_data(response=response)
return business_id
@allure.step("创建测试数据")
@allure.step("创建测试数据-地址簿")
def case_create_addressbook(self, name, iplist, available):
response = self.addressbook_create(token=self.token, name=name, iplist=iplist, available=available, delay="1h")
self.check_code(response=response, code=0)
......@@ -313,10 +317,13 @@ class AppBase(object):
assert _msg == msg
@allure.step("校验数据")
def check_data_keyValue(self, response, _key, _value):
def check_data_keyValue(self, response, _key, _value, isIn=None):
data = UtilsResponse().get_data(response=response)
for x, y in enumerate(data["list"]):
assert y[_key] == _value
if isIn == 1:
assert _value in y[_key]
else:
assert y[_key] == _value
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, session, members_id=None, value=None):
......@@ -433,3 +440,121 @@ class AppBase(object):
api.memberIds = memberIds
api.get_response()
return api.response
@allure.step("调用接口：rule.create")
def rule_create(self, token=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 = RuleCreate(_host=self.host)
api.token = token
api.name = name
api.severity = severity
api.url = url
api.description = description
api.status = status
api.manualClose = manualClose
api.enableTimeDefault = enableTimeDefault
api.period = period
api.periodType = periodType
api.triggersList = triggersList
api.triggersList_hostid = triggersList_hostid
api.triggersList_hostName = triggersList_hostName
api.triggersList_itemid = triggersList_itemid
api.triggersList_itemName = triggersList_itemName
api.triggersList_function = triggersList_function
api.triggersList_parameters = triggersList_parameters
api.triggersList_parameters_value = triggersList_parameters_value
api.triggersList_parameters_unit = triggersList_parameters_unit
api.triggersList_operator = triggersList_operator
api.triggersList_constant = triggersList_constant
api.triggersList_logicalSymbol = triggersList_logicalSymbol
api.triggersList_onlineFlag = triggersList_onlineFlag
api.get_response()
return api.response
@allure.step("调用接口：rule.delete")
def rule_delete(self, token=None, alarmRuleidList=None):
api = RuleDelete(_host=self.host)
api.token = token
api.alarmRuleidList = alarmRuleidList
api.get_response()
return api.response
@allure.step("调用接口：rule.get")
def rule_get(self, token=None, severity=None, itemid=None, status=None,
manageLevel=None, manageIp=None, hostName=None):
api = RuleGet(_host=self.host)
api.token = token
api.severity = severity
api.status = status
api.itemid = itemid
api.manageLevel = manageLevel
api.manageIp = manageIp
api.hostName = hostName
api.get_response()
return api.response
@allure.step("调用接口：rule.update")
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
api.url = url
api.description = description
api.status = status
api.manualClose = manualClose
api.enableTimeDefault = enableTimeDefault
api.period = period
api.periodType = periodType
api.triggersList = triggersList
api.triggersList_hostid = triggersList_hostid
api.triggersList_hostName = triggersList_hostName
api.triggersList_itemid = triggersList_itemid
api.triggersList_itemName = triggersList_itemName
api.triggersList_function = triggersList_function
api.triggersList_parameters = triggersList_parameters
api.triggersList_parameters_value = triggersList_parameters_value
api.triggersList_parameters_unit = triggersList_parameters_unit
api.triggersList_operator = triggersList_operator
api.triggersList_constant = triggersList_constant
api.triggersList_logicalSymbol = triggersList_logicalSymbol
api.triggersList_onlineFlag = triggersList_onlineFlag
api.get_response()
return api.response
@allure.step("调用接口：rule.list")
def rule_list(self, token=None, ruleName=None):
api = RuleList(_host=self.host)
api.token = token
api.ruleName = ruleName
api.get_response()
return api.response
@allure.step("调用接口：file.upload")
def file_upload(self, token=None, _file=None):
api = FileUpload(_host=self.host)
api.token = token
api.file = _file
api.get_response()
return api.response
@allure.step("创建测试数据-告警规则")
def case_create_rule(self, name, status, severity, host_id, host_name, item_key, itemid, online_flag=None):
response = self.rule_create(token=self.token, name=name, severity=severity, status=status, manualClose=0, triggersList_hostid=[host_id],
triggersList_hostName=[host_name], triggersList_itemid=[itemid],
triggersList_itemName=[item_key], triggersList_function=["avg"],
triggersList_operator=["="], triggersList_constant=[888],
triggersList_parameters_unit=["m"], triggersList_parameters_value=[5], triggersList_onlineFlag=[online_flag])
self.check_code(response=response, code=0)
rule_id = UtilsResponse().get_data(response=response)
return rule_id
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