ASF-Nabd/ASF_tools_legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add branch validation to queue

Browse Source
This commit is contained in:
mahmamdouh
2025-12-22 15:04:06 +01:00
parent 26b1f256f4
commit 5f4eaaaa20
4 changed files with 394 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
asf-cloud-server/testarena_1/app/routes/jobs.py
View File

@@ -138,26 +138,61 @@ def submit_step1():
}
})
# If successful, get available scenarios (mock for now)
scenarios = [
'Scenario_1_Basic_Test',
'Scenario_2_Advanced_Test',
'Scenario_3_Integration_Test',
'Scenario_4_Performance_Test',
'Scenario_5_Security_Test'
]
# If successful, run scenario scan to get available scenarios
print(f"[DEBUG] Running scenario scan...")
scan_cmd = f"sshpass -p '{ssh_password}' ssh {ssh_options} {ssh_user}@{ssh_host} 'python3 TPF/Sensor_hub_repo/Tools/TPF/scenario_scan.py'"
print(f"[DEBUG] Executing scan command: {scan_cmd.replace(ssh_password, '***')}")
scan_result = subprocess.run(scan_cmd, shell=True, capture_output=True, text=True, timeout=120)
print(f"[DEBUG] Scan result - Return code: {scan_result.returncode}")
print(f"[DEBUG] Scan stdout: {scan_result.stdout}")
print(f"[DEBUG] Scan stderr: {scan_result.stderr}")
# Parse the JSON response from scenario scan
try:
scan_data = json.loads(scan_result.stdout)
organized_data = scan_data.get('organized_data', {})
scenario_map = scan_data.get('scenario_map', {})
print(f"[DEBUG] Found {len(scenario_map)} scenarios in {len(organized_data)} categories")
except json.JSONDecodeError as e:
print(f"[ERROR] Failed to parse scenario scan JSON: {e}")
print(f"[ERROR] Raw output: {scan_result.stdout}")
# Fallback to mock data if scan fails
organized_data = {
"drivers": {
"diag_protocol_stack": ["diag_protocol_stack_init_test"]
},
"application_layer": {
"data_pool": ["data_pool_init_test"],
"actuator_manager": ["actuator_manager_init_test2", "actuator_manager_init_test"],
"event_system": ["event_system_init_test"]
}
}
scenario_map = {
"diag_protocol_stack_init_test": "drivers/diag_protocol_stack/test/diag_protocol_stack_init_test.test_scenario.xml",
"data_pool_init_test": "application_layer/DP_stack/data_pool/test/data_pool_init_test.test_scenario.xml",
"actuator_manager_init_test2": "application_layer/business_stack/actuator_manager/test/actuator_manager_init_test2.test_scenario.xml",
"actuator_manager_init_test": "application_layer/business_stack/actuator_manager/test/actuator_manager_init_test.test_scenario.xml",
"event_system_init_test": "application_layer/business_stack/event_system/test/event_system_init_test.test_scenario.xml"
}
print(f"[DEBUG] Branch validation successful for: {branch_name}")
return jsonify({
'success': True,
'scenarios': scenarios,
'organized_data': organized_data,
'scenario_map': scenario_map,
'message': f'Branch "{branch_name}" validated successfully',
'output': full_output,
'debug': {
'clone_returncode': clone_result.returncode,
'checkout_returncode': checkout_result.returncode,
'branch_name': branch_name
'scan_returncode': scan_result.returncode if 'scan_result' in locals() else 'N/A',
'branch_name': branch_name,
'scenarios_found': len(scenario_map)
}
})
@@ -182,15 +217,20 @@ def submit_step1():
@login_required
def submit_step2_validated():
branch_name = request.form.get('branch_name')
scenarios_json = request.form.get('scenarios')
organized_data_json = request.form.get('organized_data')
scenario_map_json = request.form.get('scenario_map')
try:
scenarios = json.loads(scenarios_json)
except:
flash('Invalid scenarios data', 'error')
organized_data = json.loads(organized_data_json) if organized_data_json else {}
scenario_map = json.loads(scenario_map_json) if scenario_map_json else {}
except json.JSONDecodeError:
flash('Invalid scenario data', 'error')
return redirect(url_for('jobs.submit'))
return render_template('jobs/submit_step2.html', branch_name=branch_name, scenarios=scenarios)
return render_template('jobs/submit_step2.html',
branch_name=branch_name,
organized_data=organized_data,
scenario_map=scenario_map)
@login_required
def submit_step2():
branch_name = request.form.get('branch_name')