This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 79d52f5ad08 [v3-3-test] Improve SQLite test DB health checks (#68021)
(#70733)
79d52f5ad08 is described below
commit 79d52f5ad0849c8b86c5c81e94dab68f3d8c7a56
Author: Jarek Potiuk <[email protected]>
AuthorDate: Thu Jul 30 11:26:59 2026 +0200
[v3-3-test] Improve SQLite test DB health checks (#68021) (#70733)
(cherry picked from commit 84ac434aaa19d520f1c19f4321a3443f4b8f5737)
Co-authored-by: Henry Chen <[email protected]>
---
airflow-core/src/airflow/utils/db_manager.py | 8 +
airflow-core/tests/unit/utils/test_db_manager.py | 8 +
devel-common/src/tests_common/pytest_plugin.py | 178 +++++++-
.../tests/unit/tests_common/test_pytest_plugin.py | 463 +++++++++++++++++++++
4 files changed, 651 insertions(+), 6 deletions(-)
diff --git a/airflow-core/src/airflow/utils/db_manager.py
b/airflow-core/src/airflow/utils/db_manager.py
index 57173341e88..3fcc0a06ab2 100644
--- a/airflow-core/src/airflow/utils/db_manager.py
+++ b/airflow-core/src/airflow/utils/db_manager.py
@@ -295,6 +295,14 @@ class RunDBManager(LoggingMixin):
for manager in self._managers:
RunDBManager._validate(manager)
+ def get_required_table_names(self) -> set[str]:
+ """Return table names required by the configured external DB
managers."""
+ required_tables: set[str] = set()
+ for manager in self._managers:
+ required_tables.update(table.name for table in
manager.metadata.tables.values())
+ required_tables.add(manager.version_table_name)
+ return required_tables
+
@staticmethod
def _validate(manager: BaseDBManager):
"""Validate the external database migration."""
diff --git a/airflow-core/tests/unit/utils/test_db_manager.py
b/airflow-core/tests/unit/utils/test_db_manager.py
index 1f8f3e7423a..513ecf5855c 100644
--- a/airflow-core/tests/unit/utils/test_db_manager.py
+++ b/airflow-core/tests/unit/utils/test_db_manager.py
@@ -277,6 +277,14 @@ class TestBaseDBManager:
class TestRunDBManager:
+ def test_get_required_table_names(self):
+ run_db_manager = _create_run_db_manager(LegacyTablesDBManager)
+
+ assert run_db_manager.get_required_table_names() == {
+ "external_legacy_table",
+ "legacy_alembic_version",
+ }
+
def test_initdb_and_upgradedb_support_legacy_manager_signatures(self,
session):
LegacySignatureExternalManager.initdb_calls = 0
LegacySignatureExternalManager.upgradedb_calls = 0
diff --git a/devel-common/src/tests_common/pytest_plugin.py
b/devel-common/src/tests_common/pytest_plugin.py
index eaac4e37fe0..413723ec317 100644
--- a/devel-common/src/tests_common/pytest_plugin.py
+++ b/devel-common/src/tests_common/pytest_plugin.py
@@ -118,6 +118,7 @@ if not keep_env_variables:
del os.environ[env_key]
SUPPORTED_DB_BACKENDS = ("sqlite", "postgres", "mysql")
+_SQLITE_DB_HEALTH_REQUIRED_TABLES = {"alembic_version"}
# A bit of a Hack - but we need to check args before they are parsed by pytest
in order to
# configure the DB before Airflow gets initialized (which happens at airflow
import time).
@@ -447,27 +448,192 @@ def initialize_airflow_tests(request):
def _initialize_airflow_db(force_db_init: bool, airflow_home: str | Path):
db_init_lock_file = Path(airflow_home).joinpath(".airflow_db_initialised")
+ db_health_failure_reason = None
if not force_db_init and db_init_lock_file.exists():
- print(
- "Skipping initializing of the DB as it was initialized already.\n"
- "You can re-initialize the database by adding --with-db-init flag
when running tests."
- )
- return
+ db_health_failure_reason = _get_airflow_db_health_failure_reason()
+ if db_health_failure_reason:
+ print(
+ "The DB initialization marker exists, but the DB health check
failed.\n"
+ f"{db_health_failure_reason}\n"
+ "Re-initializing the DB."
+ )
+ else:
+ print(
+ "Skipping initializing of the DB as it was initialized
already.\n"
+ "You can re-initialize the database by adding --with-db-init
flag when running tests."
+ )
+ return
from tests_common.test_utils.db import initial_db_init
if force_db_init:
print("Initializing the DB - forced with --with-db-init flag.")
+ elif db_health_failure_reason:
+ print("Initializing the DB - existing DB did not pass the health
check.")
else:
print(
"Initializing the DB - first time after entering the container.\n"
"Initialization can be also forced by adding --with-db-init flag
when running tests."
)
- initial_db_init()
+ try:
+ initial_db_init()
+ except _get_db_maintenance_exception_types() as ex:
+ if sqlite_db_file := _get_configured_sqlite_db_file():
+ raise RuntimeError(
+ f"Unable to re-initialize the SQLite test DB. Remove
`{sqlite_db_file}` and rerun pytest."
+ ) from ex
+ raise
db_init_lock_file.touch(exist_ok=True)
+def _get_airflow_db_health_failure_reason() -> str | None:
+ from airflow.configuration import conf
+ from airflow.models import import_all_models
+ from airflow.models.base import Base
+
+ sql_alchemy_conn = conf.get("database", "sql_alchemy_conn")
+ import_all_models()
+ required_tables = set(Base.metadata.tables) |
_SQLITE_DB_HEALTH_REQUIRED_TABLES
+ table_health_failure_reason = _get_sqlite_db_health_failure_reason(
+ sql_alchemy_conn,
+ required_tables=required_tables,
+ )
+ if table_health_failure_reason:
+ return table_health_failure_reason
+
+ if not _is_sqlite_db(sql_alchemy_conn):
+ return None
+
+ core_migration_health_failure_reason =
_get_core_migration_health_failure_reason()
+ if core_migration_health_failure_reason:
+ return core_migration_health_failure_reason
+
+ return _get_external_db_manager_health_failure_reason(sql_alchemy_conn)
+
+
+def _is_sqlite_db(sql_alchemy_conn: str) -> bool:
+ from sqlalchemy.engine import make_url
+
+ return make_url(sql_alchemy_conn).get_backend_name() == "sqlite"
+
+
+def _get_core_migration_health_failure_reason() -> str | None:
+ from airflow.utils.db import _configured_alembic_environment
+
+ try:
+ with _configured_alembic_environment() as env:
+ context = env.get_context()
+ source_heads = set(env.script.get_heads())
+ db_heads = set(context.get_current_heads())
+ except _get_db_maintenance_exception_types() as ex:
+ return f"Unable to inspect SQLite test DB core migration state: {ex}"
+
+ if source_heads == db_heads:
+ return None
+
+ db_heads_display = ", ".join(sorted(db_heads)) or "<none>"
+ source_heads_display = ", ".join(sorted(source_heads)) or "<none>"
+ return (
+ "SQLite test DB migration revision does not match the current Airflow
migration head. "
+ f"DB heads: {db_heads_display}; source heads: {source_heads_display}."
+ )
+
+
+def _get_external_db_manager_health_failure_reason(sql_alchemy_conn: str) ->
str | None:
+ from airflow import settings
+ from airflow.utils.db_manager import RunDBManager
+
+ external_db_manager = RunDBManager()
+ required_tables = external_db_manager.get_required_table_names()
+ if not required_tables:
+ return None
+
+ table_health_failure_reason = _get_sqlite_db_health_failure_reason(
+ sql_alchemy_conn,
+ required_tables=required_tables,
+ )
+ if table_health_failure_reason:
+ return table_health_failure_reason
+
+ session_factory = settings.Session
+ if session_factory is None:
+ return "Unable to inspect SQLite test DB external DB manager migration
state: settings.Session is not configured."
+
+ session = session_factory()
+ try:
+ if external_db_manager.check_migration(session):
+ return None
+ except _get_db_maintenance_exception_types() as ex:
+ return f"Unable to inspect SQLite test DB external DB manager
migration state: {ex}"
+ finally:
+ session.close()
+
+ return "SQLite test DB external DB manager migrations do not match their
current migration heads."
+
+
+def _get_db_maintenance_exception_types() -> tuple[type[Exception], ...]:
+ from alembic.script.revision import RevisionError
+ from alembic.util.exc import CommandError
+ from sqlalchemy.exc import SQLAlchemyError
+
+ return SQLAlchemyError, CommandError, RevisionError
+
+
+def _get_sqlite_db_health_failure_reason(
+ sql_alchemy_conn: str,
+ *,
+ required_tables: set[str],
+) -> str | None:
+ from sqlalchemy import create_engine, inspect
+ from sqlalchemy.engine import make_url
+ from sqlalchemy.exc import SQLAlchemyError
+
+ url = make_url(sql_alchemy_conn)
+ if url.get_backend_name() != "sqlite":
+ return None
+
+ sqlite_db_file = _get_sqlite_db_file(sql_alchemy_conn)
+ if not sqlite_db_file:
+ return "SQLite test DB uses an in-memory URL, so it must be
re-initialized for this test run."
+
+ if not sqlite_db_file.exists():
+ return f"SQLite test DB file `{sqlite_db_file}` does not exist."
+
+ engine = create_engine(url)
+ try:
+ try:
+ existing_tables = set(inspect(engine).get_table_names())
+ except SQLAlchemyError as ex:
+ return f"Unable to inspect SQLite test DB schema: {ex}"
+ finally:
+ engine.dispose()
+
+ missing_tables = sorted(required_tables - existing_tables)
+ if missing_tables:
+ missing_table_list = ", ".join(missing_tables[:10])
+ if len(missing_tables) > 10:
+ missing_table_list = f"{missing_table_list}, ...
({len(missing_tables)} total)"
+ return f"SQLite test DB schema is missing expected tables:
{missing_table_list}."
+
+ return None
+
+
+def _get_configured_sqlite_db_file() -> Path | None:
+ from airflow.configuration import conf
+
+ return _get_sqlite_db_file(conf.get("database", "sql_alchemy_conn"))
+
+
+def _get_sqlite_db_file(sql_alchemy_conn: str) -> Path | None:
+ from sqlalchemy.engine import make_url
+
+ url = make_url(sql_alchemy_conn)
+ if url.get_backend_name() != "sqlite" or not url.database or url.database
== ":memory:":
+ return None
+ return Path(url.database)
+
+
def _initialize_kerberos():
kerberos = os.environ.get("KRB5_KTNAME")
if not kerberos:
diff --git a/devel-common/tests/unit/tests_common/test_pytest_plugin.py
b/devel-common/tests/unit/tests_common/test_pytest_plugin.py
new file mode 100644
index 00000000000..c4fed2466d6
--- /dev/null
+++ b/devel-common/tests/unit/tests_common/test_pytest_plugin.py
@@ -0,0 +1,463 @@
+#
+# 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.
+
+from __future__ import annotations
+
+import sys
+import types
+from contextlib import contextmanager
+from typing import Any
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.exc import SQLAlchemyError
+
+from tests_common import pytest_plugin
+
+
+def _create_sqlite_db(sqlite_db_url: str, table_names: set[str]) -> None:
+ engine = create_engine(sqlite_db_url)
+ try:
+ with engine.begin() as conn:
+ for table_name in table_names:
+ conn.execute(text(f"CREATE TABLE {table_name} (id INTEGER
PRIMARY KEY)"))
+ finally:
+ engine.dispose()
+
+
+class _FakeMigrationContext:
+ def __init__(self, db_heads: set[str], db_heads_exception: Exception |
None = None) -> None:
+ self.db_heads = db_heads
+ self.db_heads_exception = db_heads_exception
+
+ def get_current_heads(self) -> set[str]:
+ if self.db_heads_exception:
+ raise self.db_heads_exception
+ return self.db_heads
+
+
+class _FakeAlembicScript:
+ def __init__(self, source_heads: set[str]) -> None:
+ self.source_heads = source_heads
+
+ def get_heads(self) -> set[str]:
+ return self.source_heads
+
+
+class _FakeAlembicEnvironment:
+ def __init__(
+ self, *, db_heads: set[str], source_heads: set[str],
db_heads_exception: Exception | None = None
+ ) -> None:
+ self.script = _FakeAlembicScript(source_heads)
+ self._context = _FakeMigrationContext(db_heads, db_heads_exception)
+
+ def get_context(self) -> _FakeMigrationContext:
+ return self._context
+
+
+def _install_fake_airflow_db_module(
+ monkeypatch,
+ *,
+ db_heads: set[str],
+ source_heads: set[str],
+ db_heads_exception: Exception | None = None,
+) -> None:
+ @contextmanager
+ def configured_alembic_environment():
+ yield _FakeAlembicEnvironment(
+ db_heads=db_heads,
+ source_heads=source_heads,
+ db_heads_exception=db_heads_exception,
+ )
+
+ fake_airflow_module = types.ModuleType("airflow")
+ fake_utils_module = types.ModuleType("airflow.utils")
+ fake_db_module = types.ModuleType("airflow.utils.db")
+ setattr(fake_db_module, "_configured_alembic_environment",
configured_alembic_environment)
+ setattr(fake_airflow_module, "utils", fake_utils_module)
+ setattr(fake_utils_module, "db", fake_db_module)
+
+ monkeypatch.setitem(sys.modules, "airflow", fake_airflow_module)
+ monkeypatch.setitem(sys.modules, "airflow.utils", fake_utils_module)
+ monkeypatch.setitem(sys.modules, "airflow.utils.db", fake_db_module)
+
+
+class _FakeConf:
+ def __init__(self, sql_alchemy_conn: str) -> None:
+ self.sql_alchemy_conn = sql_alchemy_conn
+
+ def get(self, section: str, key: str) -> str:
+ return self.sql_alchemy_conn
+
+
+class _FakeMetadata:
+ tables = {"dag": object()}
+
+
+class _FakeBase:
+ metadata = _FakeMetadata()
+
+
+class _FakeTable:
+ def __init__(self, name: str) -> None:
+ self.name = name
+
+
+class _FakeSession:
+ def __init__(self) -> None:
+ self.closed = False
+
+ def close(self) -> None:
+ self.closed = True
+
+
+def _install_fake_airflow_metadata_modules(monkeypatch, *, sql_alchemy_conn:
str) -> None:
+ fake_airflow_module = types.ModuleType("airflow")
+ fake_configuration_module = types.ModuleType("airflow.configuration")
+ fake_models_module = types.ModuleType("airflow.models")
+ fake_base_module = types.ModuleType("airflow.models.base")
+
+ setattr(fake_configuration_module, "conf", _FakeConf(sql_alchemy_conn))
+ setattr(fake_models_module, "import_all_models", lambda: None)
+ setattr(fake_base_module, "Base", _FakeBase)
+ setattr(fake_airflow_module, "configuration", fake_configuration_module)
+ setattr(fake_airflow_module, "models", fake_models_module)
+ setattr(fake_models_module, "base", fake_base_module)
+
+ monkeypatch.setitem(sys.modules, "airflow", fake_airflow_module)
+ monkeypatch.setitem(sys.modules, "airflow.configuration",
fake_configuration_module)
+ monkeypatch.setitem(sys.modules, "airflow.models", fake_models_module)
+ monkeypatch.setitem(sys.modules, "airflow.models.base", fake_base_module)
+
+
+def _install_fake_external_db_manager_modules(
+ monkeypatch,
+ *,
+ managers: list[Any],
+ migration_check_result: bool,
+ session: _FakeSession,
+ migration_check_exception: Exception | None = None,
+) -> None:
+ class FakeRunDBManager:
+ def get_required_table_names(self) -> set[str]:
+ required_tables: set[str] = set()
+ for manager in managers:
+ required_tables.update(table.name for table in
manager.metadata.tables.values())
+ required_tables.add(manager.version_table_name)
+ return required_tables
+
+ def check_migration(self, received_session) -> bool:
+ assert received_session is session
+ if migration_check_exception:
+ raise migration_check_exception
+ return migration_check_result
+
+ fake_airflow_module = types.ModuleType("airflow")
+ fake_settings_module = types.ModuleType("airflow.settings")
+ fake_utils_module = types.ModuleType("airflow.utils")
+ fake_db_manager_module = types.ModuleType("airflow.utils.db_manager")
+
+ setattr(fake_settings_module, "Session", lambda: session)
+ setattr(fake_db_manager_module, "RunDBManager", FakeRunDBManager)
+ setattr(fake_airflow_module, "settings", fake_settings_module)
+ setattr(fake_airflow_module, "utils", fake_utils_module)
+ setattr(fake_utils_module, "db_manager", fake_db_manager_module)
+
+ monkeypatch.setitem(sys.modules, "airflow", fake_airflow_module)
+ monkeypatch.setitem(sys.modules, "airflow.settings", fake_settings_module)
+ monkeypatch.setitem(sys.modules, "airflow.utils", fake_utils_module)
+ monkeypatch.setitem(sys.modules, "airflow.utils.db_manager",
fake_db_manager_module)
+
+
+class TestSqliteDbHealthCheck:
+ def test_non_sqlite_db_is_not_checked(self):
+ assert (
+ pytest_plugin._get_sqlite_db_health_failure_reason(
+ "postgresql://localhost/airflow",
+ required_tables={"dag"},
+ )
+ is None
+ )
+
+ def test_missing_sqlite_db_file_fails_health_check(self, tmp_path):
+ missing_db_url = f"sqlite:///{tmp_path / 'missing.db'}"
+
+ reason = pytest_plugin._get_sqlite_db_health_failure_reason(
+ missing_db_url,
+ required_tables={"dag"},
+ )
+
+ assert reason == f"SQLite test DB file `{tmp_path / 'missing.db'}`
does not exist."
+
+ def test_in_memory_sqlite_db_fails_health_check(self):
+ reason = pytest_plugin._get_sqlite_db_health_failure_reason(
+ "sqlite:///:memory:",
+ required_tables={"dag"},
+ )
+
+ assert (
+ reason == "SQLite test DB uses an in-memory URL, so it must be
re-initialized for this test run."
+ )
+
+ def test_missing_required_tables_fail_health_check(self, tmp_path):
+ sqlite_db_url = f"sqlite:///{tmp_path / 'airflow.db'}"
+ _create_sqlite_db(sqlite_db_url, {"alembic_version"})
+
+ reason = pytest_plugin._get_sqlite_db_health_failure_reason(
+ sqlite_db_url,
+ required_tables={"alembic_version", "dag", "dag_run"},
+ )
+
+ assert reason == "SQLite test DB schema is missing expected tables:
dag, dag_run."
+
+ def test_existing_required_tables_pass_health_check(self, tmp_path):
+ sqlite_db_url = f"sqlite:///{tmp_path / 'airflow.db'}"
+ _create_sqlite_db(sqlite_db_url, {"alembic_version", "dag", "dag_run"})
+
+ assert (
+ pytest_plugin._get_sqlite_db_health_failure_reason(
+ sqlite_db_url,
+ required_tables={"alembic_version", "dag", "dag_run"},
+ )
+ is None
+ )
+
+
+class TestAirflowDbHealthCheck:
+ def test_sqlite_db_health_check_includes_core_migration_health(self,
monkeypatch):
+ _install_fake_airflow_metadata_modules(monkeypatch,
sql_alchemy_conn="sqlite:////tmp/airflow.db")
+
+ def get_sqlite_db_health_failure_reason(
+ sql_alchemy_conn: str, *, required_tables: set[str]
+ ) -> str | None:
+ assert sql_alchemy_conn == "sqlite:////tmp/airflow.db"
+ assert required_tables == {"alembic_version", "dag"}
+ return None
+
+ monkeypatch.setattr(
+ pytest_plugin,
+ "_get_sqlite_db_health_failure_reason",
+ get_sqlite_db_health_failure_reason,
+ )
+ monkeypatch.setattr(
+ pytest_plugin,
+ "_get_core_migration_health_failure_reason",
+ lambda: "SQLite test DB migration revision does not match.",
+ )
+
+ assert (
+ pytest_plugin._get_airflow_db_health_failure_reason()
+ == "SQLite test DB migration revision does not match."
+ )
+
+ def test_sqlite_db_health_check_includes_external_db_manager_health(self,
monkeypatch):
+ _install_fake_airflow_metadata_modules(monkeypatch,
sql_alchemy_conn="sqlite:////tmp/airflow.db")
+ monkeypatch.setattr(pytest_plugin,
"_get_sqlite_db_health_failure_reason", lambda *_, **__: None)
+ monkeypatch.setattr(pytest_plugin,
"_get_core_migration_health_failure_reason", lambda: None)
+ monkeypatch.setattr(
+ pytest_plugin,
+ "_get_external_db_manager_health_failure_reason",
+ lambda sql_alchemy_conn: "SQLite test DB external DB manager
migrations do not match.",
+ )
+
+ assert (
+ pytest_plugin._get_airflow_db_health_failure_reason()
+ == "SQLite test DB external DB manager migrations do not match."
+ )
+
+
+class TestCoreMigrationHealthCheck:
+ def test_matching_core_migration_heads_pass_health_check(self,
monkeypatch):
+ _install_fake_airflow_db_module(
+ monkeypatch, db_heads={"current_revision"},
source_heads={"current_revision"}
+ )
+
+ assert pytest_plugin._get_core_migration_health_failure_reason() is
None
+
+ def test_mismatched_core_migration_heads_fail_health_check(self,
monkeypatch):
+ _install_fake_airflow_db_module(
+ monkeypatch, db_heads={"old_revision"},
source_heads={"current_revision"}
+ )
+
+ reason = pytest_plugin._get_core_migration_health_failure_reason()
+
+ assert (
+ reason == "SQLite test DB migration revision does not match the
current Airflow migration head. "
+ "DB heads: old_revision; source heads: current_revision."
+ )
+
+ def test_core_migration_inspection_error_fails_health_check(self,
monkeypatch):
+ _install_fake_airflow_db_module(
+ monkeypatch,
+ db_heads=set(),
+ source_heads={"current_revision"},
+ db_heads_exception=SQLAlchemyError("malformed alembic_version"),
+ )
+
+ assert (
+ pytest_plugin._get_core_migration_health_failure_reason()
+ == "Unable to inspect SQLite test DB core migration state:
malformed alembic_version"
+ )
+
+
+class TestExternalDbManagerHealthCheck:
+ def test_missing_external_db_manager_table_fails_health_check(self,
monkeypatch):
+ class FakeExternalDbManager:
+ metadata = types.SimpleNamespace(
+ tables={"external_table": _FakeTable("external_table")},
+ )
+ version_table_name = "alembic_version_external"
+
+ _install_fake_external_db_manager_modules(
+ monkeypatch,
+ managers=[FakeExternalDbManager],
+ migration_check_result=True,
+ session=_FakeSession(),
+ )
+
+ def get_sqlite_db_health_failure_reason(
+ sql_alchemy_conn: str, *, required_tables: set[str]
+ ) -> str | None:
+ assert sql_alchemy_conn == "sqlite:////tmp/airflow.db"
+ assert required_tables == {"alembic_version_external",
"external_table"}
+ return "SQLite test DB schema is missing expected tables:
external_table."
+
+ monkeypatch.setattr(
+ pytest_plugin,
+ "_get_sqlite_db_health_failure_reason",
+ get_sqlite_db_health_failure_reason,
+ )
+
+ assert (
+
pytest_plugin._get_external_db_manager_health_failure_reason("sqlite:////tmp/airflow.db")
+ == "SQLite test DB schema is missing expected tables:
external_table."
+ )
+
+ def test_external_db_manager_migration_mismatch_fails_health_check(self,
monkeypatch):
+ class FakeExternalDbManager:
+ metadata = types.SimpleNamespace(
+ tables={"external_table": _FakeTable("external_table")},
+ )
+ version_table_name = "alembic_version_external"
+
+ session = _FakeSession()
+ _install_fake_external_db_manager_modules(
+ monkeypatch,
+ managers=[FakeExternalDbManager],
+ migration_check_result=False,
+ session=session,
+ )
+ monkeypatch.setattr(pytest_plugin,
"_get_sqlite_db_health_failure_reason", lambda *_, **__: None)
+
+ assert (
+
pytest_plugin._get_external_db_manager_health_failure_reason("sqlite:////tmp/airflow.db")
+ == "SQLite test DB external DB manager migrations do not match
their current migration heads."
+ )
+ assert session.closed is True
+
+ def
test_external_db_manager_migration_inspection_error_fails_health_check(self,
monkeypatch):
+ class FakeExternalDbManager:
+ metadata = types.SimpleNamespace(
+ tables={"external_table": _FakeTable("external_table")},
+ )
+ version_table_name = "alembic_version_external"
+
+ session = _FakeSession()
+ _install_fake_external_db_manager_modules(
+ monkeypatch,
+ managers=[FakeExternalDbManager],
+ migration_check_result=True,
+ session=session,
+ migration_check_exception=SQLAlchemyError("malformed external
version table"),
+ )
+ monkeypatch.setattr(pytest_plugin,
"_get_sqlite_db_health_failure_reason", lambda *_, **__: None)
+
+ assert (
+
pytest_plugin._get_external_db_manager_health_failure_reason("sqlite:////tmp/airflow.db")
+ == "Unable to inspect SQLite test DB external DB manager migration
state: "
+ "malformed external version table"
+ )
+ assert session.closed is True
+
+
+class TestInitializeAirflowDb:
+ def test_skips_initialization_when_marker_exists_and_db_is_healthy(self,
monkeypatch, tmp_path):
+ called = False
+
+ def initial_db_init():
+ nonlocal called
+ called = True
+
+ fake_db_module = types.ModuleType("tests_common.test_utils.db")
+ setattr(fake_db_module, "initial_db_init", initial_db_init)
+ monkeypatch.setitem(sys.modules, "tests_common.test_utils.db",
fake_db_module)
+ monkeypatch.setattr(pytest_plugin,
"_get_airflow_db_health_failure_reason", lambda: None)
+ tmp_path.joinpath(".airflow_db_initialised").touch()
+
+ pytest_plugin._initialize_airflow_db(force_db_init=False,
airflow_home=tmp_path)
+
+ assert called is False
+
+ def test_initializes_when_marker_exists_but_db_health_check_fails(self,
monkeypatch, tmp_path):
+ called = False
+
+ def initial_db_init():
+ nonlocal called
+ called = True
+
+ fake_db_module = types.ModuleType("tests_common.test_utils.db")
+ setattr(fake_db_module, "initial_db_init", initial_db_init)
+ monkeypatch.setitem(sys.modules, "tests_common.test_utils.db",
fake_db_module)
+ monkeypatch.setattr(
+ pytest_plugin,
+ "_get_airflow_db_health_failure_reason",
+ lambda: "SQLite test DB file `/tmp/airflow.db` does not exist.",
+ )
+ tmp_path.joinpath(".airflow_db_initialised").touch()
+
+ pytest_plugin._initialize_airflow_db(force_db_init=False,
airflow_home=tmp_path)
+
+ assert called is True
+
+ def
test_sqlite_initialization_failure_includes_db_file_removal_guidance(self,
monkeypatch, tmp_path):
+ sqlite_db_file = tmp_path / "airflow.db"
+
+ def initial_db_init():
+ raise SQLAlchemyError("reset failed")
+
+ fake_db_module = types.ModuleType("tests_common.test_utils.db")
+ setattr(fake_db_module, "initial_db_init", initial_db_init)
+ monkeypatch.setitem(sys.modules, "tests_common.test_utils.db",
fake_db_module)
+ monkeypatch.setattr(
+ pytest_plugin,
+ "_get_airflow_db_health_failure_reason",
+ lambda: "SQLite test DB migration revision does not match.",
+ )
+ _install_fake_airflow_metadata_modules(
+ monkeypatch,
+ sql_alchemy_conn=f"sqlite:///{sqlite_db_file}",
+ )
+ tmp_path.joinpath(".airflow_db_initialised").touch()
+
+ with pytest.raises(RuntimeError) as exc_info:
+ pytest_plugin._initialize_airflow_db(force_db_init=False,
airflow_home=tmp_path)
+
+ assert str(exc_info.value) == (
+ f"Unable to re-initialize the SQLite test DB. Remove
`{sqlite_db_file}` and rerun pytest."
+ )
+ assert isinstance(exc_info.value.__cause__, SQLAlchemyError)
