pytest_wdl/providers/dx.py
def _get_failed_task(cls, analysis: dxpy.DXAnalysis) -> dict:
        """
        Find the causal failure within an execution tree and get the logs.
        """
        query = {
            "project": dxpy.WORKSPACE_ID,
            "state": "failed",
            "describe": {
                "stateTransitions": True
            },
            "include_subjobs": True,
            "root_execution": analysis.get_id()
        }
        cause = None
        cause_ts = None
        for execution_result in dxpy.find_executions(**query):
            if "stateTransitions" in execution_result["describe"]:
                for st in execution_result["describe"]["stateTransitions"]:
                    if st["newState"] == "failed":
                        ts = st["setAt"]
                        if cause is None or ts < cause_ts:
                            cause = execution_result
                            cause_ts = ts

        if cause:
            stdout, stderr = cls._get_logs(cause["id"])
            return {
                "failed_task": cause["id"],
                "failed_task_stdout": stdout,
                "failed_task_stderr": stderr
            }
        else:
            return {
                "msg": f"Analysis {analysis.get_id()} failed but the cause could not "
                       f"be determined"
            }

        Find the causal failure within an execution tree and get the logs.
        
Find the causal failure within an execution tree and get the logs.
Find the causal failure within an execution tree and get the logs .
def _get_failed_task(cls, analysis: dxpy.DXAnalysis) -> dict:
        query = {
            "project": dxpy.WORKSPACE_ID,
            "state": "failed",
            "describe": {
                "stateTransitions": True
            },
            "include_subjobs": True,
            "root_execution": analysis.get_id()
        }
        cause = None
        cause_ts = None
        for execution_result in dxpy.find_executions(**query):
            if "stateTransitions" in execution_result["describe"]:
                for st in execution_result["describe"]["stateTransitions"]:
                    if st["newState"] == "failed":
                        ts = st["setAt"]
                        if cause is None or ts < cause_ts:
                            cause = execution_result
                            cause_ts = ts
        if cause:
            stdout, stderr = cls._get_logs(cause["id"])
            return {
                "failed_task": cause["id"],
                "failed_task_stdout": stdout,
                "failed_task_stderr": stderr
            }
        else:
            return {
                "msg": f"Analysis {analysis.get_id()} failed but the cause could not "
                       f"be determined"
            }
def _get_failed_task ( cls , analysis : dxpy . DXAnalysis ) -> dict : query = { "project" : dxpy . WORKSPACE_ID , "state" : "failed" , "describe" : { "stateTransitions" : True } , "include_subjobs" : True , "root_execution" : analysis . get_id ( ) } cause = None cause_ts = None for execution_result in dxpy . find_executions ( ** query ) : if "stateTransitions" in execution_result [ "describe" ] : for st in execution_result [ "describe" ] [ "stateTransitions" ] : if st [ "newState" ] == "failed" : ts = st [ "setAt" ] if cause is None or ts < cause_ts : cause = execution_result cause_ts = ts if cause : stdout , stderr = cls . _get_logs ( cause [ "id" ] ) return { "failed_task" : cause [ "id" ] , "failed_task_stdout" : stdout , "failed_task_stderr" : stderr } else : return { "msg" : f"Analysis {analysis.get_id()} failed but the cause could not " f"be determined" }
Find the causal failure within an execution tree and get the logs.
Find the causal failure within an execution tree and get the logs .
"""
        Find the causal failure within an execution tree and get the logs.
        """
cls
analysis
dxpy.DXAnalysis