in src/send_test_notification.py [0:0]
def parse_pytest_data():
"""
Parse pytest output to get test results.
"""
pytest_raw_data = get_pytest_output()
pytest_parsed_output = []
for file in pytest_raw_data:
pytest_file_data = {}
pytest_file_data["file_name"] = file
pytest_file_data["failed_tests"] = {}
for test in pytest_raw_data[file]["testsuites"]["testsuite"]["testcase"]:
if "failure" in test:
# Team info of the failed test is propogated from team marker added on the test function to the properties section in the pytest xml report
if "properties" in test:
team_name = test["properties"]["property"]["@value"]
print(f"Test failed for team {team_name}")
if team_name not in pytest_file_data["failed_tests"]:
print("Team name not found for the failed test")
pytest_file_data["failed_tests"][team_name] = []
test_data = {}
test_name, ecr_image, instance_name = get_test_details(test["@name"])
test_data["test_name"] = test_name
print(f"Processing information for failed test: {test_name}")
if ecr_image is not None:
test_data["ecr_image"] = ecr_image
if instance_name is not None:
test_data["instance_name"] = instance_name
test_data[test["properties"]["property"]["@name"]] = test["properties"][
"property"
]["@value"]
test_data["test_path"] = test["@classname"].replace(".", "/") + "/" + test_name
test_data["fail_message"] = test["failure"]["@message"]
fail_full_message = test["failure"]["#text"]
if check_for_infrastructure_exceptions(fail_full_message):
print("Infrastructure failure found in the test. Skipping test details")
else:
pytest_file_data["failed_tests"][team_name].append(test_data)
else:
print("Test has no team name. Skipping test details")
failed_test_for_file = pytest_file_data["failed_tests"].copy()
for team_name in failed_test_for_file:
if not failed_test_for_file[team_name]:
del pytest_file_data["failed_tests"][team_name]
if pytest_file_data["failed_tests"]:
pytest_parsed_output.append(pytest_file_data)
return pytest_parsed_output