diff --git a/tests/test_conditional_steps.py b/tests/test_conditional_steps.py index ca1fa92c..4a39584f 100644 --- a/tests/test_conditional_steps.py +++ b/tests/test_conditional_steps.py @@ -1,40 +1,81 @@ import re +import os import inspect -import pytest from universum import __main__ +from universum.configuration_support import Step -true_branch_step_name = "true_branch" -false_branch_step_name = "false_branch" +class StepsInfo: + conditional_step = None + true_branch_step = None + false_branch_step = None + is_conditional_step_passed = False def test_conditional_true_branch(tmpdir, capsys): - check_conditional_step_success(tmpdir, capsys, conditional_step_passed=True) + steps_info = get_conditional_steps_info(is_conditional_step_passed=True) + check_conditional_step(tmpdir, capsys, steps_info) def test_conditional_false_branch(tmpdir, capsys): - check_conditional_step_success(tmpdir, capsys, conditional_step_passed=False) + steps_info = get_conditional_steps_info(is_conditional_step_passed=False) + check_conditional_step(tmpdir, capsys, steps_info) -def check_conditional_step_success(tmpdir, capsys, conditional_step_passed): - config_file = build_config_file(tmpdir, conditional_step_passed) - check_conditional_step(tmpdir, capsys, config_file, conditional_step_passed) +# https://github.com/Samsung/Universum/issues/744 +# Artifact will be collected only from the second executed step, the first one will be overwritten +# Skipping checking file content to not overload tests with additional logic for incorrect behaviour check +def test_same_artifact(tmpdir, capsys): + steps_info = get_conditional_steps_info(is_conditional_step_passed=True) + conditional_step_artifact = steps_info.conditional_step["artifacts"] + steps_info.true_branch_step.command = ["touch", conditional_step_artifact] + steps_info.true_branch_step.artifacts = conditional_step_artifact -def build_config_file(tmpdir, conditional_step_passed): - conditional_step_exit_code = 0 if conditional_step_passed else 1 + check_conditional_step(tmpdir, capsys, steps_info) + +def get_conditional_steps_info(is_conditional_step_passed): + steps_info = StepsInfo() + + conditional_step_name = "conditional" + conditional_step_exit_code = 0 if is_conditional_step_passed else 1 + steps_info.conditional_step = Step( + name=conditional_step_name, + command=["bash", "-c", f"touch {conditional_step_name}; exit {conditional_step_exit_code}"], + artifacts=conditional_step_name) + steps_info.is_conditional_step_passed = is_conditional_step_passed + + true_branch_step_name = "true_branch" + steps_info.true_branch_step = Step( + name=true_branch_step_name, + command=["touch", true_branch_step_name], + artifacts=true_branch_step_name) + + false_branch_step_name = "false_branch" + steps_info.false_branch_step = Step( + name=false_branch_step_name, + command=["touch", false_branch_step_name], + artifacts=false_branch_step_name) + + return steps_info + + +def write_config_file(tmpdir, conditional_steps_info): config = inspect.cleandoc(f''' from universum.configuration_support import Configuration, Step - true_branch_step = Step(name='{true_branch_step_name}', command=['touch', '{true_branch_step_name}']) - false_branch_step = Step(name='{false_branch_step_name}', command=['touch', '{false_branch_step_name}']) - conditional_step = Configuration([dict(name='conditional', - command=['bash', '-c', 'exit {conditional_step_exit_code}'], - if_succeeded=true_branch_step, if_failed=false_branch_step)]) + true_branch_step = Step(**{str(conditional_steps_info.true_branch_step)}) + false_branch_step = Step(**{str(conditional_steps_info.false_branch_step)}) + conditional_step = Step(**{str(conditional_steps_info.conditional_step)}) + + # `true/false_branch_steps` should be Python objects from this script + conditional_step.is_conditional = True + conditional_step.if_succeeded = true_branch_step + conditional_step.if_failed = false_branch_step - configs = conditional_step + configs = Configuration([conditional_step]) ''') config_file = tmpdir.join("configs.py") @@ -43,7 +84,9 @@ def build_config_file(tmpdir, conditional_step_passed): return config_file -def check_conditional_step(tmpdir, capsys, config_file, conditional_step_passed): +def check_conditional_step(tmpdir, capsys, steps_info): + config_file = write_config_file(tmpdir, steps_info) + artifacts_dir = tmpdir.join("artifacts") params = ["-vt", "none", "-fsd", str(tmpdir), @@ -56,11 +99,20 @@ def check_conditional_step(tmpdir, capsys, config_file, conditional_step_passed) assert return_code == 0 captured = capsys.readouterr() - print(captured.out) conditional_succeeded_regexp = r"\] conditional.*Success.*\| 5\.2" assert re.search(conditional_succeeded_regexp, captured.out, re.DOTALL) - expected_log = true_branch_step_name if conditional_step_passed else false_branch_step_name - unexpected_log = false_branch_step_name if conditional_step_passed else true_branch_step_name - assert expected_log in captured.out - assert not unexpected_log in captured + is_conditional_step_passed = steps_info.is_conditional_step_passed + conditional_step_artifact = steps_info.conditional_step.artifacts + true_branch_step_artifact = steps_info.true_branch_step.artifacts + false_branch_step_artifact = steps_info.false_branch_step.artifacts + + assert os.path.exists(os.path.join(artifacts_dir, conditional_step_artifact)) + + expected_artifact = true_branch_step_artifact if is_conditional_step_passed else false_branch_step_artifact + if expected_artifact: + assert os.path.exists(os.path.join(artifacts_dir, expected_artifact)) + + unexpected_artifact = false_branch_step_artifact if is_conditional_step_passed else true_branch_step_artifact + if unexpected_artifact: + assert not os.path.exists(os.path.join(artifacts_dir, unexpected_artifact)) diff --git a/universum/modules/artifact_collector.py b/universum/modules/artifact_collector.py index e6a5c094..cfdbcdf5 100644 --- a/universum/modules/artifact_collector.py +++ b/universum/modules/artifact_collector.py @@ -4,10 +4,11 @@ import os import shutil import zipfile +from typing import List, Optional, Dict, Union, TypedDict import glob2 -from ..configuration_support import Configuration +from ..configuration_support import Configuration, Step from ..lib.ci_exception import CriticalCiException, CiException from ..lib.gravity import Dependency from ..lib.utils import make_block @@ -58,6 +59,11 @@ def make_big_archive(target, source): return filename +class ArtifactInfo(TypedDict): + path: str + clean: bool + + class ArtifactCollector(ProjectDirectory, HasOutput, HasStructure): reporter_factory = Dependency(Reporter) automation_server_factory = Dependency(AutomationServerForHostingBuild) @@ -155,14 +161,15 @@ def preprocess_artifact_list(self, artifact_list, ignore_already_existing=False) @make_block("Preprocessing artifact lists") def set_and_clean_artifacts(self, project_configs: Configuration, ignore_existing_artifacts: bool = False) -> None: self.html_output.artifact_dir_ready = True - artifact_list = [] - report_artifact_list = [] + artifact_list: List[ArtifactInfo] = [] + report_artifact_list: List[ArtifactInfo] = [] for configuration in project_configs.all(): if configuration.artifacts: - path = utils.parse_path(configuration.artifacts, self.settings.project_root) - artifact_list.append(dict(path=path, clean=configuration.artifact_prebuild_clean)) + artifact_list.append(self.get_config_artifact(configuration)) + if configuration.is_conditional: + artifact_list.extend(self.get_conditional_step_branches_artifacts(configuration)) if configuration.report_artifacts: - path = utils.parse_path(configuration.report_artifacts, self.settings.project_root) + path: str = utils.parse_path(configuration.report_artifacts, self.settings.project_root) report_artifact_list.append(dict(path=path, clean=configuration.artifact_prebuild_clean)) if artifact_list: @@ -175,6 +182,26 @@ def set_and_clean_artifacts(self, project_configs: Configuration, ignore_existin with self.structure.block(block_name=name, pass_errors=True): self.preprocess_artifact_list(report_artifact_list, ignore_existing_artifacts) + def get_conditional_step_branches_artifacts(self, step: Step) -> List[ArtifactInfo]: + succeeded_config_artifact: Optional[ArtifactInfo] = self.get_config_artifact_if_exists(step.if_succeeded) + failed_config_artifact: Optional[ArtifactInfo] = self.get_config_artifact_if_exists(step.if_failed) + + artifacts = [] + if succeeded_config_artifact: + artifacts.append(succeeded_config_artifact) + if failed_config_artifact: + artifacts.append(failed_config_artifact) + return artifacts + + def get_config_artifact_if_exists(self, step: Step) -> Optional[ArtifactInfo]: + if step and step.artifacts: + return self.get_config_artifact(step) + return None + + def get_config_artifact(self, step: Step) -> ArtifactInfo: + path = utils.parse_path(step.artifacts, self.settings.project_root) + return dict(path=path, clean=step.artifact_prebuild_clean) + def move_artifact(self, path, is_report=False): self.out.log("Processing '" + path + "'") matches = glob2.glob(path) diff --git a/universum/modules/structure_handler.py b/universum/modules/structure_handler.py index 783d22d0..73e51d8e 100644 --- a/universum/modules/structure_handler.py +++ b/universum/modules/structure_handler.py @@ -210,7 +210,7 @@ def process_one_step(self, merged_item: Step, step_executor: Callable, skip_exec with self.block(block_name=step_label, pass_errors=False): process = self.execute_one_step(merged_item, step_executor) error = process.get_error() - if error: + if error and not merged_item.is_conditional: self.fail_current_block(error) has_artifacts: bool = bool(merged_item.artifacts) or bool(merged_item.report_artifacts) if not merged_item.background and has_artifacts: @@ -245,7 +245,13 @@ def execute_steps_recursively(self, parent: Step, current_step_failed = not self.execute_steps_recursively(merged_item, child.children, step_executor, skip_execution) elif child.is_conditional: - current_step_failed = not self.execute_conditional_step(merged_item, step_executor) + conditional_step_succeeded: bool = self.process_one_step(merged_item, step_executor, + skip_execution=False) + step_to_execute: Step = merged_item.if_succeeded if conditional_step_succeeded else merged_item.if_failed + return self.execute_steps_recursively(parent=Step(), + children=Configuration([step_to_execute]), + step_executor=step_executor, + skip_execution=False) else: if merged_item.finish_background and self.active_background_steps: self.out.log("All ongoing background steps should be finished before next step execution") @@ -265,20 +271,6 @@ def execute_steps_recursively(self, parent: Step, return not some_step_failed - - def execute_conditional_step(self, step, step_executor): - self.configs_current_number += 1 - step_name = self._build_step_name(step.name) - conditional_step_succeeded = False - with self.block(block_name=step_name, pass_errors=True): - process = self.execute_one_step(step, step_executor) - conditional_step_succeeded = not process.get_error() - step_to_execute = step.if_succeeded if conditional_step_succeeded else step.if_failed - return self.execute_steps_recursively(parent=Step(), - children=Configuration([step_to_execute]), - step_executor=step_executor, - skip_execution=False) - def report_background_steps(self) -> bool: result: bool = True for item in self.active_background_steps: