data = {
            "dag_id": ti.dag_id,
            "execution_date": dag_run.execution_date.isoformat(),
            "task_id": ti.task_id,
            "state": ti.state,
            "start_date": ti.start_date.isoformat() if ti.start_date else "",
            "end_date": ti.end_date.isoformat() if ti.end_date else "",
        }
        if has_mapped_instances:
            data["map_index"] = str(ti.map_index) if ti.map_index >= 0 else ""
        return data
    AirflowConsole().print_as(data=dag_run.task_instances, output=args.output, mapper=format_task_instance)
@cli_utils.action_cli(check_db=False)
def task_test(args, dag: DAG | None = None) -> None: