fix case

tests/pytest/util/common.py

@@ -17,6 +17,7 @@ import requests
 import time
 import socket
 import json
+import toml
 from .boundary import DataBoundary
 import taos
 from util.log import *
@@ -443,7 +444,9 @@ class TDCom:
         return buildPath
 
     def getClientCfgPath(self):
-        buildPath = self.getBuildPath()
+        # buildPath = self.getBuildPath()
+        buildPath = get_path()
+
         if (buildPath == ""):
             tdLog.exit("taosd not found!")
         else:
@@ -752,4 +755,29 @@ def is_json(msg):
     else:
         return False
 
+def get_path(tool="taosd"):
+        selfPath = os.path.dirname(os.path.realpath(__file__))
+
+        if ("community" in selfPath):
+            projPath = selfPath[:selfPath.find("community")]
+        else:
+            projPath = selfPath[:selfPath.find("tests")]
+
+        paths = []
+        for root, dirs, files in os.walk(projPath):
+            if ((tool) in files or ("%s.exe"%tool) in files):
+                rootRealPath = os.path.dirname(os.path.realpath(root))
+                if ("packaging" not in rootRealPath):
+                    paths.append(os.path.join(root, tool))
+                    break
+        if (len(paths) == 0):
+                return ""
+        return paths[0]
+
+def dict2toml(in_dict: dict, file:str):
+    if not isinstance(in_dict, dict):
+        return ""
+    with open(file, 'w') as f:
+        toml.dump(in_dict, f)
+
 tdCom = TDCom()

tests/pytest/util/dnodes.py

@@ -96,9 +96,9 @@ class TDSimClient:
 
         for key, value in self.cfgDict.items():
             self.cfg(key, value)
-        
+
         try:
-            if bool(updatecfgDict) and updatecfgDict[0] and updatecfgDict[0][0]:                    
+            if bool(updatecfgDict) and updatecfgDict[0] and updatecfgDict[0][0]:
                 clientCfg = dict (updatecfgDict[0][0].get('clientCfg'))
                 for key, value in clientCfg.items():
                     self.cfg(key, value)
@@ -244,7 +244,6 @@ class TDDnode:
         # print(updatecfgDict)
         isFirstDir = 1
         if bool(updatecfgDict) and updatecfgDict[0] and updatecfgDict[0][0]:
-            print(updatecfgDict[0][0])
             for key, value in updatecfgDict[0][0].items():
                 if key == "clientCfg" and self.remoteIP == "" and not platform.system().lower() == 'windows':
                     continue
@@ -324,7 +323,6 @@ class TDDnode:
             if os.system(cmd) != 0:
                 tdLog.exit(cmd)
             self.running = 1
-            print("dnode:%d is running with %s " % (self.index, cmd))
             tdLog.debug("dnode:%d is running with %s " % (self.index, cmd))
             if self.valgrind == 0:
                 time.sleep(0.1)
@@ -358,7 +356,7 @@ class TDDnode:
                 #         break
                 #     elif bkey2 in line:
                 #         popen.kill()
-                #         break                        
+                #         break
                 #     if time.time() > timeout:
                 #         print(time.time(),timeout)
                 #         tdLog.exit('wait too long for taosd start')
@@ -407,7 +405,6 @@ class TDDnode:
             if os.system(cmd) != 0:
                 tdLog.exit(cmd)
             self.running = 1
-            print("dnode:%d is running with %s " % (self.index, cmd))
             tdLog.debug("dnode:%d is running with %s " % (self.index, cmd))
             if self.valgrind == 0:
                 time.sleep(0.1)
@@ -655,7 +652,6 @@ class TDDnodes:
     def stoptaosd(self, index):
         self.check(index)
         self.dnodes[index - 1].stoptaosd()
-        
 
     def start(self, index):
         self.check(index)

tests/pytest/util/taosadapter.py

+from fabric2 import Connection
+from util.log import *
+from util.common import *
+
+
+class TAdapter:
+    def __init__(self):
+        self.running = 0
+        self.deployed = 0
+        self.remoteIP = ""
+        self.taosadapter_cfg_dict = {
+            "debug"         : False,
+            "taosConfigDir" : "",
+            "port"          : 6041,
+            "logLevel"      : "info",
+            "cors"          : {
+                "allowAllOrigins" : True,
+            },
+            "pool"          : {
+                "maxConnect"    : 4000,
+                "maxIdle"       : 4000,
+                "idleTimeout"   : "1h"
+            },
+            "ssl"           : {
+                "enable"        : False,
+                "certFile"      : "",
+                "keyFile"       : "",
+            },
+            "log"           : {
+                "path"                  : "",
+                "rotationCount"         : 30,
+                "rotationTime"          : "24h",
+                "rotationSize"          : "1GB",
+                "enableRecordHttpSql"   : False,
+                "sqlRotationCount"      : 2,
+                "sqlRotationTime"       : "24h",
+                "sqlRotationSize"       : "1GB",
+            },
+            "monitor"       : {
+                "collectDuration"           : "3s",
+                "incgroup"                  : False,
+                "pauseQueryMemoryThreshold" : 70,
+                "pauseAllMemoryThreshold"   : 80,
+                "identity"                  : "",
+                "writeToTD"                 : True,
+                "user"                      : "root",
+                "password"                  : "taosdata",
+                "writeInterval"             : "30s"
+            }
+
+        }
+    # TODO: add taosadapter env:
+    # 1. init cfg.toml.dict ：OK
+    # 2. dump dict to toml ： OK
+    # 3. update cfg.toml.dict ：OK
+    # 4. check adapter exists ： OK
+    # 5. deploy adapter cfg ： OK
+    # 6. adapter start ：   OK
+    # 7. adapter stop
+
+    def init(self, path, remoteIP=""):
+        self.path = path
+        self.remoteIP = remoteIP
+        binPath = get_path() + "/../../../"
+        binPath = os.path.realpath(binPath)
+
+        if path == "":
+            self.path = os.path.abspath(binPath + "../../")
+        else:
+            self.path = os.path.realpath(path)
+
+        if self.remoteIP:
+            try:
+                self.config = eval(remoteIP)
+                self.remote_conn = Connection(host=self.config["host"], port=self.config["port"], user=self.config["user"], connect_kwargs={'password':self.config["password"]})
+            except Exception as e:
+                tdLog.notice(e)
+
+    def update_cfg(self, update_dict :dict):
+        if not isinstance(update_dict, dict):
+            return
+        if "log" in update_dict and "path" in update_dict["log"]:
+            del update_dict["log"]["path"]
+        for key, value in update_dict.items():
+            if key in ["cors", "pool", "ssl", "log", "monitor", "opentsdb", "influxdb", "statsd", "collectd", "opentsdb_telnet", "node_exporter", "prometheus"]:
+                if  isinstance(value, dict):
+                    for k, v in value.items():
+                        self.taosadapter_cfg_dict[key][k] = v
+            else:
+                self.taosadapter_cfg_dict[key] = value
+
+    def check_adapter(self):
+        if getPath(tool="taosadapter"):
+            return False
+        else:
+            return True
+
+    def remote_exec(self, updateCfgDict, execCmd):
+        remoteCfgDict = copy.deepcopy(updateCfgDict)
+        if "log" in remoteCfgDict and "path" in remoteCfgDict["log"]:
+            del remoteCfgDict["log"]["path"]
+
+        remoteCfgDictStr = base64.b64encode(toml.dumps(remoteCfgDict).encode()).decode()
+        execCmdStr = base64.b64encode(execCmd.encode()).decode()
+        with self.remote_conn.cd((self.config["path"]+sys.path[0].replace(self.path, '')).replace('\\','/')):
+            self.remote_conn.run(f"python3 ./test.py  -D {remoteCfgDictStr} -e {execCmdStr}" )
+
+    def cfg(self, option, value):
+        cmd = f"echo {option} = {value} >> {self.cfg_path}"
+        if os.system(cmd) != 0:
+            tdLog.exit(cmd)
+
+    def deploy(self, *update_cfg_dict):
+        self.log_dir = f"{self.path}/sim/dnode1/log"
+        self.cfg_dir = f"{self.path}/sim/dnode1/cfg"
+        self.cfg_path = f"{self.cfg_dir}/taosadapter.toml"
+
+        cmd = f"touch {self.cfg_path}"
+        if os.system(cmd) != 0:
+            tdLog.exit(cmd)
+
+        self.taosadapter_cfg_dict["log"]["path"] = self.log_dir
+        if bool(update_cfg_dict):
+            self.update_cfg(update_dict=update_cfg_dict)
+
+        if (self.remoteIP == ""):
+            dict2toml(self.taosadapter_cfg_dict, self.cfg_path)
+        else:
+            self.remote_exec(self.taosadapter_cfg_dict, "tAdapter.deploy(update_cfg_dict)")
+
+        self.deployed = 1
+
+        tdLog.debug(f"taosadapter is deployed and configured by {self.cfg_path}")
+
+    def start(self):
+        bin_path = get_path(tool="taosadapter")
+
+        if (bin_path == ""):
+            tdLog.exit("taosadapter not found!")
+        else:
+            tdLog.info(f"taosadapter found: {bin_path}")
+
+        if platform.system().lower() == 'windows':
+            cmd = f"mintty -h never {bin_path} -c {self.cfg_dir}"
+        else:
+            cmd = f"nohup {bin_path} -c {self.cfg_path} > /dev/null 2>&1 & "
+
+        if  self.remoteIP:
+            self.remote_exec(self.taosadapter_cfg_dict, f"tAdapter.deployed=1\ntAdapter.log_dir={self.log_dir}\ntAdapter.cfg_dir={self.cfg_dir}\ntAdapter.start()")
+            self.running = 1
+        else:
+            os.system(f"rm -rf {self.log_dir}/taosadapter*")
+            if os.system(cmd) != 0:
+                tdLog.exit(cmd)
+            self.running = 1
+            tdLog.debug(f"taosadapter is running with {cmd} " )
+
+            time.sleep(0.1)
+            key = 'all plugin init finish'
+            bkey = bytes(key, encoding="utf8")
+            logFile = self.log_dir + "/taosadapter*"
+            file_exists =  False
+            i = 0
+            while (not file_exists):
+                for file in os.listdir(self.log_dir):
+                    if "taosadapter" in file:
+                        file_exists = True
+                        break
+                sleep(0.1)
+                i += 1
+                if i > 50:
+                    tdLog.notice("log file is too long to create")
+                    break
+
+            tailCmdStr = 'tail -f '
+            popen = subprocess.Popen(
+                tailCmdStr + logFile,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                shell=True)
+            # pid = popen.pid
+            # print('Popen.pid:' + str(pid))
+            timeout = time.time() + 60 * 2
+            while True:
+                line = popen.stdout.readline().strip()
+                if bkey in line:
+                    popen.kill()
+                    break
+                if time.time() > timeout:
+                    tdLog.exit('wait too long for taosadapter start')
+            tdLog.debug("the taosadapter has been started.")
+
+    def start_taosadapter(self):
+        """
+            use this method, must deploy taosadapter
+        """
+        bin_path = get_path(tool="taosadapter")
+
+        if (bin_path == ""):
+            tdLog.exit("taosadapter not found!")
+        else:
+            tdLog.info(f"taosadapter found: {bin_path}")
+
+        if self.deployed == 0:
+            tdLog.exit("taosadapter is not deployed")
+
+        if platform.system().lower() == 'windows':
+            cmd = f"mintty -h never {bin_path} -c {self.cfg_dir}"
+        else:
+            cmd = f"nohup {bin_path} -c {self.cfg_path} > /dev/null 2>&1 & "
+
+        if  self.remoteIP:
+            self.remote_exec(self.taosadapter_cfg_dict, f"tAdapter.deployed=1\ntAdapter.log_dir={self.log_dir}\ntAdapter.cfg_dir={self.cfg_dir}\ntAdapter.start()")
+            self.running = 1
+        else:
+            if os.system(cmd) != 0:
+                tdLog.exit(cmd)
+            self.running = 1
+            tdLog.debug(f"taosadapter is running with {cmd} " )
+
+            time.sleep(0.1)
+
+    def stop(self, force_kill=False):
+        signal = "-SIGKILL" if force_kill else "-SIGTERM"
+
+        if  self.remoteIP:
+            self.remote_exec(self.taosadapter_cfg_dict, "tAdapter.running=1\ntAdapter.stop()")
+            tdLog.info("stop taosadapter")
+            return
+
+        toBeKilled = "taosadapter"
+
+        if self.running != 0:
+            psCmd = f"ps -ef|grep -w {toBeKilled}| grep -v grep | awk '{{print $2}}'"
+            processID = subprocess.check_output(
+                psCmd, shell=True).decode("utf-8")
+
+            while(processID):
+                killCmd = f"kill {signal} {processID} > /dev/null 2>&1"
+                os.system(killCmd)
+                time.sleep(1)
+                processID = subprocess.check_output(
+                    psCmd, shell=True).decode("utf-8")
+            if not platform.system().lower() == 'windows':
+                for port in range(6030, 6041):
+                    fuserCmd = f"fuser -k -n tcp {port} > /dev/null"
+                    os.system(fuserCmd)
+
+        self.running = 0
+        tdLog.debug(f"taosadapter is stopped by kill {signal}")
+
+
+
+tAdapter = TAdapter()
\ No newline at end of file

tests/system-test/test.py

@@ -22,11 +22,14 @@ import json
 import platform
 import socket
 import threading
+
+import toml
 sys.path.append("../pytest")
 from util.log import *
 from util.dnodes import *
 from util.cases import *
 from util.cluster import *
+from util.taosadapter import *
 
 import taos
 import taosrest
@@ -64,12 +67,13 @@ if __name__ == "__main__":
     dnodeNums = 1
     mnodeNums = 0
     updateCfgDict = {}
+    adapter_cfg_dict = {}
     execCmd = ""
     queryPolicy = 1
     createDnodeNums = 1
     restful = False
-    opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:k:e:N:M:Q:C:R', [
-        'file=', 'path=', 'master', 'logSql', 'stop', 'cluster', 'valgrind', 'help', 'restart', 'updateCfgDict', 'killv', 'execCmd','dnodeNums','mnodeNums','queryPolicy','createDnodeNums','restful'])
+    opts, args = getopt.gnu_getopt(sys.argv[1:], 'f:p:m:l:scghrd:k:e:N:M:Q:C:RD:', [
+        'file=', 'path=', 'master', 'logSql', 'stop', 'cluster', 'valgrind', 'help', 'restart', 'updateCfgDict', 'killv', 'execCmd','dnodeNums','mnodeNums','queryPolicy','createDnodeNums','restful','adaptercfgupdate'])
     for key, value in opts:
         if key in ['-h', '--help']:
             tdLog.printNoPrefix(
@@ -90,6 +94,7 @@ if __name__ == "__main__":
             tdLog.printNoPrefix('-Q set queryPolicy in one dnode')
             tdLog.printNoPrefix('-C create Dnode Numbers in one cluster')
             tdLog.printNoPrefix('-R restful realization form')
+            tdLog.printNoPrefix('-D taosadapter update cfg dict ')
 
 
             sys.exit(0)
@@ -138,7 +143,7 @@ if __name__ == "__main__":
             try:
                 execCmd = base64.b64decode(value.encode()).decode()
             except:
-                print('updateCfgDict convert fail.')
+                print('execCmd run fail.')
                 sys.exit(0)
 
         if key in ['-N', '--dnodeNums']:
@@ -156,8 +161,18 @@ if __name__ == "__main__":
         if key in ['-R', '--restful']:
             restful = True
 
+        if key in ['-D', '--adaptercfgupdate']:
+            try:
+                adaptercfgupdate = eval(base64.b64decode(value.encode()).decode())
+            except:
+                print('adapter cfg update convert fail.')
+                sys.exit(0)
+
     if not execCmd == "":
-        tdDnodes.init(deployPath)
+        if restful:
+            tAdapter.init(deployPath)
+        else:
+            tdDnodes.init(deployPath)
         print(execCmd)
         exec(execCmd)
         quit()
@@ -190,6 +205,31 @@ if __name__ == "__main__":
         if valgrind:
             time.sleep(2)
 
+        if restful:
+            toBeKilled = "taosadapter"
+
+            killCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}' | xargs kill -TERM > /dev/null 2>&1" % toBeKilled
+
+            psCmd = "ps -ef|grep -w %s| grep -v grep | awk '{print $2}'" % toBeKilled
+            processID = subprocess.check_output(psCmd, shell=True)
+
+            while(processID):
+                os.system(killCmd)
+                time.sleep(1)
+                processID = subprocess.check_output(psCmd, shell=True)
+
+            for port in range(6030, 6041):
+                usePortPID = "lsof -i tcp:%d | grep LISTEn | awk '{print $2}'" % port
+                processID = subprocess.check_output(usePortPID, shell=True)
+
+                if processID:
+                    killCmd = "kill -TERM %s" % processID
+                    os.system(killCmd)
+                fuserCmd = "fuser -k -n tcp %d" % port
+                os.system(fuserCmd)
+
+            tdLog.info('stop taosadapter')
+
         tdLog.info('stop All dnodes')
 
     if masterIp == "":
@@ -219,6 +259,7 @@ if __name__ == "__main__":
         except Exception as r:
             print(r)
         updateCfgDictStr = ''
+        # adapter_cfg_dict_str = ''
         if is_test_framework:
             moduleName = fileName.replace(".py", "").replace(os.sep, ".")
             uModule = importlib.import_module(moduleName)
@@ -227,30 +268,44 @@ if __name__ == "__main__":
                 if ((json.dumps(updateCfgDict) == '{}') and hasattr(ucase, 'updatecfgDict')):
                     updateCfgDict = ucase.updatecfgDict
                     updateCfgDictStr = "-d %s"%base64.b64encode(json.dumps(updateCfgDict).encode()).decode()
+                if ((json.dumps(adapter_cfg_dict) == '{}') and hasattr(ucase, 'taosadapter_cfg_dict')):
+                    adapter_cfg_dict = ucase.taosadapter_cfg_dict
+                    # adapter_cfg_dict_str = f"-D {base64.b64encode(toml.dumps(adapter_cfg_dict).encode()).decode()}"
             except Exception as r:
                 print(r)
         else:
             pass
+        if restful:
+            tAdapter.init(deployPath, masterIp)
+            tAdapter.stop(force_kill=True)
+
         if dnodeNums == 1 :
             tdDnodes.deploy(1,updateCfgDict)
             tdDnodes.start(1)
             tdCases.logSql(logSql)
+            if restful:
+                tAdapter.deploy(adapter_cfg_dict)
+                tAdapter.start()
+
             if queryPolicy != 1:
                 queryPolicy=int(queryPolicy)
-                conn = taos.connect(
-                host,
-                config=tdDnodes.getSimCfgPath())
-                tdSql.init(conn.cursor())
-                tdSql.execute("create qnode on dnode 1")
-                tdSql.execute('alter local "queryPolicy" "%d"'%queryPolicy)     
-                tdSql.query("show local variables;")
-                for i in range(tdSql.queryRows):
-                    if tdSql.queryResult[i][0] == "queryPolicy" :
-                        if int(tdSql.queryResult[i][1]) == int(queryPolicy):
-                            tdLog.success('alter queryPolicy to %d successfully'%queryPolicy)
-                        else :
-                            tdLog.debug(tdSql.queryResult)
-                            tdLog.exit("alter queryPolicy to  %d failed"%queryPolicy) 
+                if restful:
+                    conn = taosrest.connect(url=f"http://{host}:6041")
+                else:
+                    conn = taos.connect(host,config=tdDnodes.getSimCfgPath())
+
+                cursor = conn.cursor()
+                cursor.execute("create qnode on dnode 1")
+                cursor.execute(f'alter local "queryPolicy" "{queryPolicy}"')
+                cursor.execute("show local variables")
+                res = cursor.fetchall()
+                for i in range(cursor.rowcount):
+                    if res[i][0] == "queryPolicy" :
+                        if int(res[i][1]) == int(queryPolicy):
+                            tdLog.success(f'alter queryPolicy to {queryPolicy} successfully')
+                        else:
+                            tdLog.debug(res)
+                            tdLog.exit(f"alter queryPolicy to  {queryPolicy} failed")
         else :
             tdLog.debug("create an cluster  with %s nodes and make %s dnode as independent mnode"%(dnodeNums,mnodeNums))
             dnodeslist = cluster.configure_cluster(dnodeNums=dnodeNums,mnodeNums=mnodeNums)
@@ -264,13 +319,16 @@ if __name__ == "__main__":
             for dnode in tdDnodes.dnodes:
                 tdDnodes.starttaosd(dnode.index)
             tdCases.logSql(logSql)
+
+            if restful:
+                tAdapter.deploy(adapter_cfg_dict)
+                tAdapter.start()
+
             if not restful:
-                conn = taos.connect(
-                    host,
-                   config=tdDnodes.getSimCfgPath())
+                conn = taos.connect(host,config=tdDnodes.getSimCfgPath())
             else:
                 conn = taosrest.connect(url=f"http://{host}:6041")
-            print(tdDnodes.getSimCfgPath(),host)
+            tdLog.info(tdDnodes.getSimCfgPath(),host)
             if createDnodeNums == 1:
                 createDnodeNums=dnodeNums
             else:
@@ -285,9 +343,7 @@ if __name__ == "__main__":
             conn = None
         else:
             if not restful:
-                conn = taos.connect(
-                    host="%s"%(host),
-                    config=tdDnodes.sim.getCfgDir())
+                conn = taos.connect(host="%s"%(host), config=tdDnodes.sim.getCfgDir())
             else:
                 conn = taosrest.connect(url=f"http://{host}:6041")
         if is_test_framework:
@@ -314,18 +370,28 @@ if __name__ == "__main__":
                 ucase = uModule.TDTestCase()
                 if (json.dumps(updateCfgDict) == '{}'):
                     updateCfgDict = ucase.updatecfgDict
+                if (json.dumps(adapter_cfg_dict) == '{}'):
+                    adapter_cfg_dict = ucase.taosadapter_cfg_dict
             except:
                 pass
+
+        if restful:
+            tAdapter.init(deployPath, masterIp)
+            tAdapter.stop(force_kill=True)
+
         if dnodeNums == 1 :
             tdDnodes.deploy(1,updateCfgDict)
             tdDnodes.start(1)
             tdCases.logSql(logSql)
+
+            if restful:
+                tAdapter.deploy(adapter_cfg_dict)
+                tAdapter.start()
+
             if queryPolicy != 1:
                 queryPolicy=int(queryPolicy)
                 if not restful:
-                    conn = taos.connect(
-                    host,
-                    config=tdDnodes.getSimCfgPath())
+                    conn = taos.connect(host,config=tdDnodes.getSimCfgPath())
                 else:
                     conn = taosrest.connect(url=f"http://{host}:6041")
                 # tdSql.init(conn.cursor())
@@ -366,10 +432,13 @@ if __name__ == "__main__":
             for dnode in tdDnodes.dnodes:
                 tdDnodes.starttaosd(dnode.index)
             tdCases.logSql(logSql)
+
+            if restful:
+                tAdapter.deploy(adapter_cfg_dict)
+                tAdapter.start()
+
             if not restful:
-                conn = taos.connect(
-                    host,
-                    config=tdDnodes.getSimCfgPath())
+                conn = taos.connect(host,config=tdDnodes.getSimCfgPath())
             else:
                 conn = taosrest.connect(url=f"http://{host}:6041")
             print(tdDnodes.getSimCfgPath(),host)
@@ -394,9 +463,7 @@ if __name__ == "__main__":
         else:
             tdLog.info("Procedures for testing self-deployment")
             if not restful:
-                conn = taos.connect(
-                    host,
-                    config=tdDnodes.getSimCfgPath())
+                conn = taos.connect(host,config=tdDnodes.getSimCfgPath())
             else:
                 conn = taosrest.connect(url=f"http://{host}:6041")