start_date=DEFAULT_DATE,
            end_date=DEFAULT_DATE + datetime.timedelta(days=1),
            dag_run_state=dag_run_state,
            include_subdags=False,
            include_parentdag=False,
            session=session,
        )
        session.refresh(upstream_ti)
        session.refresh(ti)
        session.refresh(ti2)
        assert upstream_ti.state is None  
        assert ti.state is None  
        assert ti2.state == State.SUCCESS  
        dagruns = (
            session.query(
                DagRun,
            )
            .filter(
                DagRun.dag_id == dag_id,
            )
            .all()
        )
        assert len(dagruns) == 1
        dagrun: DagRun = dagruns[0]
        assert dagrun.state == dag_run_state
    def test_dag_test_basic(self):
        dag = DAG(dag_id="test_local_testing_conn_file", start_date=DEFAULT_DATE)
        mock_object = mock.MagicMock()
        @task_decorator
        def check_task():