触发器相关

.idea/dictionaries/Administrator.xml

@@ -20,6 +20,7 @@
       <w>objectid</w>
       <w>objectids</w>
       <w>preservekeys</w>
+      <w>pytest</w>
       <w>ruleid</w>
       <w>ruleids</w>
       <w>sanshi</w>

WorkCase/API/HostInterface/test_hostinterface_get.py

@@ -202,14 +202,14 @@ class TestHostInterfaceGet(object):
 
     @allure.title("hostinterface.get：sortfield=interfaceid, sortorder=ASC")
     @allure.story("查询主机接口：sortfield=interfaceid, sortorder=ASC")
-    def test_case_8(self):
+    def test_case_08(self):
         response = self.hostinterface_get(sortfield=["interfaceid"], sortorder=["ASC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="ASC")
 
     @allure.title("hostinterface.get：sortfield=interfaceid, sortorder=DESC")
     @allure.story("查询主机接口：sortfield=interfaceid, sortorder=DESC")
-    def test_case_9(self):
+    def test_case_09(self):
         response = self.hostinterface_get(sortfield=["interfaceid"], sortorder=["DESC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="DESC")

WorkCase/API/HostPrototype/test_hostprototype_get.py

@@ -211,14 +211,14 @@ class TestHostPrototypeGet(object):
 
     @allure.title("hostprototype.get：sortfield=hostid, sortorder=ASC")
     @allure.story("查询主机原型：sortfield=hostid, sortorder=ASC")
-    def test_case_8(self):
+    def test_case_08(self):
         response = self.hostprototype_get(sortfield=["hostid"], sortorder=["ASC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="ASC")
 
     @allure.title("hostprototype.get：sortfield=hostid, sortorder=DESC")
     @allure.story("查询主机原型：sortfield=hostid, sortorder=DESC")
-    def test_case_9(self):
+    def test_case_09(self):
         response = self.hostprototype_get(sortfield=["hostid"], sortorder=["DESC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="DESC")

WorkCase/API/Problem/test_problem_get.py

@@ -153,35 +153,35 @@ class TestProblemGet(object):
 
     @allure.title("problem.get：output=[source, object, objectid, name, severity]")
     @allure.story("查询触发器原型：output=[source, object, objectid, name, severity]")
-    def test_case_05(self):
+    def test_case_04(self):
         response = self.problem_get(output=["source", "object", "objectid", "name", "severity"])
         self.check_code(response=response, code=0)
         self.check_sql(response=response)
 
     @allure.title("problem.get：sortfield=eventid")
     @allure.story("查询告警：sortfield=eventid")
-    def test_case_07(self):
+    def test_case_05(self):
         response = self.problem_get(sortfield=["eventid"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible=True)
 
     @allure.title("problem.get：sortfield=eventid, sortorder=ASC")
     @allure.story("查询告警：sortfield=eventid, sortorder=ASC")
-    def test_case_8(self):
+    def test_case_06(self):
         response = self.problem_get(sortfield=["eventid"], sortorder=["ASC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="ASC")
 
     @allure.title("problem.get：sortfield=eventid, sortorder=DESC")
     @allure.story("查询告警：sortfield=eventid, sortorder=DESC")
-    def test_case_9(self):
+    def test_case_07(self):
         response = self.problem_get(sortfield=["eventid"], sortorder=["DESC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="DESC")
 
     @allure.title("problem.get：output=[source, object, objectid, name, severity]")
     @allure.story("查询触发器原型：output=[source, object, objectid, name, severity]")
-    def test_case_10(self):
+    def test_case_08(self):
         response = self.problem_get(hostids=[10272, 10275], output=["source", "object", "objectid", "name", "severity"])
         self.check_code(response=response, code=0)
         self.check_sql(response=response)

WorkCase/API/Trigger/test_trigger_create.py

+# -*- coding: utf-8 -*-
+# 测试用例
+# 作者: 陈磊
+# 时间: 2019-11-20
+
+
+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 WorkData.Zabbix.triggers import DataTriggers
+from WorkData.Zabbix.trigger_tag import DataTriggerTag
+import allure
+
+# allure.label("")
+
+
+@allure.feature("测试模块：trigger.create")
+class TestTriggerCreate(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.last()}=0"
+
+    @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_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,
+                       priority=None, _type=None, correlation_mode=None, status=None):
+        api = TriggerCreate(_host=self.host)
+        api.description = description
+        api.expression = expression
+        api.priority = priority
+        api.type = _type
+        api.correlation_mode = correlation_mode
+        api.status = status
+        api.tags_tag = tags_tag
+        api.tags_value = tags_value
+        api.get_response()
+        return api.response
+
+    @allure.step("连接数据库：zabbix")
+    def db_session(self):
+        session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
+        return session
+
+    @allure.step("查询表：triggers")
+    def select_triggers(self, triggerid=None):
+        session = self.db_session()
+        sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid)
+        return sql
+
+    @allure.step("查询表：trigger_tag")
+    def select_trigger_tag(self, triggerid=None):
+        session = self.db_session()
+        sql = DataTriggerTag().select_all_from_allKeys(session=session, triggerid=triggerid)
+        return sql
+
+    @allure.step("校验查询结果")
+    def check_select(self, triggerid, description, tags_tag, tags_value):
+        sql = self.select_triggers(triggerid=triggerid)
+        for x, y in enumerate(sql):
+            assert y.description == description
+            assert y.status == 0
+            assert y.value == 0
+
+        sql = self.select_trigger_tag(triggerid=triggerid)
+        assert sql != []
+        for x, y in enumerate(sql):
+            assert y.tag == tags_tag[x]
+            assert y.value == tags_value[x]
+
+    @allure.step("断言返回结果：校验排序")
+    def check_sortfield(self, response, possible):
+        result = UtilsResponse().get_result(response=response)
+        UtilsResponse().check_sort(_list=result, key="groupid", possible=possible)
+
+    @allure.step("断言返回结果：校验返回数据的数量")
+    def check_num(self, response, num):
+        result = UtilsResponse().get_result(response=response)
+        assert len(result) == num
+
+    @allure.step("断言返回结果")
+    def check_code(self, response, code):
+        _code = UtilsResponse().get_code(response=response)
+        assert _code == code
+
+    @allure.step("断言返回错误信息")
+    def check_msg(self, response, msg):
+        _msg = UtilsResponse().get_msg(response=response)
+        assert _msg == msg
+
+    @allure.step("断言返回错误结果")
+    def check_status(self, response, status):
+        _status = UtilsResponse().get_status(response=response)
+        assert _status == status
+
+    @allure.title("trigger.create：创建一个触发器")
+    @allure.story("创建触发器：创建一个触发器")
+    @allure.severity("blocker")
+    def test_case_01(self):
+        tags_tag = ["volume"]
+        tags_value = ["{#FSNAME}"]
+        response = self.trigger_create(description=self.base_description, expression=self.base_expression,)
+        self.check_code(response=response, code=0)
+
+        triggerid = UtilsResponse().get_result(response=response)["triggerids"][0]
+        # self.check_select(triggerid=triggerid, description=self.base_description, tags_tag=tags_tag, tags_value=tags_value)
+
+
+if __name__ == "__main__":
+    from WorkUtils.UtilsPyTest import UtilsPyTest
+    from WorkUtils.UtilsCmd import UtilsCmd
+
+    import os
+
+    # 执行自动化测试用例
+    # case_info = os.path.split(__file__)
+    # case = UtilsCmd().pytest_cmd()
+    # r = UtilsPyTest(case=case, case_info=case_info)
+    # r.run_main()
+
+    a = TestTriggerCreate()
+    a.setup_class()
+    a.test_case_01()

WorkCase/API/Trigger/test_trigger_delete.py

+# -*- coding: utf-8 -*-
+# 测试用例
+# 作者: 陈磊
+# 时间: 2019-11-13
+
+
+from __future__ import division
+
+from WorkCase import CaseBase
+from WorkUtils.UtilsLog import UtilsLog
+from WorkUtils.UtilsDataBase import UtilsDataBase
+from WorkUtils.UtilsResponse import UtilsResponse
+from WorkApi.API.Trigger.trigger_create import TriggerCreate
+from WorkApi.API.Trigger.trigger_delete import TriggerDelete
+
+from WorkData.Zabbix.triggers import DataTriggers
+
+import allure
+
+# allure.label("")
+
+
+@allure.feature("测试模块：trigger.delete")
+class TestTriggerDelete(object):
+    log = UtilsLog()
+    env = CaseBase().environment
+    host = env["host"]
+    db_url = env["db_url"]
+    db_port = env["db_port"]
+    db_user = env["db_user"]
+    db_pw = env["db_pw"]
+    db_base = env["db_base"]
+    base_name = "SS测试主机原型"
+    base_description = "SS测试触发器"
+    base_expression = "{Zabbix server:vfs.fs.size[{#FSNAME},pfree].last()}<20"
+
+    @classmethod
+    def setup_class(cls):
+        cls.log.debug("开始执行测试套件.......")
+        session = UtilsDataBase().conn_mysql(db_url=cls.db_url, db_port=cls.db_port, db_base=cls.db_base, db_user=cls.db_user, db_pw=cls.db_pw)
+        DataTriggers().delete_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):
+        tags_tag = ["volume"]
+        tags_value = ["{#FSNAME}"]
+        response = self.trigger_create(description=self.base_description, expression=self.base_expression, tags_tag=tags_tag, tags_value=tags_value)
+        self.check_code(response=response, code=0)
+        triggerid = UtilsResponse().get_result(response=response)["triggerids"][0]
+        return triggerid
+
+    @allure.step("调用接口：trigger.delete")
+    def trigger_delete(self, triggerids=None):
+        api = TriggerDelete(_host=self.host)
+        api.triggerids = triggerids
+        api.get_response()
+        return api.response
+
+    @allure.step("连接数据库：zabbix")
+    def db_session(self):
+        session = UtilsDataBase().conn_mysql(db_url=self.db_url, db_port=self.db_port, db_base=self.db_base, db_user=self.db_user, db_pw=self.db_pw)
+        return session
+
+    @allure.step("查询表：triggers")
+    def select_hosts(self, triggerid=None):
+        session = self.db_session()
+        sql = DataTriggers().select_all_from_allKeys(session=session, triggerid=triggerid)
+        return sql
+
+    @allure.step("校验查询结果")
+    def check_select(self, triggerid):
+        sql = self.select_hosts(triggerid=triggerid)
+        assert sql == []
+
+    @allure.step("断言返回结果：校验排序")
+    def check_sortfield(self, response, possible):
+        result = UtilsResponse().get_result(response=response)
+        UtilsResponse().check_sort(_list=result, key="groupid", possible=possible)
+
+    @allure.step("断言返回结果：校验返回数据的数量")
+    def check_num(self, response, num):
+        result = UtilsResponse().get_result(response=response)
+        assert len(result) == num
+
+    @allure.step("断言返回结果")
+    def check_code(self, response, code):
+        _code = UtilsResponse().get_code(response=response)
+        assert _code == code
+
+    @allure.step("断言返回错误信息")
+    def check_msg(self, response, msg):
+        _msg = UtilsResponse().get_msg(response=response)
+        assert _msg == msg
+
+    @allure.step("断言返回错误结果")
+    def check_status(self, response, status):
+        _status = UtilsResponse().get_status(response=response)
+        assert _status == status
+
+    @allure.title("trigger.delete：删除的ID存在")
+    @allure.story("删除触发器：删除的ID存在")
+    @allure.severity("blocker")
+    def test_case_01(self):
+        triggerid = self.case_create()
+
+        response = self.trigger_delete(triggerids=[triggerid])
+        self.check_code(response=response, code=0)
+        self.check_select(triggerid=triggerid)
+
+    @allure.title("trigger.delete：删除的ID不存在 500")
+    @allure.story("删除触发器：删除的ID不存在 500")
+    def test_case_02(self):
+        triggerid = self.case_create()
+
+        response = self.trigger_delete(triggerids=[triggerid])
+        self.check_code(response=response, code=0)
+        self.check_select(triggerid=triggerid)
+
+        response = self.trigger_delete(triggerids=[triggerid])
+        self.check_code(response=response, code=100)
+
+    @allure.title("trigger.delete：删除多个ID")
+    @allure.story("删除触发器：删除多个ID")
+    def test_case_03(self):
+        triggerid_1 = self.case_create()
+        triggerid_2 = self.case_create()
+        triggerid_3 = self.case_create()
+
+        response = self.trigger_delete(triggerids=[triggerid_1, triggerid_2, triggerid_3])
+        self.check_code(response=response, code=0)
+        self.check_select(triggerid=triggerid_1)
+        self.check_select(triggerid=triggerid_2)
+        self.check_select(triggerid=triggerid_3)
+
+
+if __name__ == "__main__":
+    from WorkUtils.UtilsPyTest import UtilsPyTest
+    from WorkUtils.UtilsCmd import UtilsCmd
+
+    import os
+
+    # 执行自动化测试用例
+    case_info = os.path.split(__file__)
+    case = UtilsCmd().pytest_cmd()
+    r = UtilsPyTest(case=case, case_info=case_info)
+    r.run_main()
+
+    # a = TestTriggerDelete()
+    # a.setup_class()
+    # a.test_case_03()
+

WorkCase/API/TriggerPrototype/test_triggerprototype_get.py

@@ -215,14 +215,14 @@ class TestTriggerPrototypeGet(object):
 
     @allure.title("triggerprototype.get：sortfield=triggerid, sortorder=ASC")
     @allure.story("查询触发器原型：sortfield=triggerid, sortorder=ASC")
-    def test_case_8(self):
+    def test_case_08(self):
         response = self.triggerprototype_get(sortfield=["triggerid"], sortorder=["ASC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="ASC")
 
     @allure.title("triggerprototype.get：sortfield=triggerid, sortorder=DESC")
     @allure.story("查询触发器原型：sortfield=triggerid, sortorder=DESC")
-    def test_case_9(self):
+    def test_case_09(self):
         response = self.triggerprototype_get(sortfield=["triggerid"], sortorder=["DESC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="DESC")

WorkUtils/UtilsResponse.py

@@ -103,20 +103,24 @@ class UtilsResponse:
     def check_sort(self, _list, key, possible):
         self.log.debug("校验排序")
         if possible == "ASC":
+            self.log.debug("校验升序排序...")
             start = 0
             for x, y in enumerate(_list):
                 assert y[key] > start
                 start = y[key]
         elif possible == "DESC":
+            self.log.debug("校验降序排序...")
             start = float("inf")
             for x, y in enumerate(_list):
                 assert y[key] < start
                 start = y[key]
         else:
+            self.log.debug("校验默认排序...")
             start = 0
             for x, y in enumerate(_list):
                 assert y[key] > start
                 start = y[key]
+        self.log.debug("校验成功")
 
     def get_data(self, response):
         """