Skip to content

Z
zmops-test
Commit 4ff684f3 by sanshi
Options

删除一部分用例

parent 5339f1de
Showing with 15 additions and 2979 deletions
WorkCase/API/HostGroup/test_hostgroup_massadd.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-15
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostGroup.hostGroup_create import HostGroupCreate
from WorkApi.API.HostGroup.hostGroup_massadd import HostGroupMassadd
from WorkData.Zabbix.hstgrp import DataHstGrp
from WorkData.Zabbix.hosts_groups import DataHostsGroups
import allure
import pytest
@allure.feature("测试模块：hostgroup.massadd")
class TestHostGroupMassadd(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS自动化测试主机组"
base_hostids = [10266, 10269, 10272]
@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)
DataHstGrp().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostgroup.create")
def hostgroup_create(self, name=None, groupid=None, flags=None, internal=None):
api = HostGroupCreate(_host=self.host)
api.name = name
api.groupid = groupid
api.flags = flags
api.internal = internal
api.get_response()
return api.response
@allure.step("调用接口：hostgroup.massadd")
def hostgroup_massadd(self, groups=None, hosts=None, templates=None):
api = HostGroupMassadd(_host=self.host)
api.groups = groups
api.hosts = hosts
api.templates = templates
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self, name):
response = self.hostgroup_create(name=name)
self.check_code(response=response, code=0)
groupid = UtilsResponse().get_result(response=response)["groupids"][0]
return groupid
@allure.step("连接数据库：zabbix")
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("查询表：hstgrp")
def select_hstgrp(self, groupid=None, name=None, internal=None, flags=None):
session = self.db_session()
sql = DataHstGrp().select_all_from_allKeys(session=session, groupid=groupid, name=name, internal=internal, flags=flags)
return sql
@allure.step("查询表：hosts_groups")
def select_hosts_groups(self, groupid=None, hostid=None):
session = self.db_session()
sql = DataHostsGroups().select_all_from_allKeys(session=session, groupid=groupid, hostid=hostid)
return sql
@allure.step("校验查询结果")
def check_select(self, groupids, hostids):
for x, y in enumerate(groupids):
for q, w in enumerate(hostids):
sql = self.select_hosts_groups(groupid=y, hostid=w)
assert sql != []
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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.title("hostgroup.massadd：一个主机组 批量添加 一个主机")
@allure.story("批量添加主机组：一个主机组 批量添加 一个主机")
@allure.severity("blocker")
def test_case_01(self):
name = self.base_name + "01"
groupid = self.case_create(name=name)
response = self.hostgroup_massadd(groups=[groupid], hosts=[self.base_hostids[0]])
self.check_code(response=response, code=0)
self.check_select(hostids=[self.base_hostids[0]], groupids=[groupid])
@allure.title("hostgroup.massadd：多个主机组 批量添加 一个主机")
@allure.story("批量添加主机组：多个主机组 批量添加 一个主机")
def test_case_02(self):
name_1 = self.base_name + "021"
name_2 = self.base_name + "022"
groupid_1 = self.case_create(name=name_1)
groupid_2 = self.case_create(name=name_2)
response = self.hostgroup_massadd(groups=[groupid_1, groupid_2], hosts=[self.base_hostids[0]])
self.check_code(response=response, code=0)
self.check_select(hostids=[self.base_hostids[0]], groupids=[groupid_1, groupid_2])
@allure.title("hostgroup.massadd：多个主机组 批量添加 多个主机")
@allure.story("批量添加主机组：多个主机组 批量添加 多个主机")
def test_case_03(self):
name_1 = self.base_name + "031"
name_2 = self.base_name + "032"
name_3 = self.base_name + "033"
groupid_1 = self.case_create(name=name_1)
groupid_2 = self.case_create(name=name_2)
groupid_3 = self.case_create(name=name_3)
response = self.hostgroup_massadd(groups=[groupid_1, groupid_2, groupid_3], hosts=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]])
self.check_code(response=response, code=0)
self.check_select(hostids=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]], groupids=[groupid_1, groupid_2, groupid_3])
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestHostGroupMassadd()
# a.setup_class()
# a.test_case_03()
WorkCase/API/HostGroup/test_hostgroup_massremove.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-15
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostGroup.hostGroup_create import HostGroupCreate
from WorkApi.API.HostGroup.hostGroup_massadd import HostGroupMassadd
from WorkApi.API.HostGroup.hostGroup_massremove import HostGroupMassremove
from WorkData.Zabbix.hstgrp import DataHstGrp
from WorkData.Zabbix.hosts_groups import DataHostsGroups
import allure
import pytest
@allure.feature("测试模块：hostgroup.massremove")
class TestHostGroupMassremove(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS自动化测试主机组"
base_hostids = [10266, 10272, 10296]
@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)
DataHstGrp().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostgroup.create")
def hostgroup_create(self, name=None, groupid=None, flags=None, internal=None):
api = HostGroupCreate(_host=self.host)
api.name = name
api.groupid = groupid
api.flags = flags
api.internal = internal
api.get_response()
return api.response
@allure.step("调用接口：hostgroup.massadd")
def hostgroup_massadd(self, groups=None, hosts=None, templates=None):
api = HostGroupMassadd(_host=self.host)
api.groups = groups
api.hosts = hosts
api.templates = templates
api.get_response()
return api.response
@allure.step("调用接口：hostgroup.massremove")
def hostgroup_massremove(self, groups=None, hosts=None, templates=None):
api = HostGroupMassremove(_host=self.host)
api.groups = groups
api.hosts = hosts
api.templates = templates
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self, name):
response = self.hostgroup_create(name=name)
self.check_code(response=response, code=0)
groupid = UtilsResponse().get_result(response=response)["groupids"][0]
return groupid
@allure.step("连接数据库：zabbix")
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("查询表：hstgrp")
def select_hstgrp(self, groupid=None, name=None, internal=None, flags=None):
session = self.db_session()
sql = DataHstGrp().select_all_from_allKeys(session=session, groupid=groupid, name=name, internal=internal, flags=flags)
return sql
@allure.step("查询表：hosts_groups")
def select_hosts_groups(self, groupid=None, hostid=None):
session = self.db_session()
sql = DataHostsGroups().select_all_from_allKeys(session=session, groupid=groupid, hostid=hostid)
return sql
@allure.step("校验查询结果")
def check_select(self, groupids, hostids):
for x, y in enumerate(groupids):
for q, w in enumerate(hostids):
sql = self.select_hosts_groups(groupid=y, hostid=w)
assert sql != []
@allure.step("校验查询结果")
def check_select_delete(self, groupids, hostids):
for x, y in enumerate(groupids):
for q, w in enumerate(hostids):
sql = self.select_hosts_groups(groupid=y, hostid=w)
assert sql == []
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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.title("hostgroup.massremove：一个主机组 批量删除 一个主机")
@allure.story("批量删除主机组：一个主机组 批量删除 一个主机")
@allure.severity("blocker")
def test_case_01(self):
name = self.base_name + "01"
groupid = self.case_create(name=name)
self.hostgroup_massadd(groups=[groupid], hosts=[self.base_hostids[0]])
response = self.hostgroup_massremove(groups=[groupid], hosts=[self.base_hostids[0]])
self.check_code(response=response, code=0)
self.check_select_delete(hostids=[self.base_hostids[0]], groupids=[groupid])
@allure.title("hostgroup.massremove：多个主机组 批量删除 一个主机")
@allure.story("批量删除主机组：多个主机组 批量删除 一个主机")
def test_case_02(self):
name_1 = self.base_name + "021"
name_2 = self.base_name + "022"
groupid_1 = self.case_create(name=name_1)
groupid_2 = self.case_create(name=name_2)
self.hostgroup_massadd(groups=[groupid_1, groupid_2], hosts=[self.base_hostids[0]])
response = self.hostgroup_massremove(groups=[groupid_1, groupid_2], hosts=[self.base_hostids[0]])
self.check_code(response=response, code=0)
self.check_select_delete(hostids=[self.base_hostids[0]], groupids=[groupid_1, groupid_2])
@allure.title("hostgroup.massremove：多个主机组 批量删除 多个主机")
@allure.story("批量删除主机组：多个主机组 批量删除 多个主机")
def test_case_03(self):
name_1 = self.base_name + "031"
name_2 = self.base_name + "032"
name_3 = self.base_name + "033"
groupid_1 = self.case_create(name=name_1)
groupid_2 = self.case_create(name=name_2)
groupid_3 = self.case_create(name=name_3)
self.hostgroup_massadd(groups=[groupid_1, groupid_2, groupid_3], hosts=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]])
response = self.hostgroup_massremove(groups=[groupid_1, groupid_2, groupid_3], hosts=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]])
self.check_code(response=response, code=0)
self.check_select_delete(hostids=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]], groupids=[groupid_1, groupid_2, groupid_3])
@allure.title("hostgroup.massremove：多个主机组 批量删除 部分多个主机")
@allure.story("批量删除主机组：多个主机组 批量删除 部分多个主机")
def test_case_04(self):
name_1 = self.base_name + "041"
name_2 = self.base_name + "042"
name_3 = self.base_name + "043"
groupid_1 = self.case_create(name=name_1)
groupid_2 = self.case_create(name=name_2)
groupid_3 = self.case_create(name=name_3)
self.hostgroup_massadd(groups=[groupid_1, groupid_2, groupid_3], hosts=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]])
response = self.hostgroup_massremove(groups=[groupid_1, groupid_3], hosts=[self.base_hostids[0], self.base_hostids[1]])
self.check_code(response=response, code=0)
self.check_select_delete(hostids=[self.base_hostids[0], self.base_hostids[1]], groupids=[groupid_1, groupid_3])
self.check_select(hostids=[self.base_hostids[2]], groupids=[groupid_1, groupid_2, groupid_3])
self.check_select(hostids=[self.base_hostids[0], self.base_hostids[1]], groupids=[groupid_2])
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestHostGroupMassremove()
# a.setup_class()
# a.test_case_04()
WorkCase/API/HostGroup/test_hostgroup_massupdate.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-15
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostGroup.hostGroup_create import HostGroupCreate
from WorkApi.API.HostGroup.hostGroup_massadd import HostGroupMassadd
from WorkApi.API.HostGroup.hostGroup_massupdate import HostGroupMassupdate
from WorkData.Zabbix.hstgrp import DataHstGrp
from WorkData.Zabbix.hosts_groups import DataHostsGroups
import allure
import pytest
@allure.feature("测试模块：hostgroup.massupdate")
class TestHostGroupMassupdate(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS自动化测试主机组"
base_hostids = [10266, 10272, 10296]
@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)
DataHstGrp().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostgroup.create")
def hostgroup_create(self, name=None, groupid=None, flags=None, internal=None):
api = HostGroupCreate(_host=self.host)
api.name = name
api.groupid = groupid
api.flags = flags
api.internal = internal
api.get_response()
return api.response
@allure.step("调用接口：hostgroup.massadd")
def hostgroup_massadd(self, groups=None, hosts=None, templates=None):
api = HostGroupMassadd(_host=self.host)
api.groups = groups
api.hosts = hosts
api.templates = templates
api.get_response()
return api.response
@allure.step("调用接口：hostgroup.massupdate")
def hostgroup_massupdate(self, groups=None, hosts=None, templates=None):
api = HostGroupMassupdate(_host=self.host)
api.groups = groups
api.hosts = hosts
api.templates = templates
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self, name):
response = self.hostgroup_create(name=name)
self.check_code(response=response, code=0)
groupid = UtilsResponse().get_result(response=response)["groupids"][0]
return groupid
@allure.step("连接数据库：zabbix")
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("查询表：hstgrp")
def select_hstgrp(self, groupid=None, name=None, internal=None, flags=None):
session = self.db_session()
sql = DataHstGrp().select_all_from_allKeys(session=session, groupid=groupid, name=name, internal=internal, flags=flags)
return sql
@allure.step("查询表：hosts_groups")
def select_hosts_groups(self, groupid=None, hostid=None):
session = self.db_session()
sql = DataHostsGroups().select_all_from_allKeys(session=session, groupid=groupid, hostid=hostid)
return sql
@allure.step("校验查询结果")
def check_select(self, groupids, hostids):
for x, y in enumerate(groupids):
for q, w in enumerate(hostids):
sql = self.select_hosts_groups(groupid=y, hostid=w)
assert sql != []
@allure.step("校验查询结果")
def check_select_delete(self, groupids, hostids):
import time
time.sleep(1)
for x, y in enumerate(groupids):
for q, w in enumerate(hostids):
sql = self.select_hosts_groups(groupid=y, hostid=w)
assert sql == []
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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.title("hostgroup.massupdate：一个主机组 批量更新 一个主机")
@allure.story("批量更新主机组：一个主机组 批量更新 一个主机")
@allure.severity("blocker")
def test_case_01(self):
name = self.base_name + "01"
groupid = self.case_create(name=name)
# self.hostgroup_massadd(groups=[groupid], hosts=[self.base_hostids[0]])
response = self.hostgroup_massupdate(groups=[groupid], hosts=[self.base_hostids[0]])
self.check_code(response=response, code=0)
self.check_select(hostids=[self.base_hostids[0]], groupids=[groupid])
@allure.title("hostgroup.massupdate：多个主机组 批量更新 一个主机")
@allure.story("批量更新主机组：多个主机组 批量更新 一个主机")
def test_case_02(self):
name_1 = self.base_name + "021"
name_2 = self.base_name + "022"
groupid_1 = self.case_create(name=name_1)
groupid_2 = self.case_create(name=name_2)
response = self.hostgroup_massupdate(groups=[groupid_1, groupid_2], hosts=[self.base_hostids[0]])
self.check_code(response=response, code=0)
self.check_select(hostids=[self.base_hostids[0]], groupids=[groupid_1, groupid_2])
@allure.title("hostgroup.massupdate：多个主机组 批量更新 多个主机")
@allure.story("批量更新主机组：多个主机组 批量更新 多个主机")
def test_case_03(self):
name_1 = self.base_name + "031"
name_2 = self.base_name + "032"
name_3 = self.base_name + "033"
groupid_1 = self.case_create(name=name_1)
groupid_2 = self.case_create(name=name_2)
groupid_3 = self.case_create(name=name_3)
response = self.hostgroup_massupdate(groups=[groupid_1, groupid_2, groupid_3], hosts=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]])
self.check_code(response=response, code=0)
self.check_select(hostids=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]], groupids=[groupid_1, groupid_2, groupid_3])
@allure.title("hostgroup.massupdate：多个主机组 批量更新 部分多个主机")
@allure.story("批量更新主机组：多个主机组 批量更新 部分多个主机")
def test_case_04(self):
name_1 = self.base_name + "041"
name_2 = self.base_name + "042"
name_3 = self.base_name + "043"
groupid_1 = self.case_create(name=name_1)
groupid_2 = self.case_create(name=name_2)
groupid_3 = self.case_create(name=name_3)
self.hostgroup_massadd(groups=[groupid_1, groupid_2, groupid_3], hosts=[self.base_hostids[0], self.base_hostids[1], self.base_hostids[2]])
response = self.hostgroup_massupdate(groups=[groupid_1, groupid_3], hosts=[self.base_hostids[0], self.base_hostids[1]])
self.check_code(response=response, code=0)
self.check_select(hostids=[self.base_hostids[0], self.base_hostids[1]], groupids=[groupid_1, groupid_3])
self.check_select(hostids=[self.base_hostids[2]], groupids=[groupid_1, groupid_2, groupid_3])
self.check_select(hostids=[self.base_hostids[0], self.base_hostids[1]], groupids=[groupid_2])
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestHostGroupMassupdate()
# a.setup_class()
# a.test_case_04()
WorkCase/API/HostInterface/test_hostinterface_massadd.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-12
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostInterface.hostInterface_create import HostInterfaceCreate
from WorkApi.API.HostInterface.hostInterface_massadd import HostInterfaceMassadd
from WorkData.Zabbix.interface import DataInterface
import allure
import pytest
@allure.feature("测试模块：hostinterface.massadd")
class TestHostInterfaceMassadd(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机接口"
base_ip = "10.0.0.28"
base_hostids = [10266, 10286, 10507]
base_dns = "sanshi.zmops.cn"
base_port = "127"
new_ip = "192.168.1.28"
new_port = "721"
@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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[0])
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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[1])
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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[2])
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostinterface.create")
def hostinterface_create(self, dns=None, hostid=None, ip=None, main=None, port=None, _type=None, useip=None, bulk=None):
api = HostInterfaceCreate(_host=self.host)
api.dns = dns
api.hostid = hostid
api.ip = ip
api.main = main
api.port = port
api.type = _type
api.useip = useip
api.bulk = bulk
api.get_response()
return api.response
@allure.step("调用接口：hostinterface.massadd")
def hostinterface_massadd(self, hosts=None, dns=None, ip=None, main=None, port=None, _type=None, useip=None, bulk=None):
api = HostInterfaceMassadd(_host=self.host)
api.hosts = hosts
api.interfaces_dns = dns
api.interfaces_ip = ip
api.interfaces_main = main
api.interfaces_port = port
api.interfaces_type = _type
api.interfaces_useip = useip
api.interfaces_bulk = bulk
api.get_response()
return api.response
@allure.step("连接数据库：zabbix")
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 case_create(self, ip, hostid, main, port, _type, useip, bulk, dns):
response = self.hostinterface_create(ip=ip, hostid=hostid, main=main, port=port, _type=_type, useip=useip, bulk=bulk, dns=dns)
self.check_code(response=response, code=0)
interfaceid = UtilsResponse().get_result(response=response)["interfaceids"][0]
return interfaceid
@allure.step("查询表：interface")
def select_interface(self, interfaceid=None, hostid=None):
session = self.db_session()
sql = DataInterface().select_all_from_allKeys(session=session, interfaceid=interfaceid, hostid=hostid)
return sql
@allure.step("公共测试方法")
def case_main(self, hosts, ips, dnss, mains, ports, _types, useips, bulks):
response = self.hostinterface_massadd(hosts=hosts, ip=ips, dns=dnss, main=mains,
port=ports, _type=_types, useip=useips, bulk=bulks)
self.check_code(response=response, code=0)
interfaceids = UtilsResponse().get_result(response=response)["interfaceids"]
i = 0
for q, w in enumerate(hosts):
for x, y in enumerate(ips):
self.check_select(interfaceid=interfaceids[i], hostid=hosts[q], ip=ips[x], dns=dnss[x], main=mains[x], port=ports[x],
_type=_types[x], useip=useips[x], bulk=bulks[x])
i = i + 1
@allure.step("校验查询结果")
def check_select(self, interfaceid, hostid, main, port, _type, useip, bulk, dns, ip):
sql = self.select_interface(hostid=hostid, interfaceid=interfaceid)
for x, y in enumerate(sql):
assert y.main == main
assert y.port == port
assert y.type == _type
assert y.useip == useip
assert y.bulk == bulk
if dns != "" and ip == "":
assert y.dns == dns
elif ip != "" and dns == "":
assert y.ip == ip
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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("断言返回错误结果")
def check_status(self, response, status):
_status = UtilsResponse().get_status(response=response)
assert _status == status
@allure.title("hostinterface.massadd：一个主机批量创建多个主机接口")
@allure.story("批量创建主机接口：一个主机批量创建多个主机接口")
@allure.severity("blocker")
def test_case_01(self):
self.case_main(hosts=[self.base_hostids[0]], ips=[self.base_ip, self.base_ip], dnss=["", ""], mains=[1, 0], ports=["110", "111"], _types=[1, 1],
useips=[1, 1], bulks=[1, 1])
@allure.title("hostinterface.massadd：多个主机批量创建多个主机接口")
@allure.story("批量创建主机接口：多个主机批量创建多个主机接口")
def test_case_02(self):
self.case_main(hosts=[self.base_hostids[1], self.base_hostids[2]], ips=[self.base_ip, self.base_ip], dnss=["", ""], mains=[1, 0], ports=["110", "111"], _types=[1, 1],
useips=[1, 1], bulks=[1, 1])
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
# case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case)
# r.run_main()
a = TestHostInterfaceMassadd()
a.setup_class()
a.test_case_02()
WorkCase/API/HostInterface/test_hostinterface_massremove.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-14
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostInterface.hostInterface_create import HostInterfaceCreate
from WorkApi.API.HostInterface.hostInterface_massadd import HostInterfaceMassadd
from WorkApi.API.HostInterface.hostInterface_massremove import HostInterfaceMassremove
from WorkData.Zabbix.interface import DataInterface
import allure
import pytest
@allure.feature("测试模块：hostinterface.massremove")
class TestHostInterfaceMassremove(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机接口"
base_ip = "10.0.0.28"
base_hostids = [10266, 10286, 10507]
base_dns = "sanshi.zmops.cn"
base_port = "127"
new_ip = "192.168.1.28"
new_port = "721"
@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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[0])
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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[1])
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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[2])
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostinterface.create")
def hostinterface_create(self, dns=None, hostid=None, ip=None, main=None, port=None, _type=None, useip=None, bulk=None):
api = HostInterfaceCreate(_host=self.host)
api.dns = dns
api.hostid = hostid
api.ip = ip
api.main = main
api.port = port
api.type = _type
api.useip = useip
api.bulk = bulk
api.get_response()
return api.response
@allure.step("调用接口：hostinterface.massadd")
def hostinterface_massadd(self, hosts=None, dns=None, ip=None, main=None, port=None, _type=None, useip=None, bulk=None):
api = HostInterfaceMassadd(_host=self.host)
api.hosts = hosts
api.interfaces_dns = dns
api.interfaces_ip = ip
api.interfaces_main = main
api.interfaces_port = port
api.interfaces_type = _type
api.interfaces_useip = useip
api.interfaces_bulk = bulk
api.get_response()
return api.response
@allure.step("调用接口：hostinterface.massremove")
def hostinterface_massremove(self, hostids=None, dns=None, ip=None, port=None, _type=None):
api = HostInterfaceMassremove(_host=self.host)
api.hostids = hostids
api.interfaces_dns = dns
api.interfaces_ip = ip
api.interfaces_port = port
api.get_response()
return api.response
@allure.step("连接数据库：zabbix")
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 case_create(self, ip, hostid, main, port, _type, useip, bulk, dns):
response = self.hostinterface_create(ip=ip, hostid=hostid, main=main, port=port, _type=_type, useip=useip, bulk=bulk, dns=dns)
self.check_code(response=response, code=0)
interfaceid = UtilsResponse().get_result(response=response)["interfaceids"][0]
return interfaceid
@allure.step("查询表：interface")
def select_interface(self, interfaceid=None, hostid=None):
session = self.db_session()
sql = DataInterface().select_all_from_allKeys(session=session, interfaceid=interfaceid, hostid=hostid)
return sql
@allure.step("公共测试方法")
def case_main(self, hosts, ips, dnss, mains, ports, _types, useips, bulks):
response = self.hostinterface_massadd(hosts=hosts, ip=ips, dns=dnss, main=mains,
port=ports, _type=_types, useip=useips, bulk=bulks)
self.check_code(response=response, code=0)
interfaceids = UtilsResponse().get_result(response=response)["interfaceids"]
i = 0
for q, w in enumerate(hosts):
for x, y in enumerate(ips):
self.check_select(interfaceid=interfaceids[i], hostid=hosts[q], ip=ips[x], dns=dnss[x], main=mains[x], port=ports[x],
_type=_types[x], useip=useips[x], bulk=bulks[x])
i = i + 1
return interfaceids
@allure.step("校验查询结果")
def check_select(self, interfaceid, hostid, main, port, _type, useip, bulk, dns, ip):
sql = self.select_interface(hostid=hostid, interfaceid=interfaceid)
for x, y in enumerate(sql):
assert y.main == main
assert y.port == port
assert y.type == _type
assert y.useip == useip
assert y.bulk == bulk
if dns != "" and ip == "":
assert y.dns == dns
elif ip != "" and dns == "":
assert y.ip == ip
@allure.step("校验查询结果")
def check_delete(self, interfaceids):
for x, y in enumerate(interfaceids):
sql = self.select_interface(interfaceid=y)
assert sql == []
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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("断言返回错误结果")
def check_status(self, response, status):
_status = UtilsResponse().get_status(response=response)
assert _status == status
@allure.title("hostinterface.massadd：一个主机批量创建多个主机接口")
@allure.story("批量创建主机接口：一个主机批量创建多个主机接口")
@allure.severity("blocker")
def test_case_01(self):
interfaceids = self.case_main(hosts=[self.base_hostids[0]], ips=[self.base_ip, self.base_ip, self.base_ip], dnss=["", "", ""],
mains=[1, 0, 0], ports=["110", "111", "112"], _types=[1, 1, 1], useips=[1, 1, 1], bulks=[1, 1, 1])
self.hostinterface_massremove(hostids=[self.base_hostids[0]], ip=[self.base_ip, self.base_ip], dns=["", ""], port=["111", "112"])
self.hostinterface_massremove(hostids=[self.base_hostids[0]], ip=[self.base_ip], dns=[""], port=["110"])
self.check_delete(interfaceids=interfaceids)
@allure.title("hostinterface.massadd：多个主机批量创建多个主机接口")
@allure.story("批量创建主机接口：多个主机批量创建多个主机接口")
def test_case_02(self):
interfaceids = self.case_main(hosts=[self.base_hostids[1], self.base_hostids[2]], ips=[self.base_ip, self.base_ip], dnss=["", ""],
mains=[1, 0], ports=["110", "111"], _types=[1, 1], useips=[1, 1], bulks=[1, 1])
# 先删除所有非默认的接口
self.hostinterface_massremove(hostids=[self.base_hostids[1], self.base_hostids[2]], ip=[self.base_ip], dns=[""], port=["111"])
# 才能删除所有默认的接口
self.hostinterface_massremove(hostids=[self.base_hostids[1], self.base_hostids[2]], ip=[self.base_ip], dns=[""], port=["110"])
self.check_delete(interfaceids=interfaceids)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
# case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case, case_info=case_info)
# r.run_main()
a = TestHostInterfaceMassremove()
a.setup_class()
a.test_case_01()
WorkCase/API/HostInterface/test_hostinterface_replace.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-14
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostInterface.hostInterface_create import HostInterfaceCreate
from WorkApi.API.HostInterface.hostInterface_massadd import HostInterfaceMassadd
from WorkApi.API.HostInterface.hostInterface_replace import HostInterfaceReplace
from WorkData.Zabbix.interface import DataInterface
import allure
import pytest
@allure.feature("测试模块：hostinterface.replace")
class TestHostInterfaceReplace(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机接口"
base_ip = "10.0.0.28"
base_hostids = [10266, 10286, 10507]
base_dns = "sanshi.zmops.cn"
base_port = "127"
new_ip = "192.168.1.28"
new_port = "721"
@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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[0])
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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[1])
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)
DataInterface().delete_hostid(session=session, hostid=cls.base_hostids[2])
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostinterface.create")
def hostinterface_create(self, dns=None, hostid=None, ip=None, main=None, port=None, _type=None, useip=None, bulk=None):
api = HostInterfaceCreate(_host=self.host)
api.dns = dns
api.hostid = hostid
api.ip = ip
api.main = main
api.port = port
api.type = _type
api.useip = useip
api.bulk = bulk
api.get_response()
return api.response
@allure.step("调用接口：hostinterface.massadd")
def hostinterface_massadd(self, hosts=None, dns=None, ip=None, main=None, port=None, _type=None, useip=None, bulk=None):
api = HostInterfaceMassadd(_host=self.host)
api.hosts = hosts
api.interfaces_dns = dns
api.interfaces_ip = ip
api.interfaces_main = main
api.interfaces_port = port
api.interfaces_type = _type
api.interfaces_useip = useip
api.interfaces_bulk = bulk
api.get_response()
return api.response
@allure.step("调用接口：hostinterface.replace")
def hostinterface_replace(self, hostids=None, dns=None, ip=None, port=None, _type=None):
api = HostInterfaceReplace(_host=self.host)
api.hostids = hostids
api.interfaces_dns = dns
api.interfaces_ip = ip
api.interfaces_port = port
api.get_response()
return api.response
@allure.step("连接数据库：zabbix")
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 case_create(self, ip, hostid, main, port, _type, useip, bulk, dns):
response = self.hostinterface_create(ip=ip, hostid=hostid, main=main, port=port, _type=_type, useip=useip, bulk=bulk, dns=dns)
self.check_code(response=response, code=0)
interfaceid = UtilsResponse().get_result(response=response)["interfaceids"][0]
return interfaceid
@allure.step("查询表：interface")
def select_interface(self, interfaceid=None, hostid=None):
session = self.db_session()
sql = DataInterface().select_all_from_allKeys(session=session, interfaceid=interfaceid, hostid=hostid)
return sql
@allure.step("公共测试方法")
def case_main(self, hosts, ips, dnss, mains, ports, _types, useips, bulks):
response = self.hostinterface_massadd(hosts=hosts, ip=ips, dns=dnss, main=mains,
port=ports, _type=_types, useip=useips, bulk=bulks)
self.check_code(response=response, code=0)
interfaceids = UtilsResponse().get_result(response=response)["interfaceids"]
i = 0
for q, w in enumerate(hosts):
for x, y in enumerate(ips):
self.check_select(interfaceid=interfaceids[i], hostid=hosts[q], ip=ips[x], dns=dnss[x], main=mains[x], port=ports[x],
_type=_types[x], useip=useips[x], bulk=bulks[x])
i = i + 1
return interfaceids
@allure.step("校验查询结果")
def check_select(self, interfaceid, hostid, main, port, _type, useip, bulk, dns, ip):
sql = self.select_interface(hostid=hostid, interfaceid=interfaceid)
for x, y in enumerate(sql):
assert y.main == main
assert y.port == port
assert y.type == _type
assert y.useip == useip
assert y.bulk == bulk
if dns != "" and ip == "":
assert y.dns == dns
elif ip != "" and dns == "":
assert y.ip == ip
@allure.step("校验查询结果")
def check_delete(self, interfaceids):
for x, y in enumerate(interfaceids):
sql = self.select_interface(interfaceid=y)
assert sql == []
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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("断言返回错误结果")
def check_status(self, response, status):
_status = UtilsResponse().get_status(response=response)
assert _status == status
@allure.title("hostinterface.massadd：一个主机批量创建多个主机接口")
@allure.story("批量创建主机接口：一个主机批量创建多个主机接口")
@allure.severity("blocker")
def test_case_01(self):
interfaceids = self.case_main(hosts=[self.base_hostids[0]], ips=[self.base_ip, self.base_ip], dnss=["", ""], mains=[1, 0],
ports=["110", "111"], _types=[1, 1], useips=[1, 1], bulks=[1, 1])
self.hostinterface_replace(hostids=[self.base_hostids[0]], ip=[self.base_ip], dns=[""], port=["112"])
# self.check_delete(interfaceids=interfaceids)
@allure.title("hostinterface.massadd：多个主机批量创建多个主机接口")
@allure.story("批量创建主机接口：多个主机批量创建多个主机接口")
def test_case_02(self):
interfaceids = self.case_main(hosts=[self.base_hostids[1], self.base_hostids[2]], ips=[self.base_ip, self.base_ip], dnss=["", ""],
mains=[1, 0], ports=["110", "111"], _types=[1, 1], useips=[1, 1], bulks=[1, 1])
# 先删除所有非默认的接口
self.hostinterface_replace(hostids=[self.base_hostids[1], self.base_hostids[2]], ip=[self.base_ip], dns=[""], port=["111"])
# 才能删除所有默认的接口
self.hostinterface_replace(hostids=[self.base_hostids[1], self.base_hostids[2]], ip=[self.base_ip], dns=[""], port=["110"])
self.check_delete(interfaceids=interfaceids)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
# case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case, case_info=case_info)
# r.run_main()
a = TestHostInterfaceReplace()
a.setup_class()
a.test_case_01()
WorkCase/API/HostPrototype/test_hostprototype_create.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-11
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostPrototype.hostPrototype_create import HostPrototypeCreate
from WorkData.Zabbix.hosts import DataHosts
from WorkData.Zabbix.group_prototype import DataGroupPrototype
from WorkData.Zabbix.host_discovery import DataHostDiscovery
import allure
import pytest
@allure.feature("测试模块：hostprototype.create")
class TestHostPrototypeCreate(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS自动化测试主机原型"
base_ip = "10.0.0.28"
base_hostid = 10001
base_dns = "sanshi.zmops.cn"
base_port = "127"
base_hosts = ["{#VM.NAME}", "{#HM.NAME}", "{#VM.UUID}", "{#HV.UUID}"]
base_ruleid = 10009
groupLinks_groupid = 46
new_name = "修改SS测试主机原型"
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_name(session=session, name=cls.base_name)
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_name(session=session, name=cls.new_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostprototype.create")
def hostprototype_create(self, host=None, name=None, ruleid=None, templateids=None, groupLinks_groupid=None, groupPrototypes_name=None):
api = HostPrototypeCreate(_host=self.host)
api.host = host
api.name = name
api.ruleid = ruleid
api.templateids = templateids
api.groupLinks_groupid = groupLinks_groupid
api.groupPrototypes_name = groupPrototypes_name
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self, host, group_prototypes_name, ruleid):
response = self.hostprototype_create(host=host, name=self.base_name, ruleid=ruleid, groupLinks_groupid=[self.groupLinks_groupid],
groupPrototypes_name=[group_prototypes_name])
self.check_code(response=response, code=0)
hostid = UtilsResponse().get_result(response=response)["hostids"][0]
return hostid
@allure.step("连接数据库：zabbix")
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("查询表：hosts")
def select_hosts(self, hostid=None):
session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid)
return sql
@allure.step("查询表：group_prototype")
def select_group_prototype(self, hostid=None):
session = self.db_session()
sql = DataGroupPrototype().select_all_from_allKeys(session=session, hostid=hostid)
return sql
@allure.step("查询表：host_discovery")
def select_host_discovery(self, hostid=None):
session = self.db_session()
sql = DataHostDiscovery().select_all_from_allKeys(session=session, hostid=hostid)
return sql
@allure.step("校验查询结果")
def check_select(self, hostid, host, name, flags, groupPrototypes_name, groupid, parent_itemid):
sql = self.select_hosts(hostid=hostid)
for x, y in enumerate(sql):
assert y.host == host
assert y.name == name
assert y.flags == flags
sql = self.select_group_prototype(hostid=hostid)
assert sql != []
for x, y in enumerate(sql):
if y.name is not "":
# self.log.debug(y.name)
# self.log.debug(groupPrototypes_name)
assert y.name == groupPrototypes_name
elif y.groupid is not "":
assert y.groupid == groupid
sql = self.select_host_discovery(hostid=hostid)
for x, y in enumerate(sql):
assert y.parent_itemid == parent_itemid
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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("断言返回错误结果")
def check_status(self, response, status):
_status = UtilsResponse().get_status(response=response)
assert _status == status
@allure.title("hostprototype.create：在指定规则下新增主机原型，并链接到主机组")
@allure.story("创建主机原型：在指定规则下新增主机原型，并链接到主机组")
@allure.severity("blocker")
def test_case_01(self):
host = self.base_hosts[0]
group_prototypes_name = self.base_hosts[0]
response = self.hostprototype_create(host=host, name=self.base_name, ruleid=self.base_ruleid, groupLinks_groupid=[self.groupLinks_groupid],
groupPrototypes_name=[group_prototypes_name])
self.check_code(response=response, code=0)
hostid = UtilsResponse().get_result(response=response)["hostids"][0]
self.check_select(hostid=hostid, host=host, name=self.base_name, flags=2,
groupPrototypes_name=group_prototypes_name, groupid=self.groupLinks_groupid, parent_itemid=self.base_ruleid)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestHostPrototypeCreate()
# a.setup_class()
# a.test_case_01()
# a.test_case_02()
# a.test_case_03()
# a.test_case_06()
WorkCase/API/HostPrototype/test_hostprototype_delete.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-13
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostPrototype.hostPrototype_create import HostPrototypeCreate
from WorkApi.API.HostPrototype.hostPrototype_delete import HostPrototypeDelete
from WorkData.Zabbix.hosts import DataHosts
import allure
import pytest
@allure.feature("测试模块：hostprototype.delete")
class TestHostPrototypeDelete(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机原型"
base_ip = "10.0.0.28"
base_hostid = 10266
base_dns = "sanshi.zmops.cn"
base_port = "127"
base_hosts = ["{#VM.NAME}", "{#HM.NAME}", "{#VM.UUID}", "{#HV.UUID}"]
base_ruleids = [10009, 10010, 10013, 10014]
groupLinks_groupid = 46
new_name = "修改SS测试主机原型"
new_ip = "192.168.1.28"
new_port = "721"
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_name(session=session, name=cls.base_name)
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_name(session=session, name=cls.new_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostprototype.create")
def hostprototype_create(self, host=None, name=None, ruleid=None, templateids=None, groupLinks_groupid=None, groupPrototypes_name=None):
api = HostPrototypeCreate(_host=self.host)
api.host = host
api.name = name
api.ruleid = ruleid
api.templateids = templateids
api.groupLinks_groupid = groupLinks_groupid
api.groupPrototypes_name = groupPrototypes_name
api.get_response()
return api.response
@allure.step("调用接口：hostprototype.delete")
def hostprototype_delete(self, hostids=None):
api = HostPrototypeDelete(_host=self.host)
api.hostids = hostids
api.get_response()
return api.response
@allure.step("连接数据库：zabbix")
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 case_create(self, host, group_prototypes_name, ruleid):
response = self.hostprototype_create(host=host, name=self.base_name, ruleid=ruleid, groupLinks_groupid=[self.groupLinks_groupid],
groupPrototypes_name=[group_prototypes_name])
self.check_code(response=response, code=0)
hostid = UtilsResponse().get_result(response=response)["hostids"][0]
return hostid
@allure.step("查询表：hosts")
def select_hosts(self, hostid=None):
session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid)
return sql
@allure.step("校验查询结果")
def check_select(self, hostid):
sql = self.select_hosts(hostid=hostid)
assert sql == []
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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("断言返回错误结果")
def check_status(self, response, status):
_status = UtilsResponse().get_status(response=response)
assert _status == status
@allure.title("hostprototype.delete：删除的ID存在")
@allure.story("删除主机原型：删除的ID存在")
@allure.severity("blocker")
def test_case_01(self):
host = self.base_hosts[0]
group_prototypes_name = self.base_hosts[0]
ruleid = self.base_ruleids[0]
hostid = self.case_create(host=host, group_prototypes_name=group_prototypes_name, ruleid=ruleid)
response = self.hostprototype_delete(hostids=[hostid])
self.check_code(response=response, code=0)
self.check_select(hostid=hostid)
@allure.title("hostprototype.delete：删除的ID不存在 500")
@allure.story("删除主机原型：删除的ID不存在 500")
def test_case_02(self):
host = self.base_hosts[1]
group_prototypes_name = self.base_hosts[1]
ruleid = self.base_ruleids[1]
hostid = self.case_create(host=host, group_prototypes_name=group_prototypes_name, ruleid=ruleid)
response = self.hostprototype_delete(hostids=[hostid])
self.check_code(response=response, code=0)
self.check_select(hostid=hostid)
response = self.hostprototype_delete(hostids=[hostid])
self.check_code(response=response, code=100)
@allure.title("hostprototype.delete：删除多个ID")
@allure.story("删除主机原型：删除多个ID")
def test_case_03(self):
hostid_1 = self.case_create(host=self.base_hosts[2], group_prototypes_name=self.base_hosts[2], ruleid=self.base_ruleids[2])
hostid_2 = self.case_create(host=self.base_hosts[3], group_prototypes_name=self.base_hosts[3], ruleid=self.base_ruleids[3])
response = self.hostprototype_delete(hostids=[hostid_1, hostid_2])
self.check_code(response=response, code=0)
self.check_select(hostid=hostid_1)
self.check_select(hostid=hostid_2)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestHostPrototypeDelete()
# a.setup_class()
# a.test_case_01()
# a.test_case_02()
# a.test_case_03()
# a.test_case_06()
WorkCase/API/HostPrototype/test_hostprototype_get.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-13
from __future__ import division
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostPrototype.hostPrototype_create import HostPrototypeCreate
from WorkApi.API.HostPrototype.hostPrototype_get import HostPrototypeGet
from WorkData.Zabbix.hosts import DataHosts
from WorkCase import CaseBase
import allure
import pytest
@allure.feature("测试模块：hostprototype.get")
class TestHostPrototypeGet(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机原型"
base_ip = "10.0.0.28"
base_hostid = 10266
base_dns = "sanshi.zmops.cn"
base_port = "127"
base_hosts = ["{#VM.NAME}", "{#HM.NAME}", "{#VM.UUID}", "{#HV.UUID}"]
base_ruleids = [10009, 10010]
groupLinks_groupid = 46
new_name = "修改SS测试主机原型"
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_name(session=session, name=cls.base_name)
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_name(session=session, name=cls.new_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostprototype.create")
def hostprototype_create(self, host=None, name=None, ruleid=None, templateids=None, groupLinks_groupid=None, groupPrototypes_name=None):
api = HostPrototypeCreate(_host=self.host)
api.host = host
api.name = name
api.ruleid = ruleid
api.templateids = templateids
api.groupLinks_groupid = groupLinks_groupid
api.groupPrototypes_name = groupPrototypes_name
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self, host, group_prototypes_name, ruleid):
response = self.hostprototype_create(host=host, name=self.base_name, ruleid=ruleid, groupLinks_groupid=[self.groupLinks_groupid],
groupPrototypes_name=[group_prototypes_name])
self.check_code(response=response, code=0)
hostid = UtilsResponse().get_result(response=response)["hostids"][0]
return hostid
@allure.step("调用接口：hostprototype.get")
def hostprototype_get(self, hostids=None, discoveryids=None, inherited=None, selectDiscoveryRule=None, selectGroupLinks=None,
selectGroupPrototypes=None, selectInventory=None, selectParentHost=None, selectTemplates=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 = HostPrototypeGet(_host=self.host)
api.hostids = hostids
api.discoveryids = discoveryids
api.inherited = inherited
api.selectDiscoveryRule = selectDiscoveryRule
api.selectGroupLinks = selectGroupLinks
api.selectGroupPrototypes = selectGroupPrototypes
api.selectInventory = selectInventory
api.selectParentHost = selectParentHost
api.selectTemplates = selectTemplates
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("连接数据库：zabbix")
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("查询表：hosts")
def select_hosts(self, hostid=None):
session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid)
return sql
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="hostid", 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_sql(self, response, hostid):
sql = self.select_hosts(hostid=hostid)
result = UtilsResponse().get_result(response=response)
for x, y in enumerate(sql):
assert y.host in UtilsResponse().get_values_from_key(_dict=result, key="host")
assert y.name in UtilsResponse().get_values_from_key(_dict=result, key="name")
assert y.status in UtilsResponse().get_values_from_key(_dict=result, key="status")
for x, y in enumerate(result):
assert y["host"] in UtilsResponse().get_values_from_sql(_sql=sql, key="host")
assert y["name"] in UtilsResponse().get_values_from_sql(_sql=sql, key="name")
assert y["status"] in UtilsResponse().get_values_from_sql(_sql=sql, key="status")
@allure.title("hostprototype.get：无参数调用")
@allure.story("查询主机原型：无参数调用")
@allure.severity("blocker")
def test_case_01(self):
response = self.hostprototype_get()
self.check_code(response=response, code=0)
@allure.title("hostprototype.get：countOutput=true")
@allure.story("查询主机原型：countOutput=true")
def test_case_02(self):
response = self.hostprototype_get(countOutput="true")
self.check_code(response=response, code=0)
@allure.title("hostprototype.get：hostids=特定IDs")
@allure.story("查询主机原型：hostids=特定IDs")
def test_case_03(self):
response = self.hostprototype_get(hostids=[self.base_hostid])
self.check_code(response=response, code=0)
@allure.title("hostprototype.get：output=[name]")
@allure.story("查询主机原型：output=[name]")
def test_case_04(self):
response = self.hostprototype_get(output=["name"])
self.check_code(response=response, code=0)
@allure.title("hostprototype.get：hostids=特定IDs, output=[ip, port, type, main, useip, bulk]")
@allure.story("查询主机原型：hostids=特定IDs, output=[ip, port, type, main, useip, bulk]")
def test_case_05(self):
host = self.base_hosts[0]
group_prototypes_name = self.base_hosts[0]
ruleid = self.base_ruleids[0]
hostid = self.case_create(host=host, group_prototypes_name=group_prototypes_name, ruleid=ruleid)
response = self.hostprototype_get(hostids=[hostid], output=["host", "name", "status"])
self.check_code(response=response, code=0)
self.check_sql(response=response, hostid=hostid)
@allure.title("hostprototype.get：limit=1")
@allure.story("查询主机原型：limit=1")
def test_case_06(self):
host = self.base_hosts[1]
group_prototypes_name = self.base_hosts[1]
ruleid = self.base_ruleids[1]
self.case_create(host=host, group_prototypes_name=group_prototypes_name, ruleid=ruleid)
response = self.hostprototype_get(limit=1)
self.check_code(response=response, code=0)
result = UtilsResponse().get_result(response=response)
assert len(result) == 1
@allure.title("hostprototype.get：sortfield=hostid")
@allure.story("查询主机原型：sortfield=hostid")
def test_case_07(self):
response = self.hostprototype_get(sortfield=["hostid"])
self.check_code(response=response, code=0)
self.check_sortfield(response=response, possible=True)
@allure.title("hostprototype.get：sortfield=hostid, sortorder=ASC")
@allure.story("查询主机原型：sortfield=hostid, sortorder=ASC")
def test_case_08(self):
response = self.hostprototype_get(sortfield=["hostid"], sortorder=["ASC"])
self.check_code(response=response, code=0)
self.check_sortfield(response=response, possible="ASC")
@allure.title("hostprototype.get：sortfield=hostid, sortorder=DESC")
@allure.story("查询主机原型：sortfield=hostid, sortorder=DESC")
def test_case_09(self):
response = self.hostprototype_get(sortfield=["hostid"], sortorder=["DESC"])
self.check_code(response=response, code=0)
self.check_sortfield(response=response, possible="DESC")
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestHostPrototypeGet()
# a.setup_class()
# a.test_case_07()
WorkCase/API/HostPrototype/test_hostprototype_update.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-12
from __future__ import division
from WorkCase import CaseBase
from WorkData.Zabbix.group_prototype import DataGroupPrototype
from WorkData.Zabbix.host_discovery import DataHostDiscovery
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.HostPrototype.hostPrototype_create import HostPrototypeCreate
from WorkApi.API.HostPrototype.hostPrototype_update import HostPrototypeUpdate
from WorkData.Zabbix.hosts import DataHosts
import allure
import pytest
@allure.feature("测试模块：hostprototype.update")
class TestHostPrototypeUpdate(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机原型"
base_ip = "10.0.0.28"
base_hostid = 10266
base_dns = "sanshi.zmops.cn"
base_port = "127"
base_hosts = ["{#VM.NAME}", "{#HM.NAME}", "{#VM.UUID}", "{#HV.UUID}"]
base_ruleids = [10009, 10010, 10013, 10014]
groupLinks_groupid = 46
new_name = "修改SS测试主机原型"
new_port = "721"
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_name(session=session, name=cls.base_name)
session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
DataHosts().delete_like_name(session=session, name=cls.new_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：hostprototype.create")
def hostprototype_create(self, host=None, name=None, ruleid=None, templateids=None, groupLinks_groupid=None, groupPrototypes_name=None):
api = HostPrototypeCreate(_host=self.host)
api.host = host
api.name = name
api.ruleid = ruleid
api.templateids = templateids
api.groupLinks_groupid = groupLinks_groupid
api.groupPrototypes_name = groupPrototypes_name
api.get_response()
return api.response
@allure.step("调用接口：hostprototype.update")
def hostprototype_update(self, hostid=None, host=None, name=None, status=None, templateids=None,
groupLinks_groupid=None, groupPrototypes_name=None):
api = HostPrototypeUpdate(_host=self.host)
api.hostid = hostid
api.host = host
api.name = name
api.status = status
api.templateids = templateids
api.groupLinks_groupid = groupLinks_groupid
api.groupPrototypes_name = groupPrototypes_name
api.get_response()
return api.response
@allure.step("连接数据库：zabbix")
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 case_create(self, host, group_prototypes_name, ruleid):
response = self.hostprototype_create(host=host, name=self.base_name, ruleid=ruleid, groupLinks_groupid=[self.groupLinks_groupid],
groupPrototypes_name=[group_prototypes_name])
self.check_code(response=response, code=0)
hostid = UtilsResponse().get_result(response=response)["hostids"][0]
return hostid
@allure.step("查询表：hosts")
def select_hosts(self, hostid=None):
session = self.db_session()
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid)
return sql
@allure.step("查询表：group_prototype")
def select_group_prototype(self, hostid=None):
session = self.db_session()
sql = DataGroupPrototype().select_all_from_allKeys(session=session, hostid=hostid)
return sql
@allure.step("查询表：host_discovery")
def select_host_discovery(self, hostid=None):
session = self.db_session()
sql = DataHostDiscovery().select_all_from_allKeys(session=session, hostid=hostid)
return sql
@allure.step("校验查询结果")
def check_select(self, hostid, host, name, status, flags, groupPrototypes_name, groupid, parent_itemid):
sql = self.select_hosts(hostid=hostid)
for x, y in enumerate(sql):
assert y.host == host
assert y.name == name
assert y.flags == flags
assert y.status == status
sql = self.select_group_prototype(hostid=hostid)
assert sql != []
for x, y in enumerate(sql):
if y.name is not "":
self.log.debug(y.name)
self.log.debug(groupPrototypes_name)
assert y.name == groupPrototypes_name
elif y.groupid is not "":
assert y.groupid == groupid
sql = self.select_host_discovery(hostid=hostid)
for x, y in enumerate(sql):
assert y.parent_itemid == parent_itemid
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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("断言返回错误结果")
def check_status(self, response, status):
_status = UtilsResponse().get_status(response=response)
assert _status == status
@allure.title("hostprototype.update：修改name")
@allure.story("更新主机原型：修改name")
@allure.severity("blocker")
def test_case_01(self):
host = self.base_hosts[0]
group_prototypes_name = self.base_hosts[0]
ruleid = self.base_ruleids[0]
hostid = self.case_create(host=host, group_prototypes_name=group_prototypes_name, ruleid=ruleid)
response = self.hostprototype_update(hostid=hostid, name=self.new_name, host=host)
self.check_code(response=response, code=0)
self.check_select(hostid=hostid, host=host, name=self.new_name, flags=2, status=0,
groupPrototypes_name=group_prototypes_name, groupid=self.groupLinks_groupid, parent_itemid=ruleid)
@allure.title("hostprototype.update：修改status")
@allure.story("更新主机原型：修改status")
def test_case_02(self):
host = self.base_hosts[1]
group_prototypes_name = self.base_hosts[1]
ruleid = self.base_ruleids[1]
hostid = self.case_create(host=host, group_prototypes_name=group_prototypes_name, ruleid=ruleid)
response = self.hostprototype_update(hostid=hostid, name=self.base_name, host=host, status=1)
self.check_code(response=response, code=0)
self.check_select(hostid=hostid, host=host, name=self.base_name, flags=2, status=1,
groupPrototypes_name=group_prototypes_name, groupid=self.groupLinks_groupid, parent_itemid=ruleid)
@allure.title("hostprototype.update：修改host, groupPrototypes_name")
@allure.story("更新主机原型：修改host, groupPrototypes_name")
def test_case_03(self):
host = self.base_hosts[2]
group_prototypes_name = self.base_hosts[2]
ruleid = self.base_ruleids[2]
hostid = self.case_create(host=host, group_prototypes_name=group_prototypes_name, ruleid=ruleid)
response = self.hostprototype_update(hostid=hostid, name=self.base_name, host=self.base_hosts[3], status=1, groupPrototypes_name=[self.base_hosts[3]])
self.check_code(response=response, code=0)
self.check_select(hostid=hostid, host=self.base_hosts[3], name=self.base_name, flags=2, status=1,
groupPrototypes_name=self.base_hosts[3], groupid=self.groupLinks_groupid, parent_itemid=self.base_ruleids[2])
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestHostPrototypeUpdate()
# a.setup_class()
# a.test_case_03()
# a.test_case_02()
# a.test_case_03()
# a.test_case_06()
WorkCase/API/MediaType/test_mediatype_create.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-01
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.MediaType.mediatype_create import MediaTypeCreate
import allure
# allure.label("")
@allure.feature("测试模块：mediatype.create")
class TestMediaTypeCreate(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_description = "SS测试媒体"
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：mediatype.create")
def mediatype_create(self, description=None, _type=None, exec_path=None, gsm_modem=None, passwd=None, smtp_email=None,
smtp_helo=None, smtp_server=None, smtp_port=None, smtp_security=None, smtp_verify_host=None, smtp_verify_peer=None,
smtp_authentication=None, status=None, username=None, exec_params=None, maxsessions=None, maxattempts=None,
attempt_interval=None, content_type=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 = MediaTypeCreate(_host=self.host)
api.description = description
api.type = _type
api.exec_path = exec_path
api.gsm_modem = gsm_modem
api.passwd = passwd
api.smtp_email = smtp_email
api.smtp_helo = smtp_helo
api.smtp_server = smtp_server
api.smtp_port = smtp_port
api.smtp_security = smtp_security
api.smtp_verify_host = smtp_verify_host
api.smtp_verify_peer = smtp_verify_peer
api.smtp_authentication = smtp_authentication
api.status = status
api.username = username
api.exec_params = exec_params
api.maxsessions = maxsessions
api.maxattempts = maxattempts
api.attempt_interval = attempt_interval
api.content_type = content_type
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("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.title("mediatype.create：description=字符串, _type=0")
@allure.story("创建媒体：description=字符串, _type=0")
@allure.severity("blocker")
def test_case_01(self):
description = self.base_description + "01"
_type = 0
response = self.mediatype_create(description=description, _type=_type)
self.check_code(response=response, code=0)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestMediaTypeCreate()
# a.test_case_01()
WorkCase/API/MediaType/test_mediatype_get.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-01
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.MediaType.mediatype_get import MediaTypeGet
import allure
# allure.label("")
@allure.feature("测试模块：mediatype.get")
class TestMediaTypeGet(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：mediatype.get")
def mediatype_get(self, mediatypeids=None, mediaids=None, userids=None, selectUsers=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 = MediaTypeGet(_host=self.host)
api.mediatypeids = mediatypeids
api.mediaids = mediaids
api.userids = userids
api.selectUsers = selectUsers
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("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.title("mediatype.get：无参数调用")
@allure.story("查询维护：无参数调用")
@allure.severity("blocker")
def test_case_01(self):
response = self.mediatype_get()
self.check_code(response=response, code=0)
@allure.title("mediatype.get：countOutput=true")
@allure.story("查询维护：countOutput=true")
def test_case_02(self):
response = self.mediatype_get(countOutput="true")
self.check_code(response=response, code=0)
@allure.title("mediatype.get：mediatypeids=特定IDs")
@allure.story("查询维护：mediatypeids=特定IDs")
def test_case_03(self):
response = self.mediatype_get(mediatypeids=[1, 5])
self.check_code(response=response, code=0)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestMediaTypeGet()
# a.test_case_01()
WorkCase/API/Problem/test_problem_get.py
...@@ -17,9 +17,6 @@ from WorkCase import CaseBase ...@@ -17,9 +17,6 @@ from WorkCase import CaseBase
import allure import allure
# allure.label("")
@allure.feature("测试模块：problem.get") @allure.feature("测试模块：problem.get")
class TestProblemGet(object): class TestProblemGet(object):
log = UtilsLog() log = UtilsLog()
...@@ -30,15 +27,10 @@ class TestProblemGet(object): ...@@ -30,15 +27,10 @@ class TestProblemGet(object):
db_user = env["db_user"] db_user = env["db_user"]
db_pw = env["db_pw"] db_pw = env["db_pw"]
db_base = env["db_base"] db_base = env["db_base"]
base_name = "SS测试主机原型"
base_description = "SS测试触发器原型"
base_expression = "{Zabbix server:vfs.fs.size[{#FSNAME},pfree].last()}<20"
@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)
# DataTriggers().delete_description(session=session, description=cls.base_description)
@classmethod @classmethod
def teardown_class(cls): def teardown_class(cls):
...@@ -126,11 +118,13 @@ class TestProblemGet(object): ...@@ -126,11 +118,13 @@ class TestProblemGet(object):
result = UtilsResponse().get_result(response=response) result = UtilsResponse().get_result(response=response)
for x, y in enumerate(result): for x, y in enumerate(result):
sql = self.select_problem(eventid=y["eventid"]) sql = self.select_problem(eventid=y["eventid"])
assert y["source"] in UtilsResponse().get_values_from_sql(_sql=sql, key="source") for q, w in enumerate(sql):
assert y["object"] in UtilsResponse().get_values_from_sql(_sql=sql, key="object") assert y["source"] == w.source
assert y["objectid"] in UtilsResponse().get_values_from_sql(_sql=sql, key="objectid") assert y["object"] == w.object
assert y["name"] in UtilsResponse().get_values_from_sql(_sql=sql, key="name") assert y["objectid"] == w.objectid
assert y["severity"] in UtilsResponse().get_values_from_sql(_sql=sql, key="severity") assert y["name"] == w.name
assert y["severity"] == w.severity
assert q == 0
@allure.title("problem.get：无参数调用") @allure.title("problem.get：无参数调用")
@allure.story("查询告警：无参数调用") @allure.story("查询告警：无参数调用")
...@@ -201,4 +195,4 @@ if __name__ == "__main__": ...@@ -201,4 +195,4 @@ if __name__ == "__main__":
# a = TestProblemGet() # a = TestProblemGet()
# a.setup_class() # a.setup_class()
# a.test_case_10() # a.test_case_04()
WorkCase/API/Problem/test_problem_getLatest.py
...@@ -17,8 +17,6 @@ from WorkCase import CaseBase ...@@ -17,8 +17,6 @@ from WorkCase import CaseBase
import allure import allure
# allure.label("")
@allure.feature("测试模块：problem.getLatest") @allure.feature("测试模块：problem.getLatest")
class TestProblemGetLatest(object): class TestProblemGetLatest(object):
...@@ -121,11 +119,13 @@ class TestProblemGetLatest(object): ...@@ -121,11 +119,13 @@ class TestProblemGetLatest(object):
result = UtilsResponse().get_result(response=response) result = UtilsResponse().get_result(response=response)
for x, y in enumerate(result["data"]): for x, y in enumerate(result["data"]):
sql = self.select_problem(eventid=y["eventid"]) sql = self.select_problem(eventid=y["eventid"])
assert y["source"] in UtilsResponse().get_values_from_sql(_sql=sql, key="source") for q, w in enumerate(sql):
assert y["object"] in UtilsResponse().get_values_from_sql(_sql=sql, key="object") assert y["source"] == w.source
assert y["objectid"] in UtilsResponse().get_values_from_sql(_sql=sql, key="objectid") assert y["object"] == w.object
assert y["name"] in UtilsResponse().get_values_from_sql(_sql=sql, key="name") assert y["objectid"] == w.objectid
assert y["severity"] in UtilsResponse().get_values_from_sql(_sql=sql, key="severity") assert y["name"] == w.name
assert y["severity"] == w.severity
assert q == 0
assert result["total"] >= 0 assert result["total"] >= 0
......
WorkCase/API/Proxy/test_proxy_get.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-04
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.Proxy.proxy_get import ProxyGet
import allure
# allure.label("")
@allure.feature("测试模块：proxy.get")
class TestProxyGet(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
@classmethod
def setup_class(cls):
cls.log.debug("开始执行测试套件.......")
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：proxy.get")
def proxy_get(self, proxyids=None, selectHosts=None, selectInterface=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 = ProxyGet(_host=self.host)
api.proxyids = proxyids
api.selectHosts = selectHosts
api.selectInterface = selectInterface
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("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.title("proxy.get：无参数调用")
@allure.story("查询代理：无参数调用")
@allure.severity("blocker")
def test_case_01(self):
response = self.proxy_get()
self.check_code(response=response, code=0)
@allure.title("proxy.get：countOutput=true")
@allure.story("查询代理：countOutput=true")
def test_case_02(self):
response = self.proxy_get(countOutput="true")
self.check_code(response=response, code=0)
@allure.title("proxy.get：proxyids=特定IDs")
@allure.story("查询代理：proxyids=特定IDs")
def test_case_03(self):
response = self.proxy_get(proxyids=[10283, 10284])
self.check_code(response=response, code=0)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestProxyGet()
# a.test_case_01()
WorkCase/API/TriggerPrototype/test_triggerprototype_create.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-11
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.TriggerPrototype.triggerPrototype_create import TriggerPrototypeCreate
from WorkData.Zabbix.triggers import DataTriggers
from WorkData.Zabbix.trigger_tag import DataTriggerTag
import allure
# allure.label("")
@allure.feature("测试模块：triggerprototype.create")
class TestTriggerPrototypeCreate(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机原型"
base_description = "SS测试触发器原型"
base_expression = "{Zabbix server:vfs.fs.size[{#FSNAME},pfree].last()}<20"
@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)
DataTriggers().delete_like_description(session=session, description=cls.base_description)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：triggerprototype.create")
def triggerprototype_create(self, description=None, expression=None, tags_tag=None, tags_value=None):
api = TriggerPrototypeCreate(_host=self.host)
api.description = description
api.expression = expression
api.tags_tag = tags_tag
api.tags_value = tags_value
api.get_response()
return api.response
@allure.step("连接数据库：zabbix")
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("查询表：triggers")
def select_triggers(self, triggerid=None):
session = self.db_session()
sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid)
return sql
@allure.step("查询表：trigger_tag")
def select_trigger_tag(self, triggerid=None):
session = self.db_session()
sql = DataTriggerTag().select_all_from_allKeys(session=session, triggerid=triggerid)
return sql
@allure.step("校验查询结果")
def check_select(self, triggerid, description, tags_tag, tags_value):
sql = self.select_triggers(triggerid=triggerid)
for x, y in enumerate(sql):
assert y.description == description
assert y.status == 0
assert y.value == 0
sql = self.select_trigger_tag(triggerid=triggerid)
assert sql != []
for x, y in enumerate(sql):
assert y.tag == tags_tag[x]
assert y.value == tags_value[x]
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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("断言返回错误结果")
def check_status(self, response, status):
_status = UtilsResponse().get_status(response=response)
assert _status == status
@allure.title("triggerprototype.create：创建一个触发器")
@allure.story("创建触发器原型：创建一个触发器")
@allure.severity("blocker")
def test_case_01(self):
tags_tag = ["volume"]
tags_value = ["{#FSNAME}"]
response = self.triggerprototype_create(description=self.base_description, expression=self.base_expression, tags_tag=tags_tag, tags_value=tags_value)
self.check_code(response=response, code=0)
triggerid = UtilsResponse().get_result(response=response)["triggerids"][0]
self.check_select(triggerid=triggerid, description=self.base_description, tags_tag=tags_tag, tags_value=tags_value)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestTriggerPrototypeCreate()
# a.setup_class()
# a.test_case_01()
# a.test_case_02()
# a.test_case_03()
# a.test_case_06()
WorkCase/API/TriggerPrototype/test_triggerprototype_delete.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-13
from __future__ import division
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.TriggerPrototype.triggerPrototype_create import TriggerPrototypeCreate
from WorkApi.API.TriggerPrototype.triggerPrototype_delete import TriggerPrototypeDelete
from WorkData.Zabbix.triggers import DataTriggers
import allure
# allure.label("")
@allure.feature("测试模块：triggerprototype.delete")
class TestTriggerPrototypeDelete(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机原型"
base_description = "SS测试触发器原型"
base_expression = "{Zabbix server:vfs.fs.size[{#FSNAME},pfree].last()}<20"
@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)
DataTriggers().delete_like_description(session=session, description=cls.base_description)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：triggerprototype.create")
def triggerprototype_create(self, description=None, expression=None, tags_tag=None, tags_value=None):
api = TriggerPrototypeCreate(_host=self.host)
api.description = description
api.expression = expression
api.tags_tag = tags_tag
api.tags_value = tags_value
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self):
tags_tag = ["volume"]
tags_value = ["{#FSNAME}"]
response = self.triggerprototype_create(description=self.base_description, expression=self.base_expression, tags_tag=tags_tag, tags_value=tags_value)
self.check_code(response=response, code=0)
triggerid = UtilsResponse().get_result(response=response)["triggerids"][0]
return triggerid
@allure.step("调用接口：triggerprototype.delete")
def triggerprototype_delete(self, triggerids=None):
api = TriggerPrototypeDelete(_host=self.host)
api.triggerids = triggerids
api.get_response()
return api.response
@allure.step("连接数据库：zabbix")
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("查询表：triggers")
def select_hosts(self, triggerid=None):
session = self.db_session()
sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid)
return sql
@allure.step("校验查询结果")
def check_select(self, triggerid):
sql = self.select_hosts(triggerid=triggerid)
assert sql == []
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="groupid", 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("断言返回错误结果")
def check_status(self, response, status):
_status = UtilsResponse().get_status(response=response)
assert _status == status
@allure.title("triggerprototype.delete：删除的ID存在")
@allure.story("删除触发器原型：删除的ID存在")
@allure.severity("blocker")
def test_case_01(self):
triggerid = self.case_create()
response = self.triggerprototype_delete(triggerids=[triggerid])
self.check_code(response=response, code=0)
self.check_select(triggerid=triggerid)
@allure.title("triggerprototype.delete：删除的ID不存在 500")
@allure.story("删除触发器原型：删除的ID不存在 500")
def test_case_02(self):
triggerid = self.case_create()
response = self.triggerprototype_delete(triggerids=[triggerid])
self.check_code(response=response, code=0)
self.check_select(triggerid=triggerid)
response = self.triggerprototype_delete(triggerids=[triggerid])
self.check_code(response=response, code=100)
@allure.title("triggerprototype.delete：删除多个ID")
@allure.story("删除触发器原型：删除多个ID")
def test_case_03(self):
triggerid_1 = self.case_create()
triggerid_2 = self.case_create()
triggerid_3 = self.case_create()
response = self.triggerprototype_delete(triggerids=[triggerid_1, triggerid_2, triggerid_3])
self.check_code(response=response, code=0)
self.check_select(triggerid=triggerid_1)
self.check_select(triggerid=triggerid_2)
self.check_select(triggerid=triggerid_3)
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestTriggerPrototypeDelete()
# a.setup_class()
# a.test_case_03()
WorkCase/API/TriggerPrototype/test_triggerprototype_get.py deleted 100644 → 0
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-14
from __future__ import division
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
from WorkApi.API.TriggerPrototype.triggerPrototype_create import TriggerPrototypeCreate
from WorkApi.API.TriggerPrototype.triggerPrototype_get import TriggerPrototypeGet
from WorkData.Zabbix.triggers import DataTriggers
from WorkCase import CaseBase
import allure
# allure.label("")
@allure.feature("测试模块：triggerprototype.get")
class TestTriggerPrototypeGet(object):
log = UtilsLog()
env = CaseBase().environment
host = env["host"]
db_url = env["db_url"]
db_port = env["db_port"]
db_user = env["db_user"]
db_pw = env["db_pw"]
db_base = env["db_base"]
base_name = "SS测试主机原型"
base_description = "SS测试触发器原型"
base_expression = "{Zabbix server:vfs.fs.size[{#FSNAME},pfree].last()}<20"
@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)
DataTriggers().delete_like_description(session=session, description=cls.base_description)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
def teardown_method(self):
self.log.debug("测试用例执行结束...")
@allure.step("调用接口：triggerprototype.create")
def triggerprototype_create(self, description=None, expression=None, tags_tag=None, tags_value=None):
api = TriggerPrototypeCreate(_host=self.host)
api.description = description
api.expression = expression
api.tags_tag = tags_tag
api.tags_value = tags_value
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create(self):
tags_tag = ["volume"]
tags_value = ["{#FSNAME}"]
response = self.triggerprototype_create(description=self.base_description, expression=self.base_expression, tags_tag=tags_tag, tags_value=tags_value)
self.check_code(response=response, code=0)
triggerid = UtilsResponse().get_result(response=response)["triggerids"][0]
return triggerid
@allure.step("调用接口：triggerprototype.get")
def triggerprototype_get(self, active=None, applicationids=None, discoveryids=None, functions=None, group=None,
groupids=None, host=None, hostids=None, inherited=None, maintenance=None, min_severity=None,
monitored=None, templated=None, templateids=None, triggerids=None, expandExpression=None,
selectDiscoveryRule=None, selectFunctions=None, selectGroups=None, selectHosts=None,
selectItems=None, selectDependencies=None, selectTags=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 = TriggerPrototypeGet(_host=self.host)
api.active = active
api.applicationids = applicationids
api.discoveryids = discoveryids
api.functions = functions
api.group = group
api.groupids = groupids
api.host = host
api.hostids = hostids
api.inherited = inherited
api.maintenance = maintenance
api.min_severity = min_severity
api.monitored = monitored
api.templated = templated
api.templateids = templateids
api.triggerids = triggerids
api.expandExpression = expandExpression
api.selectDiscoveryRule = selectDiscoveryRule
api.selectFunctions = selectFunctions
api.selectGroups = selectGroups
api.selectHosts = selectHosts
api.selectItems = selectItems
api.selectDependencies = selectDependencies
api.selectTags = selectTags
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("连接数据库：zabbix")
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("查询表：triggers")
def select_hosts(self, triggerid=None):
session = self.db_session()
sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid)
return sql
@allure.step("断言返回结果：校验排序")
def check_sortfield(self, response, possible):
result = UtilsResponse().get_result(response=response)
UtilsResponse().check_sort(_list=result, key="triggerid", 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_sql(self, response, triggerid):
sql = self.select_hosts(triggerid=triggerid)
result = UtilsResponse().get_result(response=response)
for x, y in enumerate(sql):
assert y.description in UtilsResponse().get_values_from_key(_dict=result, key="description")
assert y.status in UtilsResponse().get_values_from_key(_dict=result, key="status")
assert y.value in UtilsResponse().get_values_from_key(_dict=result, key="value")
assert y.flags in UtilsResponse().get_values_from_key(_dict=result, key="flags")
for x, y in enumerate(result):
assert y["description"] in UtilsResponse().get_values_from_sql(_sql=sql, key="description")
assert y["status"] in UtilsResponse().get_values_from_sql(_sql=sql, key="status")
assert y["value"] in UtilsResponse().get_values_from_sql(_sql=sql, key="value")
assert y["flags"] in UtilsResponse().get_values_from_sql(_sql=sql, key="flags")
@allure.title("triggerprototype.get：无参数调用")
@allure.story("查询触发器原型：无参数调用")
@allure.severity("blocker")
def test_case_01(self):
response = self.triggerprototype_get()
self.check_code(response=response, code=0)
@allure.title("triggerprototype.get：countOutput=true")
@allure.story("查询触发器原型：countOutput=true")
def test_case_02(self):
response = self.triggerprototype_get(countOutput="true")
self.check_code(response=response, code=0)
@allure.title("triggerprototype.get：output=[description]")
@allure.story("查询触发器原型：output=[description]")
def test_case_03(self):
response = self.triggerprototype_get(output=["description"])
self.check_code(response=response, code=0)
@allure.title("triggerprototype.get：triggerids=特定IDs, output=[description]")
@allure.story("查询触发器原型：triggerids=特定IDs, output=[description]")
def test_case_04(self):
triggerid = self.case_create()
response = self.triggerprototype_get(triggerids=[triggerid], output=["description"])
self.check_code(response=response, code=0)
@allure.title("triggerprototype.get：triggerids=特定IDs, output=[description, status, value, flags]")
@allure.story("查询触发器原型：triggerids=特定IDs, output=[description, status, value, flags]")
def test_case_05(self):
triggerid = self.case_create()
response = self.triggerprototype_get(triggerids=[triggerid], output=["description", "status", "value", "flags"])
self.check_code(response=response, code=0)
self.check_sql(response=response, triggerid=triggerid)
@allure.title("triggerprototype.get：limit=1")
@allure.story("查询触发器原型：limit=1")
def test_case_06(self):
self.case_create()
response = self.triggerprototype_get(limit=1)
self.check_code(response=response, code=0)
result = UtilsResponse().get_result(response=response)
assert len(result) == 1
@allure.title("triggerprototype.get：sortfield=hostid")
@allure.story("查询触发器原型：sortfield=hostid")
def test_case_07(self):
response = self.triggerprototype_get(sortfield=["triggerid"])
self.check_code(response=response, code=0)
self.check_sortfield(response=response, possible=True)
@allure.title("triggerprototype.get：sortfield=triggerid, sortorder=ASC")
@allure.story("查询触发器原型：sortfield=triggerid, sortorder=ASC")
def test_case_08(self):
response = self.triggerprototype_get(sortfield=["triggerid"], sortorder=["ASC"])
self.check_code(response=response, code=0)
self.check_sortfield(response=response, possible="ASC")
@allure.title("triggerprototype.get：sortfield=triggerid, sortorder=DESC")
@allure.story("查询触发器原型：sortfield=triggerid, sortorder=DESC")
def test_case_09(self):
response = self.triggerprototype_get(sortfield=["triggerid"], sortorder=["DESC"])
self.check_code(response=response, code=0)
self.check_sortfield(response=response, possible="DESC")
if __name__ == "__main__":
from WorkUtils.UtilsPyTest import UtilsPyTest
from WorkUtils.UtilsCmd import UtilsCmd
import os
# 执行自动化测试用例
case_info = os.path.split(__file__)
case = UtilsCmd().pytest_cmd()
r = UtilsPyTest(case=case, case_info=case_info)
r.run_main()
# a = TestTriggerPrototypeGet()
# a.setup_class()
# a.test_case_04()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment