Commit d6e9b8f1 by sanshi

查询经验手册

parent 90e1cedb
...@@ -26,7 +26,7 @@ class ExperienceGet(object): ...@@ -26,7 +26,7 @@ class ExperienceGet(object):
self._data = {} self._data = {}
self.response = "" self.response = ""
self.local_json = {} self.token = {}
self.hostType = None self.hostType = None
self.hostName = None self.hostName = None
...@@ -34,13 +34,17 @@ class ExperienceGet(object): ...@@ -34,13 +34,17 @@ class ExperienceGet(object):
self.manageIp = None self.manageIp = None
self.businessid = None self.businessid = None
self.alarmRuleid = None self.alarmRuleid = None
self.itemid = None
self.title = None self.title = None
self.tagid = None self.tagName = None
self.description = None self.description = None
self.api = UtilsRequest() self.api = UtilsRequest()
def get_response(self): def get_response(self):
base = ApiBase()
base.dict_add_key(_key="token", value=self.token)
self._headers = base._json
base = ApiBase() base = ApiBase()
base.dict_add_key(_key="hostType", value=self.hostType) base.dict_add_key(_key="hostType", value=self.hostType)
...@@ -49,11 +53,9 @@ class ExperienceGet(object): ...@@ -49,11 +53,9 @@ class ExperienceGet(object):
base.dict_add_key(_key="manageIp", value=self.manageIp) base.dict_add_key(_key="manageIp", value=self.manageIp)
base.dict_add_key(_key="businessid", value=self.businessid) base.dict_add_key(_key="businessid", value=self.businessid)
base.dict_add_key(_key="alarmRuleid", value=self.alarmRuleid) base.dict_add_key(_key="alarmRuleid", value=self.alarmRuleid)
base.dict_add_key(_key="itemid", value=self.itemid)
base.dict_add_key(_key="title", value=self.title) base.dict_add_key(_key="title", value=self.title)
base.dict_add_key(_key="tagid", value=self.tagid) base.dict_add_key(_key="tagName", value=self.tagName)
base.dict_add_key(_key="description", value=self.description) base.dict_add_key(_key="description", value=self.description)
self._params = base._json
self.local_json = base._json self.response = self.api.get(url=self._url, headers=self._headers, params=self._params)
self._json = self.local_json
self.response = self.api.post(url=self._url, json=self._json)
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# 测试用例 # 测试用例
# 作者: 陈磊 # 作者: 陈磊
# 时间: 2019-11-18 # 时间: 2019-12-04
from __future__ import division from __future__ import division
from WorkApi.APP.Api.api_login import ApiLogin
from WorkCase import CaseBase from WorkCase import CaseBase
from WorkData.Argus.members import DataMembers
from WorkUtils.UtilsLog import UtilsLog from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse from WorkUtils.UtilsResponse import UtilsResponse
from WorkUtils.UtilsDataBase import UtilsDataBase from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkApi.APP.Experience.experience_create import ExperienceCreate
from WorkApi.APP.Experience.experience_get import ExperienceGet from WorkApi.APP.Experience.experience_get import ExperienceGet
from WorkData.Argus.experience import DataExperience
from WorkData.Argus.experience_alarm_rule import DataExperienceAlarmRule
from WorkData.Argus.experience_businesses import DataExperienceBusinesses
from WorkData.Argus.experience_hosts import DataExperienceHosts
from WorkData.Argus.experience_hosts_type import DataExperienceHostsType
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 import DataAlarmRule
from WorkData.Argus.businesses import DataBusinesses
from WorkData.Argus.hosts import DataHosts from WorkData.Argus.hosts import DataHosts
from WorkData.Argus.hosts_items import DataHostsItems
from WorkData.Argus.problem import DataProblem
from WorkData.Argus.tag import DataTag
import allure import allure
...@@ -28,16 +47,28 @@ class TestExperienceGet(object): ...@@ -28,16 +47,28 @@ class TestExperienceGet(object):
db_pw = env["db_pw"] db_pw = env["db_pw"]
db_base = env["db_base"] db_base = env["db_base"]
base_host_name = "SS测试主机名" login_name = "SS测试用户名"
base_asset_code = "SS测试主机资产编号" base_id = 10
base_manage_ip = "10.0.28." base_password = "123456"
base_parent_host = "SS测试所属主机"
base_type_agent = 126
base_type_snmp = 127
base_title = "SS测试经验标题"
base_description = "SS测试经验描述"
base_content = "SS测试经验内容"
base_businessName = "SS测试经验业务名称"
base_monitorName = "SS测试经验监控名称"
base_hostid = 10266
base_name = "经验手册-测试数据"
base_ip = "172.99.1.99"
@classmethod @classmethod
def setup_class(cls): def setup_class(cls):
cls.log.debug("开始执行测试套件.......") cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw) 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) DataExperience().delete_like_title(session=session, title=cls.base_title)
@classmethod @classmethod
def teardown_class(cls): def teardown_class(cls):
...@@ -50,17 +81,20 @@ class TestExperienceGet(object): ...@@ -50,17 +81,20 @@ class TestExperienceGet(object):
self.log.debug("测试用例执行结束...") self.log.debug("测试用例执行结束...")
@allure.step("调用接口:experience.get") @allure.step("调用接口:experience.get")
def experience_get(self, hostType=None, hostName=None, manageLevel=None, manageIp=None, businessid=None, alarmRuleid=None, def experience_get(self, token=None, hostType=None, hostName=None, manageLevel=None, manageIp=None, businessid=None, alarmRuleid=None,
title=None, tagid=None, description=None): itemid=None, title=None, tagid=None, tagName=None, description=None):
api = ExperienceGet(_host=self.host) api = ExperienceGet(_host=self.host)
api.token = token
api.hostType = hostType api.hostType = hostType
api.hostName = hostName api.hostName = hostName
api.manageLevel = manageLevel api.manageLevel = manageLevel
api.manageIp = manageIp api.manageIp = manageIp
api.businessid = businessid api.businessid = businessid
api.alarmRuleid = alarmRuleid api.alarmRuleid = alarmRuleid
api.itemid = itemid
api.title = title api.title = title
api.tagid = tagid api.tagid = tagid
api.tagName = tagName
api.description = description api.description = description
api.get_response() api.get_response()
return api.response return api.response
...@@ -75,102 +109,343 @@ class TestExperienceGet(object): ...@@ -75,102 +109,343 @@ class TestExperienceGet(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) 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 return session
@allure.step("查询表:hosts") @allure.step("查询表:experience")
def select_hosts(self, hostid=None, host_name=None, available=None, monitor_status=None): def select_experience(self, experienceid=None, title=None, content=None):
session = self.db_session()
sql = DataExperience().select_all_from_allKeys(session=session, experienceid=experienceid, title=title, content=content)
return sql
@allure.step("查询表:experience_alarm_rule")
def select_experience_alarm_rule(self, experienceid=None):
session = self.db_session() session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid, host_name=host_name, available=available, monitor_status=monitor_status) sql = DataExperienceAlarmRule().select_all_from_allKeys(session=session, experienceid=experienceid)
return sql return sql
@allure.step("新增表数据:hosts") @allure.step("查询表:experience_businesses")
def insert_hosts(self, host_name=None, asset_code=None, manage_ip=None, parent_host=None, available=None, monitor_status=None): def select_experience_businesses(self, experienceid=None):
session = self.db_session() session = self.db_session()
hostid = DataHosts().insert_one(session=session, host_name=host_name, asset_code=asset_code, manage_ip=manage_ip, sql = DataExperienceBusinesses().select_all_from_allKeys(session=session, experienceid=experienceid)
parent_host=parent_host, available=available, monitor_status=monitor_status) return sql
return hostid
@allure.title("experience.get:无参数调用") @allure.step("查询表:experience_hosts")
@allure.story("查询主机监控:无参数调用") def select_experience_hosts(self, experienceid=None):
session = self.db_session()
sql = DataExperienceHosts().select_all_from_allKeys(session=session, experienceid=experienceid)
return sql
@allure.step("查询表:experience_hosts_type")
def select_experience_hosts_type(self, experienceid=None):
session = self.db_session()
sql = DataExperienceHostsType().select_all_from_allKeys(session=session, experienceid=experienceid)
return sql
@allure.step("查询表:experience_tag")
def select_experience_tag(self, experienceid=None):
session = self.db_session()
sql = DataExperienceTag().select_all_from_allKeys(session=session, experienceid=experienceid)
return sql
@allure.step("查询表:tag")
def select_tag(self, tag=None):
session = self.db_session()
sql = DataTag().select_all_from_allKeys(session=session, tag=tag)
return sql
@allure.step("查询表:experience_items")
def select_experience_items(self, experienceid=None):
session = self.db_session()
sql = DataExperienceItems().select_all_from_allKeys(session=session, experienceid=experienceid)
return sql
@allure.step("查询表:experience_ips")
def select_experience_ips(self, experienceid=None):
session = self.db_session()
sql = DataExperienceIps().select_all_from_allKeys(session=session, experienceid=experienceid)
return sql
@allure.step("查询表:experience_problem")
def select_experience_problem(self, experienceid=None):
session = self.db_session()
sql = DataExperienceProblem().select_all_from_allKeys(session=session, experienceid=experienceid)
return sql
@allure.step("调用接口:experience.create")
def experience_create(self, token=None, title=None, content=None, businessids=None, hostTypes=None, hostids=None,
itemids=None, manageIpList=None, alarmRuleids=None, description=None, tagNameList=None, problemidList=None,
hostTypeTree=None):
api = ExperienceCreate(_host=self.host)
api.token = token
api.title = title
api.content = content
api.businessids = businessids
api.hostTypes = hostTypes
api.hostids = hostids
api.itemids = itemids
api.manageIpList = manageIpList
api.alarmRuleids = alarmRuleids
api.description = description
api.tagNameList = tagNameList
api.problemidList = problemidList
api.hostTypeTree = hostTypeTree
api.get_response()
return api.response
@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("获取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)
@allure.step("创建测试数据")
def case_create(self, num):
# 删除第一个表
session = self.db_session()
DataAlarmRule().delete_like_name(session=session, name=self.base_name)
# 删除第二个表
session = self.db_session()
DataBusinesses().delete_like_name(session=session, name=self.base_name)
session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.base_name)
for x, y in enumerate(sql):
# 删除第三个表
session = self.db_session()
DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表
session = self.db_session()
DataProblem().delete_hostid(session=session, hostid=y.hostid)
# 删除第五个表
session = self.db_session()
DataHosts().delete_like_host_name(session=session, host_name=self.base_name)
# 删除第六个表
session = self.db_session()
DataTag().delete_like_tag(session=session, tag=self.base_name)
session = self.db_session()
self.alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.base_name, status=1, severity=1,
manual_close=0, enable_time_default=1, create_time=1)
session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.base_name, _type=1)
session = self.db_session()
self.hostid = DataHosts().insert_one(session=session, host_name=self.base_name, del_flag=0, monitor_status=1,
manage_ip=self.base_ip)
session = self.db_session()
self.itemid = DataHostsItems().insert_one(session=session, host_id=self.hostid, name=self.base_name, key_="key_")
session = self.db_session()
self.eventid = DataProblem().insert_one(session=session, eventid=999999, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.base_name)
self.title = self.base_title + num
self.content = self.base_content + num
self.description = self.base_description + num
response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_agent],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid],
tagNameList=[self.base_name], description=self.description)
self.check_code(response=response, code=0)
self.experienceid = UtilsResponse().get_data(response=response)
@allure.step("校验数据")
def check_select(self, response):
data = UtilsResponse().get_data(response=response)
ids = []
for x, y in enumerate(data["list"]):
ids.append(y["id"])
sql = self.select_experience(experienceid=y["id"])
for q, w in enumerate(sql):
assert y["title"] == w.title
assert y["content"] == w.content
if y["id"] == self.experienceid:
for e, d in enumerate(y["businessList"]):
assert d["experienceid"] == self.experienceid
assert d["businessid"] == self.businessid
assert d["name"] == self.base_name
for e, d in enumerate(y["hostsTypeList"]):
assert d["typeId"] == self.base_type_agent
for e, d in enumerate(y["hostsList"]):
assert d["experienceid"] == self.experienceid
assert d["hostid"] == self.hostid
assert d["host_name"] == self.base_name
for e, d in enumerate(y["itemList"]):
assert d["itemId"] == self.itemid
assert d["itemName"] == self.base_name
for e, d in enumerate(y["manageIpList"]):
assert d["experienceid"] == self.experienceid
assert d["hostid"] == self.hostid
for e, d in enumerate(y["problemDescriptionList"]):
assert d["problemId"] == self.eventid
for e, d in enumerate(y["alarmRuleList"]):
assert d["experienceid"] == self.experienceid
assert d["alarm_ruleid"] == self.alarm_ruleid
assert d["name"] == self.base_name
for e, d in enumerate(y["tagList"]):
assert d["experienceid"] == self.experienceid
assert d["tag"] == self.base_name
assert self.experienceid in ids
@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.title("experience.get:token未传")
@allure.story("查询经验手册:token未传")
@allure.severity("blocker") @allure.severity("blocker")
def test_case_01(self): def test_case_01(self):
host_name = self.base_host_name + "01" self.get_base_token()
asset_code = self.base_asset_code + "01" self.update_members_del_flag(members_id=self.base_id, value=1)
manage_ip = self.base_manage_ip + "1" response = self.experience_get(token=self.token)
parent_host = self.base_parent_host + "01" self.check_code(response=response, code=2004)
self.insert_hosts(host_name=host_name, asset_code=asset_code, manage_ip=manage_ip, parent_host=parent_host, available=1, monitor_status=1) self.check_msg(response=response, msg="用户不存在")
response = self.experience_get()
self.check_code(response=response, code=0)
@allure.title("experience.get:模糊查询hostName") @allure.title("experience.get:token的用户已删除")
@allure.story("查询主机监控:模糊查询hostName") @allure.story("查询经验手册:token的用户已删除")
def test_case_02(self): def test_case_02(self):
host_name = self.base_host_name + "02" self.get_base_token()
asset_code = self.base_asset_code + "02" self.update_members_del_flag(members_id=self.base_id, value=1)
manage_ip = self.base_manage_ip + "2" response = self.experience_get(token=self.token)
parent_host = self.base_parent_host + "02" self.check_code(response=response, code=2004)
self.insert_hosts(host_name=host_name, asset_code=asset_code, manage_ip=manage_ip, parent_host=parent_host, available=1, monitor_status=1) self.check_msg(response=response, msg="用户不存在")
response = self.experience_get(hostName=host_name)
self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
@allure.title("experience.get:查询assetCode") @allure.title("experience.get:token的用户已禁用")
@allure.story("查询主机监控:查询assetCode") @allure.story("查询经验手册:token的用户已禁用")
def test_case_03(self): def test_case_03(self):
host_name = self.base_host_name + "03" self.get_base_token()
asset_code = self.base_asset_code + "03" self.update_members_status(members_id=self.base_id, value=1)
manage_ip = self.base_manage_ip + "3" response = self.experience_get(token=self.token)
parent_host = self.base_parent_host + "03" self.check_code(response=response, code=2003)
self.insert_hosts(host_name=host_name, asset_code=asset_code, manage_ip=manage_ip, parent_host=parent_host, available=1, monitor_status=1) self.check_msg(response=response, msg="账户已被禁用")
response = self.experience_get(assetCode=asset_code)
self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
@allure.title("experience.get:模糊查询manageIp") @allure.title("experience.get:无参数调用")
@allure.story("查询主机监控:模糊查询manageIp") @allure.story("查询经验手册:无参数调用")
def test_case_04(self): def test_case_04(self):
host_name = self.base_host_name + "04" self.get_base_token()
asset_code = self.base_asset_code + "04" self.case_create(num="04")
manage_ip = self.base_manage_ip + "4" response = self.experience_get(token=self.token)
parent_host = self.base_parent_host + "04"
self.insert_hosts(host_name=host_name, asset_code=asset_code, manage_ip=manage_ip, parent_host=parent_host, available=1, monitor_status=1)
response = self.experience_get(manageIp=manage_ip)
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
@allure.title("experience.get:模糊查询parentHost") @allure.title("experience.get:hostType查询")
@allure.story("查询主机监控:模糊查询parentHost") @allure.story("查询经验手册:hostType查询")
def test_case_05(self): def test_case_05(self):
host_name = self.base_host_name + "05" self.get_base_token()
asset_code = self.base_asset_code + "05" self.case_create(num="05")
manage_ip = self.base_manage_ip + "5" response = self.experience_get(token=self.token, hostType=self.base_type_agent)
parent_host = self.base_parent_host + "05"
self.insert_hosts(host_name=host_name, asset_code=asset_code, manage_ip=manage_ip, parent_host=parent_host, available=1, monitor_status=1)
response = self.experience_get(parentHost=parent_host)
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response) self.check_select(response=response)
@allure.title("experience.get:模糊查询available") @allure.title("experience.get:hostName查询")
@allure.story("查询主机监控:模糊查询available") @allure.story("查询经验手册:hostName查询")
def test_case_06(self): def test_case_06(self):
host_name = self.base_host_name + "06" self.get_base_token()
asset_code = self.base_asset_code + "06" self.case_create(num="06")
manage_ip = self.base_manage_ip + "6" response = self.experience_get(token=self.token, hostName=self.base_name)
parent_host = self.base_parent_host + "06"
self.insert_hosts(host_name=host_name, asset_code=asset_code, manage_ip=manage_ip, parent_host=parent_host, available=0, monitor_status=1)
response = self.experience_get(available=0)
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response) self.check_select(response=response)
@allure.title("experience.get:模糊查询monitor_status") @allure.title("experience.get:manageLevel查询")
@allure.story("查询主机监控:模糊查询monitor_status") @allure.story("查询经验手册:manageLevel查询")
def test_case_07(self): def test_case_07(self):
host_name = self.base_host_name + "07" self.get_base_token()
asset_code = self.base_asset_code + "07" self.case_create(num="07")
manage_ip = self.base_manage_ip + "7" response = self.experience_get(token=self.token, manageLevel=1)
parent_host = self.base_parent_host + "07"
self.insert_hosts(host_name=host_name, asset_code=asset_code, manage_ip=manage_ip, parent_host=parent_host, available=1, monitor_status=0)
response = self.experience_get(status=0)
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response) self.check_select(response=response)
@allure.title("experience.get:manageIp查询")
@allure.story("查询经验手册:manageIp查询")
def test_case_08(self):
self.get_base_token()
self.case_create(num="08")
response = self.experience_get(token=self.token, manageIp=self.base_ip)
self.check_code(response=response, code=0)
self.check_select(response=response)
@allure.title("experience.get:businessid查询")
@allure.story("查询经验手册:businessid查询")
def test_case_09(self):
self.get_base_token()
self.case_create(num="09")
response = self.experience_get(token=self.token, businessid=self.businessid)
self.check_code(response=response, code=0)
self.check_select(response=response)
@allure.title("experience.get:alarmRuleid查询")
@allure.story("查询经验手册:alarmRuleid查询")
def test_case_10(self):
self.get_base_token()
self.case_create(num="10")
response = self.experience_get(token=self.token, alarmRuleid=self.alarm_ruleid)
self.check_code(response=response, code=0)
self.check_select(response=response)
@allure.title("experience.get:itemid查询")
@allure.story("查询经验手册:itemid查询")
def test_case_11(self):
self.get_base_token()
self.case_create(num="11")
response = self.experience_get(token=self.token, itemid=self.itemid)
self.check_code(response=response, code=0)
self.check_select(response=response)
@allure.title("experience.get:title查询")
@allure.story("查询经验手册:title查询")
def test_case_12(self):
self.get_base_token()
self.case_create(num="12")
response = self.experience_get(token=self.token, title=self.title)
self.check_code(response=response, code=0)
self.check_select(response=response)
@allure.title("experience.get:tagName查询")
@allure.story("查询经验手册:tagName查询")
def test_case_13(self):
self.get_base_token()
self.case_create(num="13")
response = self.experience_get(token=self.token, tagName=self.base_name)
self.check_code(response=response, code=0)
self.check_select(response=response)
@allure.title("experience.get:description查询")
@allure.story("查询经验手册:description查询")
def test_case_14(self):
self.get_base_token()
self.case_create(num="14")
response = self.experience_get(token=self.token, description=self.description)
self.check_code(response=response, code=0)
self.check_select(response=response)
@allure.title("experience.get:所有字段查询")
@allure.story("查询经验手册:所有字段查询")
def test_case_15(self):
self.get_base_token()
self.case_create(num="15")
response = self.experience_get(token=self.token, hostType=self.base_type_agent, manageLevel=1,
manageIp=self.base_ip, businessid=self.businessid, alarmRuleid=self.alarm_ruleid,
itemid=self.itemid, title=self.title, tagName=self.base_name,
description=self.description)
self.check_code(response=response, code=0)
self.check_select(response=response)
if __name__ == "__main__": if __name__ == "__main__":
...@@ -187,4 +462,4 @@ if __name__ == "__main__": ...@@ -187,4 +462,4 @@ if __name__ == "__main__":
# a = TestExperienceGet() # a = TestExperienceGet()
# a.setup_class() # a.setup_class()
# a.test_case_04() # a.test_case_15()
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