From f810c813e13fc8182b710f9d50700a33e7830b4c Mon Sep 17 00:00:00 2001 From: "Guan-Ming (Wesley) Chiu" <105915352+guan404ming@users.noreply.github.com> Date: Fri, 3 Jul 2026 21:39:33 +0800 Subject: [PATCH 1/5] Add integration tests for coordinator-mode TypeScript tasks --- .../workflows/additional-prod-image-tests.yml | 16 ++ .github/workflows/ci-amd.yml | 2 + .github/workflows/ci-arm.yml | 2 + airflow-e2e-tests/docker/ts.yml | 45 +++++ .../tests/airflow_e2e_tests/conftest.py | 117 ++++++++++++ .../tests/airflow_e2e_tests/constants.py | 12 ++ .../ts_sdk_tests/__init__.py | 16 ++ .../ts_sdk_tests/test_ts_sdk_dag.py | 169 ++++++++++++++++++ dev/breeze/doc/ci/04_selective_checks.md | 1 + .../airflow_breeze/utils/selective_checks.py | 13 ++ dev/breeze/tests/test_selective_checks.py | 15 ++ ts-sdk/example/.gitignore | 1 + 12 files changed, 409 insertions(+) create mode 100644 airflow-e2e-tests/docker/ts.yml create mode 100644 airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/__init__.py create mode 100644 airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py diff --git a/.github/workflows/additional-prod-image-tests.yml b/.github/workflows/additional-prod-image-tests.yml index 99a3a4c872858..a4ec751e5f4ef 100644 --- a/.github/workflows/additional-prod-image-tests.yml +++ b/.github/workflows/additional-prod-image-tests.yml @@ -72,6 +72,10 @@ on: # yamllint disable-line rule:truthy description: "Whether to run the OpenLineage older-Airflow compat matrix (true/false)" required: true type: string + run-ts-sdk-e2e-tests: + description: "Whether to run TypeScript SDK e2e tests (true/false)" + required: true + type: string providers-compatibility-tests-matrix: description: "JSON matrix of released Airflow versions for the OpenLineage e2e compat tests." required: true @@ -338,6 +342,18 @@ jobs: e2e_test_mode: "go_sdk" if: inputs.canary-run == 'true' || inputs.run-go-sdk-e2e-tests == 'true' + test-e2e-integration-tests-ts-sdk: + name: "TypeScript SDK e2e tests with PROD image" + uses: ./.github/workflows/airflow-e2e-tests.yml + with: + workflow-name: "TypeScript SDK e2e test" + runners: ${{ inputs.runners }} + platform: ${{ inputs.platform }} + default-python-version: "${{ inputs.default-python-version }}" + use-uv: ${{ inputs.use-uv }} + e2e_test_mode: "ts_sdk" + if: inputs.canary-run == 'true' || inputs.run-ts-sdk-e2e-tests == 'true' + test-e2e-integration-tests-openlineage: name: "OpenLineage e2e tests with PROD image" uses: ./.github/workflows/airflow-e2e-tests.yml diff --git a/.github/workflows/ci-amd.yml b/.github/workflows/ci-amd.yml index 9edb394a6cc7c..c2a477b694590 100644 --- a/.github/workflows/ci-amd.yml +++ b/.github/workflows/ci-amd.yml @@ -149,6 +149,7 @@ jobs: run-go-sdk-e2e-tests: ${{ steps.selective-checks.outputs.run-go-sdk-e2e-tests }} run-openlineage-e2e-tests: ${{ steps.selective-checks.outputs.run-openlineage-e2e-tests }} run-openlineage-e2e-compat-tests: ${{ steps.selective-checks.outputs.run-openlineage-e2e-compat-tests }} + run-ts-sdk-e2e-tests: ${{ steps.selective-checks.outputs.run-ts-sdk-e2e-tests }} run-system-tests: ${{ steps.selective-checks.outputs.run-system-tests }} run-task-sdk-tests: ${{ steps.selective-checks.outputs.run-task-sdk-tests }} run-task-sdk-integration-tests: ${{ steps.selective-checks.outputs.run-task-sdk-integration-tests }} @@ -930,6 +931,7 @@ jobs: # yamllint disable rule:line-length providers-compatibility-tests-matrix: > ${{ needs.build-info.outputs.providers-compatibility-tests-matrix }} + run-ts-sdk-e2e-tests: ${{ needs.build-info.outputs.run-ts-sdk-e2e-tests }} use-uv: ${{ needs.build-info.outputs.use-uv }} run-ui-e2e-tests: ${{ needs.build-info.outputs.run-ui-e2e-tests }} run-airflow-ctl-integration-tests: ${{ needs.build-info.outputs.run-airflow-ctl-integration-tests }} diff --git a/.github/workflows/ci-arm.yml b/.github/workflows/ci-arm.yml index 8f5ce3ac07987..42ae7e5215b8c 100644 --- a/.github/workflows/ci-arm.yml +++ b/.github/workflows/ci-arm.yml @@ -138,6 +138,7 @@ jobs: run-go-sdk-e2e-tests: ${{ steps.selective-checks.outputs.run-go-sdk-e2e-tests }} run-openlineage-e2e-tests: ${{ steps.selective-checks.outputs.run-openlineage-e2e-tests }} run-openlineage-e2e-compat-tests: ${{ steps.selective-checks.outputs.run-openlineage-e2e-compat-tests }} + run-ts-sdk-e2e-tests: ${{ steps.selective-checks.outputs.run-ts-sdk-e2e-tests }} run-system-tests: ${{ steps.selective-checks.outputs.run-system-tests }} run-task-sdk-tests: ${{ steps.selective-checks.outputs.run-task-sdk-tests }} run-task-sdk-integration-tests: ${{ steps.selective-checks.outputs.run-task-sdk-integration-tests }} @@ -919,6 +920,7 @@ jobs: # yamllint disable rule:line-length providers-compatibility-tests-matrix: > ${{ needs.build-info.outputs.providers-compatibility-tests-matrix }} + run-ts-sdk-e2e-tests: ${{ needs.build-info.outputs.run-ts-sdk-e2e-tests }} use-uv: ${{ needs.build-info.outputs.use-uv }} run-ui-e2e-tests: ${{ needs.build-info.outputs.run-ui-e2e-tests }} run-airflow-ctl-integration-tests: ${{ needs.build-info.outputs.run-airflow-ctl-integration-tests }} diff --git a/airflow-e2e-tests/docker/ts.yml b/airflow-e2e-tests/docker/ts.yml new file mode 100644 index 0000000000000..6ea502425056d --- /dev/null +++ b/airflow-e2e-tests/docker/ts.yml @@ -0,0 +1,45 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# Docker Compose override for ts_sdk E2E test mode. +# +# NodeCoordinator launches the TypeScript bundle with a Node.js runtime, which +# the stock Airflow worker image does not ship. The node-provider service +# copies the node binary from the official Node image (the same image +# conftest._setup_ts_sdk_integration builds the bundle with) into a shared +# volume, the bundle built by conftest is bind-mounted into the directory +# NodeCoordinator scans, and the worker consumes the "typescript" Celery queue +# where @task.stub tasks are routed. +--- +services: + node-provider: + image: ${NODE_IMAGE:-node:22-slim} + command: ["cp", "/usr/local/bin/node", "/opt/nodejs/node"] + volumes: + - nodejs-bin:/opt/nodejs + + airflow-worker: + volumes: + - ./ts-bundles:/opt/airflow/ts-bundles:ro + - nodejs-bin:/opt/nodejs:ro + command: celery worker -q typescript,default + depends_on: + node-provider: + condition: service_completed_successfully + +volumes: + nodejs-bin: diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py index 72687795ae1f9..1e0a6e0c465e5 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py @@ -18,6 +18,7 @@ import json import os +import re import subprocess from concurrent.futures import ThreadPoolExecutor from datetime import datetime @@ -55,6 +56,7 @@ LANG_SDK_NATIVE_TOOLCHAIN, LOCALSTACK_PATH, LOGS_FOLDER, + NODE_IMAGE, OPENLINEAGE_COMPOSE_PATH, OPENSEARCH_PATH, PROVIDERS_MOUNT_CONTAINER_PATH, @@ -62,6 +64,10 @@ SCALA_SPARK_EXAMPLE_DAGS_PATH, SCALA_SPARK_EXAMPLE_LIBS_PATH, TEST_REPORT_FILE, + TS_COMPOSE_PATH, + TS_SDK_BUILD_HOME_PATH, + TS_SDK_EXAMPLE_PATH, + TS_SDK_ROOT_PATH, XCOM_BUCKET, ) @@ -640,6 +646,114 @@ def _setup_openlineage_integration(dot_env_file, tmp_dir): copyfile(OPENLINEAGE_COMPOSE_PATH, tmp_dir / "openlineage.yml") +def _setup_ts_sdk_integration(dot_env_file, tmp_dir): + """Set up the ts_sdk E2E test mode. + + Builds the ts-sdk example bundle with pnpm inside an ephemeral Node + container (the host needs no Node toolchain), drops it with its + airflow-metadata.yaml sidecar into the directory ``NodeCoordinator`` + scans, copies the Python stub Dag, and writes the coordinator + configuration. + + The worker has no Node runtime of its own; the node-provider service in + ``ts.yml`` copies the node binary out of the same image used here for the + build, so the bundle executes on the runtime it was built for. + """ + # --user keeps build outputs owned by the current user (not root); HOME + # points at a writable, gitignored dir so the pnpm store and corepack + # cache persist between runs. corepack resolves the pnpm version pinned + # by ts-sdk/package.json's packageManager field. + TS_SDK_BUILD_HOME_PATH.mkdir(parents=True, exist_ok=True) + # Shims go into a writable dir on PATH (the container user cannot write to + # /usr/local/bin) so nested `pnpm run ...` invocations resolve pnpm too. + build_script = ( + 'export PATH="$HOME/bin:$PATH"' + ' && mkdir -p "$HOME/bin"' + ' && corepack enable --install-directory "$HOME/bin"' + " && cd /repo/ts-sdk" + " && pnpm install --frozen-lockfile" + " && pnpm run build" + " && cd example" + " && pnpm install" + " && pnpm run build" + ) + console.print(f"[yellow]Building TypeScript SDK example bundle ({NODE_IMAGE})...") + subprocess.run( + [ + "docker", + "run", + "--rm", + "--user", + f"{os.getuid()}:{os.getgid()}", + "-e", + "HOME=/repo/files/pnpm-home", + "-e", + "COREPACK_ENABLE_DOWNLOAD_PROMPT=0", + "-e", + # Non-interactive pnpm: never prompt (e.g. before replacing a + # node_modules left behind by a host-side install). + "CI=true", + "-v", + f"{AIRFLOW_ROOT_PATH}:/repo", + NODE_IMAGE, + "bash", + "-c", + build_script, + ], + check=True, + ) + + # Copy the compose override into the temp directory. + copyfile(TS_COMPOSE_PATH, tmp_dir / "ts.yml") + + # Place the bundle and its metadata sidecar where the compose bind-mount + # (./ts-bundles) exposes them to the worker at /opt/airflow/ts-bundles. + ts_bundles_dir = tmp_dir / "ts-bundles" + ts_bundles_dir.mkdir() + copyfile(TS_SDK_EXAMPLE_PATH / "dist" / "bundle.mjs", ts_bundles_dir / "bundle.mjs") + supervisor_ts = (TS_SDK_ROOT_PATH / "src" / "generated" / "supervisor.ts").read_text() + version_match = re.search(r'SUPERVISOR_API_VERSION = "(\d{4}-\d{2}-\d{2})"', supervisor_ts) + if not version_match: + raise RuntimeError("Cannot find SUPERVISOR_API_VERSION in ts-sdk/src/generated/supervisor.ts") + (ts_bundles_dir / "airflow-metadata.yaml").write_text( + f'sdk:\n supervisor_schema_version: "{version_match.group(1)}"\n' + ) + + # Copy the TS SDK example stub Dag so Airflow can discover and serialize it. + copyfile( + TS_SDK_EXAMPLE_PATH / "dags" / "typescript_example.py", tmp_dir / "dags" / "typescript_example.py" + ) + + # Coordinator registry: maps the logical name "ts" to NodeCoordinator, which + # scans bundles_root for bundle.mjs and launches it with the node binary + # provided by the node-provider service. + # Queue mapping: routes tasks on the "typescript" queue to "ts". + coordinator_config = json.dumps( + { + "ts": { + "classpath": "airflow.sdk.coordinators.node.NodeCoordinator", + "kwargs": { + "bundles_root": ["/opt/airflow/ts-bundles"], + "node_executable": "/opt/nodejs/node", + }, + } + } + ) + queue_to_coordinator = json.dumps({"typescript": "ts"}) + + dot_env_file.write_text( + f"AIRFLOW_UID={os.getuid()}\n" + f"NODE_IMAGE={NODE_IMAGE}\n" + # Single-quote the JSON values so Docker Compose reads them literally. + f"AIRFLOW__SDK__COORDINATORS='{coordinator_config}'\n" + f"AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{queue_to_coordinator}'\n" + # Connection and variable read by the TS example bundle tasks. + "AIRFLOW_CONN_TYPESCRIPT_EXAMPLE_HTTP=http://user:pass@example.com/\n" + "AIRFLOW_VAR_TYPESCRIPT_EXAMPLE_GREETING=greetings from e2e\n" + ) + os.environ["ENV_FILE_PATH"] = str(dot_env_file) + + def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory): tmp_dir = tmp_path_factory.mktemp("breeze-airflow-e2e-tests") @@ -694,6 +808,9 @@ def spin_up_airflow_environment(tmp_path_factory: pytest.TempPathFactory): elif E2E_TEST_MODE == "openlineage": compose_file_names.append("openlineage.yml") _setup_openlineage_integration(dot_env_file, tmp_dir) + elif E2E_TEST_MODE == "ts_sdk": + compose_file_names.append("ts.yml") + _setup_ts_sdk_integration(dot_env_file, tmp_dir) # # Please Do not use this Fernet key in any deployments! Please generate your own key. diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py index 6b8d7c42d0c83..2b245a0021cc1 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py @@ -90,6 +90,18 @@ # and module fetches go through the HTTPS proxy (no git/gcc needed). GO_BUILDER_IMAGE = os.environ.get("GO_BUILDER_IMAGE", "golang:1.25-alpine") +# TypeScript SDK E2E test paths +TS_SDK_ROOT_PATH = AIRFLOW_ROOT_PATH / "ts-sdk" +TS_SDK_EXAMPLE_PATH = TS_SDK_ROOT_PATH / "example" +TS_COMPOSE_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "ts.yml" +# Node image used both to build the bundle and to provide the node binary to the +# worker (ts.yml); keeping them identical guarantees the bundle runs on the same +# runtime it was built for. +NODE_IMAGE = os.environ.get("NODE_IMAGE", "node:22-slim") +# Writable HOME for the containerized pnpm build; keeps the pnpm store and +# corepack cache between runs (files/ is gitignored). +TS_SDK_BUILD_HOME_PATH = AIRFLOW_ROOT_PATH / "files" / "pnpm-home" + # Local provider sources are mounted into the airflow containers under this directory so # ``_PIP_ADDITIONAL_REQUIREMENTS`` can install the in-tree (latest, possibly unreleased) # provider rather than the published one from PyPI. diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/__init__.py b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/__init__.py new file mode 100644 index 0000000000000..13a83393a9124 --- /dev/null +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/__init__.py @@ -0,0 +1,16 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py new file mode 100644 index 0000000000000..542bbf387cd4d --- /dev/null +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py @@ -0,0 +1,169 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""End-to-end tests for coordinator-mode TypeScript tasks. + +Run with:: + + E2E_TEST_MODE=ts_sdk uv run --project airflow-e2e-tests pytest \\ + tests/airflow_e2e_tests/ts_sdk_tests/ -xvs + +What is verified +---------------- +``conftest._setup_ts_sdk_integration`` builds the ``ts-sdk/example`` bundle with +pnpm inside an ephemeral Node container and drops ``bundle.mjs`` plus its +``airflow-metadata.yaml`` sidecar into the directory ``NodeCoordinator`` scans. +The ``typescript_example`` Dag (``ts-sdk/example/dags/typescript_example.py``) +mixes a native Python task with TypeScript stub tasks:: + + python_start >> build_message; read_connection + +* ``build_message`` / ``read_connection`` are ``@task.stub(queue="typescript")`` + tasks whose handlers live in the bundle (``ts-sdk/example/src/main.ts``). The + ``typescript`` queue is routed to ``NodeCoordinator``, which launches the + bundle with the node binary provided by the ``node-provider`` compose service + and drives it through the msgpack-over-IPC coordinator protocol. +* ``python_start`` (Python) pushes an XCom; ``build_message`` (TS) pulls it, + reads the ``typescript_example_greeting`` Variable, pushes the + ``typescript_message`` XCom, and returns a message object; ``read_connection`` + (TS) fetches the ``typescript_example_http`` Connection. + +The Dag is triggered exactly once by the module-scoped ``completed_run`` +fixture; each test asserts a different facet of that single run. Together they +confirm, end-to-end: + +1. ``NodeCoordinator`` locates the bundle, launches it with the volume-provided + Node runtime, and every TypeScript task reports ``SucceedTask``. +2. Variable / Connection reads and XCom writes work through the Task Execution + API, and XCom crosses the Python <-> TypeScript boundary in both directions. +3. Structured task logs emitted by the TypeScript runtime over the coordinator + logs channel reach Airflow's task-log store. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from datetime import datetime, timezone + +import pytest + +from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient + +# Coordinator startup only needs to launch node with the prebuilt bundle; +# allow room for scheduling and the Python upstream task. +_TS_TASK_TIMEOUT = 600 +# Task logs are written when the task finishes; allow a little slack for them +# to become retrievable through the API after the run reaches a terminal state. +_LOG_FETCH_TIMEOUT = 120 + +_DAG_ID = "typescript_example" + + +@dataclass +class _CompletedRun: + """The single ``typescript_example`` run shared across the assertions in this module.""" + + client: AirflowClient + run_id: str + state: str + ti_states: dict[str, str] + + def xcom(self, task_id: str, key: str = "return_value"): + return self.client.get_xcom_value(dag_id=_DAG_ID, task_id=task_id, run_id=self.run_id, key=key).get( + "value" + ) + + def logs(self, task_id: str, try_number: int = 1) -> str: + """Return the concatenated task-log records for *task_id*, retrying until present.""" + deadline = time.monotonic() + _LOG_FETCH_TIMEOUT + while True: + resp = self.client.get_task_logs( + dag_id=_DAG_ID, run_id=self.run_id, task_id=task_id, try_number=try_number + ) + text = "\n".join(str(entry) for entry in resp.get("content", [])) + if text.strip() or time.monotonic() > deadline: + return text + time.sleep(3) + + +@pytest.fixture(scope="module") +def completed_run() -> _CompletedRun: + """Trigger ``typescript_example`` once and wait for it to finish. + + Module-scoped so the run happens a single time; every test in this module + inspects the resulting states, XComs, and logs. + """ + client = AirflowClient() + resp = client.trigger_dag(_DAG_ID, json={"logical_date": datetime.now(timezone.utc).isoformat()}) + run_id = resp["dag_run_id"] + state = client.wait_for_dag_run(dag_id=_DAG_ID, run_id=run_id, timeout=_TS_TASK_TIMEOUT) + ti_resp = client.get_task_instances(dag_id=_DAG_ID, run_id=run_id) + ti_states = {ti["task_id"]: ti.get("state") for ti in ti_resp.get("task_instances", [])} + return _CompletedRun(client=client, run_id=run_id, state=state, ti_states=ti_states) + + +def test_dag_run_succeeded(completed_run: _CompletedRun): + assert completed_run.state == "success", ( + f"expected the run to succeed; got {completed_run.state!r}. task states: {completed_run.ti_states}" + ) + + +def test_task_states(completed_run: _CompletedRun): + """The Python task and both coordinator-mode TypeScript tasks end ``success``.""" + expected = { + "python_start": "success", + "build_message": "success", + "read_connection": "success", + } + for task_id, want in expected.items(): + assert completed_run.ti_states.get(task_id) == want, ( + f"{task_id!r} expected {want!r}. all task states: {completed_run.ti_states}" + ) + + +def test_build_message_xcom_round_trip(completed_run: _CompletedRun): + """XCom crosses Python -> TypeScript -> XCom, including the Variable read. + + ``build_message`` pulls ``python_start``'s XCom, prefixes it with the + ``typescript_example_greeting`` Variable, pushes the combined text under the + ``typescript_message`` key, and returns it as ``return_value``. + """ + assert completed_run.xcom("python_start") == "hello from Python" + + message = "greetings from e2e; upstream=hello from Python" + value = completed_run.xcom("build_message") + assert value == {"message": message, "upstream": "hello from Python"}, ( + f"unexpected 'build_message' return_value: {value!r}" + ) + assert completed_run.xcom("build_message", key="typescript_message") == message + + +def test_read_connection_xcom(completed_run: _CompletedRun): + """The TypeScript task reads the ``typescript_example_http`` Connection.""" + value = completed_run.xcom("read_connection") + assert value == { + "id": "typescript_example_http", + "type": "http", + "host": "example.com", + "login": "user", + "hasPassword": True, + }, f"unexpected 'read_connection' return_value: {value!r}" + + +def test_coordinator_logs_reach_task_log_store(completed_run: _CompletedRun): + """Runtime logs emitted over the coordinator logs channel land in the task logs.""" + assert "[ts-sdk.runtime] Coordinator runtime started" in completed_run.logs("build_message") diff --git a/dev/breeze/doc/ci/04_selective_checks.md b/dev/breeze/doc/ci/04_selective_checks.md index 59d55d1117840..7475d15490414 100644 --- a/dev/breeze/doc/ci/04_selective_checks.md +++ b/dev/breeze/doc/ci/04_selective_checks.md @@ -584,6 +584,7 @@ GitHub Actions to pass the list of parameters to a command to execute | run-mypy | Whether mypy check is supposed to run in this build | true | | | run-system-tests | Whether system tests should be run ("true"/"false") | true | | | run-task-sdk-tests | Whether Task SDK tests should be run ("true"/"false") | true | | +| run-ts-sdk-e2e-tests | Whether TypeScript SDK e2e tests should be run — on `ts-sdk/`, TS e2e test, or Node coordinator changes ("true"/"false") | true | | | run-ui-tests | Whether UI tests should be run ("true"/"false") | true | | | run-unit-tests | Whether unit tests should be run ("true"/"false") | true | | | run-www-tests | Whether Legacy WWW tests should be run ("true"/"false") | true | | diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py b/dev/breeze/src/airflow_breeze/utils/selective_checks.py index 970c90e606fce..ae0ca7586035d 100644 --- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py +++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py @@ -135,6 +135,7 @@ class FileGroupForCi(Enum): GO_SDK_E2E_FILES = auto() OPENLINEAGE_E2E_FILES = auto() OPENLINEAGE_E2E_COMPAT_FILES = auto() + TS_SDK_E2E_FILES = auto() ALL_PYPROJECT_TOML_FILES = auto() ALL_PYTHON_FILES = auto() ALL_SOURCE_FILES = auto() @@ -271,6 +272,13 @@ def __hash__(self): r"^airflow-e2e-tests/tests/airflow_e2e_tests/constants\.py$", r"^airflow-e2e-tests/docker/openlineage-compat\.Dockerfile$", ], + FileGroupForCi.TS_SDK_E2E_FILES: [ + r"^ts-sdk/.*", + r"^airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/.*", + r"^airflow-e2e-tests/docker/ts\.yml$", + r"^task-sdk/src/airflow/sdk/coordinators/_subprocess\.py$", + r"^task-sdk/src/airflow/sdk/coordinators/node/.*", + ], FileGroupForCi.PYTHON_PRODUCTION_FILES: [ # Production Python source the runtime ships — excludes tests, docs, # dev tooling, and generated files within those trees. Used by @@ -1078,6 +1086,10 @@ def run_openlineage_e2e_compat_tests(self) -> bool: return True return self._should_be_run(FileGroupForCi.OPENLINEAGE_E2E_COMPAT_FILES) + @cached_property + def run_ts_sdk_e2e_tests(self) -> bool: + return self._should_be_run(FileGroupForCi.TS_SDK_E2E_FILES) + @cached_property def run_amazon_tests(self) -> bool: if self.providers_test_types_list_as_strings_in_json == "[]": @@ -1222,6 +1234,7 @@ def prod_image_build(self) -> bool: or self.run_go_sdk_e2e_tests or self.run_openlineage_e2e_tests or self.run_openlineage_e2e_compat_tests + or self.run_ts_sdk_e2e_tests or self.run_ui_e2e_tests ) diff --git a/dev/breeze/tests/test_selective_checks.py b/dev/breeze/tests/test_selective_checks.py index 3698d41d51b33..c901496baec3d 100644 --- a/dev/breeze/tests/test_selective_checks.py +++ b/dev/breeze/tests/test_selective_checks.py @@ -811,6 +811,21 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str): "Task SDK integration tests and prod image build should run but no other tests", ) ), + ( + pytest.param( + ("ts-sdk/src/coordinator/runtime.ts",), + { + "prod-image-build": "true", + "run-unit-tests": "false", + "run-task-sdk-tests": "false", + "run-task-sdk-integration-tests": "false", + "run-ts-sdk-e2e-tests": "true", + "run-go-sdk-e2e-tests": "false", + "full-tests-needed": "false", + }, + id="TypeScript SDK files changed - TS SDK e2e tests and prod image build should run", + ) + ), ( pytest.param( ("airflow-ctl/src/airflowctl/random.py",), diff --git a/ts-sdk/example/.gitignore b/ts-sdk/example/.gitignore index 8ee10058ff835..1652dc9c671bd 100644 --- a/ts-sdk/example/.gitignore +++ b/ts-sdk/example/.gitignore @@ -1,3 +1,4 @@ .pnpm-store/ dist/ node_modules/ +pnpm-lock.yaml From c898038cf6937ea950619d089f24ab5e6344406e Mon Sep 17 00:00:00 2001 From: "Guan-Ming (Wesley) Chiu" <105915352+guan404ming@users.noreply.github.com> Date: Fri, 3 Jul 2026 23:02:07 +0800 Subject: [PATCH 2/5] Register ts_sdk e2e test mode in breeze CLI --- .github/workflows/airflow-e2e-tests.yml | 4 ++-- dev/breeze/src/airflow_breeze/commands/testing_commands.py | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/airflow-e2e-tests.yml b/.github/workflows/airflow-e2e-tests.yml index 8cb55568e3e5b..26f94e9740f39 100644 --- a/.github/workflows/airflow-e2e-tests.yml +++ b/.github/workflows/airflow-e2e-tests.yml @@ -49,7 +49,7 @@ on: # yamllint disable-line rule:truthy type: string required: true e2e_test_mode: - description: "Test mode - basic, remote_log, remote_log_elasticsearch, remote_log_opensearch, xcom_object_storage, event_driven, java_sdk, or go_sdk" # yamllint disable-line rule:line-length + description: "Test mode - basic, remote_log, remote_log_elasticsearch, remote_log_opensearch, xcom_object_storage, event_driven, java_sdk, go_sdk, or ts_sdk" # yamllint disable-line rule:line-length type: string default: "basic" java-sdk-version: @@ -84,7 +84,7 @@ on: # yamllint disable-line rule:truthy type: string default: "" e2e_test_mode: - description: "Test mode - basic, remote_log, remote_log_elasticsearch, remote_log_opensearch, xcom_object_storage, event_driven, java_sdk, or go_sdk" # yamllint disable-line rule:line-length + description: "Test mode - basic, remote_log, remote_log_elasticsearch, remote_log_opensearch, xcom_object_storage, event_driven, java_sdk, go_sdk, or ts_sdk" # yamllint disable-line rule:line-length type: string default: "basic" java-sdk-version: diff --git a/dev/breeze/src/airflow_breeze/commands/testing_commands.py b/dev/breeze/src/airflow_breeze/commands/testing_commands.py index fd3f0444bd895..7731c25a3548b 100644 --- a/dev/breeze/src/airflow_breeze/commands/testing_commands.py +++ b/dev/breeze/src/airflow_breeze/commands/testing_commands.py @@ -1513,6 +1513,7 @@ def _build_openlineage_e2e_compat_image(airflow_version: str, python: str) -> st "java_sdk", "go_sdk", "openlineage", + "ts_sdk", ], case_sensitive=False, ), From a94077a1a4fcc0f60064453b6fabe0a1e5601ecb Mon Sep 17 00:00:00 2001 From: "Guan-Ming (Wesley) Chiu" <105915352+guan404ming@users.noreply.github.com> Date: Sat, 18 Jul 2026 21:53:26 +0800 Subject: [PATCH 3/5] Exercise real airflow-ts-pack packing in TypeScript SDK e2e tests The reviewer asked the e2e tests to wait for airflow-ts-pack and use its real packing path. The example build now runs airflow-ts-pack, which embeds the airflow metadata in bundle.mjs, so the hand-written airflow-metadata.yaml sidecar would bypass the path users actually take. --- .../tests/airflow_e2e_tests/conftest.py | 24 +++++++------------ .../ts_sdk_tests/test_ts_sdk_dag.py | 5 ++-- .../output_testing_airflow-e2e-tests.svg | 2 +- .../output_testing_airflow-e2e-tests.txt | 2 +- 4 files changed, 14 insertions(+), 19 deletions(-) diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py index 1e0a6e0c465e5..e4ab0027220a1 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py @@ -18,7 +18,6 @@ import json import os -import re import subprocess from concurrent.futures import ThreadPoolExecutor from datetime import datetime @@ -67,7 +66,6 @@ TS_COMPOSE_PATH, TS_SDK_BUILD_HOME_PATH, TS_SDK_EXAMPLE_PATH, - TS_SDK_ROOT_PATH, XCOM_BUCKET, ) @@ -650,10 +648,11 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir): """Set up the ts_sdk E2E test mode. Builds the ts-sdk example bundle with pnpm inside an ephemeral Node - container (the host needs no Node toolchain), drops it with its - airflow-metadata.yaml sidecar into the directory ``NodeCoordinator`` - scans, copies the Python stub Dag, and writes the coordinator - configuration. + container (the host needs no Node toolchain); the example build runs + ``airflow-ts-pack``, which embeds the airflow metadata in the bundle + itself. The bundle is dropped into the directory ``NodeCoordinator`` + scans, the Python stub Dag is copied, and the coordinator + configuration is written. The worker has no Node runtime of its own; the node-provider service in ``ts.yml`` copies the node binary out of the same image used here for the @@ -706,18 +705,13 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir): # Copy the compose override into the temp directory. copyfile(TS_COMPOSE_PATH, tmp_dir / "ts.yml") - # Place the bundle and its metadata sidecar where the compose bind-mount - # (./ts-bundles) exposes them to the worker at /opt/airflow/ts-bundles. + # Place the bundle where the compose bind-mount (./ts-bundles) exposes it + # to the worker at /opt/airflow/ts-bundles. airflow-ts-pack embeds the + # airflow metadata in bundle.mjs itself, so no sidecar is written — the + # coordinator must resolve the schema version from the embedded manifest. ts_bundles_dir = tmp_dir / "ts-bundles" ts_bundles_dir.mkdir() copyfile(TS_SDK_EXAMPLE_PATH / "dist" / "bundle.mjs", ts_bundles_dir / "bundle.mjs") - supervisor_ts = (TS_SDK_ROOT_PATH / "src" / "generated" / "supervisor.ts").read_text() - version_match = re.search(r'SUPERVISOR_API_VERSION = "(\d{4}-\d{2}-\d{2})"', supervisor_ts) - if not version_match: - raise RuntimeError("Cannot find SUPERVISOR_API_VERSION in ts-sdk/src/generated/supervisor.ts") - (ts_bundles_dir / "airflow-metadata.yaml").write_text( - f'sdk:\n supervisor_schema_version: "{version_match.group(1)}"\n' - ) # Copy the TS SDK example stub Dag so Airflow can discover and serialize it. copyfile( diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py index 542bbf387cd4d..8fd7d4835ca4c 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py @@ -24,8 +24,9 @@ What is verified ---------------- ``conftest._setup_ts_sdk_integration`` builds the ``ts-sdk/example`` bundle with -pnpm inside an ephemeral Node container and drops ``bundle.mjs`` plus its -``airflow-metadata.yaml`` sidecar into the directory ``NodeCoordinator`` scans. +pnpm inside an ephemeral Node container — the example build runs +``airflow-ts-pack``, which embeds the airflow metadata in ``bundle.mjs`` — and +drops the bundle into the directory ``NodeCoordinator`` scans. The ``typescript_example`` Dag (``ts-sdk/example/dags/typescript_example.py``) mixes a native Python task with TypeScript stub tasks:: diff --git a/dev/breeze/doc/images/output_testing_airflow-e2e-tests.svg b/dev/breeze/doc/images/output_testing_airflow-e2e-tests.svg index 7f2f3b4a26aa7..942306b371c88 100644 --- a/dev/breeze/doc/images/output_testing_airflow-e2e-tests.svg +++ b/dev/breeze/doc/images/output_testing_airflow-e2e-tests.svg @@ -148,7 +148,7 @@ │(TEXT)│ │--e2e-test-mode               Specify the mode to use for E2E tests. [default: basic]│ │(basic|remote_log|remote_log_elasticsearch|remote_log_opensearch|xcom_object_sto│ -│rage|event_driven|java_sdk|go_sdk|openlineage)│ +│rage|event_driven|java_sdk|go_sdk|openlineage|ts_sdk)│ │--airflow-version             Run the openlineage mode against a released Airflow version (e.g. 3.1.8) with   │ │current providers from main, instead of the default PROD image. Builds a        │ │lightweight image from apache/airflow:<version>. Only used by the openlineage   │ diff --git a/dev/breeze/doc/images/output_testing_airflow-e2e-tests.txt b/dev/breeze/doc/images/output_testing_airflow-e2e-tests.txt index 123969fa33710..31336273b3b1a 100644 --- a/dev/breeze/doc/images/output_testing_airflow-e2e-tests.txt +++ b/dev/breeze/doc/images/output_testing_airflow-e2e-tests.txt @@ -1 +1 @@ -45750ff1a736395d4f563fdca433886f +3005f4343ce3d9c33960042d8d7e2d20 From a498f5da934e23937d800ac0f656eef5ebb106c4 Mon Sep 17 00:00:00 2001 From: "Guan-Ming (Wesley) Chiu" <105915352+guan404ming@users.noreply.github.com> Date: Sat, 18 Jul 2026 22:44:26 +0800 Subject: [PATCH 4/5] Trim ts_sdk e2e docstrings and comments to essentials --- airflow-e2e-tests/docker/ts.yml | 11 ++-- .../tests/airflow_e2e_tests/conftest.py | 42 ++++--------- .../tests/airflow_e2e_tests/constants.py | 8 +-- .../ts_sdk_tests/test_ts_sdk_dag.py | 60 ++++--------------- 4 files changed, 31 insertions(+), 90 deletions(-) diff --git a/airflow-e2e-tests/docker/ts.yml b/airflow-e2e-tests/docker/ts.yml index 6ea502425056d..056106ff9d54a 100644 --- a/airflow-e2e-tests/docker/ts.yml +++ b/airflow-e2e-tests/docker/ts.yml @@ -17,13 +17,10 @@ # Docker Compose override for ts_sdk E2E test mode. # -# NodeCoordinator launches the TypeScript bundle with a Node.js runtime, which -# the stock Airflow worker image does not ship. The node-provider service -# copies the node binary from the official Node image (the same image -# conftest._setup_ts_sdk_integration builds the bundle with) into a shared -# volume, the bundle built by conftest is bind-mounted into the directory -# NodeCoordinator scans, and the worker consumes the "typescript" Celery queue -# where @task.stub tasks are routed. +# The stock worker image ships no Node.js runtime, so node-provider copies the +# node binary from the same image conftest builds the bundle with into a +# shared volume. The bundle is bind-mounted where NodeCoordinator scans, and +# the worker consumes the "typescript" queue where @task.stub tasks are routed. --- services: node-provider: diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py index e4ab0027220a1..df4bc7ea08362 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py @@ -647,24 +647,17 @@ def _setup_openlineage_integration(dot_env_file, tmp_dir): def _setup_ts_sdk_integration(dot_env_file, tmp_dir): """Set up the ts_sdk E2E test mode. - Builds the ts-sdk example bundle with pnpm inside an ephemeral Node - container (the host needs no Node toolchain); the example build runs - ``airflow-ts-pack``, which embeds the airflow metadata in the bundle - itself. The bundle is dropped into the directory ``NodeCoordinator`` - scans, the Python stub Dag is copied, and the coordinator - configuration is written. - - The worker has no Node runtime of its own; the node-provider service in - ``ts.yml`` copies the node binary out of the same image used here for the - build, so the bundle executes on the runtime it was built for. + Builds the ts-sdk example bundle in an ephemeral Node container (no host + Node toolchain needed); the example build runs ``airflow-ts-pack``, which + embeds the airflow metadata in ``bundle.mjs``. The node-provider service + in ``ts.yml`` gives the worker the node binary from the same image, so + the bundle runs on the runtime it was built for. """ - # --user keeps build outputs owned by the current user (not root); HOME - # points at a writable, gitignored dir so the pnpm store and corepack - # cache persist between runs. corepack resolves the pnpm version pinned - # by ts-sdk/package.json's packageManager field. + # --user keeps build outputs owned by the current user; HOME is a + # writable, gitignored dir so pnpm/corepack caches persist between runs. TS_SDK_BUILD_HOME_PATH.mkdir(parents=True, exist_ok=True) - # Shims go into a writable dir on PATH (the container user cannot write to - # /usr/local/bin) so nested `pnpm run ...` invocations resolve pnpm too. + # corepack shims go in $HOME/bin (on PATH) because the container user + # cannot write to /usr/local/bin. build_script = ( 'export PATH="$HOME/bin:$PATH"' ' && mkdir -p "$HOME/bin"' @@ -689,8 +682,6 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir): "-e", "COREPACK_ENABLE_DOWNLOAD_PROMPT=0", "-e", - # Non-interactive pnpm: never prompt (e.g. before replacing a - # node_modules left behind by a host-side install). "CI=true", "-v", f"{AIRFLOW_ROOT_PATH}:/repo", @@ -702,26 +693,18 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir): check=True, ) - # Copy the compose override into the temp directory. copyfile(TS_COMPOSE_PATH, tmp_dir / "ts.yml") - # Place the bundle where the compose bind-mount (./ts-bundles) exposes it - # to the worker at /opt/airflow/ts-bundles. airflow-ts-pack embeds the - # airflow metadata in bundle.mjs itself, so no sidecar is written — the - # coordinator must resolve the schema version from the embedded manifest. + # Deliberately no metadata sidecar: the coordinator must resolve the schema + # version from the metadata airflow-ts-pack embedded in the bundle. ts_bundles_dir = tmp_dir / "ts-bundles" ts_bundles_dir.mkdir() copyfile(TS_SDK_EXAMPLE_PATH / "dist" / "bundle.mjs", ts_bundles_dir / "bundle.mjs") - # Copy the TS SDK example stub Dag so Airflow can discover and serialize it. copyfile( TS_SDK_EXAMPLE_PATH / "dags" / "typescript_example.py", tmp_dir / "dags" / "typescript_example.py" ) - # Coordinator registry: maps the logical name "ts" to NodeCoordinator, which - # scans bundles_root for bundle.mjs and launches it with the node binary - # provided by the node-provider service. - # Queue mapping: routes tasks on the "typescript" queue to "ts". coordinator_config = json.dumps( { "ts": { @@ -738,10 +721,9 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir): dot_env_file.write_text( f"AIRFLOW_UID={os.getuid()}\n" f"NODE_IMAGE={NODE_IMAGE}\n" - # Single-quote the JSON values so Docker Compose reads them literally. + # single-quoted so Docker Compose reads the JSON literally f"AIRFLOW__SDK__COORDINATORS='{coordinator_config}'\n" f"AIRFLOW__SDK__QUEUE_TO_COORDINATOR='{queue_to_coordinator}'\n" - # Connection and variable read by the TS example bundle tasks. "AIRFLOW_CONN_TYPESCRIPT_EXAMPLE_HTTP=http://user:pass@example.com/\n" "AIRFLOW_VAR_TYPESCRIPT_EXAMPLE_GREETING=greetings from e2e\n" ) diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py index 2b245a0021cc1..e389a8bc6b9bf 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py @@ -94,12 +94,10 @@ TS_SDK_ROOT_PATH = AIRFLOW_ROOT_PATH / "ts-sdk" TS_SDK_EXAMPLE_PATH = TS_SDK_ROOT_PATH / "example" TS_COMPOSE_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" / "ts.yml" -# Node image used both to build the bundle and to provide the node binary to the -# worker (ts.yml); keeping them identical guarantees the bundle runs on the same -# runtime it was built for. +# Builds the bundle and provides the worker's node binary (ts.yml), so the +# bundle runs on the runtime it was built for. NODE_IMAGE = os.environ.get("NODE_IMAGE", "node:22-slim") -# Writable HOME for the containerized pnpm build; keeps the pnpm store and -# corepack cache between runs (files/ is gitignored). +# Writable HOME for the containerized pnpm build; gitignored, caches persist. TS_SDK_BUILD_HOME_PATH = AIRFLOW_ROOT_PATH / "files" / "pnpm-home" # Local provider sources are mounted into the airflow containers under this directory so diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py index 8fd7d4835ca4c..3c30810065472 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py @@ -21,37 +21,14 @@ E2E_TEST_MODE=ts_sdk uv run --project airflow-e2e-tests pytest \\ tests/airflow_e2e_tests/ts_sdk_tests/ -xvs -What is verified ----------------- -``conftest._setup_ts_sdk_integration`` builds the ``ts-sdk/example`` bundle with -pnpm inside an ephemeral Node container — the example build runs -``airflow-ts-pack``, which embeds the airflow metadata in ``bundle.mjs`` — and -drops the bundle into the directory ``NodeCoordinator`` scans. -The ``typescript_example`` Dag (``ts-sdk/example/dags/typescript_example.py``) -mixes a native Python task with TypeScript stub tasks:: - - python_start >> build_message; read_connection - -* ``build_message`` / ``read_connection`` are ``@task.stub(queue="typescript")`` - tasks whose handlers live in the bundle (``ts-sdk/example/src/main.ts``). The - ``typescript`` queue is routed to ``NodeCoordinator``, which launches the - bundle with the node binary provided by the ``node-provider`` compose service - and drives it through the msgpack-over-IPC coordinator protocol. -* ``python_start`` (Python) pushes an XCom; ``build_message`` (TS) pulls it, - reads the ``typescript_example_greeting`` Variable, pushes the - ``typescript_message`` XCom, and returns a message object; ``read_connection`` - (TS) fetches the ``typescript_example_http`` Connection. - -The Dag is triggered exactly once by the module-scoped ``completed_run`` -fixture; each test asserts a different facet of that single run. Together they -confirm, end-to-end: - -1. ``NodeCoordinator`` locates the bundle, launches it with the volume-provided - Node runtime, and every TypeScript task reports ``SucceedTask``. -2. Variable / Connection reads and XCom writes work through the Task Execution - API, and XCom crosses the Python <-> TypeScript boundary in both directions. -3. Structured task logs emitted by the TypeScript runtime over the coordinator - logs channel reach Airflow's task-log store. +The ``typescript_example`` Dag mixes a Python task with ``@task.stub`` +TypeScript tasks whose handlers live in the ``airflow-ts-pack`` bundle built +by ``conftest._setup_ts_sdk_integration``. Triggered once via the +module-scoped ``completed_run`` fixture, the run confirms end-to-end that +``NodeCoordinator`` launches the bundle on the volume-provided Node runtime, +Variable/Connection reads and Python <-> TypeScript XCom round-trips work +through the Task Execution API, and coordinator-channel logs reach the +task-log store. """ from __future__ import annotations @@ -76,8 +53,6 @@ @dataclass class _CompletedRun: - """The single ``typescript_example`` run shared across the assertions in this module.""" - client: AirflowClient run_id: str state: str @@ -89,7 +64,7 @@ def xcom(self, task_id: str, key: str = "return_value"): ) def logs(self, task_id: str, try_number: int = 1) -> str: - """Return the concatenated task-log records for *task_id*, retrying until present.""" + """Fetch task logs, retrying until present (log upload is async).""" deadline = time.monotonic() + _LOG_FETCH_TIMEOUT while True: resp = self.client.get_task_logs( @@ -103,11 +78,7 @@ def logs(self, task_id: str, try_number: int = 1) -> str: @pytest.fixture(scope="module") def completed_run() -> _CompletedRun: - """Trigger ``typescript_example`` once and wait for it to finish. - - Module-scoped so the run happens a single time; every test in this module - inspects the resulting states, XComs, and logs. - """ + """Trigger ``typescript_example`` once; every test inspects the same run.""" client = AirflowClient() resp = client.trigger_dag(_DAG_ID, json={"logical_date": datetime.now(timezone.utc).isoformat()}) run_id = resp["dag_run_id"] @@ -124,7 +95,6 @@ def test_dag_run_succeeded(completed_run: _CompletedRun): def test_task_states(completed_run: _CompletedRun): - """The Python task and both coordinator-mode TypeScript tasks end ``success``.""" expected = { "python_start": "success", "build_message": "success", @@ -137,12 +107,8 @@ def test_task_states(completed_run: _CompletedRun): def test_build_message_xcom_round_trip(completed_run: _CompletedRun): - """XCom crosses Python -> TypeScript -> XCom, including the Variable read. - - ``build_message`` pulls ``python_start``'s XCom, prefixes it with the - ``typescript_example_greeting`` Variable, pushes the combined text under the - ``typescript_message`` key, and returns it as ``return_value``. - """ + """``build_message`` combines ``python_start``'s XCom with the Variable, + pushes it under ``typescript_message``, and returns it.""" assert completed_run.xcom("python_start") == "hello from Python" message = "greetings from e2e; upstream=hello from Python" @@ -154,7 +120,6 @@ def test_build_message_xcom_round_trip(completed_run: _CompletedRun): def test_read_connection_xcom(completed_run: _CompletedRun): - """The TypeScript task reads the ``typescript_example_http`` Connection.""" value = completed_run.xcom("read_connection") assert value == { "id": "typescript_example_http", @@ -166,5 +131,4 @@ def test_read_connection_xcom(completed_run: _CompletedRun): def test_coordinator_logs_reach_task_log_store(completed_run: _CompletedRun): - """Runtime logs emitted over the coordinator logs channel land in the task logs.""" assert "[ts-sdk.runtime] Coordinator runtime started" in completed_run.logs("build_message") From 7b2f74b01648988ae7a87f055c54de94b979d83e Mon Sep 17 00:00:00 2001 From: "Guan-Ming (Wesley) Chiu" <105915352+guan404ming@users.noreply.github.com> Date: Sun, 19 Jul 2026 17:38:22 +0800 Subject: [PATCH 5/5] Build TS SDK e2e example natively in CI and skip doc-only changes --- .github/workflows/airflow-e2e-tests.yml | 20 +++++++++++-- .../tests/airflow_e2e_tests/conftest.py | 29 ++++++++++--------- .../airflow_breeze/utils/selective_checks.py | 2 +- dev/breeze/tests/test_selective_checks.py | 11 +++++++ 4 files changed, 44 insertions(+), 18 deletions(-) diff --git a/.github/workflows/airflow-e2e-tests.yml b/.github/workflows/airflow-e2e-tests.yml index 26f94e9740f39..3ce03733da04f 100644 --- a/.github/workflows/airflow-e2e-tests.yml +++ b/.github/workflows/airflow-e2e-tests.yml @@ -121,8 +121,9 @@ jobs: use-uv: ${{ inputs.use-uv }} make-mnt-writeable-and-cleanup: true id: breeze - # The java_sdk mode runs the Gradle builds and the go_sdk mode runs the Go - # build; every other e2e mode skips the toolchain steps below. + # The java_sdk mode runs the Gradle builds, the go_sdk mode runs the Go + # build and the ts_sdk mode runs the pnpm build; every other e2e mode + # skips the toolchain steps below. # LANG_SDK_NATIVE_TOOLCHAIN=true (the same switch the lang-SDK k8s job uses) # makes the e2e conftest build with the host toolchain provisioned here # instead of an ephemeral toolchain container, skipping the image pull; local @@ -172,12 +173,25 @@ jobs: key: e2e-go-sdk-go-v1-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('go-sdk/go.mod', 'go-sdk/go.sum') }} # yamllint disable-line rule:line-length restore-keys: | e2e-go-sdk-go-v1-${{ runner.os }}-${{ runner.arch }}- + - name: "Setup pnpm for the TypeScript SDK build" + if: inputs.e2e_test_mode == 'ts_sdk' + uses: pnpm/action-setup@0ebf47130e4866e96fce0953f49152a61190b271 # v6.0.9 + with: + package_json_file: ts-sdk/package.json + run_install: false + - name: "Setup Node for the TypeScript SDK build" + if: inputs.e2e_test_mode == 'ts_sdk' + uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 + with: + node-version: 22 + cache: 'pnpm' + cache-dependency-path: 'ts-sdk/pnpm-lock.yaml' - name: "Test e2e integration tests" run: breeze testing airflow-e2e-tests env: DOCKER_IMAGE: "${{ inputs.docker-image-tag }}" E2E_TEST_MODE: "${{ inputs.e2e_test_mode }}" - LANG_SDK_NATIVE_TOOLCHAIN: "${{ (inputs.e2e_test_mode == 'java_sdk' || inputs.e2e_test_mode == 'go_sdk') && 'true' || 'false' }}" # yamllint disable-line rule:line-length + LANG_SDK_NATIVE_TOOLCHAIN: "${{ (inputs.e2e_test_mode == 'java_sdk' || inputs.e2e_test_mode == 'go_sdk' || inputs.e2e_test_mode == 'ts_sdk') && 'true' || 'false' }}" # yamllint disable-line rule:line-length - name: "Save Java SDK Gradle dependency cache" # Saved even when the e2e tests fail: the Gradle warm-up is independent of # test outcome, and actions/cache's post step would drop it on every red run. diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py index df4bc7ea08362..b97add595fd11 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py @@ -66,6 +66,7 @@ TS_COMPOSE_PATH, TS_SDK_BUILD_HOME_PATH, TS_SDK_EXAMPLE_PATH, + TS_SDK_ROOT_PATH, XCOM_BUCKET, ) @@ -644,15 +645,14 @@ def _setup_openlineage_integration(dot_env_file, tmp_dir): copyfile(OPENLINEAGE_COMPOSE_PATH, tmp_dir / "openlineage.yml") -def _setup_ts_sdk_integration(dot_env_file, tmp_dir): - """Set up the ts_sdk E2E test mode. - - Builds the ts-sdk example bundle in an ephemeral Node container (no host - Node toolchain needed); the example build runs ``airflow-ts-pack``, which - embeds the airflow metadata in ``bundle.mjs``. The node-provider service - in ``ts.yml`` gives the worker the node binary from the same image, so - the bundle runs on the runtime it was built for. - """ +def _build_ts_sdk_example_bundle(*, native=False): + build_commands = ( + "pnpm install --frozen-lockfile && pnpm run build && cd example && pnpm install && pnpm run build" + ) + if native: + console.print("[yellow]Building TypeScript SDK example bundle (host toolchain)...") + subprocess.run(["bash", "-c", build_commands], cwd=TS_SDK_ROOT_PATH, check=True) + return # --user keeps build outputs owned by the current user; HOME is a # writable, gitignored dir so pnpm/corepack caches persist between runs. TS_SDK_BUILD_HOME_PATH.mkdir(parents=True, exist_ok=True) @@ -663,11 +663,7 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir): ' && mkdir -p "$HOME/bin"' ' && corepack enable --install-directory "$HOME/bin"' " && cd /repo/ts-sdk" - " && pnpm install --frozen-lockfile" - " && pnpm run build" - " && cd example" - " && pnpm install" - " && pnpm run build" + f" && {build_commands}" ) console.print(f"[yellow]Building TypeScript SDK example bundle ({NODE_IMAGE})...") subprocess.run( @@ -693,6 +689,11 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir): check=True, ) + +def _setup_ts_sdk_integration(dot_env_file, tmp_dir): + """Set up the ts_sdk E2E test mode.""" + _build_ts_sdk_example_bundle(native=LANG_SDK_NATIVE_TOOLCHAIN) + copyfile(TS_COMPOSE_PATH, tmp_dir / "ts.yml") # Deliberately no metadata sidecar: the coordinator must resolve the schema diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py b/dev/breeze/src/airflow_breeze/utils/selective_checks.py index ae0ca7586035d..b0204836b7169 100644 --- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py +++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py @@ -273,7 +273,7 @@ def __hash__(self): r"^airflow-e2e-tests/docker/openlineage-compat\.Dockerfile$", ], FileGroupForCi.TS_SDK_E2E_FILES: [ - r"^ts-sdk/.*", + r"^ts-sdk/(?!.*\.md$).*", r"^airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/.*", r"^airflow-e2e-tests/docker/ts\.yml$", r"^task-sdk/src/airflow/sdk/coordinators/_subprocess\.py$", diff --git a/dev/breeze/tests/test_selective_checks.py b/dev/breeze/tests/test_selective_checks.py index c901496baec3d..384ca066f9a48 100644 --- a/dev/breeze/tests/test_selective_checks.py +++ b/dev/breeze/tests/test_selective_checks.py @@ -826,6 +826,17 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str): id="TypeScript SDK files changed - TS SDK e2e tests and prod image build should run", ) ), + ( + pytest.param( + ("ts-sdk/README.md",), + { + "prod-image-build": "false", + "run-ts-sdk-e2e-tests": "false", + "full-tests-needed": "false", + }, + id="TypeScript SDK README-only change - TS SDK e2e tests should be skipped", + ) + ), ( pytest.param( ("airflow-ctl/src/airflowctl/random.py",),