import os
import sys
import json
from scenario_exe_parser import parse_test_scenario
import subprocess
import os
import sys
import json
import subprocess
# Assuming parse_test_scenario is imported correctly
# from scenario_exe_parser import parse_test_scenario
# --- Global Paths ---
current_directory = os.path.dirname(os.path.abspath(__file__))
REPO_PATH = os.path.join(current_directory, "Sensor_hub_repo")
COMPONENT_DIR = os.path.join(REPO_PATH, "components")
RESULT_PATH = "/home/asf/testarena"
# The HTML Template
REPORT_TEMPLATE = """
ESP32 Test Execution Report
Overall Scenario Summary
| Scenario Name |
Final Result |
| {{scenario_name}} |
{{overall_status}} |
Detailed Test Cases
| Test Case ID |
Result |
Execution Log |
{{test_case_rows}}
"""
def run_test_suite(tasks):
aggregated_results = {}
shell_script = "./TPF/test_execution.sh"
if os.name != 'nt':
subprocess.run(["chmod", "+x", shell_script])
for task in tasks:
print(f"--- Starting Task: {task['id']} ---")
result = subprocess.run(
[shell_script, task['id'], task['cmd'], task['path'], REPO_PATH],
capture_output=True, text=True
)
print(result.stdout)
json_found = False
for line in result.stdout.splitlines():
if line.startswith("FINAL_JSON_OUTPUT:"):
json_string = line.replace("FINAL_JSON_OUTPUT:", "").strip()
try:
task_json = json.loads(json_string)
aggregated_results.update(task_json)
json_found = True
except json.JSONDecodeError as e:
print(f"!!! JSON Parsing Error: {e}")
if not json_found:
aggregated_results[task['id']] = ["ERROR", "N/A"]
return aggregated_results
def generate_html_report(scenario_name, results, output_path):
all_passed = all(info[0] == "PASS" for info in results.values())
overall_status = "PASS" if all_passed else "FAIL"
overall_class = "status-pass" if all_passed else "status-fail"
test_case_rows = ""
for tc_id, info in results.items():
status = info[0]
log_url = info[1]
status_class = "status-pass" if status == "PASS" else "status-fail"
test_case_rows += f"""
| {tc_id} |
{status} |
View Log |
"""
# Use the global REPORT_TEMPLATE
report_content = REPORT_TEMPLATE.replace("{{scenario_name}}", scenario_name) \
.replace("{{overall_status}}", overall_status) \
.replace("{{overall_class}}", overall_class) \
.replace("{{test_case_rows}}", test_case_rows)
report_file = os.path.join(output_path, "execution_report.html")
with open(report_file, "w") as f:
f.write(report_content)
print(f"HTML Report generated at: {report_file}")
def save_summary(results, task_id_path):
json_path = os.path.join(task_id_path, "final_summary.json")
with open(json_path, "w") as f:
json.dump(results, f, indent=4)
print(f"\nFinal results saved to {json_path}")
if __name__ == "__main__":
queue_id = "1234"
scenario_path = "application_layer/business_stack/actuator_manager/test/actuator_manager_init_test.test_scenario.xml"
task_id = "56754"
# Path logic
queue_path = os.path.join(RESULT_PATH, queue_id)
task_id_path = os.path.join(queue_path, task_id) # Corrected pathing
os.makedirs(task_id_path, exist_ok=True)
# Note: Ensure parse_test_scenario is defined or imported
scenario_data = parse_test_scenario(os.path.join(COMPONENT_DIR, scenario_path))
my_tasks = []
sub_tasks_data = scenario_data['test_cases']
for case_id, exec_cmd in sub_tasks_data.items():
my_tasks.append({
"id": case_id,
"cmd": exec_cmd,
"path": task_id_path
})
final_data = run_test_suite(my_tasks)
save_summary(final_data, task_id_path)
# Generate report INSIDE the task folder
generate_html_report(os.path.basename(scenario_path), final_data, task_id_path)