This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch airflow-ctl/v0-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/airflow-ctl/v0-1-test by this
push:
new 3c33fbc2404 [airflow-ctl/v0-1-test] Skip airflowctl state store tests
on servers older than 3.3 (#75753) (#76122)
3c33fbc2404 is described below
commit 3c33fbc2404903f0c17854d580632a75a573d30c
Author: Andrew Chang <[email protected]>
AuthorDate: Sat Oct 31 17:07:07 2026 +0900
[airflow-ctl/v0-1-test] Skip airflowctl state store tests on servers older
than 3.3 (#75753) (#76122)
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.
On this branch, #70724 skipped the asset state store commands
unconditionally. This replaces that skip with the version gate, so
the commands run again against a 3.3 or newer server. The
taskinstances state-store entry is dropped because this branch has
no such commands.
(cherry picked from commit fccfbf8ad2b65388861013a4a58fa5bd76ecf544)
---
.../tests/airflowctl_tests/conftest.py | 8 +++
.../airflowctl_tests/test_airflowctl_commands.py | 71 +++++++++++++---------
2 files changed, 50 insertions(+), 29 deletions(-)
diff --git a/airflow-ctl-tests/tests/airflowctl_tests/conftest.py
b/airflow-ctl-tests/tests/airflowctl_tests/conftest.py
index c232b8b22b0..13fdee8d97e 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 39ed27cb25d..aa3da0d8140 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():
@@ -56,6 +57,12 @@ TEST_COMMANDS = [
"assets list",
"assets get 1",
"assets create-event --asset-id=1",
+ "assets set-state-store 1 test_key test_value",
+ "assets get-state-store 1 test_key",
+ "assets list-state-store 1",
+ "assets delete-state-store 1 test_key",
+ "assets set-state-store 1 clear_key value",
+ "assets clear-state-store 1",
# Backfill commands
"backfill list example_bash_operator",
# Config commands
@@ -140,19 +147,14 @@ TEST_COMMANDS = [
"plugins list-import-errors",
]
-# This release branch targets core 3.2.x, which has no asset state store API,
so these commands stay
-# skipped here. Integration coverage for them lives on `main`.
-ASSET_STATE_STORE_TEST_COMMANDS = [
- "assets set-state-store 1 test_key test_value",
- "assets get-state-store 1 test_key",
- "assets list-state-store 1",
- "assets delete-state-store 1 test_key",
- "assets set-state-store 1 clear_key value",
- "assets clear-state-store 1",
-]
-_ASSET_STATE_STORE_SKIP = pytest.mark.skip(
- reason="core 3.2.x has no asset state store API; covered on main, skipped
for this release branch"
-)
+# 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"),
+}
NO_AUTH_TEST_COMMANDS = [
"version --remote",
@@ -164,23 +166,30 @@ DATE_PARAM_2 = date_param()
# Unique xcom key per test run to avoid "already exists" errors from leftover
state
_XCOM_KEY_1 = f"test_xcom_key_{DATE_PARAM_1.replace(':', '').replace('+',
'').replace('-', '')[:16]}"
_XCOM_KEY_2 = f"test_xcom_key_{DATE_PARAM_2.replace(':', '').replace('+',
'').replace('-', '')[:16]}"
-TEST_COMMANDS_DEBUG_MODE = (
- [LOGIN_COMMAND]
- + [test.format(date_param=DATE_PARAM_1, xcom_key=_XCOM_KEY_1) for test in
TEST_COMMANDS]
- + [pytest.param(test, marks=_ASSET_STATE_STORE_SKIP) for test in
ASSET_STATE_STORE_TEST_COMMANDS]
-)
-TEST_COMMANDS_SKIP_KEYRING = (
- [LOGIN_COMMAND_SKIP_KEYRING]
- + [test.format(date_param=DATE_PARAM_2, xcom_key=_XCOM_KEY_2) for test in
TEST_COMMANDS]
- + [pytest.param(test, marks=_ASSET_STATE_STORE_SKIP) for test in
ASSET_STATE_STORE_TEST_COMMANDS]
-)
+TEST_COMMANDS_DEBUG_MODE = [LOGIN_COMMAND] + [
+ test.format(date_param=DATE_PARAM_1, xcom_key=_XCOM_KEY_1) for test in
TEST_COMMANDS
+]
+TEST_COMMANDS_SKIP_KEYRING = [LOGIN_COMMAND_SKIP_KEYRING] + [
+ test.format(date_param=DATE_PARAM_2, xcom_key=_XCOM_KEY_2) for test in
TEST_COMMANDS
+]
-def _command_id(command) -> str:
- command = command.values[0] if hasattr(command, "values") else command
+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
@@ -205,20 +214,24 @@ def test_hardcoded_xcom_key_would_collide():
@pytest.mark.parametrize(
"command",
TEST_COMMANDS_DEBUG_MODE,
- ids=[_command_id(command) 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=[_command_id(command) 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={
