def sanitize()

in aws_lambda_builders/workflow.py [0:0]

• 47 lines of code
• 14 McCabe index (conditional complexity)

def sanitize(func):  # pylint: disable=too-many-statements
    """
    sanitize the executable path of the runtime specified by validating it.
    :param func: Workflow's run method is sanitized
    """

    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):  # pylint: disable=too-many-statements
        valid_paths = {}
        invalid_paths = {}
        validation_errors = []
        # NOTE: we need to access binaries to get paths and resolvers, before validating.
        for binary, binary_checker in self.binaries.items():
            invalid_paths[binary] = []
            try:
                exec_paths = (
                    binary_checker.resolver.exec_paths
                    if not binary_checker.path_provided
                    else binary_checker.binary_path
                )
            except ValueError as ex:
                raise WorkflowFailedError(workflow_name=self.NAME, action_name="Resolver", reason=str(ex))
            for executable_path in exec_paths:
                try:
                    valid_path = binary_checker.validator.validate(executable_path)
                    if valid_path:
                        valid_paths[binary] = valid_path
                except MisMatchRuntimeError as ex:
                    LOG.debug("Invalid executable for %s at %s", binary, executable_path, exc_info=str(ex))
                    invalid_paths[binary].append(executable_path)

                except RuntimeValidatorError as ex:
                    LOG.debug("Runtime validation error for %s", binary, exc_info=str(ex))
                    if str(ex) not in validation_errors:
                        validation_errors.append(str(ex))

                if valid_paths.get(binary, None):
                    binary_checker.binary_path = valid_paths[binary]
                    break
        if validation_errors:
            raise WorkflowFailedError(
                workflow_name=self.NAME, action_name="Validation", reason="\n".join(validation_errors)
            )

        if len(self.binaries) != len(valid_paths):
            validation_failed_binaries = set(self.binaries.keys()).difference(valid_paths.keys())
            for validation_failed_binary in validation_failed_binaries:
                message = "Binary validation failed for {0}, searched for {0} in following locations  : {1} which did not satisfy constraints for runtime: {2}. Do you have {0} for runtime: {2} on your PATH?".format(
                    validation_failed_binary, invalid_paths[validation_failed_binary], self.runtime
                )
                validation_errors.append(message)
            raise WorkflowFailedError(
                workflow_name=self.NAME, action_name="Validation", reason="\n".join(validation_errors)
            )
        func(self, *args, **kwargs)

    return wrapper