This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 60b1848e4c6 Add integration tests for coordinator-mode TypeScript
tasks (#69328)
60b1848e4c6 is described below
commit 60b1848e4c65ecfe79b1d5bba4d611203576df6c
Author: Guan-Ming Chiu <[email protected]>
AuthorDate: Mon Jul 20 14:33:20 2026 +0800
Add integration tests for coordinator-mode TypeScript tasks (#69328)
* Add integration tests for coordinator-mode TypeScript tasks
* Register ts_sdk e2e test mode in breeze CLI
* 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.
* Trim ts_sdk e2e docstrings and comments to essentials
* Build TS SDK e2e example natively in CI and skip doc-only changes
---
.github/workflows/additional-prod-image-tests.yml | 16 +++
.github/workflows/airflow-e2e-tests.yml | 24 +++-
.github/workflows/ci-amd.yml | 2 +
.github/workflows/ci-arm.yml | 2 +
airflow-e2e-tests/docker/ts.yml | 42 +++++++
.../tests/airflow_e2e_tests/conftest.py | 94 +++++++++++++++
.../tests/airflow_e2e_tests/constants.py | 10 ++
.../airflow_e2e_tests/ts_sdk_tests/__init__.py | 16 +++
.../ts_sdk_tests/test_ts_sdk_dag.py | 134 +++++++++++++++++++++
dev/breeze/doc/ci/04_selective_checks.md | 1 +
.../images/output_testing_airflow-e2e-tests.svg | 2 +-
.../images/output_testing_airflow-e2e-tests.txt | 2 +-
.../airflow_breeze/commands/testing_commands.py | 1 +
.../src/airflow_breeze/utils/selective_checks.py | 13 ++
dev/breeze/tests/test_selective_checks.py | 26 ++++
ts-sdk/example/.gitignore | 1 +
16 files changed, 379 insertions(+), 7 deletions(-)
diff --git a/.github/workflows/additional-prod-image-tests.yml
b/.github/workflows/additional-prod-image-tests.yml
index 99a3a4c8728..a4ec751e5f4 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/airflow-e2e-tests.yml
b/.github/workflows/airflow-e2e-tests.yml
index 8cb55568e3e..3ce03733da0 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:
@@ -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/.github/workflows/ci-amd.yml b/.github/workflows/ci-amd.yml
index 9edb394a6cc..c2a477b6945 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 8f5ce3ac079..42ae7e5215b 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 00000000000..056106ff9d5
--- /dev/null
+++ b/airflow-e2e-tests/docker/ts.yml
@@ -0,0 +1,42 @@
+# 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.
+#
+# 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:
+ 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 72687795ae1..b97add595fd 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
@@ -55,6 +55,7 @@ from airflow_e2e_tests.constants import (
LANG_SDK_NATIVE_TOOLCHAIN,
LOCALSTACK_PATH,
LOGS_FOLDER,
+ NODE_IMAGE,
OPENLINEAGE_COMPOSE_PATH,
OPENSEARCH_PATH,
PROVIDERS_MOUNT_CONTAINER_PATH,
@@ -62,6 +63,10 @@ from airflow_e2e_tests.constants import (
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 +645,92 @@ def _setup_openlineage_integration(dot_env_file, tmp_dir):
copyfile(OPENLINEAGE_COMPOSE_PATH, tmp_dir / "openlineage.yml")
+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)
+ # 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"'
+ ' && corepack enable --install-directory "$HOME/bin"'
+ " && cd /repo/ts-sdk"
+ f" && {build_commands}"
+ )
+ 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",
+ "CI=true",
+ "-v",
+ f"{AIRFLOW_ROOT_PATH}:/repo",
+ NODE_IMAGE,
+ "bash",
+ "-c",
+ build_script,
+ ],
+ 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
+ # 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")
+
+ copyfile(
+ TS_SDK_EXAMPLE_PATH / "dags" / "typescript_example.py", tmp_dir /
"dags" / "typescript_example.py"
+ )
+
+ 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-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"
+ "AIRFLOW_CONN_TYPESCRIPT_EXAMPLE_HTTP=http://user:[email protected]/\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 +785,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 6b8d7c42d0c..e389a8bc6b9 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
@@ -90,6 +90,16 @@ GO_COMPOSE_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" /
"docker" / "go.yml"
# 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"
+# 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; 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
# ``_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 00000000000..13a83393a91
--- /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 00000000000..3c308100654
--- /dev/null
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_dag.py
@@ -0,0 +1,134 @@
+# 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
+
+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
+
+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:
+ 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:
+ """Fetch task logs, retrying until present (log upload is async)."""
+ 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)
+
+
[email protected](scope="module")
+def completed_run() -> _CompletedRun:
+ """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"]
+ 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):
+ 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):
+ """``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"
+ 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):
+ 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):
+ 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 f63a118b2ba..5e3c231b5fe 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/doc/images/output_testing_airflow-e2e-tests.svg
b/dev/breeze/doc/images/output_testing_airflow-e2e-tests.svg
index 7f2f3b4a26a..942306b371c 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><text class="breeze-testing-airflow-e2e-tests-r5" x="0" y="312.8"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-12)">│</text><text
class="breeze-testing-airflow-e2e-tests-r7" x="463.6" y="312.8"
textLength="73.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-12)">(TEXT)</text><text
class="breeze-testing-airflow-e2e-tests-r5" x="1451.8" y="312.8"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-12)">│</text><text
class="breez [...]
</text><text class="breeze-testing-airflow-e2e-tests-r5" x="0" y="337.2"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-13)">│</text><text
class="breeze-testing-airflow-e2e-tests-r4" x="24.4" y="337.2" textLength="366"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-13)">--e2e-test-mode               </text><text
class="breeze-testing-airflow-e2e-tests-r1" x="463.6" y="337.2" textLen [...]
</text><text class="breeze-testing-airflow-e2e-tests-r5" x="0" y="361.6"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-14)">│</text><text
class="breeze-testing-airflow-e2e-tests-r7" x="463.6" y="361.6"
textLength="976"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-14)">(basic|remote_log|remote_log_elasticsearch|remote_log_opensearch|xcom_object_sto</text><text
class="breeze-testing-airflow-e2e-tests-r5" x="1451.8" y="361.6"
textLength="12.2" clip-path="u [...]
-</text><text class="breeze-testing-airflow-e2e-tests-r5" x="0" y="386"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-15)">│</text><text
class="breeze-testing-airflow-e2e-tests-r7" x="463.6" y="386"
textLength="561.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-15)">rage|event_driven|java_sdk|go_sdk|openlineage)</text><text
class="breeze-testing-airflow-e2e-tests-r5" x="1451.8" y="386"
textLength="12.2" clip-path="url(#breeze-testing-airflow-e2e-tests-l [...]
+</text><text class="breeze-testing-airflow-e2e-tests-r5" x="0" y="386"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-15)">│</text><text
class="breeze-testing-airflow-e2e-tests-r7" x="463.6" y="386"
textLength="646.6"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-15)">rage|event_driven|java_sdk|go_sdk|openlineage|ts_sdk)</text><text
class="breeze-testing-airflow-e2e-tests-r5" x="1451.8" y="386"
textLength="12.2" clip-path="url(#breeze-testing-airflow-e2e- [...]
</text><text class="breeze-testing-airflow-e2e-tests-r5" x="0" y="410.4"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-16)">│</text><text
class="breeze-testing-airflow-e2e-tests-r4" x="24.4" y="410.4" textLength="366"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-16)">--airflow-version             </text><text
class="breeze-testing-airflow-e2e-tests-r1" x="463.6" y="410.4"
textLength="976" [...]
</text><text class="breeze-testing-airflow-e2e-tests-r5" x="0" y="434.8"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-17)">│</text><text
class="breeze-testing-airflow-e2e-tests-r1" x="463.6" y="434.8"
textLength="976"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-17)">current providers from main, instead of the default PROD image. Builds a        </text><text
[...]
</text><text class="breeze-testing-airflow-e2e-tests-r5" x="0" y="459.2"
textLength="12.2"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-18)">│</text><text
class="breeze-testing-airflow-e2e-tests-r1" x="463.6" y="459.2"
textLength="463.6"
clip-path="url(#breeze-testing-airflow-e2e-tests-line-18)">lightweight image from apache/airflow:</text><text
class="breeze-testing-airflow-e2e-tests-r8" x="927.2" y="459.2"
textLength="109.8" clip-path="url(#breeze-testing-airflo [...]
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 123969fa337..31336273b3b 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
diff --git a/dev/breeze/src/airflow_breeze/commands/testing_commands.py
b/dev/breeze/src/airflow_breeze/commands/testing_commands.py
index fd3f0444bd8..7731c25a354 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 @@ option_e2e_test_mode = click.option(
"java_sdk",
"go_sdk",
"openlineage",
+ "ts_sdk",
],
case_sensitive=False,
),
diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py
b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
index 589b286d75e..f03d468e7ca 100644
--- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py
+++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
@@ -136,6 +136,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()
@@ -272,6 +273,13 @@ CI_FILE_GROUP_MATCHES: HashableDict[FileGroupForCi] =
HashableDict(
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/(?!.*\.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$",
+ 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
@@ -1079,6 +1087,10 @@ class SelectiveChecks:
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 == "[]":
@@ -1229,6 +1241,7 @@ class SelectiveChecks:
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 7fe29df1bda..158bef04a31 100644
--- a/dev/breeze/tests/test_selective_checks.py
+++ b/dev/breeze/tests/test_selective_checks.py
@@ -811,6 +811,32 @@ 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(
+ ("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",),
diff --git a/ts-sdk/example/.gitignore b/ts-sdk/example/.gitignore
index 8ee10058ff8..1652dc9c671 100644
--- a/ts-sdk/example/.gitignore
+++ b/ts-sdk/example/.gitignore
@@ -1,3 +1,4 @@
.pnpm-store/
dist/
node_modules/
+pnpm-lock.yaml
