Check Variable write conflicts against the team being written (#72042)

Author: ferruzzi

airflow-core/newsfragments/72042.bugfix.rst

@@ -0,0 +1 @@
+The write-conflict check performed when setting or updating a team-scoped Variable now resolves against that team rather than the global scope, so a secrets backend shadowing the key within the team is detected and a global-only definition no longer warns about a conflict that would not shadow the read.

airflow-core/src/airflow/models/variable.py

@@ -258,7 +258,7 @@ def set(
         # check if the secret exists in the custom secrets' backend.
         from airflow.sdk import SecretCache
 
-        Variable.check_for_write_conflict(key=key)
+        Variable.check_for_write_conflict(key=key, team_name=team_name)
         if serialize_json:
             stored_value = json.dumps(value, indent=2)
         else:
@@ -342,7 +342,7 @@ def update(
                 "Multi-team mode is not configured in the Airflow environment. To assign a team to a variable, multi-mode must be enabled."
             )
 
-        Variable.check_for_write_conflict(key=key)
+        Variable.check_for_write_conflict(key=key, team_name=team_name)
 
         if Variable.get_variable_from_secrets(key=key, team_name=team_name) is None:
             raise KeyError(f"Variable {key} does not exist")
@@ -430,7 +430,7 @@ def rotate_fernet_key(self):
             self._val = fernet.rotate(self._val.encode("utf-8")).decode()
 
     @staticmethod
-    def check_for_write_conflict(key: str) -> None:
+    def check_for_write_conflict(key: str, team_name: str | None = None) -> None:
         """
         Log a warning if a variable exists outside the metastore.
 
@@ -439,11 +439,15 @@ def check_for_write_conflict(key: str) -> None:
         subsequent reads will not read the set value.
 
         :param key: Variable Key
+        :param team_name: Team name the variable is being written for, so the check resolves
+            against the same scope the write will use
         """
         for secrets_backend in ensure_secrets_loaded():
             if not isinstance(secrets_backend, MetastoreBackend):
                 try:
-                    var_val = secrets_backend.get_variable(key=key)
+                    var_val = call_secrets_backend_method(
+                        secrets_backend.get_variable, team_name=team_name, key=key
+                    )
                     if var_val is not None:
                         _backend_name = type(secrets_backend).__name__
                         log.warning(

airflow-core/tests/unit/models/test_variable.py

@@ -28,6 +28,7 @@
 
 from airflow.models import Variable, crypto, variable
 from airflow.sdk import SecretCache
+from airflow.secrets import BaseSecretsBackend
 from airflow.secrets.metastore import MetastoreBackend
 
 from tests_common.test_utils import db
@@ -41,6 +42,30 @@
 pytestmark = pytest.mark.db_test
 
 
+class _TeamUnawareVariableBackend(BaseSecretsBackend):
+    """A custom backend whose ``get_variable`` override predates the ``team_name`` keyword."""
+
+    def __init__(self):
+        self.was_called = False
+
+    # The signature mismatch with the base class is the point of this fixture, so mypy's
+    # override check has to be waived here rather than fixed.
+    def get_variable(self, key: str) -> str | None:  # type: ignore[override]
+        self.was_called = True
+        return "secret_val"
+
+
+class _TeamAwareVariableBackend(BaseSecretsBackend):
+    """A custom backend whose ``get_variable`` override accepts ``team_name``."""
+
+    def __init__(self):
+        self.received_team_name: str | None = None
+
+    def get_variable(self, key: str, team_name: str | None = None) -> str | None:
+        self.received_team_name = team_name
+        return "secret_val"
+
+
 class TestVariable:
     @pytest.fixture(autouse=True)
     def setup_test_cases(self):
@@ -192,6 +217,43 @@ def test_variable_set_with_extra_secret_backend(self, mock_ensure_secrets, caplo
         )
         Variable.delete(key="key", session=session)
 
+    @mock.patch("airflow.models.variable.ensure_secrets_loaded")
+    def test_write_conflict_check_forwards_team_name(self, mock_ensure_secrets):
+        """The check must resolve against the scope being written, not the global one."""
+        backend = _TeamAwareVariableBackend()
+        mock_ensure_secrets.return_value = [backend, MetastoreBackend()]
+
+        Variable.check_for_write_conflict(key="key", team_name="team_a")
+
+        assert backend.received_team_name == "team_a"
+
+    @mock.patch("airflow.models.variable.ensure_secrets_loaded")
+    def test_write_conflict_check_tolerates_team_unaware_backend(self, mock_ensure_secrets):
+        """A backend whose override predates ``team_name`` must still be consulted, not error out."""
+        backend = _TeamUnawareVariableBackend()
+        mock_ensure_secrets.return_value = [backend, MetastoreBackend()]
+
+        Variable.check_for_write_conflict(key="key", team_name="team_a")
+
+        assert backend.was_called
+
+    @conf_vars({("core", "multi_team"): "True"})
+    @mock.patch.object(Variable, "check_for_write_conflict")
+    def test_set_forwards_team_name_to_write_conflict_check(self, mock_check, testing_team, session):
+        Variable.set(key="key", value="db-value", team_name=testing_team.name, session=session)
+
+        assert mock_check.call_args.kwargs["team_name"] == testing_team.name
+
+    @conf_vars({("core", "multi_team"): "True"})
+    @mock.patch.object(Variable, "check_for_write_conflict")
+    def test_update_forwards_team_name_to_write_conflict_check(self, mock_check, testing_team, session):
+        Variable.set(key="key", value="db-value", team_name=testing_team.name, session=session)
+        SecretCache.invalidate_variable("key")
+
+        Variable.update(key="key", value="new-value", team_name=testing_team.name, session=session)
+
+        assert mock_check.call_args.kwargs["team_name"] == testing_team.name
+
     def test_variable_set_get_round_trip_json(self):
         value = {"a": 17, "b": 47}
         Variable.set(key="tested_var_set_id", value=value, serialize_json=True)