### Body

We have several tests marked in the code base with `@pytest.mark.quarantined`
It means that the tests are flaky and if fail in CI it does not fail the build.
The goal is to fix the tests and make them stable.

This task is to gather all of them under the same issue instead of dedicated issue per test.

- [x] [TestImpersonation](https://github.com/apache/airflow/blob/bfcae349b88fd959e32bfacd027a5be976fe2132/tests/core/test_impersonation_tests.py#L117)
- [x] [TestImpersonationWithCustomPythonPath](https://github.com/apache/airflow/blob/bfcae349b88fd959e32bfacd027a5be976fe2132/tests/core/test_impersonation_tests.py#L181)
- [x] [test_exception_propagation](https://github.com/apache/airflow/blob/76f81cd4a7433b7eeddb863b2ae6ee59176cf816/tests/jobs/test_local_task_job.py#L772)
- [x] [test_localtaskjob_maintain_heart_rate](https://github.com/apache/airflow/blob/76f81cd4a7433b7eeddb863b2ae6ee59176cf816/tests/jobs/test_local_task_job.py#L402)
- [x] [test_exception_propagation](https://github.com/apache/airflow/blob/4d0fd8ef6adc35f683c7561f05688a65fd7451f4/tests/executors/test_celery_executor.py#L103)
- [x] [test_process_sigterm_works_with_retries](https://github.com/apache/airflow/blob/65010fda091242870a410c65478eae362899763b/tests/jobs/test_local_task_job.py#L770)


### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
self.dagbag = dagbag
        with patch("airflow.jobs.base_job.sleep") as self.mock_base_job_sleep:
            yield
    def validate_ti_states(self, dag_run, ti_state_mapping, error_message):
        for task_id, expected_state in ti_state_mapping.items():
            task_instance = dag_run.get_task_instance(task_id=task_id)
            task_instance.refresh_from_db()
            assert task_instance.state == expected_state, error_message
    def test_localtaskjob_essential_attr(self, dag_maker):
        """
        Check whether essential attributes
        of LocalTaskJob can be assigned with
        proper values without intervention
        """
        with dag_maker("test_localtaskjob_essential_attr"):
            op1 = EmptyOperator(task_id="op1")
        dr = dag_maker.create_dagrun()
        ti = dr.get_task_instance(task_id=op1.task_id)
        job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
        essential_attr = ["dag_id", "job_type", "start_date", "hostname"]
        check_result_1 = [hasattr(job1, attr) for attr in essential_attr]
        assert all(check_result_1)
        check_result_2 = [getattr(job1, attr) is not None for attr in essential_attr]
        assert all(check_result_2)
    def test_localtaskjob_heartbeat(self, dag_maker):