import pytest from calculate.scripts.scripts import Var, Task, Static, TaskError, Done,\ DoneAny, Success, SuccessAny, Failed,\ FailedAny, Block, For, While, Until,\ Handler, Script, Run, ScriptError,\ ActionError from calculate.variables.datavars import Namespace, Variable, IntegerType,\ StringType, BooleanType, ListType,\ HashType from calculate.utils.io_module import IOModule @pytest.mark.tasks class TestTasks(): def test_if_arguments_of_the_Var_objects_are_correct_and_they_are_called_with_datavars_namespace_and_script_namespace_arguments__this_objects_can_be_used_for_different_checks(self): Namespace.reset() datavars = Namespace.datavars script_namespace = Script.make_script_variables('test_script', [], datavars) with Namespace('os'): Variable('var_1', source='value', type=StringType) Variable('var_2', source=2, type=IntegerType) Variable('var_3', source=True, type=BooleanType) Variable('var_4', source=False, type=BooleanType) Variable('var_5', source=['a', 'b', 'c'], type=ListType) Variable('var_6', source=2, type=IntegerType) Variable('var_7', source='a', type=StringType) Variable('var_8', source='text matchtext', type=StringType) with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) Variable('var_2', source=2, type=IntegerType) assert Var('os.var_3')(datavars, None, script_namespace) assert ~Var('os.var_4')(datavars, None, script_namespace) assert (Var('os.var_2') == 2)(datavars, None, script_namespace) assert ((Var("os.var_2") == 2) & (Var("os.var_1") == "value"))(datavars, None, script_namespace) assert ~((Var("os.var_2") == 2) & (Var("os.var_1") == "other_value"))(datavars, None, script_namespace) assert ((Var("os.var_2") == 3) | (Var("os.var_1") == "value"))(datavars, None, script_namespace) assert ~((Var("os.var_2") == 3) | (Var("os.var_1") == "other_value"))(datavars, None, script_namespace) assert (Var("os.var_2") < 3)(datavars, None, script_namespace) assert (Var("os.var_2") > 1)(datavars, None, script_namespace) assert (Var("os.var_2") <= 3)(datavars, None, script_namespace) assert (Var("os.var_2") <= 2)(datavars, None, script_namespace) assert (Var("os.var_2") >= 1)(datavars, None, script_namespace) assert (Var("os.var_2") >= 2)(datavars, None, script_namespace) assert Var("os.var_5").has('a')(datavars, None, script_namespace) assert (Var("os.var_5") << 'a')(datavars, None, script_namespace) assert ~(Var("os.var_5") << 'd')(datavars, None, script_namespace) assert (Var('os.var_6') == Var('os.var_2'))(datavars, None, script_namespace) assert (Var('os.var_5') << Var('os.var_7'))(datavars, None, script_namespace) assert (Var('.var_2') == 2)(datavars, datavars.os.linux, script_namespace) assert Var('.var_8').match('match')(datavars, datavars.os, script_namespace) assert ~Var('.var_8').match(r'match\b')(datavars, datavars.os, script_namespace) assert (Var('.var_8').regex('match', 'text ') == 'text text text' )(datavars, datavars.os, script_namespace) assert (Var('.var_8').regex('match\b', 'text ') != 'text text text' )(datavars, datavars.os, script_namespace) assert (Var('.var_8').replace('match', 'text ') == 'text text text' )(datavars, datavars.os, script_namespace) def test_if_script_object_is_created_with_one_correct_task_without_any_conditions__the_task_will_be_successfully_completed(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var', source=2, type=IntegerType) def action(output, arg1): return f'os.var = {arg1}' Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=["os.var"]) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task' in datavars.scripts.test_script assert datavars.scripts.test_script.task.get_hash() ==\ {'result': 'os.var = 2', 'success': True, 'error_message': None} def test_if_script_object_is_created_with_one_correct_task_with_a_fulfilled_condition__the_task_will_be_successfully_completed(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source='var_value', type=StringType) Variable('var_2', source=2, type=IntegerType) Variable('var_3', source=True, type=BooleanType) def action(output, arg1, arg2): return arg1 + ' and ' + arg2 Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=[Static('static_value'), "os.var_1"], when=(Var('os.var_2') & Var('os.var_3'))) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task' in datavars.scripts.test_script assert datavars.scripts.test_script.task.get_hash() ==\ {'result': 'static_value and var_value', 'success': True, 'error_message': None} def test_if_script_object_is_created_with_one_correct_task_with_a_unfulfilled_condition__the_task_will_be_successfully_completed(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(output: IOModule, arg1: str): return arg1 Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=[Static('value')], when=~Var('os.linux.var_1')) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task' not in datavars.scripts.test_script def test_if_script_object_is_created_with_one_failing_not_essential_task__the_script_execution_will_be_continued_and_task_result_will_contain_a_failed_status_and_error_message(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(output: IOModule, arg1: str): raise Exception('Test error') return arg1 Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=[Static('value')], essential=False, when=Var('os.linux.var_1')) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task' in datavars.scripts.test_script assert datavars.scripts.test_script.task.get_hash() ==\ {'result': None, 'success': False, 'error_message': 'Test error'} def test_if_script_object_is_created_with_one_failing_essential_task__the_script_execution_will_be_interrupted(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(output: IOModule, arg1: str): raise Exception('Test error') return arg1 with pytest.raises(TaskError): Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=[Static('value')], essential=True, when=Var('os.linux.var_1')) ).initialize(IOModule(), datavars, None)() def test_if_script_object_is_created_with_some_essential_tasks_with_conditions_and_some_conditions_uses_done_to_check_if_tasks_from_Done_or_DoneAny_arguments_were_completed__the_script_executes_tasks_which_done_condition_is_fulfilled(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' Script('test_script' ).tasks( Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')], when=~Var('os.linux.var_1')), Task(id='task_4', name="Task 4", action=action, args=[Static('value_4')], when=(Done('task_1', 'task_2') & Var('os.linux.var_1'))), Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')], when=(Var('os.linux.var_1') & Done('task_1', 'task_3'))), Task(id='task_6', name="Task 4", action=action, args=[Static('value_4')], when=DoneAny('task_1', 'task_3')) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_3' not in datavars.scripts.test_script assert 'task_4' in datavars.scripts.test_script assert 'task_5' not in datavars.scripts.test_script assert 'task_6' in datavars.scripts.test_script def test_if_script_object_is_created_with_some_essential_tasks_with_conditions_and_some_conditions_uses_done_to_check_if_tasks_from_Success_or_SuccessAny_arguments_were_completed__the_script_executes_tasks_witch_done_condition_is_fulfilled(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' def bad_action(arg1: str): raise Exception('Error lol') return f'bad action -> {arg1}' Script('test_script' ).tasks( Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), Task(id='task_3', name="Task 3", essential=False, action=bad_action, args=[Static('value_3')]), Task(id='task_4', name="Task 4", action=action, args=[Static('value_4')], when=~Var('os.linux.var_1')), Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')], when=(Success('task_1', 'task_2') & Var('os.linux.var_1'))), Task(id='task_6', name="Task 6", action=action, args=[Static('value_6')], when=(Var('os.linux.var_1') & Success('task_1', 'task_3'))), Task(id='task_7', name="Task 7", action=action, args=[Static('value_7')], when=(Success('task_1', 'task_4') & Var('os.linux.var_1'))), Task(id='task_8', name="Task 8", action=action, args=[Static('value_8')], when=(SuccessAny('task_1', 'task_2'))), Task(id='task_9', name="Task 9", action=action, args=[Static('value_9')], when=(SuccessAny('task_1', 'task_3') & Var('os.linux.var_1'))), Task(id='task_10', name="Task 10", action=action, args=[Static('value_10')], when=(Var('os.linux.var_1') & SuccessAny('task_3', 'task_4'))) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_4' not in datavars.scripts.test_script assert 'task_5' in datavars.scripts.test_script assert 'task_6' not in datavars.scripts.test_script assert 'task_7' not in datavars.scripts.test_script assert 'task_8' in datavars.scripts.test_script assert 'task_9' in datavars.scripts.test_script assert 'task_10' not in datavars.scripts.test_script def test_if_script_object_is_created_with_some_essential_tasks_with_conditions_and_some_conditions_uses_done_to_check_if_tasks_from_Failed_or_FailedAny_arguments_were_completed__the_script_executes_tasks_witch_done_condition_is_fulfilled(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' def bad_action(arg1: str): raise Exception('Error lol') return f'bad action -> {arg1}' Script('test_script' ).tasks( Task(id='task_0', name="Task 0", action=action, args=[Static('value_0')]), Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", essential=False, action=bad_action, args=[Static('value_2')]), Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')], when=~Var('os.linux.var_1')), Task(id='task_4', name="Task 4", action=action, args=[Static('value_4')], when=(Failed('task_2', 'task_3') & Var('os.linux.var_1'))), Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')], when=(Var('os.linux.var_1') & Failed('task_1', 'task_2'))), Task(id='task_6', name="Task 6", action=action, args=[Static('value_6')], when=Failed('task_2', 'task_3')), Task(id='task_7', name="Task 7", action=action, args=[Static('value_7')], when=FailedAny('task_0', 'task_1')), Task(id='task_8', name="Task 8", action=action, args=[Static('value_8')], when=(FailedAny('task_1', 'task_2') & Var('os.linux.var_1'))), Task(id='task_9', name="Task 9", action=action, args=[Static('value_9')], when=(Var('os.linux.var_1') & FailedAny('task_2', 'task_3'))) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert not datavars.scripts.test_script.task_2.success assert 'task_3' not in datavars.scripts.test_script assert 'task_4' in datavars.scripts.test_script assert 'task_5' not in datavars.scripts.test_script assert 'task_6' in datavars.scripts.test_script assert 'task_7' not in datavars.scripts.test_script assert 'task_8' in datavars.scripts.test_script assert 'task_9' in datavars.scripts.test_script def test_if_script_object_is_created_with_essential_task_which_action_function_has_some_type_annotations_and_task_s_arguments_have_incorrect_types__the_script_interrupts_with_TaskError(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var', source={'key1': 'value1', 'key2': 'value2'}, type=HashType) def action(arg1: str): return arg1 with pytest.raises(TaskError): Script('test_script' ).tasks( Task(id='task_1', name="Task 1", action=action, args=['os.linux.var']) ).initialize(IOModule(), datavars, None)() def test_if_script_object_is_created_with_essential_task_having_set_parameter_containing_a_string_with_path_to_an_existing_variable__the_task_s_execution_result_will_be_set_to_the_variable_from_set_parameter(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=1, type=IntegerType) assert datavars.os.linux.var_2 == 1 def action(arg1: int): return arg1 * arg1 Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=['os.linux.var_1'], set='os.linux.var_2') ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task' in datavars.scripts.test_script assert datavars.os.linux.var_2 == 144 def test_if_script_object_is_created_with_essential_task_having_set_parameter_containing_a_list_of_paths_to_existing_variables__the_task_s_execution_result_will_be_set_to_the_variables_from_this_list(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=1, type=IntegerType) with Namespace('linux'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=1, type=IntegerType) assert datavars.os.linux.var_2 == 1 def action(arg1: int): return arg1 * arg1 Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=['os.linux.var_1'], set=['os.var_1', 'os.linux.var_2']) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task' in datavars.scripts.test_script assert datavars.os.linux.var_2 == 144 assert datavars.os.var_1 == 144 def test_if_script_object_is_created_with_essential_task_having_set_parameter_containing_a_tuple_of_paths_to_existing_variables__the_task_s_execution_result_will_be_set_to_the_variables_from_this_tuple(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=1, type=IntegerType) with Namespace('linux'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=1, type=IntegerType) assert datavars.os.linux.var_2 == 1 def action(arg1: int) -> int: return arg1 * arg1 Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=['os.linux.var_1'], set=('os.var_1', 'os.linux.var_2')) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task' in datavars.scripts.test_script assert datavars.os.linux.var_2 == 144 assert datavars.os.var_1 == 144 def test_if_script_object_is_created_with_essential_task_having_set_parameter_containing_a_dict_with_field_names_and_paths_to_existing_variables_and_action_function_returns_dictionary_with_some_fields_from_the_first_dict__the_values_from_dict_returned_by_action_function_will_be_set_to_the_variables_from_the_same_field(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source='value_1', type=StringType) Variable('var_2', source='value_2', type=StringType) with Namespace('linux'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=24, type=IntegerType) def action(arg1: int, arg2: int): return {'field_1': f'value_{arg1}', 'field_2': f'value_{arg2}'} Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=['os.linux.var_1', 'os.linux.var_2'], set={'field_1': 'os.var_1', 'field_2': 'os.var_2'}) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task' in datavars.scripts.test_script assert datavars.os.var_1 == 'value_12' assert datavars.os.var_2 == 'value_24' def test_if_script_object_is_created_with_essential_task_having_action_function_with_a_return_type_annotation_and_action_function_returns_value_of_different_type__the_task_fails(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source='value', type=StringType) def action(arg1: str) -> str: return len(arg1) with pytest.raises(TaskError): Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=['os.linux.var_1']) ).initialize(IOModule(), datavars, None)() def test_if_script_object_is_created_with_essential_task_having_action_function_with_a_return_type_annotation_and_set_parameter_containing_variable_of_the_different_type__the_task_fails_to_set_returned_value_to_this_variable(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source='value', type=StringType) Variable('var_2', source=12, type=IntegerType) def action(arg1: str) -> str: return arg1 with pytest.raises(TaskError): Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=['os.linux.var_1'], set='os.linux.var_2') ).initialize(IOModule(), datavars, None)() def test_if_script_object_is_created_with_plain_block_of_correct_tasks__the_script_will_be_executed_successfully(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' Script('test_script' ).tasks( Block(Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')])) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert datavars.scripts.test_script.task_1.success assert 'task_2' in datavars.scripts.test_script assert datavars.scripts.test_script.task_2.success def test_if_script_object_is_created_with_block_with_fulfilled_condition__the_script_will_execute_block_s_tasks_successfully(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' Script('test_script' ).tasks( Block(Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), when=Var('os.linux.var_1')) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_2' in datavars.scripts.test_script def test_if_script_object_is_created_with_block_with_unfulfilled_condition__the_script_will_ignore_block_s_tasks(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' Script('test_script' ).tasks( Block(Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), when=~Var('os.linux.var_1')) ).initialize(IOModule(), datavars, None)() assert 'task_3' not in datavars.scripts.test_script assert 'task_4' not in datavars.scripts.test_script def test_if_script_object_is_created_with_block_of_essential_tasks_having_rescue_tasks_and_one_of_them_fails__the_script_interrupts_execution_of_the_block_s_tasks__executes_tasks_from_the_rescue_and_interrupts_whole_execution(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' def bad_action(arg1: str): raise Exception('Error lol') return f'bad action -> {arg1}' with pytest.raises(TaskError): Script('test_script' ).tasks( Block(Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=bad_action, args=[Static('value_2')]) ).rescue( Task(id='task_3', name="Task 3", action=action, args=[Static('rescue_value')])) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_2' not in datavars.scripts.test_script assert 'task_3' in datavars.scripts.test_script def test_if_script_object_is_created_with_block_of_essential_tasks_having_loop_parameter_and_there_is_For_loop_with_static_list_in_it__the_block_tasks_will_be_iterated_using_values_from_this_list(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) output = [] expected = ['action_1 -> value_1', 'action_2 -> value_1', 'action_1 -> value_2', 'action_2 -> value_2', 'action_1 -> value_3', 'action_2 -> value_3'] def action_1(arg1: str): output.append(f'action_1 -> {arg1}') return f'action_1 -> {arg1}' def action_2(arg1: str): output.append(f'action_2 -> {arg1}') return f'action_2 -> {arg1}' Script('test_script' ).tasks( Block(Task(id='task_1', name="Task 1", action=action_1, args=['scripts.test_script.value']), Task(id='task_2', name="Task 2", action=action_2, args=['scripts.test_script.value']), loop=For('value', ['value_1', 'value_2', 'value_3'])) ).initialize(IOModule(), datavars, None)() for l_value, r_value in zip(output, expected): assert l_value == r_value def test_if_script_object_is_created_with_block_of_essential_tasks_having_loop_parameter_and_there_is_For_loop_with_a_list_variable_in_it__the_block_tasks_will_be_iterated_using_values_from_this_variable(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=['value_1', 'value_2', 'value_3'], type=ListType) output = [] expected = ['action_1 -> value_1', 'action_2 -> value_1', 'action_1 -> value_2', 'action_2 -> value_2', 'action_1 -> value_3', 'action_2 -> value_3'] def action_1(arg1: str): output.append(f'action_1 -> {arg1}') return f'action_1 -> {arg1}' def action_2(arg1: str): output.append(f'action_2 -> {arg1}') return f'action_2 -> {arg1}' Script('test_script' ).tasks( Block(Task(id='task_1', name="Task 1", action=action_1, args=['scripts.test_script.value']), Task(id='task_2', name="Task 2", action=action_2, args=['scripts.test_script.value']), loop=For('value', 'os.linux.var_1')) ).initialize(IOModule(), datavars, None)() for l_value, r_value in zip(output, expected): assert l_value == r_value def test_if_script_object_is_created_with_block_of_one_essential_task_having_loop_parameter_and_there_is_While_loop_with_a_condition_in_it__the_block_tasks_will_be_iterated_while_condition_is_true(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('counter', source=0, type=IntegerType) Variable('ceiling', source=5, type=IntegerType) Variable('output', source=[], type=ListType) expected = ['counter = 0', 'counter = 1', 'counter = 2', 'counter = 3', 'counter = 4'] def action(counter: int, out: list): out.append(f'counter = {counter}') return {'counter': counter + 1, 'output': out} Script('test_script' ).tasks( Block(Task(id='task', name="Task", action=action, args=['os.counter', 'os.output'], set={'counter': 'os.counter', 'output': 'os.output'}), loop=While(Var('os.counter') < Var('os.ceiling'))) ).initialize(IOModule(), datavars, None)() assert datavars.os.output == expected def test_if_script_object_is_created_with_block_of_two_essential_tasks_having_loop_parameter_and_there_is_While_loop_with_a_condition_in_it__the_block_tasks_will_be_iterated_while_condition_is_true(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('counter', source=0, type=IntegerType) Variable('ceiling', source=5, type=IntegerType) Variable('flag', source=False, type=BooleanType) Variable('output', source=[], type=ListType) expected = ['counter = 0', 'counter = 1', 'counter = 2', 'counter = 3', 'counter = 4'] def action_1(counter: int, out: list): out.append(f'counter = {counter}') return {'counter': counter + 1, 'output': out} def action_2(counter: int, ceiling: int): return not (counter < ceiling) Script('test_script' ).tasks( Block(Task(id='task_1', name="Task 1", action=action_1, args=['os.counter', 'os.output'], set={'counter': 'os.counter', 'output': 'os.output'}), Task(id='task_2', name="Task 2", action=action_2, args=['os.counter', 'os.ceiling'], set='os.flag'), loop=While(~Var('os.flag'))) ).initialize(IOModule(), datavars, None)() assert datavars.os.output == expected def test_if_script_object_is_created_with_block_of_two_essential_tasks_having_loop_parameter_and_there_is_Until_loop_with_a_unfulfilled_condition_in_it__the_block_tasks_will_be_iterated_one_time(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('counter', source=0, type=IntegerType) Variable('flag', source=False, type=BooleanType) Variable('output', source=[], type=ListType) def action(counter: int, out: list): out.append(f'counter = {counter}') return {'counter': counter + 1, 'output': out} Script('test_script' ).tasks( Block(Task(id='task', name="Task", action=action, args=['os.counter', 'os.output'], set={'counter': 'os.counter', 'output': 'os.output'}), loop=Until(Var('os.flag'))) ).initialize(IOModule(), datavars, None)() assert datavars.os.output == ['counter = 0'] def test_if_script_object_is_created_with_block_of_two_essential_tasks_having_loop_parameter_and_there_is_Until_loop_with_a_condition_in_it__the_block_tasks_will_be_iterated_several_times(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('counter', source=0, type=IntegerType) Variable('ceiling', source=5, type=IntegerType) Variable('flag', source=False, type=BooleanType) Variable('output', source=[], type=ListType) expected = ['counter = 0', 'counter = 1', 'counter = 2', 'counter = 3', 'counter = 4'] def action_1(counter: int, out: list): out.append(f'counter = {counter}') return {'counter': counter + 1, 'output': out} def action_2(counter: int, ceiling: int): return not (counter < ceiling) Script('test_script' ).tasks( Block(Task(id='task_1', name="Task 1", action=action_1, args=['os.counter', 'os.output'], set={'counter': 'os.counter', 'output': 'os.output'}), Task(id='task_2', name="Task 2", action=action_2, args=['os.counter', 'os.ceiling'], set='os.flag'), loop=Until(~Var('os.flag'))) ).initialize(IOModule(), datavars, None)() assert datavars.os.output == expected def test_if_script_object_is_created_with_a_handler_and_some_task_has_notify_parameter_with_this_handler_s_id_in_it__the_script_executes_tasks_from_this_handler(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): with Namespace('linux'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=[Static('value_1')], notify='handler'), Handler('handler', Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]) ) ).initialize(IOModule(), datavars, None)() assert 'task_1' in datavars.scripts.test_script assert 'task_2' in datavars.scripts.test_script def test_if_script_object_is_created_with_some_tasks_block_with_rescue_and_handler_which_id_is_in_the_notify_parameter_of_the_correct_tasks__the_script_will_execute_tasks_from_script_root__block_tasks_and_tasks_from_handler_but_tasks_from_rescue_will_be_not_executed(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' Script('test_script', ).tasks( Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), Block(Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]), Task(id='task_4', name="Task 4", action=action, args=[Static('value_4')], notify='handler_1'), when=Var('os.var_1') ).rescue(Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')])), Task(id='task_6', name="Task 6", action=action, args=[Static('value_6')]), Handler('handler_1', Task(id='task_7', name="Task 7", action=action, args=[Static('value_7')]), Task(id='task_8', name="Task 8", action=action, args=[Static('value_8')]) ) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_2' in datavars.scripts.test_script assert 'task_3' in datavars.scripts.test_script assert 'task_4' in datavars.scripts.test_script assert 'task_5' not in datavars.scripts.test_script assert 'task_6' in datavars.scripts.test_script assert 'task_7' in datavars.scripts.test_script assert 'task_8' in datavars.scripts.test_script def test_if_script_object_is_created_with_some_essential_tasks_block_with_failing_task_and_rescue_and_handler_which_id_is_in_the_notify_parameter_of_the_failing_task__the_script_will_execute_tasks_before_block_and_first_task_of_block_and_then_will_fail_and_execute_rescue_tasks(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' def bad_action(arg1: str): raise Exception('very bad error') return f'bad_{arg1}' with pytest.raises(TaskError): Script('test_script', ).tasks( Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), Block(Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]), Task(id='task_4', name="Task 4", action=bad_action, args=[Static('value_4')], notify='handler_1'), when=Var('os.var_1') ).rescue(Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')])), Task(id='task_6', name="Task 6", action=action, args=[Static('value_6')]), Handler('handler_1', Task(id='task_7', name="Task 7", action=action, args=[Static('value_7')]), Task(id='task_8', name="Task 8", action=action, args=[Static('value_8')]) ) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_2' in datavars.scripts.test_script assert 'task_3' in datavars.scripts.test_script assert 'task_4' not in datavars.scripts.test_script assert 'task_5' in datavars.scripts.test_script assert 'task_6' not in datavars.scripts.test_script assert 'task_7' not in datavars.scripts.test_script assert 'task_8' not in datavars.scripts.test_script def test_if_script_object_is_created_with_some_essential_tasks_block_with_failing_task_and_rescue_and_handler_which_id_is_in_the_notify_parameter_of_the_block_task_before_failing_task__the_script_will_execute_tasks_before_block_and_first_task_of_block_and_then_will_fail_execute_rescue_tasks_and_tasks_from_handler(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' def bad_action(arg1: str): raise Exception('very bad error') return f'bad_{arg1}' with pytest.raises(TaskError): Script('test_script', ).tasks( Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), Block(Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')], notify='handler_1'), Task(id='task_4', name="Task 4", action=bad_action, args=[Static('value_4')]), when=Var('os.var_1') ).rescue(Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')])), Task(id='task_6', name="Task 6", action=action, args=[Static('value_6')]), Handler('handler_1', Task(id='task_7', name="Task 7", action=action, args=[Static('value_7')]), Task(id='task_8', name="Task 8", action=action, args=[Static('value_8')]) ) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_2' in datavars.scripts.test_script assert 'task_3' in datavars.scripts.test_script assert 'task_4' not in datavars.scripts.test_script assert 'task_5' in datavars.scripts.test_script assert 'task_6' not in datavars.scripts.test_script assert 'task_7' in datavars.scripts.test_script assert 'task_8' in datavars.scripts.test_script def test_if_script_object_is_created_with_some_essential_tasks_block_with_failing_unessential_task_and_rescue_and_handler_which_id_is_in_the_notify_parameter_of_the_failing_unessential_task__the_script_will_execute_tasks_before_block_and_all_tasks_of_block_except_failing_task(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' def bad_action(arg1: str): raise Exception('not so bad error') return f'bad_{arg1}' Script('test_script', ).tasks( Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), Block(Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]), Task(id='task_4', name="Task 4", essential=False, action=bad_action, args=[Static('value_4')], notify='handler_1'), when=Var('os.var_1') ).rescue(Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')])), Task(id='task_6', name="Task 6", action=action, args=[Static('value_6')]), Handler('handler_1', Task(id='task_7', name="Task 7", action=action, args=[Static('value_7')]), Task(id='task_8', name="Task 8", action=action, args=[Static('value_8')]) ) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_2' in datavars.scripts.test_script assert 'task_3' in datavars.scripts.test_script assert 'task_4' in datavars.scripts.test_script assert not datavars.scripts.test_script.task_4.success assert datavars.scripts.test_script.task_4.error_message ==\ 'not so bad error' assert 'task_5' not in datavars.scripts.test_script assert 'task_6' in datavars.scripts.test_script assert 'task_7' not in datavars.scripts.test_script assert 'task_8' not in datavars.scripts.test_script def test_if_script_object_is_created_with_some_essential_tasks_one_task_with_unfulfilled_condition_block_with_tasks_and_rescue_and_handler_which_id_is_in_the_notify_parameter_of_the_task_with_unfulfilled_condition__the_script_will_execute_tasks_from_script_root_except_task_with_condition_tasks_and_all_tasks_of_block(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' def bad_action(arg1: str): raise Exception('not so bad error') return f'bad_{arg1}' Script('test_script', ).tasks( Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')], notify='handler_1', when=~Var('os.var_1')), Block(Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]), Task(id='task_4', name="Task 4", action=action, args=[Static('value_4')]), when=Var('os.var_1') ).rescue(Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')])), Task(id='task_6', name="Task 6", action=action, args=[Static('value_6')]), Handler('handler_1', Task(id='task_7', name="Task 7", action=action, args=[Static('value_7')]), Task(id='task_8', name="Task 8", action=action, args=[Static('value_8')]) ) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_2' not in datavars.scripts.test_script assert 'task_3' in datavars.scripts.test_script assert 'task_4' in datavars.scripts.test_script assert 'task_5' not in datavars.scripts.test_script assert 'task_6' in datavars.scripts.test_script assert 'task_7' not in datavars.scripts.test_script assert 'task_8' not in datavars.scripts.test_script def test_if_script_object_is_created_with_some_essential_tasks_one_block_of_tasks_having_fulfilled_condition_with_Done_and_rescue_block_and_handler_which_id_is_in_the_notify_parameter_of_the_successful_task__the_script_will_execute_all_tasks_except_task_from_rescue_block(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=True, type=BooleanType) def action(arg1: str): return f'action -> {arg1}' def bad_action(arg1: str): raise Exception('not so bad error') return f'bad_{arg1}' Script('test_script', args=['arg1', 'arg_2'], ).tasks( Task(id='task_1', name="Task 1", action=action, args=[Static('value_1')]), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')], notify='handler_1', when=Var('scripts.test_script.arg1')), Block(Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]), Task(id='task_4', name="Task 4", action=action, args=[Static('value_4')]), when=Done('task_1', 'task_2') ).rescue(Task(id='task_5', name="Task 5", action=action, args=[Static('value_5')])), Task(id='task_6', name="Task 6", action=action, args=[Static('value_6')]), Handler('handler_1', Task(id='task_7', name="Task 7", action=action, args=[Static('value_7')]), Task(id='task_8', name="Task 8", action=action, args=[Static('value_8')]) ) ).initialize(IOModule(), datavars, None)(True, False) assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_2' in datavars.scripts.test_script assert 'task_3' in datavars.scripts.test_script assert 'task_4' in datavars.scripts.test_script assert 'task_5' not in datavars.scripts.test_script assert 'task_6' in datavars.scripts.test_script assert 'task_7' in datavars.scripts.test_script assert 'task_8' in datavars.scripts.test_script def test_if_script_object_is_created_with_any_number_of_tasks_and_Run_that_is_special_object_for_running_other_scripts_and_Run_have_no_arguments_for_running_script__the_script_executes_script_from_the_Run_object_and_its_own_tasks(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' script_1 = Script('test_script_1', ).tasks(Task(id='task_1', name="Task", action=action, args=[Static('value_1')])) Script('test_script_2', ).tasks(Run(script_1, namespace='os', when=Var('os.var_1')), Task(id='task_2', name="Task", action=action, args=[Static('value_2')]), ).initialize(IOModule(), datavars, None)() assert 'test_script_1' in datavars.scripts assert 'test_script_2' in datavars.scripts assert 'task_1' in datavars.scripts.test_script_1 assert 'task_2' in datavars.scripts.test_script_2 assert datavars.scripts.test_script_1.task_1.result ==\ 'value = value_1' assert datavars.scripts.test_script_2.task_2.result ==\ 'value = value_2' def test_if_script_object_is_created_with_any_number_of_tasks_and_Run_object_is_used_to_run_other_script_with_static_argument__the_script_executes_script_from_the_Run_object_with_specified_static_arguments_and_its_own_tasks(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' script_1 = Script('test_script_1', args=['arg1'] ).tasks(Task(id='task_1', name="Task", action=action, args=["scripts.test_script_1.arg1"])) Script('test_script_2', ).tasks(Run(script_1, namespace='os', args=[Static('arg_value')], when=Var('os.var_2')), Task(id='task_2', name="Task", action=action, args=[Static('value_2')]), ).initialize(IOModule(), datavars, None)() assert 'test_script_1' in datavars.scripts assert 'test_script_2' in datavars.scripts assert 'task_1' in datavars.scripts.test_script_1 assert 'task_2' in datavars.scripts.test_script_2 assert datavars.scripts.test_script_1.task_1.result ==\ 'value = arg_value' assert datavars.scripts.test_script_2.task_2.result ==\ 'value = value_2' def test_if_script_object_is_created_with_any_number_of_tasks_and_Run_object_is_used_to_run_other_script_with_variable_as_its_only_argument__the_script_executes_script_from_the_Run_object_with_specified_variable_arguments_and_its_own_tasks(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' script_1 = Script('test_script_1', args=['arg1'] ).tasks(Task(id='task_1', name="Task 1", action=action, args=["scripts.test_script_1.arg1"])) Script('test_script_2', ).tasks(Run(script_1, namespace='os', args=['os.var_1'], when=Var('os.var_2')), Task(id='task_2', name="Task", action=action, args=[Static('value_2')]), ).initialize(IOModule(), datavars, None)() assert 'test_script_1' in datavars.scripts assert 'test_script_2' in datavars.scripts assert 'task_1' in datavars.scripts.test_script_1 assert 'task_2' in datavars.scripts.test_script_2 assert datavars.scripts.test_script_1.task_1.result ==\ 'value = 12' assert datavars.scripts.test_script_2.task_2.result ==\ 'value = value_2' def test_if_script_object_is_created_with_any_number_of_tasks_and_Run_object_is_used_to_run_other_script_with_two_variables_as_its_arguments__the_script_executes_script_from_the_Run_object_with_specified_variables_arguments_and_its_own_tasks(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source='value', type=StringType) Variable('var_3', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' script_1 = Script('test_script_1', args=['arg1', 'arg2'] ).tasks(Task(id='task_1', name="Task 1", action=action, args=["scripts.test_script_1.arg1"]), Task(id='task_2', name="Task 2", action=action, args=["scripts.test_script_1.arg2"])) Script('test_script_2', ).tasks(Run(script_1, namespace='os', args=['os.var_1', 'os.var_2'], when=Var('os.var_3')), Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]), ).initialize(IOModule(), datavars, None)() assert 'test_script_1' in datavars.scripts assert 'test_script_2' in datavars.scripts assert 'task_1' in datavars.scripts.test_script_1 assert 'task_2' in datavars.scripts.test_script_1 assert 'task_3' in datavars.scripts.test_script_2 assert datavars.scripts.test_script_1.task_1.result ==\ 'value = 12' assert datavars.scripts.test_script_1.task_2.result ==\ 'value = value' assert datavars.scripts.test_script_2.task_3.result ==\ 'value = value_3' def test_if_script_object_is_created_with_any_number_of_tasks_and_Run_object_is_used_to_run_other_script_with_insufficient_number_of_variables_as_its_arguments__the_outer_script_interrupts_its_execution_with_error(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source='value', type=StringType) Variable('var_3', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' script_1 = Script('test_script_1', args=['arg1', 'arg2'] ).tasks(Task(id='task_1', name="Task 1", action=action, args=["scripts.test_script_1.arg1"]), Task(id='task_2', name="Task 2", action=action, args=["scripts.test_script_1.arg2"])) with pytest.raises(ScriptError): Script('test_script_2', ).tasks(Run(script_1, namespace='os', args=['os.var_1'], when=Var('os.var_3')), Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]), ).initialize(IOModule(), datavars, None)() def test_if_script_object_is_created_with_any_number_of_tasks_and_Run_object_is_used_to_run_other_script_with_excess_number_of_variables_as_its_arguments__the_outer_script_interrupts_its_execution_with_error(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source='value', type=StringType) Variable('var_3', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' script_1 = Script('test_script_1', args=['arg1', 'arg2'] ).tasks(Task(id='task_1', name="Task 1", action=action, args=["scripts.test_script_1.arg1"]), Task(id='task_2', name="Task 2", action=action, args=["scripts.test_script_1.arg2"])) with pytest.raises(ScriptError): Script('test_script_2', ).tasks(Run(script_1, namespace='os', args=['os.var_1', 'os.var2', Static('value')], when=Var('os.var_3')), Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]), ).initialize(IOModule(), datavars, None)() def test_if_script_object_is_created_with_any_number_of_tasks_and_two_Run_objects_that_are_used_to_run_the_same_scripts_with_different_arguments__the_outer_script_runs_script_from_Run_objects_two_times_with_different_arguments(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' script_1 = Script('test_script_1', args=['arg1'] ).tasks(Task(id='task_1', name="Task 1", action=action, args=["scripts.test_script_1.arg1"])) Script('test_script_2', ).tasks(Run(script_1, namespace='os', args=[Static(12)], when=Var('os.var_2')), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]), Run(script_1, namespace='os', args=[Static(13)]) ).initialize(IOModule(), datavars, None)() assert 'test_script_1' in datavars.scripts assert 'test_script_2' in datavars.scripts assert 'task_1' in datavars.scripts.test_script_1 assert 'task_2' in datavars.scripts.test_script_2 assert datavars.scripts.test_script_1.arg1 == 13 assert datavars.scripts.test_script_1.task_1.result ==\ 'value = 13' assert datavars.scripts.test_script_2.task_2.result ==\ 'value = value_2' def test_if_script_object_is_created_with_any_number_of_tasks_and_essential_Run_object_is_used_to_run_other_failing_script__the_outer_script_interrupts_its_execution_with_error(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' def bad_action(arg): raise Exception('error lol') return f'value = {arg}' script_1 = Script('test_script_1', args=['arg1'] ).tasks(Task(id='task_1', name="Task 1", action=bad_action, args=["scripts.test_script_1.arg1"])) with pytest.raises(ScriptError): Script('test_script_2', ).tasks(Run(script_1, namespace='os', args=[Static(12)], when=Var('os.var_2'), essential=True), Task(id='task_2', name="Task 2", action=action, args=[Static('value_2')]) ).initialize(IOModule(), datavars, None)() def test_if_script_object_is_created_with_any_number_of_tasks_and_not_essential_Run_object_is_used_to_run_other_failing_script__the_outer_script_skips_failing_script_and_continues_to_execute_its_own_tasks(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' def bad_action(arg): raise Exception('error lol') return f'value = {arg}' script_1 = Script('test_script_1', args=['arg1'] ).tasks(Task(id='task_1', name="Task 1", essential=False, action=bad_action, args=["scripts.test_script_1.arg1"]), Task(id='task_2', name="Task 2", action=bad_action, args=["scripts.test_script_1.arg1"])) Script('test_script_2', ).tasks(Run(script_1, namespace='os', args=[Static(12)], when=Var('os.var_2'), essential=False), Task(id='task_3', name="Task 3", action=action, args=[Static('value_3')]) ).initialize(IOModule(), datavars, None)() assert 'task_1' in datavars.scripts.test_script_1 assert not datavars.scripts.test_script_1.task_1.success assert datavars.scripts.test_script_1.task_1.error_message ==\ "error lol" assert 'task_3' in datavars.scripts.test_script_2 def test_if_script_object_is_created_with_essential_task_which_action_raises_ActionError_exception_or_an_exception_inheriting_it__the_outer_script_skips_failing_task_as_if_it_is_not_essential_task(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=True, type=BooleanType) class ValueError(ActionError): pass def action(arg1): if arg1 < 15: raise ValueError('action based error') return f'value = {arg1}' Script('test_script', ).tasks(Task(id='task_1', name="Task 1", action=action, args=['os.var_1']) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script assert 'task_1' in datavars.scripts.test_script def test_if_script_object_is_created_with_not_essential_task_which_action_raises_ActionError_exception_or_an_exception_inheriting_it__the_outer_script_skips_failing_task(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=12, type=IntegerType) Variable('var_2', source=True, type=BooleanType) class ValueError(ActionError): pass def action(arg1): if arg1 < 15: raise ValueError('action based error') return f'value = {arg1}' Script('test_script', ).tasks(Task(id='task_1', name="Task 1", essential=False, action=action, args=['os.var_1']) ).initialize(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'task_1' in datavars.scripts.test_script def test_if_script_object_is_created_with_some_tasks_and_Run_objects_are_used_to_run_scripts_and_different_namespaces_is_set_in_it__the_outer_script_runs_scripts_from_Run_objects_for_different_namespaces_and_its_own_tasks(self): Namespace.reset() datavars = Namespace.datavars with Namespace('os'): Variable('var_1', source=1349, type=IntegerType) Variable('var_2', source='value_2', type=StringType) with Namespace('linux'): Variable('var_1', source=1507, type=IntegerType) Variable('var_3', source='value_3', type=StringType) Variable('var_4', source=True, type=BooleanType) def action(arg1): return f'value = {arg1}' script_1 = Script('test_script_1', args=['arg'] ).tasks(Task(id='task_1', name="Task 1", action=action, args=['.var_1'], when=Var('scripts.test_script_1.arg')), Task(id='task_2', name="Task 2", action=action, args=['.var_1'], when=~Var('scripts.test_script_1.arg'))) Script('test_script_2', ).tasks(Task(id='task_1', name="Task 1", essential=False, action=action, args=['.var_1']), Block(Task(id='task_2', name="Task 2", essential=False, action=action, args=['.var_2']), Task(id='task_3', name="Task 3", essential=False, action=action, args=['.linux.var_3']), when=Var('.linux.var_4')), Run(script_1, namespace='os.linux', args=[True], when=Var('.linux.var_4')), Run(script_1, namespace='os', args=[False], when=Var('.linux.var_4')) ).initialize(IOModule(), datavars, datavars.os)() assert 'test_script_1' in datavars.scripts assert 'test_script_2' in datavars.scripts assert 'task_1' in datavars.scripts.test_script_2 assert datavars.scripts.test_script_2.task_1.result == 'value = 1349' assert 'task_2' in datavars.scripts.test_script_2 assert datavars.scripts.test_script_2.task_2.result ==\ 'value = value_2' assert 'task_3' in datavars.scripts.test_script_2 assert datavars.scripts.test_script_2.task_3.result ==\ 'value = value_3' assert 'task_1' in datavars.scripts.test_script_1 assert datavars.scripts.test_script_1.task_1.result == 'value = 1507' assert 'task_2' in datavars.scripts.test_script_1 assert datavars.scripts.test_script_1.task_2.result == 'value = 1349'