Skip to content

Z
zmops-test
Commit ba9dc48c by sanshi
Options

整个资产代码重构

parent d54c67cc
Showing with 604 additions and 589 deletions
WorkApi/APP/Device/device_update.py 0 → 100644
# -*- coding: utf-8 -*-
# 更新资产
# 作者: 陈磊
# 时间: 2019-12-13
from WorkUtils.UtilsRequest import UtilsRequest
from WorkUtils.UtilsLog import UtilsLog
from WorkApi.ApiBase import ApiBase, GetBase
class DeviceUpdate(object):
def __init__(self, _host):
"""
:param _host: 域名
:return:
"""
self.log = UtilsLog()
self.log.info("调用更新资产")
self.log.info(self.__class__)
self._host = _host
self._headers = {}
self._path = "/device/update"
self._url = self._host + self._path
self._params = {}
self._json = {}
self._data = {}
self.response = ""
self.token = None
self.hostid = None
self.hostName = None
self.hostType = None
self.manageLevel = None
self.iplist = None
self.port = None
self.dns = None
self.monitorInterface = None
self.monitorType = None
self.parentHost = None
self.businessIds = None
self.businessTree = None
self.opsPerson = None
self.snmpCommunity = None
self.ipmiAuthtype = None
self.ipmiPrivilege = None
self.ipmiUsername = None
self.ipmiPassword = None
self.factoryId = None
self.model = None
self.version = None
self.serialnumber = None
self.description = None
self.monitorStatus = None
self.api = UtilsRequest()
def get_response(self):
base = ApiBase()
base.dict_add_key(_key="token", value=self.token)
self._headers = base._json
base = ApiBase()
base.dict_add_key(_key="hostid", value=self.hostid)
base.dict_add_key(_key="hostName", value=self.hostName)
base.dict_add_key(_key="hostType", value=self.hostType)
base.dict_add_key(_key="manageLevel", value=self.manageLevel)
base.dict_add_key(_key="iplist", value=self.iplist)
base.dict_add_key(_key="port", value=self.port)
base.dict_add_key(_key="dns", value=self.dns)
base.dict_add_key(_key="monitorInterface", value=self.monitorInterface)
base.dict_add_key(_key="monitorType", value=self.monitorType)
base.dict_add_key(_key="parentHost", value=self.parentHost)
base.dict_add_key(_key="businessIds", value=self.businessIds)
base.dict_add_key(_key="businessTree", value=self.businessTree)
base.dict_add_key(_key="opsPerson", value=self.opsPerson)
base.dict_add_key(_key="snmpCommunity", value=self.snmpCommunity)
base.dict_add_key(_key="ipmiAuthtype", value=self.ipmiAuthtype)
base.dict_add_key(_key="ipmiPrivilege", value=self.ipmiPrivilege)
base.dict_add_key(_key="ipmiUsername", value=self.ipmiUsername)
base.dict_add_key(_key="ipmiPassword", value=self.ipmiPassword)
base.dict_add_key(_key="factoryId", value=self.factoryId)
base.dict_add_key(_key="model", value=self.model)
base.dict_add_key(_key="version", value=self.version)
base.dict_add_key(_key="serialnumber", value=self.serialnumber)
base.dict_add_key(_key="description", value=self.description)
base.dict_add_key(_key="monitorStatus", value=self.monitorStatus)
self._json = base._json
self.response = self.api.post(url=self._url, headers=self._headers, json=self._json)
WorkCase/APP/Device/test_device_asset_getFactoryList.py
......@@ -9,6 +9,7 @@ from __future__ import division
import pytest
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -42,9 +43,11 @@ class TestDeviceAssetGetFactoryList(object):
base_type_snmp = 127
base_ip = "172.16.3.171"
base_dns = "sanshi.zmops.com"
base_port = 999
base_port = 161
base_name = "SS测试资产-资产IPS"
base_name = "SS 自动化专用"
base_business_id_1 = "12"
base_business_id_2 = "13"
@classmethod
def setup_class(cls):
......@@ -58,79 +61,17 @@ class TestDeviceAssetGetFactoryList(object):
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("调用接口：device.asset.getFactoryList")
def device_asset_getFactoryList(self, token=None, hostType=None):
api = DeviceAssetGetFactoryList(_host=self.host)
api.token = token
api.hostType = hostType
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
......@@ -143,91 +84,67 @@ class TestDeviceAssetGetFactoryList(object):
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid, host_name=host_name, available=available, monitor_status=monitor_status)
return sql
@allure.step("创建测试数据")
def case_create(self, num):
self.name = self.base_name + num
response = self.device_create(token=self.token, hostName=self.name, hostType=self.base_type_agent, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=1, opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.check_code(response=response, code=0)
self.device_id = UtilsResponse().get_data(response=response)
@allure.step("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.step("断言返回结果数据")
def check_msg(self, response, msg):
_msg = UtilsResponse().get_msg(response=response)
assert _msg == msg
@allure.title("device.asset.getFactoryList：token未传")
@allure.story("查询品牌下拉框：token未传")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_01(self):
self.get_base_token()
response = self.device_asset_getFactoryList()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.device_asset_getFactoryList()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("device.asset.getFactoryList：token的用户已删除")
@allure.story("查询品牌下拉框：token的用户已删除")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.device_asset_getFactoryList(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.device_asset_getFactoryList(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("device.asset.getFactoryList：token的用户已禁用")
@allure.story("查询品牌下拉框：token的用户已禁用")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.device_asset_getFactoryList(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.device_asset_getFactoryList(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("device.asset.getFactoryList：无参数调用")
@allure.story("查询品牌下拉框：无参数调用")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
self.case_create(num="04")
response = self.device_asset_getFactoryList(token=self.token)
self.check_code(response=response, code=0)
response = self.app.device_asset_getFactoryList(token=self.token)
self.app.check_code(response=response, code=0)
@allure.title("device.asset.getFactoryList：hostType查询")
@allure.story("查询品牌下拉框：hostType查询")
def test_case_05(self):
self.get_base_token()
name = self.base_name + "05"
self.app.case_create_host(name=name, host_type=self.base_type_agent,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.device_asset_getFactoryList(token=self.token, hostType=self.base_type_agent)
self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
assert data[0]["name"] == "SS专用"
assert data[0]["id"] == 3
response = self.app.device_asset_getFactoryList(token=self.token, hostType=self.base_type_agent)
self.app.check_code(response=response, code=0)
self.app.check_keyValue(UtilsResponse().get_data(response=response), _key="name", _value="SS专用")
self.app.check_keyValue(UtilsResponse().get_data(response=response), _key="id", _value=3)
@allure.title("device.asset.getFactoryList：hostType查询")
@allure.story("查询品牌下拉框：hostType查询")
def test_case_06(self):
self.get_base_token()
name = self.base_name + "06"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.device_asset_getFactoryList(token=self.token, hostType=self.base_type_snmp)
self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
assert data[0]["name"] == "SS专用"
assert data[0]["id"] == 3
response = self.app.device_asset_getFactoryList(token=self.token, hostType=self.base_type_snmp)
self.app.check_code(response=response, code=0)
self.app.check_keyValue(UtilsResponse().get_data(response=response), _key="name", _value="SS专用")
self.app.check_keyValue(UtilsResponse().get_data(response=response), _key="id", _value=3)
if __name__ == "__main__":
......@@ -244,4 +161,5 @@ if __name__ == "__main__":
# a = TestDeviceAssetGetFactoryList()
# a.setup_class()
# a.setup_method()
# a.test_case_05()
WorkCase/APP/Device/test_device_asset_ips.py
......@@ -9,6 +9,7 @@ from __future__ import division
import pytest
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -42,9 +43,11 @@ class TestDeviceAssetIps(object):
base_type_snmp = 127
base_ip = "172.16.3.171"
base_dns = "sanshi.zmops.com"
base_port = 999
base_port = 161
base_name = "SS测试资产-资产IPS"
base_name = "SS 自动化专用"
base_business_id_1 = "12"
base_business_id_2 = "13"
@classmethod
def setup_class(cls):
......@@ -58,79 +61,17 @@ class TestDeviceAssetIps(object):
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("调用接口：device.asset.ips")
def device_asset_ips(self, token=None, ipName=None):
api = DeviceAssetIps(_host=self.host)
api.token = token
api.ipName = ipName
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
......@@ -154,74 +95,53 @@ class TestDeviceAssetIps(object):
self.check_code(response=response, code=0)
self.device_id = UtilsResponse().get_data(response=response)
@allure.step("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.step("断言返回结果数据")
def check_msg(self, response, msg):
_msg = UtilsResponse().get_msg(response=response)
assert _msg == msg
@allure.title("device.asset.ips：token未传")
@allure.story("查询资产IPS下拉框：token未传")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_01(self):
self.get_base_token()
response = self.device_asset_ips()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.device_asset_ips()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("device.asset.ips：token的用户已删除")
@allure.story("查询资产IPS下拉框：token的用户已删除")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.device_asset_ips(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.device_asset_ips(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("device.asset.ips：token的用户已禁用")
@allure.story("查询资产IPS下拉框：token的用户已禁用")
@allure.severity("blocker")
@pytest.mark.skip(reason="此接口未加入token验证")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.device_asset_ips(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.device_asset_ips(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("device.asset.ips：无参数调用")
@allure.story("查询资产IPS下拉框：无参数调用")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
self.case_create(num="04")
response = self.device_asset_ips(token=self.token)
self.check_code(response=response, code=0)
response = self.app.device_asset_ips(token=self.token)
self.app.check_code(response=response, code=0)
@allure.title("device.asset.ips：ipName查询")
@allure.story("查询资产IPS下拉框：ipName查询")
def test_case_05(self):
self.get_base_token()
self.case_create(num="05")
name = self.base_name + "05"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.device_asset_ips(token=self.token, ipName=self.base_ip)
self.check_code(response=response, code=0)
data = UtilsResponse().get_data(response=response)
ids = []
for x, y in enumerate(data):
ids.append(y["id"])
sql = self.select_hosts(hostid=y["id"])
for q, w in enumerate(sql):
assert y["name"] == w.manage_ip
assert self.device_id in ids
response = self.app.device_asset_ips(token=self.token, ipName=self.base_ip)
self.app.check_code(response=response, code=0)
self.app.check_keyValue(_json=UtilsResponse().get_data(response=response), _key="name", _value=self.base_ip)
if __name__ == "__main__":
......@@ -238,4 +158,5 @@ if __name__ == "__main__":
# a = TestDeviceAssetIps()
# a.setup_class()
# a.setup_method()
# a.test_case_05()
WorkCase/APP/Device/test_device_delete.py
......@@ -70,6 +70,7 @@ class TestDeviceDelete(object):
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
......@@ -135,7 +136,8 @@ class TestDeviceDelete(object):
@allure.story("删除资产：删除一个")
@allure.severity("blocker")
def test_case_04(self):
host_id_1 = self.app.case_create_host(base_name=self.base_name, num="04", host_type=self.base_type_snmp,
name = self.base_name + "04"
host_id_1 = self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_delete(token=self.token, hostids=[host_id_1])
......@@ -146,13 +148,16 @@ class TestDeviceDelete(object):
@allure.story("删除资产：删除多个")
@allure.severity("blocker")
def test_case_05(self):
host_id_1 = self.app.case_create_host(base_name=self.base_name, num="051", host_type=self.base_type_snmp,
name_1 = self.base_name + "051"
name_2 = self.base_name + "052"
name_3 = self.base_name + "053"
host_id_1 = self.app.case_create_host(name=name_1, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
host_id_2 = self.app.case_create_host(base_name=self.base_name, num="052", host_type=self.base_type_snmp,
host_id_2 = self.app.case_create_host(name=name_2, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
host_id_3 = self.app.case_create_host(base_name=self.base_name, num="053", host_type=self.base_type_snmp,
host_id_3 = self.app.case_create_host(name=name_3, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_delete(token=self.token, hostids=[host_id_1, host_id_2, host_id_3])
......
WorkCase/APP/Device/test_device_get.py
......@@ -20,6 +20,7 @@ from WorkData.Argus.hosts import DataHosts
from WorkData.Argus.alarm_rule import DataAlarmRule
import allure
from WorkCase.APP import AppBase
@allure.feature("测试模块：device.get")
......@@ -41,13 +42,13 @@ class TestDeviceGet(object):
base_type_snmp = 127
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 999
base_port = 161
base_name = "SS测试资产-查询"
base_name = "SS 自动化专用"
base_business_id_1 = "12"
base_business_id_2 = "13"
base_business_name_1 = "SS专用业务1"
base_business_name_2 = "SS专用业务2"
base_business_name_1 = "SS 自动化专用业务1"
base_business_name_2 = "SS 自动化专用业务2"
@classmethod
def setup_class(cls):
......@@ -61,94 +62,16 @@ class TestDeviceGet(object):
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None, port=None, businessTree=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.port = port
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.businessTree = businessTree
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("调用接口：device.get")
def device_get(self, token=None, menuId=None, name=None, _type=None, manageLevel=None, monitorStatus=None,
parentHost=None, businessIds=None, available=None, memberId=None, factoryId=None,
assetCode=None, manageIp=None):
api = DeviceGet(_host=self.host)
api.token = token
api.menuId = menuId
api.name = name
api.type = _type
api.manageLevel = manageLevel
api.monitorStatus = monitorStatus
api.available = available
api.parentHost = parentHost
api.businessIds = businessIds
api.memberId = memberId
api.factoryId = factoryId
api.assetCode = assetCode
api.manageIp = manageIp
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
......@@ -161,163 +84,143 @@ class TestDeviceGet(object):
sql = DataHosts().select_all_from_allKeys(session=session, hostid=hostid, host_name=host_name, available=available, monitor_status=monitor_status)
return sql
@allure.step("创建测试数据")
def case_create(self, num):
self.name = self.base_name + num
response = self.device_create(token=self.token, hostName=self.name, hostType=self.base_type_agent,
iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
businessTree="businessTree", opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.check_code(response=response, code=0)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@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_data(self, response, _key, _value):
data = UtilsResponse().get_data(response=response)
for x, y in enumerate(data["list"]):
assert y[_key] == _value
@allure.title("device.get：token未传")
@allure.story("查询资产：token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.device_get()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.device_get()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("device.get：token的用户已删除")
@allure.story("查询资产：token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.device_get(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.device_get(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("device.get：token的用户已禁用")
@allure.story("查询资产：token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.device_get(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.device_get(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("device.get：无参数调用")
@allure.story("查询资产：无参数调用")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
self.case_create(num="04")
response = self.device_get(token=self.token)
self.check_code(response=response, code=0)
response = self.app.device_get(token=self.token)
self.app.check_code(response=response, code=0)
@allure.title("device.get：name查询")
@allure.story("查询资产：name查询")
def test_case_05(self):
self.get_base_token()
self.case_create(num="05")
response = self.device_get(token=self.token, name=self.name)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="name", _value=self.name)
name = self.base_name + "05"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, name=self.base_name + "05")
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="name", _value=self.base_name + "05")
@allure.title("device.get：type查询")
@allure.story("查询资产：type查询")
def test_case_06(self):
self.get_base_token()
self.case_create(num="06")
response = self.device_get(token=self.token, _type=self.base_type_agent)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="hostType", _value=self.base_type_agent)
name = self.base_name + "06"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, _type=self.base_type_snmp)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="hostType", _value=self.base_type_snmp)
@allure.title("device.get：manageLevel查询")
@allure.story("查询资产：manageLevel查询")
def test_case_07(self):
self.get_base_token()
self.case_create(num="07")
response = self.device_get(token=self.token, manageLevel=1)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="manageLevel", _value=1)
name = self.base_name + "07"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, manageLevel=1)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="manageLevel", _value=1)
@allure.title("device.get：monitorStatus查询")
@allure.story("查询资产：monitorStatus查询")
def test_case_08(self):
self.get_base_token()
self.case_create(num="08")
response = self.device_get(token=self.token, monitorStatus=1)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="monitorStatus", _value=1)
name = self.base_name + "08"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, monitorStatus=1)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="monitorStatus", _value=1)
@allure.title("device.get：available查询")
@allure.story("查询资产：available查询")
def test_case_09(self):
self.get_base_token()
self.case_create(num="09")
response = self.device_get(token=self.token, available=1)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="available", _value=1)
name = self.base_name + "09"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, available=1)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="available", _value=1)
@allure.title("device.get：businessIds查询")
@allure.story("查询资产：businessIds查询")
def test_case_10(self):
self.get_base_token()
self.case_create(num="10")
response = self.device_get(token=self.token, businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="businessesNames", _value=self.base_business_name_1 + "," + self.base_business_name_2)
name = self.base_name + "10"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="businessesNames", _value=self.base_business_name_1 + "," + self.base_business_name_2)
@allure.title("device.get：memberId查询")
@allure.story("查询资产：memberId查询")
def test_case_11(self):
self.get_base_token()
self.case_create(num="11")
response = self.device_get(token=self.token, memberId=11)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="memberName", _value="sanshi")
name = self.base_name + "11"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, memberId=11)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="memberName", _value="sanshi")
@allure.title("device.get：factoryId查询")
@allure.story("查询资产：factoryId查询")
def test_case_12(self):
self.get_base_token()
self.case_create(num="12")
response = self.device_get(token=self.token, factoryId=1)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="factoryId", _value=1)
name = self.base_name + "12"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, factoryId=1)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="factoryId", _value=1)
@allure.title("device.get：manageIp查询")
@allure.story("查询资产：manageIp查询")
def test_case_13(self):
self.get_base_token()
self.case_create(num="13")
response = self.device_get(token=self.token, manageIp=self.base_ip)
self.check_code(response=response, code=0)
self.check_data(response=response, _key="manageIp", _value=self.base_ip)
name = self.base_name + "13"
self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2)
response = self.app.device_get(token=self.token, manageIp=self.base_ip)
self.app.check_code(response=response, code=0)
self.app.check_data_keyValue(response=response, _key="manageIp", _value=self.base_ip)
@allure.title("device.get：menuId查询")
@allure.story("查询资产：menuId查询")
def test_case_14(self):
self.get_base_token()
response = self.device_get(token=self.token, menuId=1)
self.check_code(response=response, code=0)
response = self.app.device_get(token=self.token, menuId=1)
self.app.check_code(response=response, code=0)
if __name__ == "__main__":
......
WorkCase/APP/Device/test_device_status.py
......@@ -7,6 +7,7 @@
from __future__ import division
from WorkCase import CaseBase
from WorkCase.APP import AppBase
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
from WorkUtils.UtilsResponse import UtilsResponse
......@@ -43,9 +44,9 @@ class TestDeviceStatus(object):
base_type_snmp = 127
base_ip = "172.16.3.170"
base_dns = "sanshi.zmops.com"
base_port = 999
base_port = 161
base_name = "SS测试资产-状态变更"
base_name = "SS 自动化专用"
base_business_id_1 = "12"
base_business_id_2 = "13"
......@@ -61,80 +62,17 @@ class TestDeviceStatus(object):
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
self.app = AppBase(host=self.host)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
self.token = self.app.get_base_token(login_name=self.login_name, base_password=self.base_password)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.update_members_status(members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("调用接口：api.login")
def api_login(self, name=None, password=None):
api = ApiLogin(_host=self.host)
api.name = name
api.password = password
api.get_response()
return api.response
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
return sql
@allure.step("更新表数据：members -- status")
def update_members_status(self, members_id=None, value=None):
session = self.db_session()
sql = DataMembers().update_status(session=session, _id=members_id, value=value)
return sql
@allure.step("调用接口：device.create")
def device_create(self, token=None, hostName=None, hostType=None, manageLevel=None, iplist=None, dns=None,
monitorInterface=None, monitorType=None,
parentHost=None, businessIds=None, opsPerson=None, snmpCommunity=None, ipmiAuthtype=None,
ipmiPrivilege=None, ipmiUsername=None, ipmiPassword=None,
factoryId=None, model=None, version=None, serialnumber=None,
description=None, monitorStatus=None):
api = DeviceCreate(_host=self.host)
api.token = token
api.hostName = hostName
api.hostType = hostType
api.manageLevel = manageLevel
api.iplist = iplist
api.dns = dns
api.monitorInterface = monitorInterface
api.monitorType = monitorType
api.parentHost = parentHost
api.businessIds = businessIds
api.opsPerson = opsPerson
api.snmpCommunity = snmpCommunity
api.ipmiAuthtype = ipmiAuthtype
api.ipmiPrivilege = ipmiPrivilege
api.ipmiUsername = ipmiUsername
api.ipmiPassword = ipmiPassword
api.factoryId = factoryId
api.model = model
api.version = version
api.serialnumber = serialnumber
api.description = description
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("调用接口：device.status")
def device_status(self, token=None, hostids=None, monitorStatus=None):
api = DeviceStatus(_host=self.host)
api.token = token
api.hostids = hostids
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["OFF"])
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["ON"])
@allure.step("连接数据库：Argus")
def db_session(self):
......@@ -166,122 +104,119 @@ class TestDeviceStatus(object):
assert x == 0
assert sql != []
@allure.step("断言返回结果")
def check_code(self, response, code):
_code = UtilsResponse().get_code(response=response)
assert _code == code
@allure.step("断言返回结果数据")
def check_msg(self, response, msg):
_msg = UtilsResponse().get_msg(response=response)
assert _msg == msg
@allure.step("创建测试数据")
def case_create(self, num):
name = self.base_name + num
response = self.device_create(token=self.token, hostName=name, hostType=self.base_type_agent, iplist=self.base_ip, monitorInterface=self.base_port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
self.device_id = UtilsResponse().get_data(response=response)
return self.device_id
@allure.title("device.status：token未传")
@allure.story("变更状态：token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.device_status()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token，请重新登录")
response = self.app.device_status()
self.app.check_code(response=response, code=2001)
self.app.check_msg(response=response, msg="无token，请重新登录")
@allure.title("device.status：token的用户已删除")
@allure.story("变更状态：token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.device_status(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
self.app.update_members_del_flag(session=self.db_session(), members_id=self.base_id, value=CaseBase().del_flag["ON"])
response = self.app.device_status(token=self.token)
self.app.check_code(response=response, code=2004)
self.app.check_msg(response=response, msg="用户不存在")
@allure.title("device.status：token的用户已禁用")
@allure.story("变更状态：token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.device_status(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
self.app.update_members_status(session=self.db_session(), members_id=self.base_id, value=CaseBase().status["OFF"])
response = self.app.device_status(token=self.token)
self.app.check_code(response=response, code=2003)
self.app.check_msg(response=response, msg="账户已被禁用")
@allure.title("device.status：变更一个monitorStatus=0")
@allure.story("变更状态：变更一个monitorStatus=0")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
self.case_create(num="04")
response = self.device_status(token=self.token, hostids=[self.device_id], monitorStatus=0)
self.check_code(response=response, code=0)
self.check_select(device_id=self.device_id, status=0)
name = self.base_name + "04"
host_id = self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
monitorStatus=0)
response = self.app.device_status(token=self.token, hostids=[host_id], monitorStatus=0)
self.app.check_code(response=response, code=0)
self.check_select(device_id=host_id, status=0)
@allure.title("device.status：变更一个monitorStatus=1")
@allure.story("变更状态：变更一个monitorStatus=1")
@allure.severity("blocker")
def test_case_05(self):
self.get_base_token()
self.case_create(num="05")
response = self.device_status(token=self.token, hostids=[self.device_id], monitorStatus=0)
self.check_code(response=response, code=0)
self.check_select(device_id=self.device_id, status=0)
response = self.device_status(token=self.token, hostids=[self.device_id], monitorStatus=1)
self.check_code(response=response, code=0)
self.check_select(device_id=self.device_id, status=1)
name = self.base_name + "05"
host_id = self.app.case_create_host(name=name, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
monitorStatus=1)
response = self.app.device_status(token=self.token, hostids=[host_id], monitorStatus=0)
self.app.check_code(response=response, code=0)
self.check_select(device_id=host_id, status=0)
response = self.app.device_status(token=self.token, hostids=[host_id], monitorStatus=1)
self.app.check_code(response=response, code=0)
self.check_select(device_id=host_id, status=1)
@allure.title("device.status：变更多个monitorStatus=0")
@allure.story("变更状态：变更多个monitorStatus=0")
@allure.severity("blocker")
def test_case_06(self):
self.get_base_token()
device_id_1 = self.case_create(num="061")
device_id_2 = self.case_create(num="062")
device_id_3 = self.case_create(num="063")
response = self.device_status(token=self.token, hostids=[device_id_1, device_id_2, device_id_3], monitorStatus=0)
self.check_code(response=response, code=0)
self.check_select(device_id=device_id_1, status=0)
self.check_select(device_id=device_id_2, status=0)
self.check_select(device_id=device_id_3, status=0)
name_1 = self.base_name + "061"
name_2 = self.base_name + "062"
name_3 = self.base_name + "063"
host_id_1 = self.app.case_create_host(name=name_1, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
monitorStatus=1)
host_id_2 = self.app.case_create_host(name=name_2, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
monitorStatus=1)
host_id_3 = self.app.case_create_host(name=name_3, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
monitorStatus=1)
response = self.app.device_status(token=self.token, hostids=[host_id_1, host_id_2, host_id_3], monitorStatus=0)
self.app.check_code(response=response, code=0)
self.check_select(device_id=host_id_1, status=0)
self.check_select(device_id=host_id_2, status=0)
self.check_select(device_id=host_id_3, status=0)
@allure.title("device.status：变更多个monitorStatus=1")
@allure.story("变更状态：变更多个monitorStatus=1")
@allure.severity("blocker")
def test_case_07(self):
self.get_base_token()
device_id_1 = self.case_create(num="071")
device_id_2 = self.case_create(num="072")
device_id_3 = self.case_create(num="073")
response = self.device_status(token=self.token, hostids=[device_id_1, device_id_2, device_id_3], monitorStatus=0)
self.check_code(response=response, code=0)
self.check_select(device_id=device_id_1, status=0)
self.check_select(device_id=device_id_2, status=0)
self.check_select(device_id=device_id_3, status=0)
device_id_1 = self.case_create(num="071")
device_id_2 = self.case_create(num="072")
device_id_3 = self.case_create(num="073")
response = self.device_status(token=self.token, hostids=[device_id_1, device_id_2, device_id_3], monitorStatus=1)
self.check_code(response=response, code=0)
self.check_select(device_id=device_id_1, status=1)
self.check_select(device_id=device_id_2, status=1)
self.check_select(device_id=device_id_3, status=1)
name_1 = self.base_name + "071"
name_2 = self.base_name + "072"
name_3 = self.base_name + "073"
host_id_1 = self.app.case_create_host(name=name_1, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
monitorStatus=1)
host_id_2 = self.app.case_create_host(name=name_2, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
monitorStatus=1)
host_id_3 = self.app.case_create_host(name=name_3, host_type=self.base_type_snmp,
iplist=self.base_ip, port=self.base_port,
businessIds=self.base_business_id_1 + "," + self.base_business_id_2,
monitorStatus=1)
response = self.app.device_status(token=self.token, hostids=[host_id_1, host_id_2, host_id_3], monitorStatus=0)
self.app.check_code(response=response, code=0)
self.check_select(device_id=host_id_1, status=0)
self.check_select(device_id=host_id_2, status=0)
self.check_select(device_id=host_id_3, status=0)
response = self.app.device_status(token=self.token, hostids=[host_id_1, host_id_2, host_id_3], monitorStatus=1)
self.app.check_code(response=response, code=0)
self.check_select(device_id=host_id_1, status=1)
self.check_select(device_id=host_id_2, status=1)
self.check_select(device_id=host_id_3, status=1)
if __name__ == "__main__":
......@@ -298,4 +233,5 @@ if __name__ == "__main__":
# a = TestDeviceStatus()
# a.setup_class()
# a.setup_method()
# a.test_case_04()
WorkCase/APP/__init__.py
......@@ -14,12 +14,27 @@ from WorkApi.APP.Businesses.businesses_delete import BusinessesDelete
from WorkApi.APP.Businesses.businesses_get import BusinessesGet
from WorkApi.APP.Businesses.businesses_tree import BusinessesTree
from WorkApi.APP.Businesses.businesses_update import BusinessesUpdate
from WorkApi.APP.Device.Asset.device_asset_getFactoryList import DeviceAssetGetFactoryList
from WorkApi.APP.Device.Asset.device_asset_ips import DeviceAssetIps
from WorkApi.APP.Device.device_create import DeviceCreate
import allure
from WorkApi.APP.Device.device_delete import DeviceDelete
from WorkApi.APP.Device.device_get import DeviceGet
from WorkApi.APP.Device.device_status import DeviceStatus
from WorkApi.APP.Experience.Items.experience_items_list import ExperienceItemsList
from WorkApi.APP.Experience.Problem.experience_problem_list import ExperienceProblemList
from WorkApi.APP.Experience.experience_create import ExperienceCreate
from WorkApi.APP.Experience.experience_delete import ExperienceDelete
from WorkApi.APP.Experience.experience_get import ExperienceGet
from WorkApi.APP.Experience.experience_update import ExperienceUpdate
from WorkApi.APP.File.file_upload import FileUpload
from WorkApi.APP.HostType.hostType_create import HostTypeCreate
from WorkApi.APP.HostType.hostType_delete import HostTypeDelete
from WorkApi.APP.HostType.hostType_get import HostTypeGet
from WorkApi.APP.HostType.hostType_list import HostTypeList
from WorkApi.APP.HostType.hostType_tree import HostTypeTree
from WorkApi.APP.HostType.hostType_update import HostTypeUpdate
from WorkApi.APP.Members.Get.members_get_role import MembersGetRole
from WorkApi.APP.Members.Update.members_update_info import MembersUpdateInfo
from WorkApi.APP.Members.Update.members_update_status import MembersUpdateStatus
......@@ -262,15 +277,50 @@ class AppBase(object):
api.get_response()
return api.response
@allure.step("调用接口：device.status")
def device_status(self, token=None, hostids=None, monitorStatus=None):
api = DeviceStatus(_host=self.host)
api.token = token
api.hostids = hostids
api.monitorStatus = monitorStatus
api.get_response()
return api.response
@allure.step("调用接口：device.asset.ips")
def device_asset_ips(self, token=None, ipName=None):
api = DeviceAssetIps(_host=self.host)
api.token = token
api.ipName = ipName
api.get_response()
return api.response
@allure.step("调用接口：device.asset.getFactoryList")
def device_asset_getFactoryList(self, token=None, hostType=None):
api = DeviceAssetGetFactoryList(_host=self.host)
api.token = token
api.hostType = hostType
api.get_response()
return api.response
@allure.step("创建测试数据-资产")
def case_create_host(self, name, host_type, iplist, port, businessIds):
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, monitorInterface=port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
def case_create_host(self, name, host_type, iplist, port, businessIds, monitorStatus=None):
if monitorStatus is not None:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, monitorInterface=port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=1)
else:
response = self.device_create(
token=self.token, hostName=name, hostType=host_type, iplist=iplist, monitorInterface=port,
monitorType=1, manageLevel=1, parentHost=1, businessIds=businessIds,
opsPerson=11, snmpCommunity="snmpCommunity",
ipmiAuthtype=1, ipmiPrivilege=1, ipmiUsername="ipmiUsername", ipmiPassword="ipmiPassword",
factoryId=1, model="model", version="version", serialnumber="serialnumber", description="description",
monitorStatus=monitorStatus)
self.check_code(response=response, code=0)
host_id = UtilsResponse().get_data(response=response)
return host_id
......@@ -325,6 +375,14 @@ class AppBase(object):
else:
assert y[_key] == _value
@allure.step("校验数据")
def check_keyValue(self, _json, _key, _value, isIn=None):
for x, y in enumerate(_json):
if isIn == 1:
assert _value in y[_key]
else:
assert y[_key] == _value
@allure.step("更新表数据：members -- del_flag")
def update_members_del_flag(self, session, members_id=None, value=None):
sql = DataMembers().update_del_flag(session=session, _id=members_id, value=value)
......@@ -558,3 +616,168 @@ class AppBase(object):
self.check_code(response=response, code=0)
rule_id = UtilsResponse().get_data(response=response)
return rule_id
@allure.step("调用接口：hosttype.create")
def hostType_create(self, token=None, title=None, parentId=None, templateIds=None, interfaces=None, description=None,
sortid=None):
api = HostTypeCreate(_host=self.host)
api.token = token
api.title = title
api.parentId = parentId
api.templateIds = templateIds
api.interfaces = interfaces
api.description = description
api.sortid = sortid
api.get_response()
return api.response
@allure.step("调用接口：hosttype.delete")
def hostType_delete(self, token=None, hosttypeids=None):
api = HostTypeDelete(_host=self.host)
api.token = token
api.hosttypeids = hosttypeids
api.get_response()
return api.response
@allure.step("调用接口：hosttype.get")
def hostType_get(self, token=None, title=None, parentId=None, description=None):
api = HostTypeGet(_host=self.host)
api.token = token
api.parentId = parentId
api.title = title
api.description = description
api.get_response()
return api.response
@allure.step("调用接口：hosttype.list")
def hostType_list(self, token=None, parentId=None):
api = HostTypeList(_host=self.host)
api.token = token
api.parentId = parentId
api.get_response()
return api.response
@allure.step("调用接口：hosttype.tree")
def hostType_tree(self, token=None):
api = HostTypeTree(_host=self.host)
api.token = token
api.get_response()
return api.response
@allure.step("调用接口：hosttype.update")
def hostType_update(self, token=None, _id=None, title=None, parentId=None, templateIds=None, interfaces=None, description=None,
sortid=None):
api = HostTypeUpdate(_host=self.host)
api.token = token
api.id = _id
api.title = title
api.parentId = parentId
api.templateIds = templateIds
api.interfaces = interfaces
api.description = description
api.sortid = sortid
api.get_response()
return api.response
@allure.step("创建测试数据")
def case_create_hostType(self, title, parentId, description):
response = self.hostType_create(token=self.token, title=title, parentId=parentId, description=description)
host_type_id = UtilsResponse().get_data(response=response)
return host_type_id
@allure.step("调用接口：experience.create")
def experience_create(self, token=None, title=None, content=None, businessids=None, hostTypes=None, hostids=None,
itemids=None, manageIpList=None, alarmRuleids=None, description=None, tagNameList=None, problemidList=None,
hostTypeTree=None):
api = ExperienceCreate(_host=self.host)
api.token = token
api.title = title
api.content = content
api.businessids = businessids
api.hostTypes = hostTypes
api.hostids = hostids
api.itemids = itemids
api.manageIpList = manageIpList
api.alarmRuleids = alarmRuleids
api.description = description
api.tagNameList = tagNameList
api.problemidList = problemidList
api.hostTypeTree = hostTypeTree
api.get_response()
return api.response
@allure.step("调用接口：experience.delete")
def experience_delete(self, token=None, experienceids=None):
api = ExperienceDelete(_host=self.host)
api.token = token
api.experienceids = experienceids
api.get_response()
return api.response
@allure.step("调用接口：experience.get")
def experience_get(self, token=None, hostType=None, hostName=None, manageLevel=None, manageIp=None, businessid=None, alarmRuleid=None,
itemid=None, title=None, tagid=None, tagName=None, description=None):
api = ExperienceGet(_host=self.host)
api.token = token
api.hostType = hostType
api.hostName = hostName
api.manageLevel = manageLevel
api.manageIp = manageIp
api.businessid = businessid
api.alarmRuleid = alarmRuleid
api.itemid = itemid
api.title = title
api.tagid = tagid
api.tagName = tagName
api.description = description
api.get_response()
return api.response
@allure.step("调用接口：experience.items.list")
def experience_items_list(self, token=None, itemName=None):
api = ExperienceItemsList(_host=self.host)
api.token = token
api.itemName = itemName
api.get_response()
return api.response
@allure.step("调用接口：experience.problem.list")
def experience_problem_list(self, token=None, problemName=None):
api = ExperienceProblemList(_host=self.host)
api.token = token
api.problemName = problemName
api.get_response()
return api.response
@allure.step("调用接口：experience.update")
def experience_update(self, token=None, _id=None, title=None, content=None, businessids=None, hostTypes=None,
hostids=None, itemids=None, manageIpList=None, alarmRuleids=None, description=None, tagNameList=None, problemidList=None,
hostTypeTree=None):
api = ExperienceUpdate(_host=self.host)
api.token = token
api.id = _id
api.title = title
api.content = content
api.businessids = businessids
api.hostTypes = hostTypes
api.hostids = hostids
api.itemids = itemids
api.manageIpList = manageIpList
api.alarmRuleids = alarmRuleids
api.description = description
api.tagNameList = tagNameList
api.problemidList = problemidList
api.hostTypeTree = hostTypeTree
api.get_response()
return api.response
@allure.step("创建测试数据-经验手册")
def case_create_experience(self, title, content, description, host_types, alarmRuleids=None, businessids=None,
hostids=None, host_ids=None, itemids=None, problemidLists=None, tagNameLists=None):
response = self.experience_create(token=self.token, title=title, content=content, hostTypes=host_types,
alarmRuleids=alarmRuleids, businessids=businessids, hostids=hostids,
manageIpList=host_ids, itemids=itemids, problemidList=problemidLists,
tagNameList=tagNameLists, description=description)
self.check_code(response=response, code=0)
experienceid = UtilsResponse().get_data(response=response)
return experienceid
WorkData/Argus/hosts.py
......@@ -158,3 +158,21 @@ class DataHosts(object):
self.log.error(error)
session.close()
return error
def select_all_like_host_name(self, session, host_name):
"""
:param session: 指针
:param host_name:
:return: 查询结果
"""
self.log.debug("查询数据库:")
try:
sql_rep = session.query(self.table).filter(self.table.host_name.like(str(host_name) + "%")).all()
self.log.debug(sql_rep)
session.close()
return sql_rep
except UtilsDataBase().errors as error:
self.log.error("异常:")
self.log.error(error)
session.close()
return error
\ No newline at end of file
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