xiaochuan correct his bugs in multiapp examples, you can try it again now
This commit is contained in:
@@ -86,7 +86,6 @@ def get_vm_file(env, config: Dict[str, Any]) -> Union[Optional[str], List[Option
|
||||
if not config.get("multi", False):
|
||||
paths: List[str] = [config["path"]]
|
||||
dests: List[str] = [config["dest"]]
|
||||
print(config)
|
||||
if "time_suffix" in config.keys() and config["time_suffix"]:
|
||||
if "time_format" in config.keys():
|
||||
time_format = config["time_format"]
|
||||
@@ -96,8 +95,6 @@ def get_vm_file(env, config: Dict[str, Any]) -> Union[Optional[str], List[Option
|
||||
else:
|
||||
paths: List[str] = config["path"]
|
||||
dests: List[str] = config["dest"]
|
||||
print(paths)
|
||||
print(dests)
|
||||
|
||||
cache_paths: List[str] = []
|
||||
|
||||
@@ -121,9 +118,6 @@ def get_vm_file(env, config: Dict[str, Any]) -> Union[Optional[str], List[Option
|
||||
cache_paths.append(_path)
|
||||
with open(_path, "wb") as f:
|
||||
f.write(file)
|
||||
# debug
|
||||
print("cache_paths")
|
||||
print(cache_paths)
|
||||
return cache_paths[0] if len(cache_paths)==1 else cache_paths
|
||||
|
||||
|
||||
|
||||
@@ -68,7 +68,7 @@ from .general import (
|
||||
compare_time_in_speedtest_results,
|
||||
is_included_all_json_objects,
|
||||
is_gold_text_included_in_pdf,
|
||||
check_csv_line_number,
|
||||
check_line_number,
|
||||
file_contains,
|
||||
compare_terminal_and_txt,
|
||||
fuzzy_place_math,
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import csv
|
||||
import os
|
||||
import datetime
|
||||
import difflib
|
||||
import functools
|
||||
@@ -91,23 +92,24 @@ def fuzzy_match(result, rules) -> float:
|
||||
|
||||
|
||||
def fuzzy_place_math(result_file_path, rules) -> float:
|
||||
if result_file_path is None:
|
||||
return 0.
|
||||
expect = rules["expected"] # a list of possible answers
|
||||
# read list.docx, and get all texts out, overlook blank lines, remove blanks before and after each line
|
||||
doc = Document(result_file_path)
|
||||
words_list = []
|
||||
for para in doc.paragraphs:
|
||||
words_list.extend(para.text.split())
|
||||
# Print out the list of extracted words
|
||||
print("Your Answers: ")
|
||||
print(words_list)
|
||||
fuzzy_score_list = []
|
||||
for word in words_list:
|
||||
max_score = 0
|
||||
for ans in expect:
|
||||
score = fuzz.ratio(word, ans)
|
||||
score = fuzz.ratio(word, ans)/100
|
||||
max_score = max(max_score, score)
|
||||
fuzzy_score_list.append(max_score)
|
||||
return sum(fuzzy_score_list) / len(fuzzy_score_list)
|
||||
if len(fuzzy_score_list) != 3:
|
||||
return 0.
|
||||
return sum(fuzzy_score_list) / 3
|
||||
|
||||
def check_csv(result: str, rules: Dict[str, List[Dict[str, str]]]) -> float:
|
||||
"""
|
||||
@@ -304,29 +306,40 @@ def check_direct_json_object(result, rules) -> float:
|
||||
result = result.replace("'", '"')
|
||||
# load json object
|
||||
result = json.loads(result)
|
||||
print("result: ")
|
||||
print(result)
|
||||
print("expected: ")
|
||||
print(rules["expected"])
|
||||
if result is None:
|
||||
return 0.
|
||||
try:
|
||||
expect_in_result = rules.get("expect_in_result", False)
|
||||
if not expect_in_result:
|
||||
expected_json = rules["expected"]
|
||||
for key in expected_json.keys():
|
||||
expected_value = expected_json.get(key)
|
||||
if expected_value != result.get(key):
|
||||
return 0.
|
||||
return 1.0
|
||||
else:
|
||||
expected_json = rules["expected"]
|
||||
|
||||
expect_in_result = rules.get("expect_in_result", False)
|
||||
if not expect_in_result:
|
||||
expected_json = rules["expected"]
|
||||
for key in expected_json.keys():
|
||||
expected_value = expected_json.get(key)
|
||||
if expected_value != result.get(key):
|
||||
return 0.
|
||||
return 1.0
|
||||
else:
|
||||
expected_json = rules["expected"]
|
||||
for key in expected_json.keys():
|
||||
expected_value = expected_json.get(key)
|
||||
if expected_value not in result.get(key):
|
||||
return 0.
|
||||
return 1.0
|
||||
|
||||
for key in expected_json.keys():
|
||||
if isinstance(expected_json.get(key), List):
|
||||
flag = 0
|
||||
expected_value_list = expected_json.get(key)
|
||||
for each_expected_value in expected_value_list:
|
||||
if each_expected_value in result.get(key):
|
||||
flag = 1
|
||||
break
|
||||
if flag == 0:
|
||||
return 0.
|
||||
elif isinstance(expected_json.get(key), str):
|
||||
if expected_json.get(key) not in result.get(key):
|
||||
return 0.
|
||||
else:
|
||||
logger.debug("check_direct_json_object: expected value type not supported")
|
||||
return 0.
|
||||
return 1.0
|
||||
except:
|
||||
logger.debug("check_direct_json_object: result is not a valid json object")
|
||||
return 0.
|
||||
|
||||
def compare_time_in_speedtest_results(speedtest_result_path, time_diff):
|
||||
if not speedtest_result_path:
|
||||
@@ -397,24 +410,37 @@ def is_gold_text_included_in_pdf(pdf_file_path, gold_text_path):
|
||||
def file_contains(file_path, config):
|
||||
# file_path ends with .txt
|
||||
if not file_path:
|
||||
return 1
|
||||
with open(file_path, 'r') as f:
|
||||
file_text = f.read()
|
||||
for text in config["expected"]:
|
||||
if text not in file_text:
|
||||
return 0
|
||||
return 1
|
||||
return 0.
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
file_text = f.read()
|
||||
for text in config["expected"]:
|
||||
if text not in file_text:
|
||||
logger.debug(f"file_contains: {text} not found in {file_path}")
|
||||
return 0.
|
||||
except:
|
||||
logger.debug("file_contains: file not found or not readable")
|
||||
return 0.
|
||||
return 1.
|
||||
|
||||
|
||||
def check_csv_line_number(file_path, line_number):
|
||||
# check file_path suffix
|
||||
if not file_path.endswith('.csv'):
|
||||
return 0
|
||||
# check line number
|
||||
with open(file_path, 'r') as f:
|
||||
reader = csv.reader(f)
|
||||
line_count = sum(1 for row in reader)
|
||||
return 1 if line_count == int(line_number["expected"]) else 0
|
||||
def check_line_number(file_path, line_number):
|
||||
# check if file_path exists
|
||||
if file_path is None or not os.path.isfile(file_path):
|
||||
return 0.
|
||||
timeRegex = "([01]\d|2[0-3]):[0-5]\d:[0-5]\d"
|
||||
# check if the string that matches the timeRegex in this txt file equals to line_number["expected"]
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
line_count = 0
|
||||
for line in f:
|
||||
if re.search(timeRegex, line):
|
||||
line_count += 1
|
||||
# if line_count equals to line_number["expected"], return 1, else return 0
|
||||
return 1 if line_count == int(line_number["expected"]) else 0
|
||||
except:
|
||||
logger.debug("check_line_number: file not found or not readable")
|
||||
return 0.
|
||||
|
||||
|
||||
def compare_terminal_and_txt(txt_file_path, terminal_output):
|
||||
|
||||
Reference in New Issue
Block a user