最新告警

WorkCase/API/Problem/test_problem_get.py

@@ -152,7 +152,7 @@ class TestProblemGet(object):
         self.check_code(response=response, code=0)
 
     @allure.title("problem.get：output=[source, object, objectid, name, severity]")
-    @allure.story("查询触发器原型：output=[source, object, objectid, name, severity]")
+    @allure.story("查询告警：output=[source, object, objectid, name, severity]")
     def test_case_04(self):
         response = self.problem_get(output=["source", "object", "objectid", "name", "severity"])
         self.check_code(response=response, code=0)
@@ -179,8 +179,8 @@ class TestProblemGet(object):
         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]")
+    @allure.title("problem.get：hostids=特定IDs, output=[source, object, objectid, name, severity]")
+    @allure.story("查询告警：hostids=特定IDs, output=[source, object, objectid, name, severity]")
     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)

WorkCase/API/Problem/test_problem_getLatest.py

+# -*- coding: utf-8 -*-
+# 测试用例
+# 作者: 陈磊
+# 时间: 2019-11-21
+
+
+from __future__ import division
+from WorkUtils.UtilsLog import UtilsLog
+from WorkUtils.UtilsDataBase import UtilsDataBase
+from WorkUtils.UtilsResponse import UtilsResponse
+from WorkApi.API.Problem.problem_getLatest import ProblemGetLatest
+
+from WorkData.Zabbix.problem import DataProblem
+
+from WorkCase import CaseBase
+
+import allure
+
+
+# allure.label("")
+
+
+@allure.feature("测试模块：problem.getLatest")
+class TestProblemGetLatest(object):
+    log = UtilsLog()
+    env = CaseBase().environment
+    host = env["host"]
+    db_url = env["db_url"]
+    db_port = env["db_port"]
+    db_user = env["db_user"]
+    db_pw = env["db_pw"]
+    db_base = env["db_base"]
+
+    @classmethod
+    def setup_class(cls):
+        cls.log.debug("开始执行测试套件.......")
+
+    @classmethod
+    def teardown_class(cls):
+        cls.log.debug("结束执行测试套件.......")
+
+    def setup_method(self):
+        self.log.debug("测试用例执行开始...")
+
+    def teardown_method(self):
+        self.log.debug("测试用例执行结束...")
+
+    @allure.step("调用接口：problem.getLatest")
+    def problem_getLatest(self, eventids=None, groupids=None, hostids=None, objectids=None, applicationids=None,
+                          source=None, _object=None, acknowledged=None, suppressed=None, severities=None, evaltype=None,
+                          tags=None, recent=None, eventid_from=None, eventid_till=None, time_from=None,
+                          time_till=None, selectAcknowledges=None, selectTags=None, selectSuppressionData=None,
+                          countOutput=None, editable=None, excludeSearch=None,
+                          _filter=None, limit=None, output=None, preservekeys=None, search=None, searchByAny=None,
+                          searchWildcardsEnabled=None, sortfield=None, sortorder=None, startSearch=None):
+        api = ProblemGetLatest(_host=self.host)
+        api.eventids = eventids
+        api.groupids = groupids
+        api.hostids = hostids
+        api.objectids = objectids
+        api.applicationids = applicationids
+        api.source = source
+        api.object = _object
+        api.acknowledged = acknowledged
+        api.suppressed = suppressed
+        api.severities = severities
+        api.evaltype = evaltype
+        api.tags = tags
+        api.recent = recent
+        api.eventid_from = eventid_from
+        api.eventid_till = eventid_till
+        api.time_from = time_from
+        api.time_till = time_till
+        api.selectAcknowledges = selectAcknowledges
+        api.selectTags = selectTags
+        api.selectSuppressionData = selectSuppressionData
+        api.countOutput = countOutput
+        api.editable = editable
+        api.excludeSearch = excludeSearch
+        api.filter = _filter
+        api.limit = limit
+        api.output = output
+        api.preservekeys = preservekeys
+        api.search = search
+        api.searchByAny = searchByAny
+        api.searchWildcardsEnabled = searchWildcardsEnabled
+        api.sortfield = sortfield
+        api.sortorder = sortorder
+        api.startSearch = startSearch
+        api.get_response()
+        return api.response
+
+    @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("查询表：problem")
+    def select_problem(self, eventid=None):
+        session = self.db_session()
+        sql = DataProblem().select_all_from_allKeys(session=session, eventid=eventid)
+        return sql
+
+    @allure.step("断言返回结果：校验排序")
+    def check_sortfield(self, response, possible):
+        result = UtilsResponse().get_result(response=response)
+        UtilsResponse().check_sort(_list=result, key="eventid", possible=possible)
+
+    @allure.step("断言返回结果：校验返回数据的数量")
+    def check_num(self, response, num):
+        result = UtilsResponse().get_result(response=response)
+        assert len(result) == num
+
+    @allure.step("断言返回结果")
+    def check_code(self, response, code):
+        _code = UtilsResponse().get_code(response=response)
+        assert _code == code
+
+    @allure.step("断言返回结果：与数据库结果是否一致")
+    def check_sql(self, response):
+        result = UtilsResponse().get_result(response=response)
+        for x, y in enumerate(result["data"]):
+            sql = self.select_problem(eventid=y["eventid"])
+            assert y["source"] in UtilsResponse().get_values_from_sql(_sql=sql, key="source")
+            assert y["object"] in UtilsResponse().get_values_from_sql(_sql=sql, key="object")
+            assert y["objectid"] in UtilsResponse().get_values_from_sql(_sql=sql, key="objectid")
+            assert y["name"] in UtilsResponse().get_values_from_sql(_sql=sql, key="name")
+            assert y["severity"] in UtilsResponse().get_values_from_sql(_sql=sql, key="severity")
+
+        assert result["total"] >= 0
+
+    @allure.title("problem.getLatest：无参数调用")
+    @allure.story("查询最新告警：无参数调用")
+    @allure.severity("blocker")
+    def test_case_01(self):
+        response = self.problem_getLatest()
+        self.check_code(response=response, code=0)
+
+    @allure.title("problem.getLatest：countOutput=true")
+    @allure.story("查询最新告警：countOutput=true")
+    def test_case_02(self):
+        response = self.problem_getLatest(countOutput="true")
+        self.check_code(response=response, code=0)
+
+    @allure.title("problem.getLatest：output=[name]")
+    @allure.story("查询最新告警：output=[name]")
+    def test_case_03(self):
+        response = self.problem_getLatest(output=["name"])
+        self.check_code(response=response, code=0)
+
+    @allure.title("problem.getLatest：output=[source, object, objectid, name, severity]")
+    @allure.story("查询最新告警：output=[source, object, objectid, name, severity]")
+    def test_case_04(self):
+        response = self.problem_getLatest(output=["source", "object", "objectid", "name", "severity"])
+        self.check_code(response=response, code=0)
+        self.check_sql(response=response)
+
+    @allure.title("problem.getLatest：sortfield=eventid")
+    @allure.story("查询最新告警：sortfield=eventid")
+    def test_case_05(self):
+        response = self.problem_getLatest(sortfield=["eventid"])
+        self.check_code(response=response, code=0)
+        self.check_sortfield(response=response, possible=True)
+
+    @allure.title("problem.getLatest：sortfield=eventid, sortorder=ASC")
+    @allure.story("查询最新告警：sortfield=eventid, sortorder=ASC")
+    def test_case_06(self):
+        response = self.problem_getLatest(sortfield=["eventid"], sortorder=["ASC"])
+        self.check_code(response=response, code=0)
+        self.check_sortfield(response=response, possible="ASC")
+
+    @allure.title("problem.getLatest：sortfield=eventid, sortorder=DESC")
+    @allure.story("查询最新告警：sortfield=eventid, sortorder=DESC")
+    def test_case_07(self):
+        response = self.problem_getLatest(sortfield=["eventid"], sortorder=["DESC"])
+        self.check_code(response=response, code=0)
+        self.check_sortfield(response=response, possible="DESC")
+
+    @allure.title("problem.getLatest：hostids=特定IDs, output=[source, object, objectid, name, severity]")
+    @allure.story("查询最新告警：hostids=特定IDs, output=[source, object, objectid, name, severity]")
+    def test_case_08(self):
+        response = self.problem_getLatest(hostids=[10272, 10275], output=["source", "object", "objectid", "name", "severity"])
+        self.check_code(response=response, code=0)
+        self.check_sql(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 = TestProblemGetLatest()
+    # a.setup_class()
+    # a.test_case_08()