Skip to content

Z
zmops-test
Commit bd660c3a by sanshi
Options

经验手册模块

parent c9cd22fc
Showing with 163 additions and 167 deletions
.idea/dictionaries/Administrator.xml
......@@ -22,6 +22,7 @@
<w>dingding</w>
<w>discoveryids</w>
<w>doinspection</w>
<w>doinspections</w>
<w>drule</w>
<w>druleid</w>
<w>druleids</w>
......
WorkCase/APP/Experience/test_experience_create.py
......@@ -24,6 +24,7 @@ from WorkData.Argus.experience_tag import DataExperienceTag
from WorkData.Argus.experience_items import DataExperienceItems
from WorkData.Argus.experience_ips import DataExperienceIps
from WorkData.Argus.experience_problem import DataExperienceProblem
from WorkData.Argus.alarm_rule_triggers import DataAlarmRuleTriggers
from WorkData.Argus.alarm_rule import DataAlarmRule
from WorkData.Argus.businesses import DataBusinesses
......@@ -216,6 +217,7 @@ class TestExperienceCreate(object):
self.alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.name, status=1, severity=1,
manual_close=0, enable_time_default=1, create_time=1,
description="description")
session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
......@@ -231,6 +233,10 @@ class TestExperienceCreate(object):
self.itemid = itemids[0]
self.insert_problem(eventid=self.eventid, clock=0, severity=1)
session = self.db_session()
DataAlarmRuleTriggers().insert_one(session=session, hostid=self.host_id, itemid=self.itemid, alarm_ruleid=self.alarm_ruleid,
remark="remark")
@allure.step("校验数据")
def check_select(self, response, title, content, description):
self.experienceid = UtilsResponse().get_data(response=response)
......@@ -242,39 +248,13 @@ class TestExperienceCreate(object):
assert y.remark == description
assert y.host_type_tree == "[[1,120,126],[1,120,127]]"
@allure.title("experience.create：token未传")
@allure.story("创建经验手册：token未传")
@allure.severity("blocker")
def test_case_01(self):
response = self.app.experience_create()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="token不存在")
@allure.title("experience.create：token的用户已删除")
@allure.story("创建经验手册：token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.experience_create(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("experience.create：token的用户已禁用")
@allure.story("创建经验手册：token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.experience_create(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("experience.create：必填项")
@allure.story("创建经验手册：必填项")
@allure.severity("blocker")
def test_case_04(self):
title = self.base_title + "04"
content = self.base_content + "04"
description = self.base_description + "04"
def test_case_01(self):
title = self.base_title + "01"
content = self.base_content + "01"
description = self.base_description + "01"
response = self.app.experience_create(token=self.token, title=title, content=content, description=description, hostTypes=[self.base_type_snmp],
hostTypeTree=[self.base_type_tree_1, self.base_type_tree_2])
self.app.check_code(response=response, code=0)
......@@ -283,11 +263,11 @@ class TestExperienceCreate(object):
@allure.title("experience.create：所有字段")
@allure.story("创建经验手册：所有字段")
@allure.severity("blocker")
def test_case_05(self):
title = self.base_title + "05"
content = self.base_content + "05"
description = self.base_description + "05"
self.case_create(num="05")
def test_case_02(self):
title = self.base_title + "02"
content = self.base_content + "02"
description = self.base_description + "02"
self.case_create(num="02")
response = self.app.experience_create(token=self.token, title=title, content=content, hostTypes=[self.base_type_snmp],
hostTypeTree=[self.base_type_tree_1, self.base_type_tree_2],
......@@ -361,4 +341,4 @@ if __name__ == "__main__":
# a = TestExperienceCreate()
# a.setup_class()
# a.setup_method()
# a.test_case_05()
# a.test_case_02()
WorkCase/APP/Experience/test_experience_delete.py
......@@ -10,6 +10,7 @@ from WorkApi.APP.Api.api_login import ApiLogin
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkData.Argus.alarm_rule_triggers import DataAlarmRuleTriggers
from WorkData.Argus.members import DataMembers
from WorkData.Zabbix.events import DataEvents
from WorkUtils.UtilsLog import UtilsLog
......@@ -211,43 +212,22 @@ class TestExperienceDelete(object):
DataProblem().insert_one(session=session, eventid=self.eventid, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.name)
session = self.db_session()
DataAlarmRuleTriggers().insert_one(session=session, hostid=self.host_id, itemid=self.itemid, alarm_ruleid=self.alarm_ruleid,
remark="remark")
@allure.step("校验数据")
def check_select(self, experienceid):
sql = self.select_experience(experienceid=experienceid)
assert sql == []
@allure.title("experience.delete：token未传")
@allure.story("删除经验手册：token未传")
@allure.severity("blocker")
def test_case_01(self):
response = self.app.experience_delete()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="token不存在")
@allure.title("experience.delete：token的用户已删除")
@allure.story("删除经验手册：token的用户已删除")
def test_case_02(self):
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.experience_delete(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("experience.delete：token的用户已禁用")
@allure.story("删除经验手册：token的用户已禁用")
def test_case_03(self):
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.experience_delete(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("experience.delete：删除一个")
@allure.story("删除经验手册：删除一个")
def test_case_04(self):
title = self.base_title + "04"
content = self.base_content + "04"
description = self.base_description + "04"
self.case_create(num="04")
def test_case_01(self):
title = self.base_title + "01"
content = self.base_content + "01"
description = self.base_description + "01"
self.case_create(num="01")
experienceid_1 = self.app.case_create_experience(title=title, content=content, description=description, host_types=[self.base_type_agent])
response = self.app.experience_delete(token=self.token, experienceids=[experienceid_1])
self.app.check_code(response=response, code=0)
......@@ -255,13 +235,13 @@ class TestExperienceDelete(object):
@allure.title("experience.delete：删除多个")
@allure.story("删除经验手册：删除多个")
def test_case_05(self):
title_1 = self.base_title + "051"
title_2 = self.base_title + "052"
title_3 = self.base_title + "053"
content = self.base_content + "05"
description = self.base_description + "05"
self.case_create(num="05")
def test_case_02(self):
title_1 = self.base_title + "021"
title_2 = self.base_title + "022"
title_3 = self.base_title + "023"
content = self.base_content + "02"
description = self.base_description + "02"
self.case_create(num="2")
experienceid_1 = self.app.case_create_experience(title=title_1, content=content, description=description, host_types=[self.base_type_agent])
experienceid_2 = self.app.case_create_experience(title=title_2, content=content, description=description, host_types=[self.base_type_agent])
experienceid_3 = self.app.case_create_experience(title=title_3, content=content, description=description, host_types=[self.base_type_agent])
......
WorkCase/APP/Experience/test_experience_items_list.py
......@@ -89,46 +89,17 @@ class TestExperienceItemsList(object):
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid, host_name=host_name, available=available, monitor_status=monitor_status)
return sql
@allure.title("experience.items.list：token未传")
@allure.story("查询监控项下拉框：token未传")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_01(self):
response = self.app.experience_items_list()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="token不存在")
@allure.title("experience.items.list：token的用户已删除")
@allure.story("查询监控项下拉框：token的用户已删除")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_02(self):
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.experience_items_list(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("experience.items.list：token的用户已禁用")
@allure.story("查询监控项下拉框：token的用户已禁用")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_03(self):
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.experience_items_list(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("experience.items.list：无参数调用")
@allure.story("查询监控项下拉框：无参数调用")
@allure.severity("blocker")
def test_case_04(self):
def test_case_01(self):
response = self.app.experience_items_list(token=self.token)
self.app.check_code(response=response, code=0)
@allure.title("experience.items.list：itemName查询")
@allure.story("查询监控项下拉框：itemName查询")
@allure.severity("blocker")
def test_case_05(self):
def test_case_02(self):
response = self.app.experience_items_list(token=self.token, itemName="ping")
self.app.check_code(response=response, code=0)
self.app.check_keyValue(_json=UtilsResponse().get_data(response=response), _key="name", _value="ping", isIn=1)
......
WorkCase/APP/Experience/test_experience_problem_list.py
......@@ -145,43 +145,14 @@ class TestExperienceProblemList(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.title("experience.problem.list：token未传")
@allure.story("查询告警下拉框：token未传")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_01(self):
response = self.app.experience_problem_list()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="token不存在")
@allure.title("experience.problem.list：token的用户已删除")
@allure.story("查询告警下拉框：token的用户已删除")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_02(self):
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.experience_problem_list(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("experience.problem.list：token的用户已禁用")
@allure.story("查询告警下拉框：token的用户已禁用")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_03(self):
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.experience_problem_list(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("experience.problem.list：problemName成功查询")
@allure.story("查询告警下拉框：problemName成功查询")
@allure.severity("blocker")
def test_case_04(self):
title = self.base_title + "04"
content = self.base_content + "04"
description = self.base_description + "04"
self.case_create(num="04")
def test_case_01(self):
title = self.base_title + "01"
content = self.base_content + "01"
description = self.base_description + "01"
self.case_create(num="01")
self.app.case_create_experience(title=title, content=content, description=description, host_types=[self.base_type_agent])
response = self.app.experience_problem_list(token=self.token, problemName="SS")
......
WorkCase/APP/Experience/test_experience_update.py
......@@ -10,6 +10,7 @@ from WorkApi.APP.Api.api_login import ApiLogin
from WorkApi.APP.Device.device_create import DeviceCreate
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkData.Argus.alarm_rule_triggers import DataAlarmRuleTriggers
from WorkData.Zabbix.problem import DataProblem as ZabbixDataProblem
from WorkData.Zabbix.events import DataEvents
from WorkUtils.UtilsLog import UtilsLog
......@@ -208,6 +209,9 @@ class TestExperienceUpdate(object):
DataProblem().insert_one(session=session, eventid=self.eventid, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.name)
session = self.db_session()
DataAlarmRuleTriggers().insert_one(session=session, hostid=self.host_id, itemid=self.itemid, alarm_ruleid=self.alarm_ruleid,
remark="remark")
@allure.step("创建测试数据")
def case_create_update(self, num):
......@@ -298,42 +302,18 @@ class TestExperienceUpdate(object):
for q, w in enumerate(sql):
assert w.tagid == tagids[0]
@allure.title("experience.update：token未传")
@allure.story("更新经验手册：token未传")
@allure.severity("blocker")
def test_case_01(self):
response = self.app.experience_update()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="token不存在")
@allure.title("experience.update：token的用户已删除")
@allure.story("更新经验手册：token的用户已删除")
def test_case_02(self):
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.experience_update(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("experience.update：token的用户已禁用")
@allure.story("更新经验手册：token的用户已禁用")
def test_case_03(self):
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.experience_update(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("experience.update：更新所有字段")
@allure.story("更新经验手册：更新所有字段")
def test_case_04(self):
title = self.base_title + "04"
content = self.base_content + "04"
description = self.base_description + "04"
def test_case_01(self):
title = self.base_title + "01"
content = self.base_content + "01"
description = self.base_description + "01"
self.update_title = title + "修改"
self.update_content = content + "修改"
self.update_description = description + "修改"
self.case_create(num="04")
self.case_create_update(num="041")
self.case_create(num="01")
self.case_create_update(num="011")
self.experienceid = self.app.case_create_experience(title=title, content=content, description=description, host_types=[self.base_type_agent])
response = self.app.experience_update(token=self.token, _id=self.experienceid, title=self.update_title, content=self.update_content,
description=self.update_description, businessids=[self.update_businessid], hostTypes=[self.base_type_snmp],
......
WorkData/Argus/alarm_rule_triggers.py 0 → 100644
# -*- coding: utf_8 -*-
# 表名: alarm_rule_triggers
# 作者: 陈磊
# 时间: 2019-12-23
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, VARCHAR, SMALLINT, DATETIME, DECIMAL, BigInteger, FLOAT, NUMERIC, TEXT
from sqlalchemy.orm.exc import MultipleResultsFound, NoResultFound
from sqlalchemy import or_
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
import time
Base = declarative_base()
class AlarmRuleTriggers(Base):
"""
告警与触发器关系表
"""
__tablename__ = "alarm_rule_triggers"
id = Column(Integer, primary_key=True)
hostid = Column(VARCHAR(255), comment="资产id")
itemid = Column(VARCHAR(255), comment="监控项")
alarm_ruleid = Column(Integer, comment="告警规则id")
remark = Column(Integer, comment="备注")
class DataAlarmRuleTriggers(object):
def __init__(self):
self.log = UtilsLog()
self.log.info(self.__class__)
self.table = AlarmRuleTriggers
def select_all_from_allKeys(self, session, hostid=None, itemid=None, alarm_ruleid=None):
"""
:param session: 指针
:param hostid:
:param itemid:
:param alarm_ruleid:
:return: 查询结果
"""
self.log.debug("查询数据库:")
try:
base = UtilsDataBase()
base.add_param(_key="hostid", value=hostid)
base.add_param(_key="itemid", value=itemid)
base.add_param(_key="alarm_ruleid", value=alarm_ruleid)
sql_rep = session.query(self.table).filter_by(**base.param).all()
self.log.debug(sql_rep)
session.close()
return sql_rep
except UtilsDataBase().errors as error:
self.log.error("异常:")
self.log.error(error)
session.close()
return error
def delete_hostid(self, session, hostid):
"""
:param session: 指针
:param hostid:
:return:
"""
self.log.debug("查询数据库:")
try:
self.log.debug("删除相关测试数据")
session.query(self.table).filter_by(hostid=hostid).delete(synchronize_session=False)
# session.delete(sql)
session.commit()
session.close()
self.log.debug("删除成功")
return True
except UtilsDataBase().errors as error:
self.log.error("异常:")
self.log.error(error)
session.close()
return error
def insert_one(self, session, hostid=None, itemid=None, alarm_ruleid=None, remark=None):
"""
:param session: 指针
:param hostid:
:param itemid:
:param alarm_ruleid:
:param remark:
:return:
"""
self.log.debug("新增表数据:")
try:
base = UtilsDataBase()
base.add_param(_key="hostid", value=hostid)
base.add_param(_key="itemid", value=itemid)
base.add_param(_key="alarm_ruleid", value=alarm_ruleid)
base.add_param(_key="remark", value=remark)
sql_rep = self.table(**base.param)
session.add(sql_rep)
session.commit()
_id = sql_rep.id
self.log.debug("新增的alarm_rule_triggers_id:%s" % _id)
session.close()
return _id
except UtilsDataBase().errors as error:
self.log.error("异常:")
self.log.error(error)
session.close()
return error
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