import os import shutil 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, ScriptLauncher, RunTemplates ) from calculate.variables.datavars import ( Namespace, Variable, IntegerType, StringType, BooleanType, ListType, HashType ) from calculate.variables.loader import Datavars from calculate.utils.io_module import IOModule TESTFILES_PATH = os.path.join(os.getcwd(), 'tests/scripts/testfiles') @pytest.mark.scripts class TestTasks(): def test_to_make_testfiles(self): shutil.copytree(os.path.join(TESTFILES_PATH, 'var.backup'), os.path.join(TESTFILES_PATH, 'var'), symlinks=True) shutil.copytree(os.path.join(TESTFILES_PATH, 'etc.backup'), os.path.join(TESTFILES_PATH, 'etc'), symlinks=True) 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 = ScriptLauncher.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 = Script('test_script' ).tasks( Task(id='task', name="Task", action=action, args=["os.var"]) ) launcher = script.make_launcher(IOModule(), datavars, None) launcher() 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'))) ).make_launcher(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_not_be_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')) ).make_launcher(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')) ).make_launcher(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')) ).make_launcher(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')) ).make_launcher(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'))) ).make_launcher(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'))) ).make_launcher(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']) ).make_launcher(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') ).make_launcher(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']) ).make_launcher(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')) ).make_launcher(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'}) ).make_launcher(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']) ).make_launcher(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') ).make_launcher(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')])) ).make_launcher(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')) ).make_launcher(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')) ).make_launcher(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')])) ).make_launcher(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'])) ).make_launcher(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')) ).make_launcher(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'))) ).make_launcher(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'))) ).make_launcher(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'))) ).make_launcher(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'))) ).make_launcher(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')]) ) ).make_launcher(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')]) ) ).make_launcher(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')]) ) ).make_launcher(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')]) ) ).make_launcher(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')]) ) ).make_launcher(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')]) ) ).make_launcher(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( 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')]) ), 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')]) ).make_launcher(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')]), ).make_launcher(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')]), ).make_launcher(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')]), ).make_launcher(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')]), ).make_launcher(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')]), ).make_launcher(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')]), ).make_launcher(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)]) ).make_launcher(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')]) ).make_launcher(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')]) ).make_launcher(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']) ).make_launcher(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_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']) ).make_launcher(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}' tasks_1 = [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_1 = Script('test_script_1', args=['arg'] ).tasks(*tasks_1) 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')) ).make_launcher(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' def test_run_templates_using_script(self): datavars = Datavars(variables_path=os.path.join(TESTFILES_PATH, 'variables')) Script('test_script', ).tasks(RunTemplates(id="templates_1", action='action_1', package="test-category/test-package", chroot_path=TESTFILES_PATH, root_path="/etc"), ).make_launcher(IOModule(), datavars, None)() assert 'test_script' in datavars.scripts assert 'templates_1' in datavars.scripts.test_script assert datavars.scripts.test_script.templates_1.get_hash() ==\ {"changed": {os.path.join(TESTFILES_PATH, 'etc/dir_0'): 'N', os.path.join(TESTFILES_PATH, 'etc/dir_0/file_0'): 'N'}, "skipped": []} assert os.path.exists(os.path.join(TESTFILES_PATH, 'etc/dir_0/file_0')) def test_run_templates_using_script_and_namespace_is_set(self): datavars = Datavars(variables_path=os.path.join(TESTFILES_PATH, 'variables')) assert 'linux' in datavars.os assert 'test_3' in datavars.os.linux Script('test_script', ).tasks(RunTemplates(id="templates_1", action='action_2', package="test-category/test-package", chroot_path=TESTFILES_PATH, root_path="/etc"), ).make_launcher(IOModule(), datavars, datavars.os)() assert 'test_script' in datavars.scripts assert 'templates_1' in datavars.scripts.test_script assert datavars.scripts.test_script.templates_1.get_hash() ==\ {"changed": {os.path.join(TESTFILES_PATH, 'etc/dir_1'): 'N', os.path.join(TESTFILES_PATH, 'etc/dir_1/file_0'): 'N'}, "skipped": []} assert os.path.exists(os.path.join(TESTFILES_PATH, 'etc/dir_1/file_0')) def test_for_removing_testfiles(self): shutil.rmtree(os.path.join(TESTFILES_PATH, 'var')) shutil.rmtree(os.path.join(TESTFILES_PATH, 'etc'))