Skip to content

Commit b52f31f

Browse files
jason810496claude
andcommitted
Probe a packed TypeScript bundle for its task handlers in a gated test
The coordinator and SDK tests each fake one side, so nothing showed that a bundle packed with airflow-ts-pack answers the way the Dag processor parses it. The test packs ts-sdk/example from a copy of the SDK sources, finds it as a stub task of typescript_example would, and probes it. It needs Node.js 22 and pnpm, so it runs only with AIRFLOW_LANG_SDK_REAL_PROBE_TESTS=1, and it skips with the reason when node is missing, too old, or cannot report its version. Co-Authored-By: Claude <noreply@anthropic.com>
1 parent 24a3e8e commit b52f31f

1 file changed

Lines changed: 146 additions & 0 deletions

File tree

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
#
2+
# Licensed to the Apache Software Foundation (ASF) under one
3+
# or more contributor license agreements. See the NOTICE file
4+
# distributed with this work for additional information
5+
# regarding copyright ownership. The ASF licenses this file
6+
# to you under the Apache License, Version 2.0 (the
7+
# "License"); you may not use this file except in compliance
8+
# with the License. You may obtain a copy of the License at
9+
#
10+
# http://www.apache.org/licenses/LICENSE-2.0
11+
#
12+
# Unless required by applicable law or agreed to in writing,
13+
# software distributed under the License is distributed on an
14+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
# KIND, either express or implied. See the License for the
16+
# specific language governing permissions and limitations
17+
# under the License.
18+
"""Pack the TypeScript SDK example bundle and probe it for its task handlers."""
19+
20+
from __future__ import annotations
21+
22+
import json
23+
import os
24+
import shutil
25+
import subprocess
26+
from typing import TYPE_CHECKING
27+
from unittest import mock
28+
29+
import pytest
30+
import structlog
31+
32+
from airflow.dag_processing.processor import TaskHandlerDeclaration, TaskHandlerParsingResult
33+
from airflow.dag_processing.task_handler_processor import LangSDKTaskHandlerProcessorProcess
34+
from airflow.sdk.coordinators._subprocess import supports_task_handler_parsing
35+
from airflow.sdk.execution_time import supervisor
36+
from airflow.sdk.execution_time.coordinator import get_coordinator_manager, reset_coordinator_manager
37+
38+
from tests_common.test_utils.config import conf_vars
39+
from tests_common.test_utils.paths import AIRFLOW_ROOT_PATH
40+
41+
if TYPE_CHECKING:
42+
from collections.abc import Iterator
43+
from pathlib import Path
44+
45+
TS_SDK_PATH = AIRFLOW_ROOT_PATH / "ts-sdk"
46+
# The SDK's `engines` requirement.
47+
MIN_NODE_MAJOR = 22
48+
COORDINATORS = {
49+
"ts": {
50+
"classpath": "airflow.sdk.coordinators.node.NodeCoordinator",
51+
"kwargs": {"node_executable": shutil.which("node")},
52+
}
53+
}
54+
55+
pytestmark = pytest.mark.skipif(
56+
os.environ.get("AIRFLOW_LANG_SDK_REAL_PROBE_TESTS") != "1",
57+
reason="set AIRFLOW_LANG_SDK_REAL_PROBE_TESTS=1 to pack and probe a real TypeScript bundle",
58+
)
59+
60+
61+
def _get_toolchain_problem() -> str | None:
62+
"""Return why this machine cannot pack the example bundle, or ``None`` when it can."""
63+
if not TS_SDK_PATH.is_dir():
64+
return "the TypeScript SDK sources are absent"
65+
node = shutil.which("node")
66+
if node is None or shutil.which("pnpm") is None:
67+
return "needs node and pnpm"
68+
try:
69+
version = subprocess.run(
70+
[node, "--version"], capture_output=True, text=True, check=True
71+
).stdout.strip()
72+
major = int(version.lstrip("v").split(".")[0])
73+
except (OSError, subprocess.CalledProcessError, ValueError) as e:
74+
return f"cannot read the Node.js version: {e}"
75+
if major < MIN_NODE_MAJOR:
76+
return f"needs Node.js {MIN_NODE_MAJOR} or later, found {version}"
77+
return None
78+
79+
80+
@pytest.fixture(scope="module")
81+
def example_bundle(tmp_path_factory) -> Path:
82+
"""Pack ``ts-sdk/example`` from a copy of the SDK sources, so the checkout gets no build output."""
83+
if (problem := _get_toolchain_problem()) is not None:
84+
pytest.skip(problem)
85+
sdk = tmp_path_factory.mktemp("ts-sdk") / "ts-sdk"
86+
shutil.copytree(TS_SDK_PATH, sdk, ignore=shutil.ignore_patterns("node_modules", "dist", ".pnpm-store"))
87+
env = {**os.environ, "CI": "true", "COREPACK_ENABLE_DOWNLOAD_PROMPT": "0"}
88+
# The example links the SDK, so it is installed again once the SDK is built.
89+
for cwd, args in [
90+
(sdk, ["install", "--frozen-lockfile"]),
91+
(sdk, ["run", "build"]),
92+
(sdk / "example", ["install"]),
93+
(sdk / "example", ["run", "build"]),
94+
]:
95+
subprocess.run(["pnpm", *args], cwd=cwd, env=env, check=True)
96+
return sdk / "example" / "dist" / "bundle.min.mjs"
97+
98+
99+
@pytest.fixture
100+
def fresh_coordinator_manager() -> Iterator[None]:
101+
reset_coordinator_manager()
102+
yield
103+
reset_coordinator_manager()
104+
105+
106+
def _declare(task_id: str) -> TaskHandlerDeclaration:
107+
return TaskHandlerDeclaration(task_id=task_id, binding="named", params=None)
108+
109+
110+
@pytest.mark.usefixtures("fresh_coordinator_manager")
111+
@conf_vars({("sdk", "coordinators"): json.dumps(COORDINATORS)})
112+
@mock.patch.object(supervisor, "_should_use_exec", autospec=True, return_value=False)
113+
def test_probes_the_task_handlers_of_a_packed_typescript_bundle(mock_should_use_exec, example_bundle):
114+
coordinator = get_coordinator_manager().get_coordinator("ts")
115+
artifact = coordinator._find_task_handler_artifact(
116+
bundle_path=example_bundle.parent, dag_id="typescript_example"
117+
)
118+
assert supports_task_handler_parsing(artifact.schema_version)
119+
assert artifact.path == example_bundle.resolve()
120+
121+
result = LangSDKTaskHandlerProcessorProcess.run(
122+
coordinator="ts",
123+
path=artifact.path,
124+
bundle_path=example_bundle.parent,
125+
bundle_name="ts-task-handlers",
126+
artifact_rel_path=os.fspath(artifact.path.relative_to(example_bundle.parent.resolve())),
127+
logger=structlog.get_logger(),
128+
)
129+
130+
# Types are erased, so each handler lists no params and only its presence is checked.
131+
assert result == TaskHandlerParsingResult(
132+
fileloc=os.fspath(artifact.path),
133+
task_handlers={
134+
"typescript_example": [
135+
_declare("build_message"),
136+
_declare("read_connection"),
137+
_declare("write_and_delete_variable"),
138+
],
139+
"typescript_taskflow_example": [
140+
_declare("summarize"),
141+
_declare("report"),
142+
_declare("build_message"),
143+
],
144+
},
145+
)
146+
assert list(result.task_handlers) == ["typescript_example", "typescript_taskflow_example"]

0 commit comments

Comments
 (0)