Skip to content

Z
zmops-test
Commit 0c3e1ec3 by sanshi
Options

整个经验手册重构

parent b1933529
Showing with 566 additions and 1014 deletions
WorkCase/API/__init__.py
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-27
import allure
from WorkApi.API.Problem.problem_get import ProblemGet
class ApiBase(object):
def __init__(self, host):
self.host = host
self.token = None
@allure.step("调用接口：problem.get")
def problem_get(self, eventids=None, groupids=None, hostids=None, objectids=None, applicationids=None,
source=None, _object=None, acknowledged=None, suppressed=None, severities=None, evaltype=None,
tags=None, recent=None, eventid_from=None, eventid_till=None, time_from=None,
time_till=None, selectAcknowledges=None, selectTags=None, selectSuppressionData=None,
countOutput=None, editable=None, excludeSearch=None,
_filter=None, limit=None, output=None, preservekeys=None, search=None, searchByAny=None,
searchWildcardsEnabled=None, sortfield=None, sortorder=None, startSearch=None):
api = ProblemGet(_host=self.host)
api.eventids = eventids
api.groupids = groupids
api.hostids = hostids
api.objectids = objectids
api.applicationids = applicationids
api.source = source
api.object = _object
api.acknowledged = acknowledged
api.suppressed = suppressed
api.severities = severities
api.evaltype = evaltype
api.tags = tags
api.recent = recent
api.eventid_from = eventid_from
api.eventid_till = eventid_till
api.time_from = time_from
api.time_till = time_till
api.selectAcknowledges = selectAcknowledges
api.selectTags = selectTags
api.selectSuppressionData = selectSuppressionData
api.countOutput = countOutput
api.editable = editable
api.excludeSearch = excludeSearch
api.filter = _filter
api.limit = limit
api.output = output
api.preservekeys = preservekeys
api.search = search
api.searchByAny = searchByAny
api.searchWildcardsEnabled = searchWildcardsEnabled
api.sortfield = sortfield
api.sortorder = sortorder
api.startSearch = startSearch
api.get_response()
return api.response
WorkCase/APP/Experience/test_experience_create.py
......@@ -9,6 +9,7 @@ from __future__ import division
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.members import DataMembers
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -55,19 +56,34 @@ class TestExperienceCreate(object):
base_business_id_1 = "12"
base_business_id_2 = "13"
base_type_tree_1 = [1, 120, base_type_agent]
base_type_tree_2 = [1, 120, base_type_snmp]
base_ip = "172.16.3.170"
base_port = 161
base_title = "SS测试经验标题"
base_content = "SS测试经验内容"
base_title = "SS 自动化专用 经验手册 标题"
base_content = "SS 自动化专用 经验手册 内容"
base_description = "SS 自动化专用 经验手册 描述"
base_name = "经验手册-测试数据"
base_name = "SS 自动化专用 经验手册相关"
base_eventid = 91000000
session = UtilsDataBase().conn_mysql(db_url=db_url, db_port=db_port, db_base=db_base, db_user=db_user, db_pw=db_pw)
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataExperience().delete_like_title(session=session, title=cls.base_title)
DataAlarmRule().delete_like_name(session=cls.session, name=cls.base_name)
DataBusinesses().delete_like_name(session=cls.session, name=cls.base_name)
sql = DataHosts().select_all_like_host_name(session=cls.session, host_name=cls.base_name)
for x, y in enumerate(sql):
DataHostsItems().delete_host_id(session=cls.session, host_id=y.hostid)
DataProblem().delete_like_name(session=cls.session, name=cls.base_name)
DataHosts().delete_like_host_name(session=cls.session, host_name=cls.base_name)
DataTag().delete_like_tag(session=cls.session, tag=cls.base_name)
DataExperience().delete_like_title(session=cls.session, title=cls.base_title)
@classmethod
def teardown_class(cls):
......@@ -75,69 +91,17 @@ class TestExperienceCreate(object):
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("更新表数据：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("调用接口：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("调用接口：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("断言返回结果")
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
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):
......@@ -204,53 +168,6 @@ class TestExperienceCreate(object):
sql = DataExperienceProblem().select_all_from_allKeys(session=session, experienceid=experienceid)
return sql
@allure.step("创建测试数据")
def host_create(self, num):
name = self.base_name + num
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_snmp, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None, port=None, businessTree=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.port = port
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.businessTree = businessTree
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session()
......@@ -259,44 +176,19 @@ class TestExperienceCreate(object):
@allure.step("创建测试数据")
def case_create(self, num):
self.title = self.base_title + num
self.content = self.base_content + num
self.name = self.base_name + num
self.eventid = self.base_eventid + int(num)
# 删除第一个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataAlarmRule().delete_like_name(session=session, name=self.base_name)
# 删除第二个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataBusinesses().delete_like_name(session=session, name=self.base_name)
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.name)
for x, y in enumerate(sql):
# 删除第三个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_like_name(session=session, name=self.base_name)
# 删除第五个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=self.base_name)
# 删除第六个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataTag().delete_like_tag(session=session, tag=self.base_name)
session = self.db_session()
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)
session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
self.hostid = self.host_create(num=num)
self.hostid = self.app.case_create_host(
name=self.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
)
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
......@@ -309,81 +201,70 @@ class TestExperienceCreate(object):
status=0, hostid=self.hostid, name=self.name)
@allure.step("校验数据")
def check_select(self, response):
def check_select(self, response, title, content, description):
self.experienceid = UtilsResponse().get_data(response=response)
sql = self.select_experience(experienceid=self.experienceid)
assert sql != []
for x, y in enumerate(sql):
assert y.title == self.title
assert y.content == self.content
sql = self.select_experience_alarm_rule(experienceid=self.experienceid)
for x, y in enumerate(sql):
assert y.title == self.title
assert y.content == self.content
assert y.title == title
assert y.content == content
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):
self.get_base_token()
response = self.experience_create()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
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.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.experience_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.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.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.experience_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.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):
self.get_base_token()
self.title = self.base_title + "04"
self.content = self.base_content + "04"
response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_snmp])
self.check_code(response=response, code=0)
self.experienceid = UtilsResponse().get_data(response=response)
sql = self.select_experience(experienceid=self.experienceid)
for x, y in enumerate(sql):
assert y.title == self.title
assert y.content == self.content
title = self.base_title + "04"
content = self.base_content + "04"
description = self.base_description + "04"
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)
self.check_select(response=response, title=title, content=content, description=description)
@allure.title("experience.create：所有字段")
@allure.story("创建经验手册：所有字段")
@allure.severity("blocker")
def test_case_05(self):
self.get_base_token()
title = self.base_title + "05"
content = self.base_content + "05"
description = self.base_description + "05"
self.case_create(num="05")
response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_snmp],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid], tagNameList=[self.base_name])
self.check_code(response=response, code=0)
self.experienceid = UtilsResponse().get_data(response=response)
sql = self.select_experience(experienceid=self.experienceid)
assert sql != []
for x, y in enumerate(sql):
assert y.title == self.title
assert y.content == self.content
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],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid],
tagNameList=[self.base_name], description=description)
self.app.check_code(response=response, code=0)
self.check_select(response=response, title=title, content=content, description=description)
# 校验第一个表
sql = self.select_experience_alarm_rule(experienceid=self.experienceid)
......@@ -448,4 +329,5 @@ if __name__ == "__main__":
# a = TestExperienceCreate()
# a.setup_class()
# a.setup_method()
# a.test_case_05()
WorkCase/APP/Experience/test_experience_delete.py
......@@ -9,6 +9,7 @@ from __future__ import division
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.members import DataMembers
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -59,18 +60,29 @@ class TestExperienceDelete(object):
base_ip = "172.16.3.170"
base_port = 161
base_title = "SS测试经验标题"
base_content = "SS测试经验内容"
base_description = "SS测试经验描述"
base_title = "SS 自动化专用 经验手册 标题"
base_content = "SS 自动化专用 经验手册 内容"
base_description = "SS 自动化专用 经验手册 描述"
base_name = "经验手册-测试数据"
base_name = "SS 自动化专用 经验手册相关"
base_eventid = 91000000
session = UtilsDataBase().conn_mysql(db_url=db_url, db_port=db_port, db_base=db_base, db_user=db_user, db_pw=db_pw)
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataExperience().delete_like_title(session=session, title=cls.base_title)
DataAlarmRule().delete_like_name(session=cls.session, name=cls.base_name)
DataBusinesses().delete_like_name(session=cls.session, name=cls.base_name)
sql = DataHosts().select_all_like_host_name(session=cls.session, host_name=cls.base_name)
for x, y in enumerate(sql):
DataHostsItems().delete_host_id(session=cls.session, host_id=y.hostid)
DataProblem().delete_like_name(session=cls.session, name=cls.base_name)
DataHosts().delete_like_host_name(session=cls.session, host_name=cls.base_name)
DataTag().delete_like_tag(session=cls.session, tag=cls.base_name)
DataExperience().delete_like_title(session=cls.session, title=cls.base_title)
@classmethod
def teardown_class(cls):
......@@ -78,31 +90,17 @@ class TestExperienceDelete(object):
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("调用接口：experience.delete")
def experience_delete(self, token=None, experienceids=None):
api = ExperienceDelete(_host=self.host)
api.token = token
api.experienceids = experienceids
api.get_response()
return api.response
@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
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):
......@@ -169,87 +167,6 @@ class TestExperienceDelete(object):
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 host_create(self, num):
name = self.base_name + num
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_snmp, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None, port=None, businessTree=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.port = port
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.businessTree = businessTree
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session()
......@@ -261,38 +178,16 @@ class TestExperienceDelete(object):
self.name = self.base_name + num
self.eventid = self.base_eventid + int(num)
# 删除第一个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataAlarmRule().delete_like_name(session=session, name=self.base_name)
# 删除第二个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataBusinesses().delete_like_name(session=session, name=self.base_name)
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.name)
for x, y in enumerate(sql):
# 删除第三个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_like_name(session=session, name=self.base_name)
# 删除第五个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=self.base_name)
# 删除第六个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataTag().delete_like_tag(session=session, tag=self.base_name)
session = self.db_session()
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)
session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
self.hostid = self.host_create(num=num)
self.hostid = self.app.case_create_host(
name=self.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
)
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
......@@ -304,83 +199,61 @@ class TestExperienceDelete(object):
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.name)
@allure.step("创建经验手册")
def case_create_experience(self, num):
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)
return self.experienceid
@allure.step("校验数据")
def check_select(self, experienceid):
sql = self.select_experience(experienceid=experienceid)
assert sql == []
@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.delete：token未传")
@allure.story("删除经验手册：token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.experience_delete()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
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.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.experience_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.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.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.experience_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.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):
self.get_base_token()
title = self.base_title + "04"
content = self.base_content + "04"
description = self.base_description + "04"
self.case_create(num="04")
self.case_create_experience(num="04")
response = self.experience_delete(token=self.token, experienceids=[self.experienceid])
self.check_code(response=response, code=0)
self.check_select(experienceid=self.experienceid)
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)
self.check_select(experienceid=experienceid_1)
@allure.title("experience.delete：删除多个")
@allure.story("删除经验手册：删除多个")
def test_case_06(self):
self.get_base_token()
self.case_create(num="06")
experienceid_1 = self.case_create_experience(num="061")
experienceid_2 = self.case_create_experience(num="062")
experienceid_3 = self.case_create_experience(num="063")
response = self.experience_delete(token=self.token, experienceids=[experienceid_1, experienceid_2, experienceid_3])
self.check_code(response=response, code=0)
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")
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])
response = self.app.experience_delete(token=self.token, experienceids=[experienceid_1, experienceid_2, experienceid_3])
self.app.check_code(response=response, code=0)
self.check_select(experienceid=experienceid_1)
self.check_select(experienceid=experienceid_2)
self.check_select(experienceid=experienceid_3)
......
WorkCase/APP/Experience/test_experience_get.py
......@@ -9,6 +9,7 @@ from __future__ import division
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.members import DataMembers
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -59,18 +60,29 @@ class TestExperienceGet(object):
base_ip = "172.16.3.170"
base_port = 161
base_title = "SS测试经验标题"
base_content = "SS测试经验内容"
base_description = "SS测试经验描述"
base_title = "SS 自动化专用 经验手册 标题"
base_content = "SS 自动化专用 经验手册 内容"
base_description = "SS 自动化专用 经验手册 描述"
base_name = "经验手册-测试数据"
base_name = "SS 自动化专用 经验手册相关"
base_eventid = 91000000
session = UtilsDataBase().conn_mysql(db_url=db_url, db_port=db_port, db_base=db_base, db_user=db_user, db_pw=db_pw)
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataExperience().delete_like_title(session=session, title=cls.base_title)
DataAlarmRule().delete_like_name(session=cls.session, name=cls.base_name)
DataBusinesses().delete_like_name(session=cls.session, name=cls.base_name)
sql = DataHosts().select_all_like_host_name(session=cls.session, host_name=cls.base_name)
for x, y in enumerate(sql):
DataHostsItems().delete_host_id(session=cls.session, host_id=y.hostid)
DataProblem().delete_like_name(session=cls.session, name=cls.base_name)
DataHosts().delete_like_host_name(session=cls.session, host_name=cls.base_name)
DataTag().delete_like_tag(session=cls.session, tag=cls.base_name)
DataExperience().delete_like_title(session=cls.session, title=cls.base_title)
@classmethod
def teardown_class(cls):
......@@ -78,42 +90,17 @@ class TestExperienceGet(object):
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("调用接口：experience.get")
def experience_get(self, token=None, hostType=None, hostName=None, manageLevel=None, manageIp=None, businessid=None, alarmRuleid=None,
itemid=None, title=None, tagid=None, tagName=None, description=None):
api = ExperienceGet(_host=self.host)
api.token = token
api.hostType = hostType
api.hostName = hostName
api.manageLevel = manageLevel
api.manageIp = manageIp
api.businessid = businessid
api.alarmRuleid = alarmRuleid
api.itemid = itemid
api.title = title
api.tagid = tagid
api.tagName = tagName
api.description = description
api.get_response()
return api.response
@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
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):
......@@ -180,75 +167,6 @@ class TestExperienceGet(object):
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("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None, port=None, businessTree=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.port = port
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.businessTree = businessTree
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session()
......@@ -256,57 +174,20 @@ class TestExperienceGet(object):
return sql
@allure.step("创建测试数据")
def host_create(self, num):
name = self.base_name + num
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_snmp, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@allure.step("创建测试数据")
def case_create(self, num):
self.title = self.base_title + num
self.content = self.base_content + num
self.name = self.base_name + num
self.description = self.base_description + num
self.eventid = self.base_eventid + int(num)
# 删除第一个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataAlarmRule().delete_like_name(session=session, name=self.base_name)
# 删除第二个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataBusinesses().delete_like_name(session=session, name=self.base_name)
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.name)
for x, y in enumerate(sql):
# 删除第三个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_like_name(session=session, name=self.base_name)
# 删除第五个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=self.base_name)
# 删除第六个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataTag().delete_like_tag(session=session, tag=self.base_name)
session = self.db_session()
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)
session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
self.hostid = self.host_create(num=num)
self.hostid = self.app.case_create_host(
name=self.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
)
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
......@@ -317,13 +198,6 @@ class TestExperienceGet(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)
response = self.experience_create(token=self.token, title=self.title, content=self.content, hostTypes=[self.base_type_snmp],
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)
return self.experienceid
@allure.step("校验数据")
def check_select(self, response):
......@@ -361,81 +235,84 @@ class TestExperienceGet(object):
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")
def test_case_01(self):
self.get_base_token()
response = self.experience_get()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.experience_get()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("experience.get：token的用户已删除")
@allure.story("查询经验手册：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.experience_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.experience_get(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("experience.get：token的用户已禁用")
@allure.story("查询经验手册：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.experience_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.experience_get(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("experience.get：无参数调用")
@allure.story("查询经验手册：无参数调用")
def test_case_04(self):
self.get_base_token()
self.case_create(num="04")
response = self.experience_get(token=self.token)
self.check_code(response=response, code=0)
response = self.app.experience_get(token=self.token)
self.app.check_code(response=response, code=0)
@allure.title("experience.get：hostType查询")
@allure.story("查询经验手册：hostType查询")
def test_case_05(self):
self.get_base_token()
title = self.base_title + "05"
content = self.base_content + "05"
description = self.base_description + "05"
self.case_create(num="05")
response = self.experience_get(token=self.token, hostType=self.base_type_snmp)
self.check_code(response=response, code=0)
self.check_select(response=response)
self.app.experience_create(token=self.token, title=title, content=content, hostTypes=[self.base_type_snmp],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid],
tagNameList=[self.base_name], description=description)
response = self.app.experience_get(token=self.token, hostType=self.base_type_snmp)
self.app.check_code(response=response, code=0)
self.app.check_keyValue(_json=UtilsResponse().get_data_list(response=response), _key="hostsTypeList",
_value=[{"typeId": self.base_type_snmp}])
@allure.title("experience.get：hostName查询")
@allure.story("查询经验手册：hostName查询")
def test_case_06(self):
self.get_base_token()
title = self.base_title + "06"
content = self.base_content + "06"
description = self.base_description + "06"
self.case_create(num="06")
response = self.experience_get(token=self.token, hostName=self.base_name)
self.check_code(response=response, code=0)
self.check_select(response=response)
self.app.experience_create(token=self.token, title=title, content=content, hostTypes=[self.base_type_snmp],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid],
tagNameList=[self.base_name], description=description)
response = self.app.experience_get(token=self.token, hostName=self.name)
self.app.check_code(response=response, code=0)
self.app.check_keyValue(_json=UtilsResponse().get_data_list(response=response), _key="hostsList", _value=self.base_type_snmp)
@allure.title("experience.get：manageLevel查询")
@allure.story("查询经验手册：manageLevel查询")
def test_case_07(self):
self.get_base_token()
title = self.base_title + "07"
content = self.base_content + "07"
description = self.base_description + "07"
self.case_create(num="07")
response = self.experience_get(token=self.token, manageLevel=1)
self.check_code(response=response, code=0)
self.check_select(response=response)
self.app.experience_create(token=self.token, title=title, content=content, hostTypes=[self.base_type_snmp],
alarmRuleids=[self.alarm_ruleid], businessids=[self.businessid], hostids=[self.hostid],
manageIpList=[self.hostid], itemids=[self.itemid], problemidList=[self.eventid],
tagNameList=[self.base_name], description=description)
response = self.app.experience_get(token=self.token, manageLevel=1)
self.app.check_code(response=response, code=0)
self.app.check_keyValue(_json=UtilsResponse().get_data_list(response=response), _key="hostsList", _value=self.base_type_snmp)
@allure.title("experience.get：manageIp查询")
@allure.story("查询经验手册：manageIp查询")
......@@ -520,11 +397,12 @@ if __name__ == "__main__":
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestExperienceGet()
# a.setup_class()
# a.test_case_05()
# case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case, case_info=case_info)
# r.run_main()
a = TestExperienceGet()
a.setup_class()
a.setup_method()
a.test_case_06()
WorkCase/APP/Experience/test_experience_items_list.py
......@@ -10,7 +10,14 @@ import pytest
from WorkApi.APP.Api.api_login import ApiLogin
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkData.Argus.alarm_rule import DataAlarmRule
from WorkData.Argus.businesses import DataBusinesses
from WorkData.Argus.experience import DataExperience
from WorkData.Argus.hosts_items import DataHostsItems
from WorkData.Argus.members import DataMembers
from WorkData.Argus.problem import DataProblem
from WorkData.Argus.tag import DataTag
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse
from WorkUtils.UtilsDataBase import UtilsDataBase
......@@ -44,28 +51,30 @@ class TestExperienceItemsList(object):
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
DataAlarmRule().delete_like_name(session=cls.session, name=cls.base_name)
DataBusinesses().delete_like_name(session=cls.session, name=cls.base_name)
sql = DataHosts().select_all_like_host_name(session=cls.session, host_name=cls.base_name)
for x, y in enumerate(sql):
DataHostsItems().delete_host_id(session=cls.session, host_id=y.hostid)
DataProblem().delete_like_name(session=cls.session, name=cls.base_name)
DataHosts().delete_like_host_name(session=cls.session, host_name=cls.base_name)
DataTag().delete_like_tag(session=cls.session, tag=cls.base_name)
DataExperience().delete_like_title(session=cls.session, title=cls.base_title)
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("调用接口：experience.items.list")
def experience_items_list(self, token=None, itemName=None):
api = ExperienceItemsList(_host=self.host)
api.token = token
api.itemName = itemName
api.get_response()
return api.response
@allure.step("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
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):
......@@ -78,88 +87,49 @@ 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.step("新增表数据：hosts")
def insert_hosts(self, host_name=None, asset_code=None, manage_ip=None, parent_host=None, available=None, monitor_status=None):
session = self.db_session()
hostid = DataHosts().insert_one(session=session, host_name=host_name, asset_code=asset_code, manage_ip=manage_ip,
parent_host=parent_host, available=available, monitor_status=monitor_status)
return hostid
@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("调用接口：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.title("experience.items.list：token未传")
@allure.story("查询监控项下拉框：token未传")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_01(self):
self.get_base_token()
response = self.experience_items_list()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
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.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.experience_items_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.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.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.experience_items_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.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):
self.get_base_token()
epl_response = self.experience_items_list(token=self.token)
self.check_code(response=epl_response, code=0)
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):
self.get_base_token()
epl_response = self.experience_items_list(token=self.token, itemName="ping")
self.check_code(response=epl_response, code=0)
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)
if __name__ == "__main__":
......@@ -176,4 +146,5 @@ if __name__ == "__main__":
# a = TestExperienceItemsList()
# a.setup_class()
# a.setup_method()
# a.test_case_05()
WorkCase/APP/Experience/test_experience_problem_list.py
......@@ -8,9 +8,17 @@ from __future__ import division
import pytest
from WorkApi.API.Problem.problem_get import ProblemGet
from WorkApi.APP.Api.api_login import ApiLogin
from WorkCase.APP import AppBase
from WorkCase.API import ApiBase
from WorkData.Argus.alarm_rule import DataAlarmRule
from WorkData.Argus.businesses import DataBusinesses
from WorkData.Argus.experience import DataExperience
from WorkData.Argus.hosts import DataHosts
from WorkData.Argus.hosts_items import DataHostsItems
from WorkData.Argus.members import DataMembers
from WorkData.Argus.problem import DataProblem
from WorkData.Argus.tag import DataTag
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -39,184 +47,133 @@ class TestExperienceProblemList(object):
base_id = 10
base_password = "123456"
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_port = 161
base_title = "SS 自动化专用 经验手册 标题"
base_content = "SS 自动化专用 经验手册 内容"
base_description = "SS 自动化专用 经验手册 描述"
base_name = "SS 自动化专用 经验手册相关"
base_eventid = 91000000
session = UtilsDataBase().conn_mysql(db_url=db_url, db_port=db_port, db_base=db_base, db_user=db_user, db_pw=db_pw)
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
DataAlarmRule().delete_like_name(session=cls.session, name=cls.base_name)
DataBusinesses().delete_like_name(session=cls.session, name=cls.base_name)
sql = DataHosts().select_all_like_host_name(session=cls.session, host_name=cls.base_name)
for x, y in enumerate(sql):
DataHostsItems().delete_host_id(session=cls.session, host_id=y.hostid)
DataProblem().delete_like_name(session=cls.session, name=cls.base_name)
DataHosts().delete_like_host_name(session=cls.session, host_name=cls.base_name)
DataTag().delete_like_tag(session=cls.session, tag=cls.base_name)
DataExperience().delete_like_title(session=cls.session, title=cls.base_title)
@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("调用接口：experience.problem.list")
def experience_problem_list(self, token=None, problemName=None):
api = ExperienceProblemList(_host=self.host)
api.token = token
api.problemName = problemName
api.get_response()
return api.response
@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("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="eventid", possible=possible)
@allure.step("断言返回结果：校验返回数据的数量")
def check_num(self, response, num):
result = UtilsResponse().get_result(response=response)
assert len(result) == num
@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("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
sql = DataHostsItems().select_all_from_allKeys(session=session, host_id=host_id)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
@allure.step("创建测试数据")
def case_create(self, num):
self.name = self.base_name + num
self.eventid = self.base_eventid + int(num)
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
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)
session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
self.hostid = self.app.case_create_host(
name=self.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
)
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
for x, y in enumerate(sql):
itemids.append(y.id)
self.itemid = itemids[0]
session = self.db_session()
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)
@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("调用接口：problem.get")
def problem_get(self, eventids=None, groupids=None, hostids=None, objectids=None, applicationids=None,
source=None, _object=None, acknowledged=None, suppressed=None, severities=None, evaltype=None,
tags=None, recent=None, eventid_from=None, eventid_till=None, time_from=None,
time_till=None, selectAcknowledges=None, selectTags=None, selectSuppressionData=None,
countOutput=None, editable=None, excludeSearch=None,
_filter=None, limit=None, output=None, preservekeys=None, search=None, searchByAny=None,
searchWildcardsEnabled=None, sortfield=None, sortorder=None, startSearch=None):
api = ProblemGet(_host=self.base_host)
api.eventids = eventids
api.groupids = groupids
api.hostids = hostids
api.objectids = objectids
api.applicationids = applicationids
api.source = source
api.object = _object
api.acknowledged = acknowledged
api.suppressed = suppressed
api.severities = severities
api.evaltype = evaltype
api.tags = tags
api.recent = recent
api.eventid_from = eventid_from
api.eventid_till = eventid_till
api.time_from = time_from
api.time_till = time_till
api.selectAcknowledges = selectAcknowledges
api.selectTags = selectTags
api.selectSuppressionData = selectSuppressionData
api.countOutput = countOutput
api.editable = editable
api.excludeSearch = excludeSearch
api.filter = _filter
api.limit = limit
api.output = output
api.preservekeys = preservekeys
api.search = search
api.searchByAny = searchByAny
api.searchWildcardsEnabled = searchWildcardsEnabled
api.sortfield = sortfield
api.sortorder = sortorder
api.startSearch = startSearch
api.get_response()
return api.response
@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.title("experience.problem.list：token未传")
@allure.story("查询告警下拉框：token未传")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_01(self):
self.get_base_token()
response = self.experience_problem_list()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
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.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.experience_problem_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.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.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.experience_problem_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.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):
self.get_base_token()
epl_response = self.experience_problem_list(token=self.token, problemName="No")
self.check_code(response=epl_response, code=0)
epl_ids = []
for x, y in enumerate(UtilsResponse().get_data(response=epl_response)):
epl_ids.append(y["id"])
pg_response = self.problem_get(search={"name": ["*" + "No" + "*"]}, searchByAny=True, searchWildcardsEnabled=True,
output=["source", "object", "objectid", "name", "severity"])
self.check_code(response=pg_response, code=0)
pg_ids = []
for x, y in enumerate(UtilsResponse().get_result(response=pg_response)):
pg_ids.append(y["eventid"])
title = self.base_title + "04"
content = self.base_content + "04"
description = self.base_description + "04"
self.case_create(num="04")
self.app.case_create_experience(title=title, content=content, description=description, host_types=[self.base_type_agent])
# self.log.debug("校验与底层接口返回的数据是否一致")
# for x, y in enumerate(epl_ids):
# assert y in pg_ids
# for x, y in enumerate(pg_ids):
# assert y in epl_ids
response = self.app.experience_problem_list(token=self.token, problemName="SS")
self.app.check_code(response=response, code=0)
self.app.check_keyValue(_json=UtilsResponse().get_data(response=response), _key="name", _value="SS", isIn=1)
if __name__ == "__main__":
......@@ -233,4 +190,5 @@ if __name__ == "__main__":
# a = TestExperienceProblemList()
# a.setup_class()
# a.setup_method()
# a.test_case_04()
WorkCase/APP/Experience/test_experience_update.py
......@@ -9,6 +9,7 @@ from __future__ import division
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.members import DataMembers
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -63,15 +64,27 @@ class TestExperienceUpdate(object):
base_content = "SS测试经验内容"
base_description = "SS测试经验描述"
base_name = "经验手册-测试数据"
base_update_name = "经验手册-测试数据修改"
base_name = "SS 自动化专用 经验手册相关"
base_update_name = "SS 自动化专用 经验手册相关 修改"
base_eventid = 91000000
base_update_eventid = 91000010
session = UtilsDataBase().conn_mysql(db_url=db_url, db_port=db_port, db_base=db_base, db_user=db_user, db_pw=db_pw)
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataExperience().delete_like_title(session=session, title=cls.base_title)
DataAlarmRule().delete_like_name(session=cls.session, name=cls.base_name)
DataBusinesses().delete_like_name(session=cls.session, name=cls.base_name)
sql = DataHosts().select_all_like_host_name(session=cls.session, host_name=cls.base_name)
for x, y in enumerate(sql):
DataHostsItems().delete_host_id(session=cls.session, host_id=y.hostid)
DataProblem().delete_like_name(session=cls.session, name=cls.base_name)
DataHosts().delete_like_host_name(session=cls.session, host_name=cls.base_name)
DataTag().delete_like_tag(session=cls.session, tag=cls.base_name)
DataExperience().delete_like_title(session=cls.session, title=cls.base_title)
@classmethod
def teardown_class(cls):
......@@ -79,45 +92,17 @@ class TestExperienceUpdate(object):
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("调用接口：experience.update")
def experience_update(self, token=None, _id=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 = ExperienceUpdate(_host=self.host)
api.token = token
api.id = _id
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("断言返回结果")
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
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):
......@@ -184,154 +169,21 @@ class TestExperienceUpdate(object):
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 host_create(self, num):
name = self.base_name + num
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_snmp, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None, port=None, businessTree=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.port = port
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.businessTree = businessTree
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session()
sql = DataHostsItems().select_all_from_allKeys(session=session, host_id=host_id)
return sql
@allure.step("创建测试数据")
def case_create(self, num):
self.title = self.base_title + num
self.content = self.base_content + num
self.name = self.base_name + num
self.description = self.base_description + num
self.update_title = self.title + "修改"
self.update_content = self.content + "修改"
self.update_name = self.name + "修改"
self.update_description = self.description + "修改"
self.eventid = self.base_eventid + int(num)
self.update_eventid = self.eventid + 1
# 删除第一个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataAlarmRule().delete_like_name(session=session, name=self.base_name)
# 删除第二个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataBusinesses().delete_like_name(session=session, name=self.base_name)
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.name)
for x, y in enumerate(sql):
# 删除第三个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_like_name(session=session, name=self.base_name)
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
sql = DataHosts().select_all_from_allKeys(session=session, host_name=self.update_name)
for x, y in enumerate(sql):
# 删除第三个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHostsItems().delete_host_id(session=session, host_id=y.hostid)
# 删除第四个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataProblem().delete_like_name(session=session, name=self.base_name)
# 删除第五个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataHosts().delete_like_host_name(session=session, host_name=self.base_name)
# 删除第六个表
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
DataTag().delete_like_tag(session=session, tag=self.base_name)
session = self.db_session()
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)
session = self.db_session()
self.update_alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.name, status=1, severity=1,
manual_close=0, enable_time_default=1, create_time=1)
session = self.db_session()
self.businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
session = self.db_session()
self.update_businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
self.hostid = self.host_create(num=num)
self.hostid = self.app.case_create_host(
name=self.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
)
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
......@@ -343,25 +195,37 @@ class TestExperienceUpdate(object):
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.hostid, name=self.name)
self.update_hostid = self.host_create(num=num + "1")
@allure.step("创建测试数据")
def case_create_update(self, num):
self.update_name = self.base_name + num + "修改"
self.update_eventid = self.base_update_eventid + int(num)
sql = self.select_hosts_items(host_id=self.update_hostid)
update_itemid = []
session = self.db_session()
self.update_alarm_ruleid = DataAlarmRule().insert_one(session=session, name=self.name, status=1, severity=1,
manual_close=0, enable_time_default=1, create_time=1)
session = self.db_session()
self.update_businessid = DataBusinesses().insert_one(session=session, name=self.name, _type=1)
self.update_hostid = self.app.case_create_host(
name=self.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
)
sql = self.select_hosts_items(host_id=self.hostid)
itemids = []
for x, y in enumerate(sql):
update_itemid.append(y.id)
self.update_itemid = update_itemid[0]
itemids.append(y.id)
self.update_itemid = itemids[0]
session = self.db_session()
DataProblem().insert_one(session=session, eventid=self.update_eventid, source=0, _object=0, objectid=0, clock=0, ns=0,
r_clock=0, r_ns=0, acknowledged=0, acknowledged_time=0, severity=0,
status=0, hostid=self.update_hostid, name=self.update_name)
status=0, hostid=self.hostid, name=self.update_name)
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)
return self.experienceid
@allure.step("查询表：hosts_items")
def select_hosts_items(self, host_id=None):
session = self.db_session()
sql = DataHostsItems().select_all_from_allKeys(session=session, host_id=host_id)
return sql
@allure.step("校验数据")
def check_select(self, response):
......@@ -419,55 +283,49 @@ class TestExperienceUpdate(object):
for q, w in enumerate(sql):
assert w.tagid == tagids[0]
@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.update：token未传")
@allure.story("更新经验手册：token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.experience_update()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
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.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.experience_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.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.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.experience_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.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):
self.get_base_token()
title = self.base_title + "04"
content = self.base_content + "04"
description = self.base_description + "04"
self.update_title = title + "修改"
self.update_content = content + "修改"
self.update_description = description + "修改"
self.case_create(num="04")
response = self.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],
hostids=[self.update_hostid], itemids=[self.update_itemid], manageIpList=[self.update_hostid],
alarmRuleids=[self.update_alarm_ruleid], tagNameList=[self.base_update_name], problemidList=[self.update_eventid])
self.check_code(response=response, code=0)
self.case_create_update(num="041")
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],
hostids=[self.update_hostid], itemids=[self.update_itemid], manageIpList=[self.update_hostid],
alarmRuleids=[self.update_alarm_ruleid], tagNameList=[self.base_update_name],
problemidList=[self.update_eventid])
self.app.check_code(response=response, code=0)
self.check_select(response=response)
......
WorkCase/APP/__init__.py
# -*- coding: utf-8 -*-
# Argus模块公共封装类
# 作者: 陈磊
# 时间: 2019-12-17
from WorkApi.APP.AddressBook.addressbook_create import AddressBookCreate
from WorkApi.APP.AddressBook.addressbook_delete import AddressBookDelete
from WorkApi.APP.AddressBook.addressbook_disabled import AddressBookDisabled
......@@ -31,6 +36,8 @@ from WorkApi.APP.Experience.experience_get import ExperienceGet
from WorkApi.APP.Experience.experience_update import ExperienceUpdate
from WorkApi.APP.File.file_upload import FileUpload
from WorkApi.APP.HostMonitor.hostMonitor_get import HostMonitorGet
from WorkApi.APP.HostMonitor.hostMonitor_get_hostid import HostMonitorGetHostid
from WorkApi.APP.HostMonitor.hostMonitor_thread_hostid import HostMonitorThreadHostid
from WorkApi.APP.HostType.hostType_create import HostTypeCreate
from WorkApi.APP.HostType.hostType_delete import HostTypeDelete
from WorkApi.APP.HostType.hostType_get import HostTypeGet
......@@ -342,24 +349,74 @@ class AppBase(object):
return api.response
@allure.step("创建测试数据-资产")
def case_create_host(self, name, host_type, iplist, port, businessIds, monitorStatus=None):
if monitorStatus is not None:
def case_create_host(self, name, host_type, port, businessIds, iplist=None, dns=None, parentHost=None, monitorType=None, monitorStatus=None):
if monitorStatus is not None and monitorType is not None and parentHost is not None:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, dns=dns, monitorInterface=port,
monitorType=monitorType, manageLevel=1, parentHost=parentHost, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=monitorStatus)
elif monitorStatus is not None and monitorType is not None and parentHost is None:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, dns=dns, monitorInterface=port,
monitorType=monitorType, manageLevel=1, parentHost=1, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=monitorStatus)
elif monitorStatus is not None and monitorType is None and parentHost is not None:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, dns=dns, monitorInterface=port,
monitorType=1, manageLevel=1, parentHost=parentHost, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=monitorStatus)
elif monitorStatus is None and monitorType is not None and parentHost is not None:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, dns=dns, monitorInterface=port,
monitorType=monitorType, manageLevel=1, parentHost=parentHost, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
elif monitorStatus is not None and monitorType is None and parentHost is None:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, monitorInterface=port,
token=self.token, hostName=name, hostType=host_type, iplist=iplist, dns=dns, monitorInterface=port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=monitorStatus)
elif monitorStatus is None and monitorType is not None and parentHost is None:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, dns=dns, monitorInterface=port,
monitorType=monitorType, manageLevel=1, parentHost=1, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
elif monitorStatus is None and monitorType is None and parentHost is not None:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, dns=dns, monitorInterface=port,
monitorType=1, manageLevel=1, parentHost=parentHost, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
else:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, monitorInterface=port,
token=self.token, hostName=name, hostType=host_type, iplist=iplist, dns=dns, monitorInterface=port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=monitorStatus)
monitorStatus=1)
self.check_code(response=response, code=0)
host_id = UtilsResponse().get_data(response=response)
return host_id
......@@ -407,6 +464,7 @@ class AppBase(object):
@allure.step("校验数据")
def check_keyValue(self, _json, _key, _value, isIn=None):
assert _json != []
for x, y in enumerate(_json):
if isIn == 1:
assert _value in y[_key]
......@@ -827,3 +885,19 @@ class AppBase(object):
api.size = size
api.get_response()
return api.response
@allure.step("调用接口：hostmonitor.get.hostid")
def hostmonitor_get_hostid(self, token=None, hostid=None):
api = HostMonitorGetHostid(_host=self.host)
api.token = token
api.hostid = hostid
api.get_response()
return api.response
@allure.step("调用接口：hostmonitor.thread.hostid")
def hostmonitor_thread_hostid(self, token=None, hostid=None):
api = HostMonitorThreadHostid(_host=self.host)
api.token = token
api.hostid = hostid
api.get_response()
return api.response
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