def wait_for_output()

in packages/constructs/L2/eks-constructs/lib/kubectl-handler/cmd/__init__.py [0:0]

• 23 lines of code
• 8 McCabe index (conditional complexity)

def wait_for_output(args, expected_output, timeout_seconds):

    end_time = time.time() + timeout_seconds
    error = None

    while time.time() < end_time:
        try:
            # the output is surrounded with '', so we unquote
            output = kubectl(args).decode('utf-8')[1:-1]
            logger.info(f"KubeCtl Output: {output}")
            if output:
                if not expected_output:
                    return output
                elif expected_output and output == expected_output:
                    return output
                else:
                    error = f"Output {output} does not match expected output {expected_output}"
                    logger.info(error)
        except Exception as e:
            error = str(e)
            logger.warning(f"Output Exception: {error}")
            # also a recoverable error
            if 'NotFound' in error:
                pass
        # nosemgrep
        time.sleep(10)

    raise RuntimeError(
        f'Timeout waiting for output from kubectl command: {args} (last_error={error})')