This is an automated email from the ASF dual-hosted git repository.
shahar1 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 fccfbf8ad2b Skip airflowctl state store tests on servers older than
3.3 (#75753)
fccfbf8ad2b is described below
commit fccfbf8ad2b65388861013a4a58fa5bd76ecf544
Author: Andrew Chang <[email protected]>
AuthorDate: Wed Oct 28 06:24:08 2026 +0900
Skip airflowctl state store tests on servers older than 3.3 (#75753)
The asset and task instance state store APIs first shipped in
Airflow 3.3.0. Against an older server, every state store command
fails with 404 or 405. These failures are not compatibility bugs,
and they hide the real ones in a run against an older core.
---
.../tests/airflowctl_tests/conftest.py | 8 +++++
.../airflowctl_tests/test_airflowctl_commands.py | 39 +++++++++++++++++++---
2 files changed, 43 insertions(+), 4 deletions(-)
diff --git a/airflow-ctl-tests/tests/airflowctl_tests/conftest.py
b/airflow-ctl-tests/tests/airflowctl_tests/conftest.py
index 878c2495401..bc401523f6f 100644
--- a/airflow-ctl-tests/tests/airflowctl_tests/conftest.py
+++ b/airflow-ctl-tests/tests/airflowctl_tests/conftest.py
@@ -26,6 +26,7 @@ from subprocess import PIPE, STDOUT, Popen
import pytest
import requests
+from packaging.version import Version
from python_on_whales import DockerClient, docker
from airflowctl_tests import console
@@ -173,6 +174,13 @@ def api_token():
return _get_access_token()
[email protected](scope="session")
+def airflow_version() -> Version:
+ """Fixture that provides the Airflow version of the API server under
test."""
+ # A pre-release such as 3.4.0b2 already serves the 3.4.0 API, so compare
release numbers only.
+ return Version(Version(_request_api("version")["version"]).base_version)
+
+
@pytest.fixture
def run_command(wait_for_dag_run):
"""Fixture that provides a helper to run airflowctl commands."""
diff --git
a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
index 02f96d5c038..d3f3932c09a 100644
--- a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
+++ b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
@@ -17,6 +17,7 @@
from __future__ import annotations
import pytest
+from packaging.version import Version
def date_param():
@@ -159,6 +160,16 @@ TEST_COMMANDS = [
"plugins list-import-errors",
]
+# The first Airflow version whose API serves each command, keyed by
`_get_command_id`.
+MIN_AIRFLOW_VERSIONS = {
+ "assets set-state-store": Version("3.3.0"),
+ "assets get-state-store": Version("3.3.0"),
+ "assets list-state-store": Version("3.3.0"),
+ "assets delete-state-store": Version("3.3.0"),
+ "assets clear-state-store": Version("3.3.0"),
+ "taskinstances state-store": Version("3.3.0"),
+}
+
NO_AUTH_TEST_COMMANDS = [
"version --remote",
]
@@ -177,6 +188,22 @@ TEST_COMMANDS_SKIP_KEYRING = [LOGIN_COMMAND_SKIP_KEYRING]
+ [
]
+def _get_command_id(command: str) -> str:
+ return " ".join(command.split(" ", 2)[:2])
+
+
+def _skip_if_unsupported(command: str, airflow_version: Version) -> None:
+ min_version = MIN_AIRFLOW_VERSIONS.get(_get_command_id(command))
+ if min_version and airflow_version < min_version:
+ pytest.skip(f"needs Airflow >= {min_version}, the server runs
{airflow_version}")
+
+
+def test_min_airflow_versions_match_test_commands():
+ """A renamed command must be renamed here too, or its version gate
silently stops applying."""
+ stale_command_ids = set(MIN_AIRFLOW_VERSIONS) - {_get_command_id(command)
for command in TEST_COMMANDS}
+ assert not stale_command_ids
+
+
def test_hardcoded_xcom_key_would_collide():
"""Regression: a hardcoded xcom key produces identical 'xcom add' commands
across parametrize sets, causing 'already exists' errors when both run
@@ -201,20 +228,24 @@ def test_hardcoded_xcom_key_would_collide():
@pytest.mark.parametrize(
"command",
TEST_COMMANDS_DEBUG_MODE,
- ids=[" ".join(command.split(" ", 2)[:2]) for command in
TEST_COMMANDS_DEBUG_MODE],
+ ids=[_get_command_id(command) for command in TEST_COMMANDS_DEBUG_MODE],
)
-def test_airflowctl_commands(command: str, run_command):
+def test_airflowctl_commands(command: str, airflow_version: Version,
run_command):
"""Test airflowctl commands using docker-compose environment."""
+ _skip_if_unsupported(command, airflow_version)
run_command(command=command, env_vars={"AIRFLOW_CLI_DEBUG_MODE": "true"},
skip_login=True)
@pytest.mark.parametrize(
"command",
TEST_COMMANDS_SKIP_KEYRING,
- ids=[" ".join(command.split(" ", 2)[:2]) for command in
TEST_COMMANDS_SKIP_KEYRING],
+ ids=[_get_command_id(command) for command in TEST_COMMANDS_SKIP_KEYRING],
)
-def test_airflowctl_commands_skip_keyring(command: str, api_token: str,
run_command):
+def test_airflowctl_commands_skip_keyring(
+ command: str, api_token: str, airflow_version: Version, run_command
+):
"""Test airflowctl commands using docker-compose environment without using
keyring."""
+ _skip_if_unsupported(command, airflow_version)
run_command(
command=command,
env_vars={
