监控项查询

WorkCase/API/Items/test_items_get.py

@@ -34,9 +34,8 @@ class TestItemsGet(object):
     db_base = env["db_base"]
 
     base_groupid = 101
-    base_name = "SS TEST HOST"
+    base_name = "SS TEST HOST ITEMS"
     base_ip = "10.0.0.10"
-    base_host = "SS自动化测试主机"
     base_description = "SS自动化测试主机描述, 请勿删除"
 
     @classmethod
@@ -62,6 +61,28 @@ class TestItemsGet(object):
     def teardown_method(self):
         self.log.debug("测试用例执行结束...")
 
+    @allure.step("调用接口：host.create")
+    def host_create(self, host=None, name=None, description=None, groupids=None, interfaces=None, tags=None, templateids=None, macros=None,
+                    inventory=None, status=None, inventory_mode=None):
+        api = HostCreate(_host=self.host)
+        api.host = host
+        api.name = name
+        api.description = description
+        api.groupids = groupids
+        api.interfaces = interfaces
+        api.tags = tags
+        api.templateids = templateids
+        api.macros = macros
+        api.inventory = inventory
+        api.status = status
+        api.inventory_mode = inventory_mode
+        api.tls_connect = 1
+        api.tls_accept = 1
+        api.tls_psk_identity = "Test"
+        api.tls_psk = "Test"
+        api.get_response()
+        return api.response
+
     @allure.step("创建测试数据")
     def case_create(self, num):
         templateid = 10001
@@ -123,15 +144,15 @@ class TestItemsGet(object):
         return session
 
     @allure.step("查询表：items")
-    def select_items(self, hostid):
+    def select_items(self, hostid=None, itemid=None):
         session = self.db_session()
-        sql = DataItems().select_all_from_allKeys(session=session, hostid=hostid)
+        sql = DataItems().select_all_from_allKeys(session=session, hostid=hostid, itemid=itemid)
         return sql
 
     @allure.step("断言返回结果：校验排序")
     def check_sortfield(self, response, possible):
         result = UtilsResponse().get_result(response=response)
-        UtilsResponse().check_sort(_list=result, key="groupid", possible=possible)
+        UtilsResponse().check_sort(_list=result, key="itemid", possible=possible)
 
     @allure.step("断言返回结果：校验返回数据的数量")
     def check_num(self, response, num):
@@ -145,18 +166,11 @@ class TestItemsGet(object):
 
     @allure.step("断言返回结果：与数据库结果是否一致")
     def check_sql(self, response, hostid):
-        sql = self.select_items(hostid=hostid)
         result = UtilsResponse().get_result(response=response)
-        for x, y in enumerate(sql):
-            assert y.groupid in UtilsResponse().get_values_from_key(_dict=result, key="groupid")
-            assert y.name in UtilsResponse().get_values_from_key(_dict=result, key="name")
-            assert y.internal in UtilsResponse().get_values_from_key(_dict=result, key="internal")
-            assert y.flags in UtilsResponse().get_values_from_key(_dict=result, key="flags")
         for x, y in enumerate(result):
-            assert y["groupid"] in UtilsResponse().get_values_from_sql(_sql=sql, key="groupid")
-            assert y["name"] in UtilsResponse().get_values_from_sql(_sql=sql, key="name")
-            assert y["internal"] in UtilsResponse().get_values_from_sql(_sql=sql, key="internal")
-            assert y["flags"] in UtilsResponse().get_values_from_sql(_sql=sql, key="flags")
+            sql = self.select_items(itemid=y["itemid"])
+            for q, w in enumerate(sql):
+                assert w.name == y["name"]
 
     @allure.title("items.get：无参数调用")
     @allure.story("查询监控项：无参数调用")
@@ -188,66 +202,38 @@ class TestItemsGet(object):
     @allure.title("items.get：hostids=特定IDs, output=[hostid, name]")
     @allure.story("查询监控项：hostids=特定IDs, output=[hostid, name]")
     def test_case_05(self):
-        hostid = self.case_create(num="03")
+        hostid = self.case_create(num="05")
 
-        response = self.items_get(groupids=[hostid], output=["hostid", "name"])
+        response = self.items_get(hostids=[hostid], output=["hostid", "name"])
         self.check_code(response=response, code=0)
-        self.check_sql(response=response, groupid=groupid)
+        self.check_sql(response=response, hostid=hostid)
 
     @allure.title("items.get：limit=1")
     @allure.story("查询监控项：limit=1")
     def test_case_06(self):
-        name = self.base_name + "06"
-        response = self.items_create(name=name)
-        self.check_code(response=response, code=0)
-
         response = self.items_get(limit=1)
         self.check_code(response=response, code=0)
         result = UtilsResponse().get_result(response=response)
         assert len(result) == 1
 
-    @allure.title("items.get：精确搜索")
-    @allure.story("查询监控项：精确搜索")
+    @allure.title("items.get：sortfield=itemid")
+    @allure.story("查询监控项：sortfield=itemid")
     def test_case_07(self):
-        name = self.base_name + "07"
-        response = self.items_create(name=name)
-        self.check_code(response=response, code=0)
-
-        response = self.items_get(search={"name": [name]})
-        self.check_code(response=response, code=0)
-        result = UtilsResponse().get_result(response=response)
-        assert len(result) == 1
-
-    @allure.title("items.get：模糊搜索")
-    @allure.story("查询监控项：模糊搜索")
-    def test_case_08(self):
-        name = self.base_name + "08"
-        response = self.items_create(name=name)
-        self.check_code(response=response, code=0)
-
-        response = self.items_get(search={"name": ["*" + self.base_name + "*"]}, searchByAny=True, searchWildcardsEnabled=True)
-        self.check_code(response=response, code=0)
-        result = UtilsResponse().get_result(response=response)
-        assert len(result) >= 1
-
-    @allure.title("items.get：sortfield=groupid")
-    @allure.story("查询监控项：sortfield=groupid")
-    def test_case_09(self):
-        response = self.items_get(sortfield=["groupid"])
+        response = self.items_get(sortfield=["itemid"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible=True)
 
-    @allure.title("items.get：sortfield=groupid, sortorder=ASC")
-    @allure.story("查询监控项：sortfield=groupid, sortorder=ASC")
-    def test_case_10(self):
-        response = self.items_get(sortfield=["groupid"], sortorder=["ASC"])
+    @allure.title("items.get：sortfield=itemid, sortorder=ASC")
+    @allure.story("查询监控项：sortfield=itemid, sortorder=ASC")
+    def test_case_08(self):
+        response = self.items_get(sortfield=["itemid"], sortorder=["ASC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="ASC")
 
-    @allure.title("items.get：sortfield=groupid, sortorder=DESC")
-    @allure.story("查询监控项：sortfield=groupid, sortorder=DESC")
-    def test_case_11(self):
-        response = self.items_get(sortfield=["groupid"], sortorder=["DESC"])
+    @allure.title("items.get：sortfield=itemid, sortorder=DESC")
+    @allure.story("查询监控项：sortfield=itemid, sortorder=DESC")
+    def test_case_09(self):
+        response = self.items_get(sortfield=["itemid"], sortorder=["DESC"])
         self.check_code(response=response, code=0)
         self.check_sortfield(response=response, possible="DESC")
 
@@ -259,10 +245,10 @@ if __name__ == "__main__":
     import os
 
     # 执行自动化测试用例
-    # case_info = os.path.split(__file__)
-    # case = UtilsCmd().pytest_cmd()
-    # r = UtilsPyTest(case=case, case_info=case_info)
-    # r.run_main()
+    case_info = os.path.split(__file__)
+    case = UtilsCmd().pytest_cmd()
+    r = UtilsPyTest(case=case, case_info=case_info)
+    r.run_main()
 
-    a = TestItemsGet()
-    a.test_case_01()
+    # a = TestItemsGet()
+    # a.test_case_09()

WorkData/Zabbix/items.py

@@ -95,18 +95,21 @@ class DataItems(object):
 
         self.table = Items
 
-    def select_all_from_allKeys(self, session, hostid=None, templateid=None):
+    def select_all_from_allKeys(self, session, itemid=None, hostid=None, templateid=None):
         """
         :param session: 指针
+        :param itemid:
         :param hostid:
         :param templateid:
         :return: 查询结果
         """
         self.log.debug("查询数据库:")
+        self.log.debug("itemid: %s" % itemid)
         self.log.debug("hostid: %s" % hostid)
         self.log.debug("templateid: %s" % templateid)
         try:
             base = UtilsDataBase()
+            base.add_param(_key="itemid", value=itemid)
             base.add_param(_key="hostid", value=hostid)
             base.add_param(_key="templateid", value=templateid)
             sql_rep = session.query(self.table).filter_by(**base.param).all()