-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathscript_sample_analyze_projet_scenario
More file actions
64 lines (52 loc) · 2.64 KB
/
script_sample_analyze_projet_scenario
File metadata and controls
64 lines (52 loc) · 2.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def verify_scenario_steps_and_configuration(project):
''' Verify the existence of build, clean, compute metrics, run checks steps,
and check for scenario triggers and reporters.
Parameters:
-----------
project: DSS Project
DSS project
Returns:
--------
scenario_validation: dict
Contains validation status and reason for required steps, triggers, and reporters.
'''
scenarios_list = project.list_scenarios()
#you need to define every validation criteria you want to evaluate
scenario_validation = {
'build_steps': [],
'clean_steps': [],
'compute_metrics_steps': [],
'run_checks_steps': [],
'triggers': [],
'reporters': []
}
for scenario in scenarios_list:
sc_id = scenario['id']
scenario_obj = project.get_scenario(sc_id)
settings = scenario_obj.get_settings().get_raw()
steps = settings.get('params', {}).get('steps', [])
# Check for required steps
for step in steps:
if step.get('type') == 'build_flowitem':
scenario_validation['build_steps'].append(sc_id)
if step.get('type') == 'runnable' and 'pyrunnable_clear-intermediate-datasets' in step.get('params', {}).get('runnableType', ''):
scenario_validation['clean_steps'].append(sc_id)
if step.get('type') == 'compute_metrics':
scenario_validation['compute_metrics_steps'].append(sc_id)
if step.get('type') == 'run_checks':
scenario_validation['run_checks_steps'].append(sc_id)
# Examine triggers
if 'triggers' in settings:
scenario_validation['triggers'].append({'scenario': sc_id, 'triggers': settings['triggers']})
# Examine reporters
if 'reporters' in settings:
scenario_validation['reporters'].append({'scenario': sc_id, 'reporters': settings['reporters']})
# Determine status and reason based on findings
for step_type, step_list in scenario_validation.items():
if step_type in ['build_steps', 'clean_steps', 'compute_metrics_steps', 'run_checks_steps']:
status_key = f'{step_type[:-5]}_status' # Removes '_steps' and appends '_status'
if not step_list:
scenario_validation[status_key] = {'status': 'KO', 'reason': f'No {step_type[:-5].replace("_", " ")} step found in any scenario.'}
else:
scenario_validation[status_key] = {'status': 'OK', 'reason': f'{step_type[:-5].replace("_", " ")} step found in scenarios: {step_list}'}
return scenario_validation