"""
        Ensure that datasets referenced in a dag are correctly loaded into the database.
        """
        dag_id1 = "test_dataset_dag1"
        dag_id2 = "test_dataset_dag2"
        task_id = "test_dataset_task"
        uri1 = "s3://dataset1"
        d1 = Dataset(uri1, extra={"not": "used"})
        d2 = Dataset("s3://dataset2")
        d3 = Dataset("s3://dataset3")
        dag1 = DAG(dag_id=dag_id1, start_date=DEFAULT_DATE, schedule=[d1])
        EmptyOperator(task_id=task_id, dag=dag1, outlets=[d2, d3])
        dag2 = DAG(dag_id=dag_id2, start_date=DEFAULT_DATE)
        EmptyOperator(task_id=task_id, dag=dag2, outlets=[Dataset(uri1, extra={"should": "be used"})])
        session = settings.Session()