Skip airflowctl state store tests on servers older than 3.3

airflow-ctl-tests/tests/airflowctl_tests/conftest.py

@@ -26,6 +26,7 @@
 
 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()
 
 
+@pytest.fixture(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."""

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 @@ def date_param():
     "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 @@ def date_param():
 ]
 
 
+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={