Repository: apache/airflow
File: providers/celery/tests/integration/celery/test_celery_executor.py (Python)

Code before:
        # then issues a CREATE TABLE, rather than doing CREATE TABLE IF NOT
        # EXISTS
        session = backend.ResultSession()
        session.close()

    with patch_app, patch_execute, patch_factory:
        try:
            yield test_app
        finally:
            # Clear event loop to tear down each celery instance
            set_event_loop(None)


@pytest.mark.integration("celery")
@pytest.mark.backend("postgres")
class TestCeleryExecutor:
    def setup_method(self) -> None:
        db.clear_db_runs()
        db.clear_db_jobs()

    def teardown_method(self) -> None:
        db.clear_db_runs()
        db.clear_db_jobs()


def setup_dagrun(dag_maker):
    date = timezone.utcnow()
    start_date = date - timedelta(days=2)

    with dag_maker("test_celery_integration"):
        BaseOperator(task_id="success", start_date=start_date)
        BaseOperator(task_id="fail", start_date=start_date)

    return dag_maker.create_dagrun(logical_date=date)

    @pytest.mark.flaky(reruns=5, reruns_delay=3)
    @pytest.mark.parametrize("broker_url", _prepare_test_bodies())
    @pytest.mark.parametrize(
        "executor_config",
        [
            pytest.param({}, id="no_executor_config"),
            pytest.param(
                {
                    "pod_override": k8s.V1Pod(
                        spec=k8s.V1PodSpec(
                            containers=[
                                k8s.V1Container(
                                    name="base",
                                    resources=k8s.V1ResourceRequirements(
                                        requests={
                                            "cpu": "100m",

Review comment (bug):
```suggestion
def setup_dagrun_with_success_and_fail_tasks(dag_maker):
```

Code after:
        # then issues a CREATE TABLE, rather than doing CREATE TABLE IF NOT
        # EXISTS
        session = backend.ResultSession()
        session.close()

    with patch_app, patch_execute, patch_factory:
        try:
            yield test_app
        finally:
            # Clear event loop to tear down each celery instance
            set_event_loop(None)


@pytest.mark.integration("celery")
@pytest.mark.backend("postgres")
class TestCeleryExecutor:
    def setup_method(self) -> None:
        db.clear_db_runs()
        db.clear_db_jobs()

    def teardown_method(self) -> None:
        db.clear_db_runs()
        db.clear_db_jobs()


def setup_dagrun_with_success_and_fail_tasks(dag_maker):
    date = timezone.utcnow()
    start_date = date - timedelta(days=2)

    with dag_maker("test_celery_integration"):
        BaseOperator(task_id="success", start_date=start_date)
        BaseOperator(task_id="fail", start_date=start_date)

    return dag_maker.create_dagrun(logical_date=date)

    @pytest.mark.flaky(reruns=5, reruns_delay=3)
    @pytest.mark.parametrize("broker_url", _prepare_test_bodies())
    @pytest.mark.parametrize(
        "executor_config",
        [
            pytest.param({}, id="no_executor_config"),
            pytest.param(
                {
                    "pod_override": k8s.V1Pod(
                        spec=k8s.V1PodSpec(
                            containers=[
                                k8s.V1Container(
                                    name="base",
                                    resources=k8s.V1ResourceRequirements(
                                        requests={
                                            "cpu": "100m",