def METHOD_NAME(
    db,
    workflow,
    workflow_start_tasks_factory,
    task,
    sorted_snapshot,
    success,
    next,
    redoable,
    admin_schema_executor,
):
    workflow_start_tasks_factory.create_batch(2, workflow=workflow)
    query = """
        mutation AddWorkflowFlow($input: AddWorkflowFlowInput!) {
          addWorkflowFlow(input: $input) {
            workflow {
              tasks {
                slug
              }
              flows {
                totalCount
                edges {
                  node {
                    createdByUser
                    createdByGroup
                    next
                    tasks {
                      slug
                    }
                  }
                }
              }
            }
            clientMutationId
          }
        }
    """
    variable_values = {
        "input": {
            "workflow": to_global_id(type(workflow).__name__, workflow.pk),
            "tasks": [to_global_id(type(task).__name__, task.pk)],
            "next": next,
        }
    }
    if redoable:
        variable_values["input"]["redoable"] = redoable
    result = admin_schema_executor(
        query,
        variable_values=variable_values,
    )
    assert not bool(result.errors) == success
    if success:
        assert result.data == sorted_snapshot("tasks", lambda el: el["slug"])
        assert models.TaskFlow.objects.first().redoable == redoable