xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

test: resume three functions in common.py

This commit is contained in:
jiajingbin 2023-08-23 09:25:07 +08:00 committed by Haojun Liao
parent e073e1c1ae
commit 99a0a15385
1 changed files with 34 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
tests/pytest/util/common.py
View File

@ -813,40 +813,6 @@ class TDCom:
"""
return ','.join(map(lambda i: f'{gen_type}{i} {data_type}', range(count)))
def is_json(msg):
if isinstance(msg, str):
try:
json.loads(msg)
return True
except:
return False
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)
# stream
def create_stream(self, stream_name, des_table, source_sql, trigger_mode=None, watermark=None, max_delay=None, ignore_expired=None, ignore_update=None, subtable_value=None, fill_value=None, fill_history_value=None, stb_field_name_value=None, tag_value=None, use_exist_stb=False, use_except=False):
"""create_stream
@ -1859,4 +1825,38 @@ class TDCom:
if i == 1:
self.record_history_ts = ts_value
def is_json(msg):
if isinstance(msg, str):
try:
json.loads(msg)
return True
except:
return False
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()