Create a better test framework that can test AF3 component interactions

[amoghrajesh]

Body

Create a better testing framework that can run some tests that check the interfaces between an actual task sdk worker and airflow core.

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[kaxil]

I think there might be few ways (types of tests) to do this:

1. Running certain dags via airflow dags test or dag.test(). That will use the InProcessExecutionAPI & InProcessSupervisorComms which would allow using debugger in case of failures.
2. Just run an API-server and createintegration test under Task SDK. This will allow writing XCom pull type test with an actual server (instead of running entire DAG)
3. [Future] Run certain type of backport test with n-1 version of task SDK similar to our Provider compact tests.

[potiuk]

Why not run those dags during K8S tests? We already have "latest" version of Airflow installed in K8S and it has some example dags that are triggered. Those are truly e-2-e tests. I guess what was missing there is that dags tested there were not "complex" enough to pass xcoms.

[potiuk]

In those tests, example_dags are copied to the image:

FROM {params.airflow_image_name}

USER airflow

{extra_copy_command}

COPY --chown=airflow:0 airflow-core/src/airflow/example_dags/ /opt/airflow/dags/

COPY --chown=airflow:0 providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kubernetes_executor_templates/ /opt/airflow/pod_templates/

ENV GUNICORN_CMD_ARGS='--preload'
"""

And tests are running some of those dags, for example:

    @pytest.mark.execution_timeout(300)
    def test_integration_run_dag_task_mapping(self):
        dag_id = "example_task_mapping_second_order"
        dag_run_id, logical_date = self.start_job_in_kubernetes(dag_id, self.host)
        print(f"Found the job with logical_date {logical_date}")

        # Wait some time for the operator to complete
        self.monitor_task(
            host=self.host,
            dag_run_id=dag_run_id,
            dag_id=dag_id,
            task_id="get_nums",
            expected_final_state="success",
            timeout=300,
        )

        self.ensure_dag_expected_state(
            host=self.host,
            logical_date=logical_date,
            dag_id=dag_id,
            expected_final_state="success",
            timeout=300,
        )

[amoghrajesh]

Nice idea too. Let me consider that as well.

[amoghrajesh]

Do we have e2e integration tests for celery too?

[potiuk]

Same K8S tests with Celery Executor -> those tests run with Local. Celery , Kubernetes executor currently

[kaxil]

Yeah we could partially do that but the reason for other suggestions in #53130 (comment) is to allow ad-hoc feature testing without running the entire dag and too complex of a dag: example to

a) verify what happens end-to-end after heartbeat fails
b) verifying previous dag runs
c) testing all combinations of XCom. e.g we have 254 combinations in a specific test (that's a defensive test -- but tests like that where we can just parameterize)

We don't need to test the entire system but the interaction of Task SDK Execution code with the API server -- no need to test different executors. Instead the matrix would be latest API-server with new & last released Task SDK (which will be critical for back-compat from 3.1)

These tests should be relatively very quick and allow better debugging (breakpoints).

[potiuk]

We don't need to test the entire system but the interaction of Task SDK Execution code with the API server -- no need to test different executors. Instead the matrix would be latest API-server with new & last released Task SDK (which will be critical for back-compat from 3.1)

So it's not e2e test (because that was the title) :D

[potiuk]

Tjat sounds more like TaskSDK compatibility tests ?

[potiuk]

(and yes - there is no standard on naming test types so that might be just a confusion in naming)

[kaxil]

Yup :)

[vatsrahul1001]

I think it would be great to work on both e2e tests as well TaskSDK compatibility tests.