def kubectl()

in nuvolaris/kube.py [0:0]

• 32 lines of code
• 11 McCabe index (conditional complexity)

def kubectl(*args, namespace="nuvolaris", input=None, jsonpath=None, debugresult=True):
    # support for mocked requests
    mres = mocker.invoke(*args)
    if mres:
        mocker.save(input)
        return mres

    cmd = namespace and ["kubectl", "-n", namespace] or ["kubectl"]
    cmd += list(args)
    if jsonpath:
        cmd += ["-o", "jsonpath-as-json=%s" % jsonpath]

    # if is a string, convert input in bytes
    try: input = input.encode('utf-8')
    except: pass
        
    # executing
    logging.debug(cmd)
    res = subprocess.run(cmd, capture_output=True, input=input)

    global returncode, output, error
    returncode = res.returncode
    output = res.stdout.decode()
    error = res.stderr.decode()

    if res.returncode == 0:
        if jsonpath:
                try:
                    parsed = json.loads(output)
                    if debugresult:
                        logging.debug("result: %s", json.dumps(parsed, indent=2))
                    return parsed
                except Exception as e:
                    logging.info(output)
                    logging.info(e)
                    return e
        else:
            return output
    logging.info(f"Error: kubectl f{cmd} input='{input}' output='{output}' error='{error}'")
    raise Exception(error)