import os
import sys
import json
import subprocess
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 = {}
# Use path relative to this script
script_dir = os.path.dirname(os.path.abspath(__file__))
shell_script = os.path.join(script_dir, "test_execution.sh")
if os.name != 'nt':
subprocess.run(["chmod", "+x", shell_script])
for task in tasks:
print(f"--- Starting Task: {task['id']} ---")
# Use Popen to stream output in real-time
env = os.environ.copy()
env["PYTHONUNBUFFERED"] = "1"
process = subprocess.Popen(
[shell_script, task['id'], task['cmd'], task['path'], REPO_PATH],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
universal_newlines=True,
env=env
)
full_output = ""
for line in process.stdout:
print(line, end="")
full_output += line
process.wait()
json_found = False
for line in full_output.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]
log_url = log_url.replace("file:///home/asf/testarena/", "http://asf-server.duckdns.org:8080/results/")
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__":
if len(sys.argv) > 3:
queue_id = sys.argv[1] #"1234"
scenario_path = sys.argv[2] #"application_layer/business_stack/actuator_manager/test/actuator_manager_init_test.test_scenario.xml"
task_id = sys.argv[3] #"56754"
else:
print("Usage: python scenario_execution.py ")
sys.exit(1)
# 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)