dag = DAG("triggered_dag", start_date=DEFAULT_DATE)
        dag.add_task(EmptyOperator(task_id="first_task", owner="test"))
        dagrun = self.create_dag_run(dag, session=session)
        first_ti = dagrun.get_task_instances()[0]
        assert "first_task" == first_ti.task_id
        assert first_ti.state is None
        
        
        
        
        with mock.patch.object(DagRun, "get_task_instances") as mock_gtis:
            mock_gtis.return_value = []
            dagrun.verify_integrity()
            first_ti.refresh_from_db()
            assert first_ti.state is None
    @pytest.mark.parametrize("state", State.task_states)
    @mock.patch.object(settings, "task_instance_mutation_hook", autospec=True)
    def test_task_instance_mutation_hook(self, mock_hook, session, state):