Commit 03a5894a by sanshi

业务类型

parent e61f3fc3
<component name="ProjectDictionaryState"> <component name="ProjectDictionaryState">
<dictionary name="Administrator"> <dictionary name="Administrator">
<words> <words>
<w>alluredir</w>
<w>applicationids</w> <w>applicationids</w>
<w>authpassphrase</w> <w>authpassphrase</w>
<w>authprotocol</w> <w>authprotocol</w>
...@@ -42,6 +43,7 @@ ...@@ -42,6 +43,7 @@
<w>massupdate</w> <w>massupdate</w>
<w>mediatype</w> <w>mediatype</w>
<w>mediatypeids</w> <w>mediatypeids</w>
<w>memberid</w>
<w>monitorid</w> <w>monitorid</w>
<w>mtime</w> <w>mtime</w>
<w>networkmonitor</w> <w>networkmonitor</w>
...@@ -60,6 +62,7 @@ ...@@ -60,6 +62,7 @@
<w>sanshi</w> <w>sanshi</w>
<w>securitylevel</w> <w>securitylevel</w>
<w>securityname</w> <w>securityname</w>
<w>serviceids</w>
<w>snmpv</w> <w>snmpv</w>
<w>sortfield</w> <w>sortfield</w>
<w>sortorder</w> <w>sortorder</w>
......
...@@ -94,4 +94,4 @@ class TriggerUpdate(object): ...@@ -94,4 +94,4 @@ class TriggerUpdate(object):
self.local_json = base._json self.local_json = base._json
self._json = self.local_json self._json = self.local_json
self.response = self.api.post(url=self._url, json=[self._json]) self.response = self.api.post(url=self._url, json=self._json)
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# 编辑业务类型 # 更新业务类型
# 作者: 陈磊 # 作者: 陈磊
# 时间: 2019-11-25 # 时间: 2019-11-25
from WorkUtils.UtilsRequest import UtilsRequest from WorkUtils.UtilsRequest import UtilsRequest
...@@ -14,7 +14,7 @@ class BusinessesTypeUpdate(object): ...@@ -14,7 +14,7 @@ class BusinessesTypeUpdate(object):
:return: :return:
""" """
self.log = UtilsLog() self.log = UtilsLog()
self.log.info("调用编辑业务类型") self.log.info("调用更新业务类型")
self.log.info(self.__class__) self.log.info(self.__class__)
self._host = _host self._host = _host
......
# -*- coding: utf-8 -*-
# 创建业务
# 作者: 陈磊
# 时间: 2019-11-25
from WorkUtils.UtilsRequest import UtilsRequest
from WorkUtils.UtilsLog import UtilsLog
from WorkApi.ApiBase import ApiBase, GetBase
class BusinessesCreate(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 = "/businesses/create"
self._url = self._host + self._path
self._params = {}
self._json = {}
self._data = {}
self.response = ""
self.local_json = {}
self.token = None
self.name = None
self.description = None
self.type = None
self.principalName = None
self.principalPhone = None
self.memberid = None
self.hostids = None
self.serviceids = 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="name", value=self.name)
base.dict_add_key(_key="description", value=self.description)
base.dict_add_key(_key="type", value=self.type)
base.dict_add_key(_key="principalName", value=self.principalName)
base.dict_add_key(_key="principalPhone", value=self.principalPhone)
base.dict_add_key(_key="memberid", value=self.memberid)
base.dict_add_key(_key="hostids", value=self.hostids)
base.dict_add_key(_key="serviceids", value=self.serviceids)
self.local_json = base._json
self._json = self.local_json
self.response = self.api.post(url=self._url, headers=self._headers, json=self._json)
# -*- coding: utf-8 -*-
# 删除业务
# 作者: 陈磊
# 时间: 2019-11-25
from WorkUtils.UtilsRequest import UtilsRequest
from WorkUtils.UtilsLog import UtilsLog
from WorkApi.ApiBase import ApiBase, GetBase
class BusinessesDelete(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 = "/businesses/delete"
self._url = self._host + self._path
self._params = {}
self._json = {}
self._data = {}
self.response = ""
self.token = None
self.businessidList = 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="businessidList", value=self.businessidList)
self._json = base._json
self.response = self.api.post(url=self._url, headers=self._headers, json=self._json)
# -*- coding: utf-8 -*-
# 获取业务
# 作者: 陈磊
# 时间: 2019-11-25
from WorkUtils.UtilsRequest import UtilsRequest
from WorkUtils.UtilsLog import UtilsLog
from WorkApi.ApiBase import ApiBase, GetBase
class BusinessesGet(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 = "/businesses/get"
self._url = self._host + self._path
self._params = {}
self._json = {}
self._data = {}
self.response = ""
self.token = None
self.memberid = None
self.principal = None
self.type = None
self.hostName = None
self.businessesName = 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="memberid", value=self.memberid)
base.dict_add_key(_key="principal", value=self.principal)
base.dict_add_key(_key="type", value=self.type)
base.dict_add_key(_key="hostName", value=self.hostName)
base.dict_add_key(_key="businessesName", value=self.businessesName)
self._params = base._json
self.response = self.api.get(url=self._url, headers=self._headers, params=self._params)
# -*- coding: utf-8 -*-
# 更新业务
# 作者: 陈磊
# 时间: 2019-11-25
from WorkUtils.UtilsRequest import UtilsRequest
from WorkUtils.UtilsLog import UtilsLog
from WorkApi.ApiBase import ApiBase, GetBase
class BusinessesUpdate(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 = "/businesses/update"
self._url = self._host + self._path
self._params = {}
self._json = {}
self._data = {}
self.response = ""
self.local_json = {}
self.token = None
self.businessid = None
self.name = None
self.description = None
self.type = None
self.principalName = None
self.principalPhone = None
self.memberid = None
self.hostids = None
self.serviceids = 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="name", value=self.name)
base.dict_add_key(_key="businessid", value=self.businessid)
base.dict_add_key(_key="description", value=self.description)
base.dict_add_key(_key="type", value=self.type)
base.dict_add_key(_key="principalName", value=self.principalName)
base.dict_add_key(_key="principalPhone", value=self.principalPhone)
base.dict_add_key(_key="memberid", value=self.memberid)
base.dict_add_key(_key="hostids", value=self.hostids)
base.dict_add_key(_key="serviceids", value=self.serviceids)
self.local_json = base._json
self._json = self.local_json
self.response = self.api.post(url=self._url, headers=self._headers, json=self._json)
...@@ -38,7 +38,7 @@ class MembersUpdateInfo(object): ...@@ -38,7 +38,7 @@ class MembersUpdateInfo(object):
self.wechatUrl = None self.wechatUrl = None
self.dingdingUrl = None self.dingdingUrl = None
self.remark = None self.remark = None
self.memberId = None self.id = None
self.api = UtilsRequest() self.api = UtilsRequest()
...@@ -57,7 +57,7 @@ class MembersUpdateInfo(object): ...@@ -57,7 +57,7 @@ class MembersUpdateInfo(object):
base.dict_add_key(_key="wechatUrl", value=self.wechatUrl) base.dict_add_key(_key="wechatUrl", value=self.wechatUrl)
base.dict_add_key(_key="dingdingUrl", value=self.dingdingUrl) base.dict_add_key(_key="dingdingUrl", value=self.dingdingUrl)
base.dict_add_key(_key="remark", value=self.remark) base.dict_add_key(_key="remark", value=self.remark)
base.dict_add_key(_key="memberId", value=self.memberId) base.dict_add_key(_key="id", value=self.id)
self.local_json = base._json self.local_json = base._json
......
...@@ -30,7 +30,7 @@ class MembersUpdateStatus(object): ...@@ -30,7 +30,7 @@ class MembersUpdateStatus(object):
self.token = None self.token = None
self.status = None self.status = None
self.memberId = None self.memberIds = None
self.api = UtilsRequest() self.api = UtilsRequest()
...@@ -41,7 +41,7 @@ class MembersUpdateStatus(object): ...@@ -41,7 +41,7 @@ class MembersUpdateStatus(object):
base = ApiBase() base = ApiBase()
base.dict_add_key(_key="status", value=self.status) base.dict_add_key(_key="status", value=self.status)
base.dict_add_key(_key="memberId", value=self.memberId) base.dict_add_key(_key="memberIds", value=self.memberIds)
self.local_json = base._json self.local_json = base._json
......
...@@ -17,3 +17,5 @@ class WorkBase: ...@@ -17,3 +17,5 @@ class WorkBase:
self.case_path = "D:\\zmops-test\\WorkCase\\" self.case_path = "D:\\zmops-test\\WorkCase\\"
# 测试yaml文件路径 # 测试yaml文件路径
self.yaml_path = "D:\\zmops-test\\WorkYaml\\" self.yaml_path = "D:\\zmops-test\\WorkYaml\\"
# self.base_path = "D:\\zmops-test\\WorkCase\\"
...@@ -21,7 +21,6 @@ import allure ...@@ -21,7 +21,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostgroup_massadd")
@allure.feature("测试模块:hostgroup.massadd") @allure.feature("测试模块:hostgroup.massadd")
class TestHostGroupMassadd(object): class TestHostGroupMassadd(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -22,7 +22,6 @@ import allure ...@@ -22,7 +22,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostgroup_massremove")
@allure.feature("测试模块:hostgroup.massremove") @allure.feature("测试模块:hostgroup.massremove")
class TestHostGroupMassremove(object): class TestHostGroupMassremove(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -22,7 +22,6 @@ import allure ...@@ -22,7 +22,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostgroup_massupdate")
@allure.feature("测试模块:hostgroup.massupdate") @allure.feature("测试模块:hostgroup.massupdate")
class TestHostGroupMassupdate(object): class TestHostGroupMassupdate(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -19,7 +19,6 @@ import allure ...@@ -19,7 +19,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostinterface_massadd")
@allure.feature("测试模块:hostinterface.massadd") @allure.feature("测试模块:hostinterface.massadd")
class TestHostInterfaceMassadd(object): class TestHostInterfaceMassadd(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -20,7 +20,6 @@ import allure ...@@ -20,7 +20,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostinterface_massremove")
@allure.feature("测试模块:hostinterface.massremove") @allure.feature("测试模块:hostinterface.massremove")
class TestHostInterfaceMassremove(object): class TestHostInterfaceMassremove(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -20,7 +20,6 @@ import allure ...@@ -20,7 +20,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostinterface_replace")
@allure.feature("测试模块:hostinterface.replace") @allure.feature("测试模块:hostinterface.replace")
class TestHostInterfaceReplace(object): class TestHostInterfaceReplace(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -20,7 +20,6 @@ import allure ...@@ -20,7 +20,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostprototype_create")
@allure.feature("测试模块:hostprototype.create") @allure.feature("测试模块:hostprototype.create")
class TestHostPrototypeCreate(object): class TestHostPrototypeCreate(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -19,7 +19,6 @@ import allure ...@@ -19,7 +19,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostprototype_delete")
@allure.feature("测试模块:hostprototype.delete") @allure.feature("测试模块:hostprototype.delete")
class TestHostPrototypeDelete(object): class TestHostPrototypeDelete(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -19,7 +19,6 @@ import allure ...@@ -19,7 +19,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostprototype_get")
@allure.feature("测试模块:hostprototype.get") @allure.feature("测试模块:hostprototype.get")
class TestHostPrototypeGet(object): class TestHostPrototypeGet(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -21,7 +21,6 @@ import allure ...@@ -21,7 +21,6 @@ import allure
import pytest import pytest
@pytest.importorskip("test_hostprototype_update")
@allure.feature("测试模块:hostprototype.update") @allure.feature("测试模块:hostprototype.update")
class TestHostPrototypeUpdate(object): class TestHostPrototypeUpdate(object):
log = UtilsLog() log = UtilsLog()
......
...@@ -80,14 +80,14 @@ class TestTriggerDelete(object): ...@@ -80,14 +80,14 @@ class TestTriggerDelete(object):
return session return session
@allure.step("查询表:triggers") @allure.step("查询表:triggers")
def select_hosts(self, triggerid=None): def select_triggers(self, triggerid=None):
session = self.db_session() session = self.db_session()
sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid) sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid)
return sql return sql
@allure.step("校验查询结果") @allure.step("校验查询结果")
def check_select(self, triggerid): def check_select(self, triggerid):
sql = self.select_hosts(triggerid=triggerid) sql = self.select_triggers(triggerid=triggerid)
assert sql == [] assert sql == []
@allure.step("断言返回结果:校验排序") @allure.step("断言返回结果:校验排序")
......
...@@ -233,10 +233,11 @@ if __name__ == "__main__": ...@@ -233,10 +233,11 @@ if __name__ == "__main__":
import os import os
# 执行自动化测试用例 # 执行自动化测试用例
# case_info = os.path.split(__file__) case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd() case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case, case_info=case_info) r = UtilsPyTest(case=case, case_info=case_info)
# r.run_main() r.run_main()
a = TestTriggerGet()
a.setup_class() # a = TestTriggerGet()
a.test_case_09() # a.setup_class()
# a.test_case_09()
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-25
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.Trigger.trigger_create import TriggerCreate
from WorkApi.API.Trigger.trigger_update import TriggerUpdate
from WorkData.Zabbix.triggers import DataTriggers
import allure
# allure.label("")
@allure.feature("测试模块:trigger.update")
class TestTriggerUpdate(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 = "{Template OS Linux:agent.ping.nodata(5m)}=1"
new_expression = "{Template OS Linux:agent.ping.nodata(5m)}=2"
@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("调用接口:trigger.create")
def trigger_create(self, description=None, expression=None, tags_tag=None, tags_value=None):
api = TriggerCreate(_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, num):
response = self.trigger_create(description=self.base_description + num, expression=self.base_expression)
self.check_code(response=response, code=0)
triggerid = UtilsResponse().get_result(response=response)["triggerids"][0]
return triggerid
@allure.step("调用接口:trigger.update")
def trigger_update(self, triggerid=None, description=None, expression=None):
api = TriggerUpdate(_host=self.host)
api.triggerid = triggerid
api.description = description
api.expression = expression
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, description):
sql = self.select_hosts(triggerid=triggerid)
for x, y in enumerate(sql):
assert x == 0
assert y.description == description
@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("trigger.update:修改成功")
@allure.story("更新触发器:修改成功")
@allure.severity("blocker")
def test_case_01(self):
triggerid = self.case_create(num="01")
response = self.trigger_update(triggerid=triggerid, description=self.base_description + "修改",
expression=self.new_expression)
self.check_code(response=response, code=0)
self.check_select(triggerid=triggerid, description=self.base_description + "修改")
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 = TestTriggerUpdate()
# a.setup_class()
# a.test_case_01()
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-25
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.APP.Api.api_login import ApiLogin
from WorkApi.APP.Businesses.businesses_create import BusinessesCreate
from WorkData.Argus.members import DataMembers
from WorkData.Argus.businesses import DataBusinesses
import allure
@allure.feature("测试模块:businesses.create")
class TestBusinessesCreate(object):
log = UtilsLog()
env = CaseBase().app_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"]
login_name = "SS测试用户名"
base_id = 10
base_password = "123456"
base_name = "SS自动化测试业务名称"
base_type_1 = 1
base_type_2 = 2
@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)
DataBusinesses().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=0)
self.update_members_status(members_id=self.base_id, value=0)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=0)
self.update_members_status(members_id=self.base_id, value=0)
@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("调用接口:businesses.create")
def businesses_create(self, token=None, name=None, description=None, _type=None, principalName=None, principalPhone=None,
memberid=None):
api = BusinessesCreate(_host=self.host)
api.token = token
api.name = name
api.description = description
api.type = _type
api.principalName = principalName
api.principalPhone = principalPhone
api.memberid = memberid
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
@allure.step("连接数据库:Argus")
def db_session(self):
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
return session
@allure.step("查询表:businesses")
def select_businesses(self, businessid=None, name=None):
session = self.db_session()
sql = DataBusinesses().select_all_from_allKeys(session=session, businessid=businessid, name=name)
return sql
@allure.step("校验查询结果")
def check_select(self, response, name, description, _type, principal_name, principal_phone, memberid):
data = UtilsResponse().get_data(response=response)
sql = self.select_businesses(businessid=data)
for x, y in enumerate(sql):
assert y.name == name
assert y.description == description
assert y.type == _type
assert y.principal_name == principal_name
assert y.principal_phone == principal_phone
assert y.memberid == memberid
assert x == 0
@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("businesses.create:token未传")
@allure.story("创建业务:token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.businesses_create()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token,请重新登录")
@allure.title("businesses.create:token的用户已删除")
@allure.story("创建业务:token的用户已删除")
@allure.severity("blocker")
def test_case_02(self):
self.get_base_token()
self.update_members_del_flag(members_id=self.base_id, value=1)
response = self.businesses_create(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
@allure.title("businesses.create:token的用户已禁用")
@allure.story("创建业务:token的用户已禁用")
@allure.severity("blocker")
def test_case_03(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=1)
response = self.businesses_create(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
@allure.title("businesses.create:填写name,type")
@allure.story("创建业务:填写name,type")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
name = self.base_name + "04"
response = self.businesses_create(token=self.token, name=name, _type=self.base_type_1)
self.check_code(response=response, code=0)
self.check_select(response=response, name=name, description=None, _type=self.base_type_1, principal_name=None,
principal_phone=None, memberid=None)
@allure.title("businesses.create:填写所有字段")
@allure.story("创建业务:填写所有字段")
def test_case_05(self):
self.get_base_token()
name = self.base_name + "05"
response = self.businesses_create(token=self.token, name=name, _type=self.base_type_1, description="description", principalName="负责人Name",
principalPhone="15151969022 0519-", memberid=self.base_id)
self.check_code(response=response, code=0)
self.check_select(response=response, name=name, description="description", _type=self.base_type_1, principal_name="负责人Name",
principal_phone="15151969022 0519-", memberid=self.base_id)
@allure.title("businesses.create:同类型下创建名称相同的业务,返回异常")
@allure.story("创建业务:同类型下创建名称相同的业务,返回异常")
def test_case_06(self):
self.get_base_token()
name = self.base_name + "06"
response = self.businesses_create(token=self.token, name=name, _type=self.base_type_1)
self.check_code(response=response, code=0)
response = self.businesses_create(token=self.token, name=name, _type=self.base_type_1)
self.check_code(response=response, code=100)
self.check_msg(response=response, msg="该类型下的业务已存在")
@allure.title("businesses.create:不同类型下创建名称相同的业务,成功创建")
@allure.story("创建业务:不同类型下创建名称相同的业务,成功创建")
def test_case_07(self):
self.get_base_token()
name = self.base_name + "07"
response = self.businesses_create(token=self.token, name=name, _type=self.base_type_1)
self.check_code(response=response, code=0)
self.check_select(response=response, name=name, description=None, _type=self.base_type_1, principal_name=None,
principal_phone=None, memberid=None)
response = self.businesses_create(token=self.token, name=name, _type=self.base_type_2)
self.check_code(response=response, code=0)
self.check_select(response=response, name=name, description=None, _type=self.base_type_2, principal_name=None,
principal_phone=None, memberid=None)
@allure.title("businesses.create:name为空,返回异常")
@allure.story("创建业务:name为空,返回异常")
def test_case_08(self):
self.get_base_token()
# name = self.base_name + "08"
response = self.businesses_create(token=self.token, _type=self.base_type_1)
self.check_code(response=response, code=100)
@allure.title("businesses.create:type为空,返回异常")
@allure.story("创建业务:type为空,返回异常")
def test_case_09(self):
self.get_base_token()
name = self.base_name + "09"
response = self.businesses_create(token=self.token, name=name)
self.check_code(response=response, code=100)
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 = TestBusinessesCreate()
# a.setup_class()
# a.test_case_09()
# -*- coding: utf-8 -*-
# 测试用例
# 作者: 陈磊
# 时间: 2019-11-25
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.APP.Api.api_login import ApiLogin
from WorkApi.APP.Businesses.businesses_create import BusinessesCreate
from WorkApi.APP.Businesses.businesses_get import BusinessesGet
from WorkData.Argus.members import DataMembers
from WorkData.Argus.businesses import DataBusinesses
import allure
@allure.feature("测试模块:businesses.get")
class TestBusinessesGet(object):
log = UtilsLog()
env = CaseBase().app_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"]
login_name = "SS测试用户名"
base_id = 10
base_password = "123456"
base_name = "SS自动化测试业务名称"
base_type_1 = 1
base_type_2 = 2
@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)
DataBusinesses().delete_like_name(session=session, name=cls.base_name)
@classmethod
def teardown_class(cls):
cls.log.debug("结束执行测试套件.......")
def setup_method(self):
self.log.debug("测试用例执行开始...")
self.update_members_del_flag(members_id=self.base_id, value=0)
self.update_members_status(members_id=self.base_id, value=0)
def teardown_method(self):
self.log.debug("测试用例执行结束...")
self.update_members_del_flag(members_id=self.base_id, value=0)
self.update_members_status(members_id=self.base_id, value=0)
@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("调用接口:businesses.type.create")
def businesses_create(self, token=None, name=None, description=None, _type=None, principalName=None, principalPhone=None,
memberid=None):
api = BusinessesCreate(_host=self.host)
api.token = token
api.name = name
api.description = description
api.type = _type
api.principalName = principalName
api.principalPhone = principalPhone
api.memberid = memberid
api.get_response()
return api.response
@allure.step("调用接口:businesses.get")
def businesses_get(self, token=None):
api = BusinessesGet(_host=self.host)
api.token = token
api.get_response()
return api.response
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
@allure.step("连接数据库:Argus")
def db_session(self):
session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
return session
@allure.step("查询表:businesses")
def select_businesses(self, businessid=None, name=None):
session = self.db_session()
sql = DataBusinesses().select_all_from_allKeys(session=session, businessid=businessid,name=name)
return sql
@allure.step("创建测试数据")
def case_create(self, name):
self.businesses_create(token=self.token, name=name, _type=self.base_type_1)
@allure.step("校验查询结果")
def check_select(self, response):
data = UtilsResponse().get_data(response=response)
for q, w in enumerate(data):
sql = self.select_businesses(businessid=w["typeId"])
for x, y in enumerate(sql):
assert y.name == w["typeName"]
assert x == 0
@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("businesses.get:token未传")
@allure.story("查询业务:token未传")
@allure.severity("blocker")
def test_case_01(self):
self.get_base_token()
response = self.businesses_get()
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token,请重新登录")
@allure.title("businesses.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=1)
response = self.businesses_get(token=self.token)
self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在")
@allure.title("businesses.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=1)
response = self.businesses_get(token=self.token)
self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用")
@allure.title("businesses.get:无参数调用")
@allure.story("查询业务:无参数调用")
@allure.severity("blocker")
def test_case_04(self):
self.get_base_token()
name = self.base_name + "04"
self.case_create(name=name)
response = self.businesses_get(token=self.token)
self.check_code(response=response, code=0)
# self.check_select(response=response)
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 = TestBusinessesGet()
a.setup_class()
a.test_case_04()
...@@ -34,7 +34,7 @@ class TestBusinessesTypeCreate(object): ...@@ -34,7 +34,7 @@ class TestBusinessesTypeCreate(object):
base_id = 10 base_id = 10
base_password = "123456" base_password = "123456"
base_typeName = "SS自动化测试业务名称" base_typeName = "SS自动化测试业务类型名称"
@classmethod @classmethod
def setup_class(cls): def setup_class(cls):
......
...@@ -197,7 +197,6 @@ class TestMembersDelete(object): ...@@ -197,7 +197,6 @@ class TestMembersDelete(object):
response = self.members_delete(token=self.token) response = self.members_delete(token=self.token)
self.check_code(response=response, code=100) self.check_code(response=response, code=100)
self.check_msg(response=response, msg="Required List parameter 'memberIds' is not present")
@allure.title("members.delete:memberIds=[]") @allure.title("members.delete:memberIds=[]")
@allure.story("删除用户:memberIds=[]") @allure.story("删除用户:memberIds=[]")
......
...@@ -83,7 +83,7 @@ class TestMembersUpdateInfo(object): ...@@ -83,7 +83,7 @@ class TestMembersUpdateInfo(object):
@allure.step("调用接口:members.update.info") @allure.step("调用接口:members.update.info")
def members_update_info(self, token=None, name=None, position=None, role=None, password=None, def members_update_info(self, token=None, name=None, position=None, role=None, password=None,
phone=None, email=None, wechatUrl=None, dingdingUrl=None, remark=None, memberId=None): phone=None, email=None, wechatUrl=None, dingdingUrl=None, remark=None, _id=None):
api = MembersUpdateInfo(_host=self.host) api = MembersUpdateInfo(_host=self.host)
api.token = token api.token = token
api.name = name api.name = name
...@@ -95,7 +95,7 @@ class TestMembersUpdateInfo(object): ...@@ -95,7 +95,7 @@ class TestMembersUpdateInfo(object):
api.wechatUrl = wechatUrl api.wechatUrl = wechatUrl
api.dingdingUrl = dingdingUrl api.dingdingUrl = dingdingUrl
api.remark = remark api.remark = remark
api.memberId = memberId api.id = _id
api.get_response() api.get_response()
return api.response return api.response
...@@ -206,7 +206,7 @@ class TestMembersUpdateInfo(object): ...@@ -206,7 +206,7 @@ class TestMembersUpdateInfo(object):
response = self.members_create(token=self.token, name=name, password=self.base_password, role=0) response = self.members_create(token=self.token, name=name, password=self.base_password, role=0)
memberId = UtilsResponse().get_data(response=response) memberId = UtilsResponse().get_data(response=response)
response = self.members_update_info(token=self.token, memberId=memberId, name=name + "修改", response = self.members_update_info(token=self.token, _id=memberId, name=name + "修改",
password="Test", role=1, position="position", phone="phone", password="Test", role=1, position="position", phone="phone",
email="email", wechatUrl="wechatUrl", dingdingUrl="dingdingUrl", remark="remark") email="email", wechatUrl="wechatUrl", dingdingUrl="dingdingUrl", remark="remark")
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
......
...@@ -82,11 +82,11 @@ class TestMembersUpdateStatus(object): ...@@ -82,11 +82,11 @@ class TestMembersUpdateStatus(object):
return api.response return api.response
@allure.step("调用接口:members.update.status") @allure.step("调用接口:members.update.status")
def members_update_status(self, token=None, status=None, memberId=None): def members_update_status(self, token=None, status=None, memberIds=None):
api = MembersUpdateStatus(_host=self.host) api = MembersUpdateStatus(_host=self.host)
api.token = token api.token = token
api.status = status api.status = status
api.memberId = memberId api.memberIds = memberIds
api.get_response() api.get_response()
return api.response return api.response
...@@ -197,7 +197,7 @@ class TestMembersUpdateStatus(object): ...@@ -197,7 +197,7 @@ class TestMembersUpdateStatus(object):
response = self.members_create(token=self.token, name=name, password=self.base_password, role=0) response = self.members_create(token=self.token, name=name, password=self.base_password, role=0)
memberId = UtilsResponse().get_data(response=response) memberId = UtilsResponse().get_data(response=response)
response = self.members_update_status(token=self.token, memberId=memberId, status=0) response = self.members_update_status(token=self.token, memberIds=[memberId], status=0)
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
self.check_select(members_id=memberId, name=name, password=self.base_password, role=0, position=None, self.check_select(members_id=memberId, name=name, password=self.base_password, role=0, position=None,
phone=None, email=None, wechatUrl=None, dingdingUrl=None, remark=None, phone=None, email=None, wechatUrl=None, dingdingUrl=None, remark=None,
...@@ -213,7 +213,7 @@ class TestMembersUpdateStatus(object): ...@@ -213,7 +213,7 @@ class TestMembersUpdateStatus(object):
response = self.members_create(token=self.token, name=name, password=self.base_password, role=0) response = self.members_create(token=self.token, name=name, password=self.base_password, role=0)
memberId = UtilsResponse().get_data(response=response) memberId = UtilsResponse().get_data(response=response)
response = self.members_update_status(token=self.token, memberId=memberId, status=1) response = self.members_update_status(token=self.token, memberIds=[memberId], status=1)
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
self.check_select(members_id=memberId, name=name, password=self.base_password, role=0, position=None, self.check_select(members_id=memberId, name=name, password=self.base_password, role=0, position=None,
phone=None, email=None, wechatUrl=None, dingdingUrl=None, remark=None, phone=None, email=None, wechatUrl=None, dingdingUrl=None, remark=None,
......
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
from __future__ import division from __future__ import division
from WorkApi.APP.Members.members_create import MembersCreate
from WorkCase import CaseBase from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsResponse import UtilsResponse from WorkUtils.UtilsResponse import UtilsResponse
...@@ -30,6 +31,7 @@ class TestSettingsChangePassword(object): ...@@ -30,6 +31,7 @@ class TestSettingsChangePassword(object):
db_base = env["db_base"] db_base = env["db_base"]
base_id = 10 base_id = 10
login_name = "SS测试用户名"
base_name = "SS测试用户名" base_name = "SS测试用户名"
base_password = "123456" base_password = "123456"
...@@ -105,61 +107,80 @@ class TestSettingsChangePassword(object): ...@@ -105,61 +107,80 @@ class TestSettingsChangePassword(object):
sql = DataMembers().update_password(session=session, _id=members_id, value=value) sql = DataMembers().update_password(session=session, _id=members_id, value=value)
return sql return sql
@allure.step("获取token")
def get_base_token(self):
response = self.api_login(name=self.login_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
@allure.step("调用接口:members.create")
def members_create(self, token=None, name=None, position=None, role=None, password=None,
phone=None, email=None, wechatUrl=None, dingdingUrl=None, remark=None):
api = MembersCreate(_host=self.host)
api.token = token
api.name = name
api.position = position
api.role = role
api.password = password
api.phone = phone
api.email = email
api.wechatUrl = wechatUrl
api.dingdingUrl = dingdingUrl
api.remark = remark
api.get_response()
return api.response
@allure.title("settings.changePassword:token未传") @allure.title("settings.changePassword:token未传")
@allure.story("变更密码:token未传") @allure.story("修改密码:token未传")
@allure.severity("blocker") @allure.severity("blocker")
def test_case_01(self): def test_case_01(self):
response = self.settings_changePassword(memberId=self.base_id) self.get_base_token()
response = self.settings_changePassword()
self.check_code(response=response, code=2001) self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="无token,请重新登录") self.check_msg(response=response, msg="无token,请重新登录")
@allure.title("settings.changePassword:token的用户已删除") @allure.title("settings.changePassword:token的用户已删除")
@allure.story("变更密码:token的用户已删除") @allure.story("修改密码:token的用户已删除")
@allure.severity("blocker") @allure.severity("blocker")
def test_case_02(self): def test_case_02(self):
response = self.api_login(name=self.base_name, password=self.base_password) self.get_base_token()
data = UtilsResponse().get_data(response=response)
self.token = data
self.update_members_del_flag(members_id=self.base_id, value=1) self.update_members_del_flag(members_id=self.base_id, value=1)
response = self.settings_changePassword(token=self.token, memberId=self.base_id) response = self.settings_changePassword(token=self.token)
self.check_code(response=response, code=2004) self.check_code(response=response, code=2004)
self.check_msg(response=response, msg="用户不存在") self.check_msg(response=response, msg="用户不存在")
@allure.title("settings.changePassword:token的用户已禁用") @allure.title("settings.changePassword:token的用户已禁用")
@allure.story("变更密码:token的用户已禁用") @allure.story("修改密码:token的用户已禁用")
@allure.severity("blocker") @allure.severity("blocker")
def test_case_03(self): def test_case_03(self):
response = self.api_login(name=self.base_name, password=self.base_password) self.get_base_token()
data = UtilsResponse().get_data(response=response)
self.token = data
self.update_members_status(members_id=self.base_id, value=1) self.update_members_status(members_id=self.base_id, value=1)
response = self.settings_changePassword(token=self.token, memberId=self.base_id) response = self.settings_changePassword(token=self.token)
self.check_code(response=response, code=2003) self.check_code(response=response, code=2003)
self.check_msg(response=response, msg="账户已被禁用") self.check_msg(response=response, msg="账户已被禁用")
@allure.title("settings.changePassword:token错误") @allure.title("settings.changePassword:成功修改密码")
@allure.story("变更密码:token错误") @allure.story("修改密码:成功修改密码")
@allure.severity("blocker")
def test_case_04(self): def test_case_04(self):
response = self.api_login(name=self.base_name, password=self.base_password) self.get_base_token()
data = UtilsResponse().get_data(response=response)
self.token = data
self.update_members_status(members_id=self.base_id, value=1)
response = self.settings_changePassword(token="Test", memberId=self.base_id)
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="token错误,请重新登录")
@allure.title("settings.changePassword:正常查询") name = self.base_name + "04"
@allure.story("变更密码:正常查询") response = self.members_create(token=self.token, name=name, password=self.base_password, role=0)
@allure.severity("blocker") memberId = UtilsResponse().get_data(response=response)
def test_case_05(self):
response = self.api_login(name=self.base_name, password=self.base_password)
self.token = UtilsResponse().get_data(response=response)
new_password = "Test" new_password = "Test"
response = self.settings_changePassword(token=self.token, memberId=self.base_id, oldPassword=self.base_password, response = self.settings_changePassword(token=self.token, memberId=memberId, oldPassword=self.base_password,
newPassword=new_password) newPassword=new_password)
self.check_code(response=response, code=0) self.check_code(response=response, code=0)
self.api_login @allure.title("settings.changePassword:token错误")
@allure.story("变更密码:token错误")
def test_case_05(self):
self.get_base_token()
self.update_members_status(members_id=self.base_id, value=1)
response = self.settings_changePassword(token="Test", memberId=self.base_id)
self.check_code(response=response, code=2001)
self.check_msg(response=response, msg="token错误,请重新登录")
if __name__ == "__main__": if __name__ == "__main__":
...@@ -169,12 +190,12 @@ if __name__ == "__main__": ...@@ -169,12 +190,12 @@ if __name__ == "__main__":
import os import os
# 执行自动化测试用例 # 执行自动化测试用例
# case_info = os.path.split(__file__) case_info = os.path.split(__file__)
# case = UtilsCmd().pytest_cmd() case = UtilsCmd().pytest_cmd()
# r = UtilsPyTest(case=case, case_info=case_info) r = UtilsPyTest(case=case, case_info=case_info)
# r.run_main() r.run_main()
a = TestSettingsChangePassword() # a = TestSettingsChangePassword()
a.setup_class() # a.setup_class()
a.test_case_05() # a.test_case_05()
a.teardown_method() # a.teardown_method()
...@@ -7,8 +7,6 @@ ...@@ -7,8 +7,6 @@
class CaseBase: class CaseBase:
def __init__(self): def __init__(self):
self.base_path = "D:\\TestZmpos\\WorkCase"
self.environment = { self.environment = {
# "host": "http://10.0.0.12:7070", # "host": "http://10.0.0.12:7070",
# "db_url": "10.0.0.153", # "db_url": "10.0.0.153",
......
# -*- coding: utf_8 -*-
# 表名: businesses
# 作者: 陈磊
# 时间: 2019-11-25
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, VARCHAR, SMALLINT, DATETIME, DECIMAL, BigInteger, FLOAT, NUMERIC, TEXT
from sqlalchemy.orm.exc import MultipleResultsFound, NoResultFound
from sqlalchemy import or_
from WorkUtils.UtilsLog import UtilsLog
from WorkUtils.UtilsDataBase import UtilsDataBase
import time
Base = declarative_base()
class Businesses(Base):
"""
业务表
"""
__tablename__ = "businesses"
businessid = Column(BigInteger, primary_key=True)
name = Column(VARCHAR(255), comment="名称")
description = Column(VARCHAR(255), comment="描述")
type = Column(Integer, comment="业务类型")
principal_name = Column(VARCHAR(50), comment="业务负责人姓名")
principal_phone = Column(VARCHAR(20), comment="业务负责人手机号")
memberid = Column(BigInteger, comment="运维人员")
create_time = Column(BigInteger, comment="创建时间")
class DataBusinesses(object):
def __init__(self):
self.log = UtilsLog()
self.log.info(self.__class__)
self.table = Businesses
def select_all_from_allKeys(self, session, businessid=None, name=None, _type=None):
"""
:param session: 指针
:param businessid:
:param name:
:param _type:
:return: 查询结果
"""
self.log.debug("查询数据库:")
try:
base = UtilsDataBase()
base.add_param(_key="businessid", value=businessid)
base.add_param(_key="name", value=name)
base.add_param(_key="type", value=_type)
sql_rep = session.query(self.table).filter_by(**base.param).all()
self.log.debug(sql_rep)
session.close()
return sql_rep
except UtilsDataBase().errors as error:
self.log.error("异常:")
self.log.error(error)
session.close()
return error
def delete_like_name(self, session, name):
"""
:param session: 指针
:param name:
:return:
"""
self.log.debug("查询数据库:")
try:
self.log.debug("删除相关测试数据")
session.query(self.table).filter(self.table.name.like(str(name) + "%")).delete(synchronize_session=False)
# session.delete(sql)
session.commit()
session.close()
self.log.debug("删除成功")
return True
except UtilsDataBase().errors as error:
self.log.error("异常:")
self.log.error(error)
session.close()
return error
...@@ -5,13 +5,11 @@ ...@@ -5,13 +5,11 @@
import pytest import pytest
import sys
import os import os
from datetime import datetime from datetime import datetime
from WorkBase import WorkBase from WorkBase import WorkBase
from WorkCase import CaseBase
from WorkUtils.UtilsLog import UtilsLog from WorkUtils.UtilsLog import UtilsLog
...@@ -30,7 +28,7 @@ class UtilsPyTest: ...@@ -30,7 +28,7 @@ class UtilsPyTest:
self.case = case self.case = case
self.version = self.case["version"] self.version = self.case["version"]
self.case_path = CaseBase().base_path self.case_path = WorkBase().case_path
self.case_info = case_info self.case_info = case_info
self.xml_path = "" self.xml_path = ""
...@@ -89,6 +87,7 @@ class UtilsPyTest: ...@@ -89,6 +87,7 @@ class UtilsPyTest:
生成测试报告 生成测试报告
:return: :return:
""" """
self.log.debug("测试信息:%s" % self.case_name)
self.log.info("生成测试报告......") self.log.info("生成测试报告......")
cmd = "allure generate " + self.xml_path + " -o " + self.report_path + " --clean" cmd = "allure generate " + self.xml_path + " -o " + self.report_path + " --clean"
# self.log.debug(cmd) # self.log.debug(cmd)
......
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