This is an automated email from the ASF dual-hosted git repository.
potiuk 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 e99d6c6f025 Require team names to be lower case with no doubled
underscores (#70886)
e99d6c6f025 is described below
commit e99d6c6f025eb5d01b5050432df7a9700d79223b
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Aug 17 22:54:57 2026 +0200
Require team names to be lower case with no doubled underscores (#70886)
* Require team names to be lower case with no doubled underscores
A team name is embedded in the AIRFLOW_CONN__<TEAM>___<ID> environment
variable
namespace that holds team scoped secrets, and the backend upper-cases the
name when
it builds that variable. Two names differing only in case therefore
resolved a single
namespace between them, and a name able to contain the ___ separator could
not be told
apart from it. Restricting what a name may be spelled with is what makes
the namespace
unambiguous, rather than another check at the point of use.
Stored names are migrated rather than left to rot: db migrate lower-cases
them and
carries each team's Dag bundles, Connections, Variables, Pools and Triggers
across.
Two stored names that fold together would silently merge two teams, and
nothing can
choose which survives, so that case is refused before the migration starts.
closes: #70886
* Address migration 0131 test columns through their mapped columns
Hand-written SQL spelled `variable.key` and `slot_pool.pool` unquoted, which
MySQL rejects as reserved words, so the test could not get far enough to
check
what the migration actually did.
* Let migration 0131 rename team names on MySQL
MySQL reads the two spellings as one value, which leaves the closing update
as
the statement that performs the rename -- and InnoDB refuses to update a
primary
key that rows still reference. Any deployment whose teams own a connection,
variable, pool, trigger or bundle could not run db migrate at all.
---
airflow-core/docs/migrations-ref.rst | 4 +-
airflow-core/newsfragments/70886.significant.rst | 19 +++
.../src/airflow/cli/commands/team_command.py | 41 ++++--
.../versions/0131_3_4_0_lower_case_team_names.py | 108 ++++++++++++++++
airflow-core/src/airflow/models/team.py | 21 ++++
airflow-core/src/airflow/utils/db.py | 34 ++++-
.../tests/unit/cli/commands/test_team_command.py | 89 +++++++++++++
.../migrations/test_0131_lower_case_team_names.py | 138 +++++++++++++++++++++
airflow-core/tests/unit/utils/test_db.py | 35 +++++-
9 files changed, 478 insertions(+), 11 deletions(-)
diff --git a/airflow-core/docs/migrations-ref.rst
b/airflow-core/docs/migrations-ref.rst
index b6f7fbcf879..cc18c48f493 100644
--- a/airflow-core/docs/migrations-ref.rst
+++ b/airflow-core/docs/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are
executed via when you ru
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description
|
+=========================+==================+===================+==============================================================+
-| ``76c46545c91e`` (head) | ``3c525f44bea8`` | ``3.4.0`` | Add new
index for trigger. |
+| ``c7f0a5d2e9b4`` (head) | ``76c46545c91e`` | ``3.4.0`` | Lower case
team names. |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``76c46545c91e`` | ``3c525f44bea8`` | ``3.4.0`` | Add new
index for trigger. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| ``3c525f44bea8`` | ``b2f1a9c7d4e0`` | ``3.4.0`` | Add indexes
on serialized_dag and dag_code. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
diff --git a/airflow-core/newsfragments/70886.significant.rst
b/airflow-core/newsfragments/70886.significant.rst
new file mode 100644
index 00000000000..d772b43d169
--- /dev/null
+++ b/airflow-core/newsfragments/70886.significant.rst
@@ -0,0 +1,19 @@
+Team names must now be lower case, and cannot contain two consecutive
underscores
+
+``airflow teams create`` and ``airflow teams sync`` now accept only names
matching
+``(?!.*__)[a-z0-9_-]{3,50}`` -- lower case letters, digits, hyphens and single
+underscores. Upper case letters and ``__`` were previously accepted.
+
+A team name is embedded in the ``AIRFLOW_CONN__<TEAM>___<ID>`` environment
variable
+namespace that holds team scoped secrets, and the backend upper-cases the name
when it
+builds that variable. ``data_eng`` and ``Data_Eng`` therefore resolved a single
+namespace between them, and a name containing ``___`` could not be told apart
from the
+separator.
+
+``airflow db migrate`` lower-cases every stored team name, moving that team's
Dag
+bundles, Connections, Variables, Pools and Triggers with it. Two stored names
that
+differ only in case would collapse onto one row and silently merge two teams,
so
+migration refuses to start while any such pair exists -- rename all but one
first. Names
+that stay invalid for another reason, such as containing ``__``, are migrated
as they
+are; they keep working, but ``airflow teams sync`` exits non-zero until they
are
+renamed.
diff --git a/airflow-core/src/airflow/cli/commands/team_command.py
b/airflow-core/src/airflow/cli/commands/team_command.py
index 855a42071de..3dc2b5aa748 100644
--- a/airflow-core/src/airflow/cli/commands/team_command.py
+++ b/airflow-core/src/airflow/cli/commands/team_command.py
@@ -30,7 +30,12 @@ from airflow.configuration import conf
from airflow.dag_processing.bundles.manager import DagBundlesManager
from airflow.models.connection import Connection
from airflow.models.pool import Pool
-from airflow.models.team import Team, dag_bundle_team_association_table
+from airflow.models.team import (
+ TEAM_NAME_PATTERN,
+ Team,
+ dag_bundle_team_association_table,
+ find_invalid_team_names,
+)
from airflow.models.variable import Variable
from airflow.utils import cli as cli_utils
from airflow.utils.providers_configuration_loader import
providers_configuration_loaded
@@ -54,12 +59,23 @@ def _show_teams(teams, output):
def _extract_team_name(args):
- """Extract and validate team name from args."""
+ """
+ Extract a team name from args, without holding it to the name rule.
+
+ ``teams delete`` has to accept any name a row can hold, including one
stored before the
+ current rule applied -- otherwise a team that violates it could never be
removed.
+ """
team_name = args.name.strip()
if not team_name:
raise SystemExit("Team name cannot be empty")
- if not re.match(r"^[a-zA-Z0-9_-]{3,50}$", team_name):
- raise SystemExit("Invalid team name: must match regex
^[a-zA-Z0-9_-]{3,50}$")
+ return team_name
+
+
+def _extract_new_team_name(args):
+ """Extract a team name for a creation path, holding it to the name rule."""
+ team_name = _extract_team_name(args)
+ if not re.fullmatch(TEAM_NAME_PATTERN, team_name):
+ raise SystemExit(f"Invalid team name: must match regex
{TEAM_NAME_PATTERN}")
return team_name
@@ -78,8 +94,8 @@ def _create_default_team_pool(team_name: str, *, session:
Session) -> None:
@providers_configuration_loaded
@provide_session
def team_create(args, *, session=NEW_SESSION):
- """Create a new team. Team names must be 3-50 characters long and contain
only alphanumeric characters, hyphens, and underscores."""
- team_name = _extract_team_name(args)
+ """Create a new team. Team names must be 3-50 characters long and contain
only lower case alphanumeric characters, hyphens, and single underscores."""
+ team_name = _extract_new_team_name(args)
# Check if team with this name already exists
if session.scalar(select(Team).where(Team.name == team_name)):
@@ -210,11 +226,22 @@ def team_sync(args, *, session=NEW_SESSION):
for bundle in DagBundlesManager()._bundle_config.values()
if bundle.team_name is not None
}
+ existing_teams = Team.get_all_team_names(session=session)
+
+ # The bundle config is a second creation path for teams, so it enforces
the same rule as
+ # `teams create`. Stored names are checked too: `teams sync` shipped
without any validation,
+ # so a deployment can hold a name the rule rejects, and checking only the
incoming config
+ # would miss a team created by an earlier sync and since dropped from it.
+ if invalid := find_invalid_team_names(dag_bundle_teams | existing_teams):
+ raise SystemExit(
+ f"Invalid team name(s): {', '.join(invalid)}. "
+ f"Team names must match regex {TEAM_NAME_PATTERN}. "
+ "Names already stored must be corrected before syncing."
+ )
teams_added = 0
try:
- existing_teams = Team.get_all_team_names(session=session)
for team_name in dag_bundle_teams:
if team_name not in existing_teams:
session.add(Team(name=team_name))
diff --git
a/airflow-core/src/airflow/migrations/versions/0131_3_4_0_lower_case_team_names.py
b/airflow-core/src/airflow/migrations/versions/0131_3_4_0_lower_case_team_names.py
new file mode 100644
index 00000000000..ba1253724c0
--- /dev/null
+++
b/airflow-core/src/airflow/migrations/versions/0131_3_4_0_lower_case_team_names.py
@@ -0,0 +1,108 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Lower case team names.
+
+Revision ID: c7f0a5d2e9b4
+Revises: 76c46545c91e
+Create Date: 2026-08-12 09:14:22.518233
+
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+
+if TYPE_CHECKING:
+ from sqlalchemy.sql import ClauseElement
+
+# revision identifiers, used by Alembic.
+revision = "c7f0a5d2e9b4"
+down_revision = "76c46545c91e"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+# Tables carrying a foreign key to ``team.name``.
+_REFERRING_TABLES = ("dag_bundle_team", "connection", "variable", "slot_pool",
"trigger")
+
+_TEAM = sa.table("team", sa.column("name", sa.String))
+
+
+def build_lower_casing_statements(dialect_name: str) -> list[ClauseElement]:
+ """
+ Build the statements that fold every stored team name to lower case.
+
+ Kept free of anything read back from a row so that ``db migrate
--show-sql-only`` can emit
+ it: an offline migration has no rows to read.
+
+ Their order carries both collations a database may have, without having to
tell them apart:
+
+ * Case sensitive, where ``name <> lower(name)`` selects the names to fold.
``team.name`` is
+ the primary key and the foreign keys do not cascade on update, so the
lower case row has to
+ exist before anything can point at it, and the old one can only go once
nothing does. Two
+ stored names folding onto one collide on the primary key at the insert,
deliberately --
+ merging two teams is not a migration's decision to make, and ``db
migrate`` refuses earlier
+ with an explanation.
+ * Case insensitive, where the two spellings are a single value: the insert
and the delete
+ match nothing, the referring rows keep matching their team while they
are rewritten, and
+ the closing unconditional update renames the row in place.
+
+ MySQL is the second case, which makes that closing update the statement
that does the rename
+ -- and InnoDB refuses to update a primary key while rows still reference
it, the foreign keys
+ carrying no ``ON UPDATE CASCADE``. The referring rows were folded above
and already read as
+ the new name under this collation, so the check is suspended over the
rename it would other-
+ wise block. A name that would collide still collides: the primary key is
not what is lifted.
+ """
+ lowered = sa.func.lower(_TEAM.c.name)
+ statements: list[ClauseElement] = [
+ _TEAM.insert().from_select(["name"],
sa.select(lowered).where(_TEAM.c.name != lowered))
+ ]
+ for table_name in _REFERRING_TABLES:
+ referring = sa.table(table_name, sa.column("team_name", sa.String))
+ # ``connection`` and ``trigger`` carry the rows of team-less
deployments as well; only
+ # the team scoped ones are rewritten.
+ statements.append(
+ referring.update()
+ .where(referring.c.team_name.is_not(None))
+ .values(team_name=sa.func.lower(referring.c.team_name))
+ )
+ statements.append(_TEAM.delete().where(_TEAM.c.name != lowered))
+ if dialect_name == "mysql":
+ statements.append(sa.text("SET FOREIGN_KEY_CHECKS=0"))
+ statements.append(_TEAM.update().values(name=lowered))
+ statements.append(sa.text("SET FOREIGN_KEY_CHECKS=1"))
+ else:
+ statements.append(_TEAM.update().values(name=lowered))
+ return statements
+
+
+def upgrade():
+ """Lower case team names."""
+ for statement in build_lower_casing_statements(op.get_bind().dialect.name):
+ op.execute(statement)
+
+
+def downgrade():
+ """Unapply Lower case team names."""
+ # The original casing is recorded nowhere, so it cannot be restored. Lower
case names are
+ # valid under every rule that preceded this one, so leaving them as they
are is safe.
diff --git a/airflow-core/src/airflow/models/team.py
b/airflow-core/src/airflow/models/team.py
index 64877489c66..fab0e272409 100644
--- a/airflow-core/src/airflow/models/team.py
+++ b/airflow-core/src/airflow/models/team.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import re
from typing import TYPE_CHECKING
from sqlalchemy import Column, ForeignKey, Index, String, Table, select
@@ -26,8 +27,28 @@ from airflow.models.base import Base, StringID
from airflow.utils.session import NEW_SESSION, provide_session
if TYPE_CHECKING:
+ from collections.abc import Iterable
+
from sqlalchemy.orm import Session
+# What ``airflow teams create`` and ``airflow teams sync`` accept as a team
name.
+#
+# Lower case only: the environment secrets backend upper-cases the team name
to build
+# ``AIRFLOW_CONN__<TEAM>___<ID>``, so ``data_eng`` and ``Data_Eng`` resolve
one namespace between
+# them and each team would read the other's Connections and Variables. Two
consecutive
+# underscores are excluded so that a name can never contain the ``___``
separator itself.
+#
+# Deliberately unanchored and always used with ``re.fullmatch``: ``re.match``
against a
+# ``$``-anchored pattern also accepts a trailing newline, and unlike ``teams
create`` the
+# ``teams sync`` path does not strip what the bundle config gives it.
+TEAM_NAME_PATTERN = r"(?!.*__)[a-z0-9_-]{3,50}"
+
+
+def find_invalid_team_names(names: Iterable[str]) -> list[str]:
+ """Return the names that do not satisfy :data:`TEAM_NAME_PATTERN`,
sorted."""
+ return sorted({name for name in names if not
re.fullmatch(TEAM_NAME_PATTERN, name)})
+
+
dag_bundle_team_association_table = Table(
"dag_bundle_team",
Base.metadata,
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index 4b873b4214e..19f96a2ea6e 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -117,7 +117,7 @@ _REVISION_HEADS_MAP: dict[str, str] = {
"3.1.8": "509b94a1042d",
"3.2.0": "1d6611b6ab7c",
"3.3.0": "d2f4e1b3c5a7",
- "3.4.0": "76c46545c91e",
+ "3.4.0": "c7f0a5d2e9b4",
}
# Prefix used to identify tables holding data moved during migration.
@@ -1117,10 +1117,40 @@ def reflect_tables(tables: list[MappedClassProtocol |
str] | None, session, sche
return metadata
+def check_team_names_can_be_lower_cased(*, session: Session) -> Iterable[str]:
+ """
+ Yield an error for each group of stored team names that cannot be folded
to lower case.
+
+ Team names became lower case only, and a migration rewrites the stored
ones. Names that
+ differ only in case collapse onto a single row, which would silently merge
two teams -- one
+ would inherit the other's Dag bundles, Connections, Variables, Pools and
Triggers. Nothing
+ here can choose which name survives, so migration stops and the operator
renames one first.
+ """
+ from collections import defaultdict
+
+ from airflow.models.team import Team
+
+ if not inspect(session.get_bind()).has_table(Team.__tablename__):
+ return
+
+ by_lower: dict[str, list[str]] = defaultdict(list)
+ for name in Team.get_all_team_names(session=session):
+ by_lower[name.lower()].append(name)
+
+ for lowered, names in sorted(by_lower.items()):
+ if len(names) > 1:
+ yield (
+ f"Teams {', '.join(repr(name) for name in sorted(names))}
differ only in case and "
+ f"would both become '{lowered}'. Rename all but one with
`airflow teams create` / "
+ "`airflow teams delete`, moving its Dag bundles, Connections,
Variables and Pools "
+ "across, before migrating."
+ )
+
+
@provide_session
def _check_migration_errors(*, session: Session = NEW_SESSION) ->
Iterable[str]:
""":session: session of the sqlalchemy."""
- check_functions: Iterable[Callable[..., Iterable[str]]] = ()
+ check_functions: Iterable[Callable[..., Iterable[str]]] =
(check_team_names_can_be_lower_cased,)
for check_fn in check_functions:
log.debug("running check function %s", check_fn.__name__)
yield from check_fn(session=session)
diff --git a/airflow-core/tests/unit/cli/commands/test_team_command.py
b/airflow-core/tests/unit/cli/commands/test_team_command.py
index 3af2e4ee5b8..65c1b3e1d13 100644
--- a/airflow-core/tests/unit/cli/commands/test_team_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_team_command.py
@@ -107,6 +107,34 @@ class TestCliTeams:
with pytest.raises(SystemExit, match="Invalid team name"):
team_command.team_create(self.parser.parse_args(["teams",
"create", "test with space"]))
+ @pytest.mark.parametrize(
+ "bad_name",
+ [
+ pytest.param("Data-Eng", id="upper-case"),
+ pytest.param("team__x", id="consecutive-underscores"),
+ pytest.param("team_a___prod", id="spans-the-separator"),
+ ],
+ )
+ def
test_team_create_rejects_a_name_the_secrets_namespace_cannot_hold(self,
bad_name):
+ with pytest.raises(SystemExit, match="Invalid team name"):
+ team_command.team_create(self.parser.parse_args(["teams",
"create", bad_name]))
+
+ assert self.session.scalar(select(Team).where(Team.name == bad_name))
is None
+
+ def test_team_create_allows_a_single_underscore(self):
+ team_command.team_create(self.parser.parse_args(["teams", "create",
"team_a"]))
+
+ assert self.session.scalar(select(Team).where(Team.name == "team_a"))
is not None
+
+ def test_team_delete_accepts_a_name_the_current_rule_rejects(self):
+ """A team stored before the rule tightened still has to be
removable."""
+ self.session.add(Team(name="Data-Eng"))
+ self.session.commit()
+
+ team_command.team_delete(self.parser.parse_args(["teams", "delete",
"Data-Eng", "--yes"]))
+
+ assert self.session.scalar(select(Team).where(Team.name ==
"Data-Eng")) is None
+
def test_team_create_whitespace_name(self):
"""Test team creation with whitespace-only name."""
with pytest.raises(SystemExit, match="Team name cannot be empty"):
@@ -393,6 +421,67 @@ class TestCliTeams:
assert "integration-2" not in team_names
assert "integration-3" in team_names
+ @pytest.mark.parametrize(
+ "bad_name",
+ [
+ pytest.param("a", id="too-short"),
+ pytest.param("Team1", id="upper-case"),
+ pytest.param("team__x", id="consecutive-underscores"),
+ pytest.param("team1\n", id="trailing-newline"),
+ ],
+ )
+ def test_team_sync_rejects_an_invalid_bundle_team_name(self, bad_name):
+ """``teams sync`` is a second creation path, so it holds names to the
same rule.
+
+ Unlike ``teams create`` this path does not strip its input, so a
trailing newline
+ reaches the pattern -- which is why the check is ``re.fullmatch``
against an unanchored
+ pattern rather than ``re.match`` against a ``$``-anchored one.
+ """
+ bundle_config = [
+ {
+ "name": "bundleone",
+ "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle",
+ "kwargs": {"path": "/dev/null", "refresh_interval": 0},
+ "team_name": bad_name,
+ },
+ ]
+
+ with conf_vars(
+ {
+ ("core", "multi_team"): "True",
+ ("dag_processor", "dag_bundle_config_list"):
json.dumps(bundle_config),
+ }
+ ):
+ with pytest.raises(SystemExit, match="Invalid team name"):
+ team_command.team_sync(self.parser.parse_args(["teams",
"sync"]))
+
+ assert self.session.scalars(select(Team)).all() == []
+
+ def test_team_sync_rejects_an_invalid_name_already_stored(self):
+ """``teams sync`` shipped with no validation, so stored names can
already break the rule."""
+ self.session.add(Team(name="Data-Eng"))
+ self.session.commit()
+
+ bundle_config = [
+ {
+ "name": "bundleone",
+ "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle",
+ "kwargs": {"path": "/dev/null", "refresh_interval": 0},
+ "team_name": "team1",
+ },
+ ]
+
+ with conf_vars(
+ {
+ ("core", "multi_team"): "True",
+ ("dag_processor", "dag_bundle_config_list"):
json.dumps(bundle_config),
+ }
+ ):
+ with pytest.raises(SystemExit, match="Data-Eng"):
+ team_command.team_sync(self.parser.parse_args(["teams",
"sync"]))
+
+ assert self.session.scalar(select(Team).where(Team.name == "team1"))
is None
+
def test_team_sync(self):
bundle_config = [
{
diff --git
a/airflow-core/tests/unit/migrations/test_0131_lower_case_team_names.py
b/airflow-core/tests/unit/migrations/test_0131_lower_case_team_names.py
new file mode 100644
index 00000000000..15008679853
--- /dev/null
+++ b/airflow-core/tests/unit/migrations/test_0131_lower_case_team_names.py
@@ -0,0 +1,138 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Tests for migration 0131 (c7f0a5d2e9b4), which lower-cases stored team names.
+
+``team.name`` is a primary key whose foreign keys do not cascade on update, so
the rename has
+to carry every referring row with it. A row left behind would either break the
constraint or
+strand that team's Connections, Variables and Pools under a name nothing
resolves.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+import sqlalchemy as sa
+from alembic import command
+
+from airflow import settings
+from airflow.models import Connection, Pool, Variable
+from airflow.models.team import Team
+from airflow.utils.db import _get_alembic_config
+from airflow.utils.session import create_session
+
+from tests_common.test_utils.paths import AIRFLOW_CORE_SOURCES_PATH
+
+pytestmark = pytest.mark.db_test
+
+# Migration filenames start with a digit so they cannot be imported via the
normal import
+# system; load the module by file path instead.
+_MIGRATION_PATH = (
+ Path(AIRFLOW_CORE_SOURCES_PATH) /
"airflow/migrations/versions/0131_3_4_0_lower_case_team_names.py"
+)
+_spec = importlib.util.spec_from_file_location("migration_0131",
_MIGRATION_PATH)
+_migration = importlib.util.module_from_spec(_spec) # type: ignore[arg-type]
+_spec.loader.exec_module(_migration) # type: ignore[union-attr]
+
+_OLD = "Data-Eng"
+_NEW = "data-eng"
+
+
+# Addressed through the mapped columns rather than by name: ``variable.key``
and
+# ``slot_pool.pool`` are reserved words on MySQL, which only quotes them
correctly when
+# SQLAlchemy renders the identifier for the dialect.
+_REFERRING_ROWS = (
+ (Connection, Connection.conn_id, "c_0131"),
+ (Variable, Variable.key, "v_0131"),
+ (Pool, Pool.pool, "p_0131"),
+)
+
+
+def _team_names(conn) -> set[str]:
+ return set(conn.scalars(sa.select(Team.name)))
+
+
+class TestMigration0131:
+ def test_rename_carries_referring_rows_with_the_team(self):
+ with create_session() as session:
+ session.add(Team(name=_OLD))
+ session.flush()
+ session.add_all(
+ [
+ Connection(conn_id="c_0131", conn_type="http",
team_name=_OLD),
+ Variable(key="v_0131", val="x", team_name=_OLD),
+ Pool(pool="p_0131", slots=1, description="",
include_deferred=False, team_name=_OLD),
+ ]
+ )
+ session.commit()
+
+ try:
+ with settings.engine.begin() as conn:
+ for statement in
_migration.build_lower_casing_statements(conn.dialect.name):
+ conn.execute(statement)
+
+ with settings.engine.connect() as conn:
+ assert _OLD not in _team_names(conn)
+ assert _NEW in _team_names(conn)
+ for model, key_column, key in _REFERRING_ROWS:
+ team_name =
conn.scalar(sa.select(model.team_name).where(key_column == key))
+ assert team_name == _NEW, model.__tablename__
+ finally:
+ with settings.engine.begin() as conn:
+ for model, key_column, key in _REFERRING_ROWS:
+ conn.execute(sa.delete(model).where(key_column == key))
+ conn.execute(sa.delete(Team).where(Team.name.in_([_OLD,
_NEW])))
+
+ def test_upgrade_emits_sql_offline(self, capsys):
+ """``db migrate --show-sql-only`` has no rows to read, so the
migration must not read any."""
+ command.upgrade(_get_alembic_config(),
f"{_migration.down_revision}:{_migration.revision}", sql=True)
+
+ emitted = capsys.readouterr().out
+ assert "INSERT INTO team (name) SELECT lower(team.name)" in emitted
+ assert "UPDATE team SET name=lower(team.name)" in emitted
+
+ @pytest.mark.parametrize(
+ ("dialect_name", "closing"),
+ [
+ pytest.param(
+ "mysql",
+ [
+ "SET FOREIGN_KEY_CHECKS=0",
+ "UPDATE team SET name=lower(team.name)",
+ "SET FOREIGN_KEY_CHECKS=1",
+ ],
+ id="mysql-suspends-the-check-over-the-rename",
+ ),
+ pytest.param(
+ "postgresql",
+ ["UPDATE team SET name=lower(team.name)"],
+ id="others-rename-under-the-check",
+ ),
+ ],
+ )
+ def test_only_mysql_suspends_foreign_key_checks(self, dialect_name,
closing):
+ """InnoDB refuses the closing rename while the foreign keys still
guard the old name."""
+ rendered = [
+ str(statement).strip() for statement in
_migration.build_lower_casing_statements(dialect_name)
+ ]
+
+ assert rendered[-len(closing) :] == closing
+ assert "SET FOREIGN_KEY_CHECKS=0" not in rendered[: -len(closing)]
diff --git a/airflow-core/tests/unit/utils/test_db.py
b/airflow-core/tests/unit/utils/test_db.py
index c81859edbdc..a33f7b667f4 100644
--- a/airflow-core/tests/unit/utils/test_db.py
+++ b/airflow-core/tests/unit/utils/test_db.py
@@ -31,16 +31,18 @@ from alembic.config import Config
from alembic.migration import MigrationContext
from alembic.runtime.environment import EnvironmentContext
from alembic.script import ScriptDirectory
-from sqlalchemy import Column, Integer, MetaData, Table, select
+from sqlalchemy import Column, Integer, MetaData, Table, delete, select
from airflow import settings
from airflow.models import Base as airflow_base
+from airflow.models.team import Team
from airflow.utils.db import (
AutocommitEngineForMySQL,
LazySelectSequence,
_get_alembic_config,
_get_current_revision,
check_migrations,
+ check_team_names_can_be_lower_cased,
compare_server_default,
compare_type,
create_default_connections,
@@ -50,6 +52,7 @@ from airflow.utils.db import (
upgradedb,
)
from airflow.utils.db_manager import RunDBManager
+from airflow.utils.session import create_session
from tests_common.test_utils.config import conf_vars
@@ -237,6 +240,36 @@ class TestDb:
check_migrations(0)
check_migrations(1)
+ @pytest.mark.usefixtures("initialized_db")
+ def
test_check_team_names_can_be_lower_cased_passes_when_no_names_collide(self):
+ with create_session() as session:
+ session.add_all([Team(name="data-eng"), Team(name="Platform")])
+ session.commit()
+ try:
+ assert
list(check_team_names_can_be_lower_cased(session=session)) == []
+ finally:
+ session.execute(delete(Team))
+ session.commit()
+
+ # MySQL's default collation is case insensitive, so the two rows this
needs cannot coexist
+ # there -- which is also why the collision it guards against cannot arise
on MySQL.
+ @pytest.mark.backend("postgres", "sqlite")
+ @pytest.mark.usefixtures("initialized_db")
+ def test_check_team_names_can_be_lower_cased_reports_a_collision(self):
+ """Two names folding onto one row would merge two teams, so migration
has to stop."""
+ with create_session() as session:
+ session.add_all([Team(name="data-eng"), Team(name="Data-Eng")])
+ session.commit()
+ try:
+ errors =
list(check_team_names_can_be_lower_cased(session=session))
+ finally:
+ session.execute(delete(Team))
+ session.commit()
+
+ assert len(errors) == 1
+ assert "'Data-Eng', 'data-eng'" in errors[0]
+ assert "would both become 'data-eng'" in errors[0]
+
@pytest.mark.parametrize(
("auth", "expected"),
[
