创建主机补充校验macro

.idea/dictionaries/Administrator.xml

@@ -19,6 +19,7 @@
       <w>groupids</w>
       <w>hostid</w>
       <w>hostids</w>
+      <w>hostmacro</w>
       <w>hostmonitor</w>
       <w>hostprototype</w>
       <w>hstgrp</w>

WorkCase/API/Host/test_host_create.py

@@ -17,6 +17,7 @@ from WorkData.Zabbix.items import DataItems
 from WorkData.Zabbix.triggers import DataTriggers
 from WorkData.Zabbix.functions import DataFunctions
 from WorkData.Zabbix.hosts_groups import DataHostsGroups
+from WorkData.Zabbix.hostmacro import DataHostsMacro
 from WorkData.Zabbix.interface import DataInterface
 
 import allure
@@ -121,6 +122,12 @@ class TestHostCreate(object):
         sql = DataTriggers().select_all_from_allKeys(session=session, templateid=templateid)
         return sql
 
+    @allure.step("查询表：hostmacro")
+    def select_hostmacro(self, hostid=None):
+        session = self.db_session()
+        sql = DataHostsMacro().select_all_from_allKeys(session=session, hostid=hostid)
+        return sql
+
     @allure.step("校验查询结果")
     def check_select(self, hostid, _type):
         sql = self.select_hosts_group(groupid=self.base_group_id, hostid=hostid)
@@ -258,6 +265,12 @@ class TestHostCreate(object):
             sql = self.select_functions(triggerid=y)
             assert sql != []
 
+        self.log.info("校验主机宏是否已创建")
+        sql = self.select_hostmacro(hostid=hostid)
+        for x, y in enumerate(sql):
+            assert y.macro == "{$SNMP_COMMUNITY}"
+            assert y.value == "public"
+
 
 if __name__ == "__main__":
     from WorkUtils.UtilsPyTest import UtilsPyTest
@@ -266,11 +279,11 @@ 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()
-
-    a = TestHostCreate()
-    a.setup_class()
-    a.test_case_04()
+    case_info = os.path.split(__file__)
+    case = UtilsCmd().pytest_cmd()
+    r = UtilsPyTest(case=case, case_info=case_info)
+    r.run_main()
+
+    # a = TestHostCreate()
+    # a.setup_class()
+    # a.test_case_04()

WorkCase/__init__.py

@@ -11,17 +11,17 @@ class CaseBase:
 
         self.environment = {
             # "host": "http://10.0.0.12:7070",
-            "db_url": "10.0.0.153",
-            "db_port": 3306,
-            "db_user": "zmops",
-            "db_pw": "0VMoH%W7|h",
-            "db_base": "zabbix",
-            "host": "http://localhost:7070",
-            # "db_url": "10.0.0.155",
+            # "db_url": "10.0.0.153",
             # "db_port": 3306,
-            # "db_user": "root",
-            # "db_pw": "123456",
-            # "db_base": "zabbix"
+            # "db_user": "zmops",
+            # "db_pw": "0VMoH%W7|h",
+            # "db_base": "zabbix",
+            "host": "http://localhost:7070",
+            "db_url": "10.0.0.155",
+            "db_port": 3306,
+            "db_user": "root",
+            "db_pw": "123456",
+            "db_base": "zabbix"
         }
 
         self.app_environment = {

WorkData/Zabbix/hostmacro.py

+# -*- coding: utf_8 -*-
+# 表名: hostmacro
+# 作者: 陈磊
+# 时间: 2019-11-22
+
+
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy import Column, Integer, VARCHAR, BIGINT, 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 HostsMacro(Base):
+    """
+    主机组与主机宏关系表
+    """
+    __tablename__ = "hostmacro"
+
+    hostmacroid = Column(BigInteger, primary_key=True)
+    hostid = Column(BigInteger)
+    macro = Column(VARCHAR(255))
+    value = Column(VARCHAR(255))
+
+
+class DataHostsMacro(object):
+    def __init__(self):
+        self.log = UtilsLog()
+        self.log.info(self.__class__)
+
+        self.table = HostsMacro
+
+    def select_all_from_allKeys(self, session, hostid=None):
+        """
+        :param session: 指针
+        :param hostid:
+        :return: 查询结果
+        """
+        self.log.debug("查询数据库:")
+        try:
+            base = UtilsDataBase()
+            base.add_param(_key="hostid", value=hostid)
+            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