def METHOD_NAME(
    num_tasks: int,
    seed: int = 0,
    max_num_nodes: int = 10,
    max_num_parents: int = 5,
    max_num_candidate_resources: int = 5,
    max_task_runtime: int = 3600,
    max_data_size: int = 1000,
) -> sky.Dag:
    """Generates a random Sky DAG to test Sky optimizer."""
    random.seed(seed)
    with sky.Dag() as dag:
        for i in range(num_tasks):
            op = sky.Task(name=f'task{i}')
            task_runtime = random.random() * max_task_runtime
            op.set_time_estimator(lambda _: task_runtime)
            op.num_nodes = random.randint(1, max_num_nodes)
            if i == 0:
                num_parents = 0
            else:
                num_parents = random.randint(0, min(i, max_num_parents))
            if num_parents == 0:
                src_cloud = random.choice(['s3:', 'gs:', None])
                src_volume = random.randint(0, max_data_size)
            else:
                parents = random.choices(dag.tasks[:-1], k=num_parents)
                for parent in parents:
                    parent >> op
                # NOTE: Sky only takes single input data source
                src_cloud = parents[0]
                # Sky uses the parent's output data size
                src_volume = None
            if src_cloud is not None:
                op.set_inputs(src_cloud, src_volume)
            op.set_outputs('CLOUD', random.randint(0, max_data_size))
            num_candidates = random.randint(1, max_num_candidate_resources)
            candidate_instance_types = random.choices(ALL_INSTANCE_TYPE_INFOS,
                                                      k=num_candidates)
            candidate_resources = set()
            for candidate in candidate_instance_types:
                instance_type = candidate.instance_type
                if pd.isna(instance_type):
                    assert candidate.cloud == 'GCP', candidate
                    (instance_list,
                     _) = service_catalog.get_instance_type_for_accelerator(
                         candidate.accelerator_name,
                         candidate.accelerator_count,
                         clouds='gcp')
                    assert instance_list, (candidate, instance_list)
                    instance_type = random.choice(instance_list)
                resources = sky.Resources(
                    cloud=clouds.CLOUD_REGISTRY.from_str(candidate.cloud),
                    instance_type=instance_type,
                    accelerators={
                        candidate.accelerator_name: candidate.accelerator_count
                    })
                candidate_resources.add(resources)
            op.set_resources(candidate_resources)
    return dag