Run fab provider tests in v2-11-test

Since we are releasing fab provider from v2-11-test we should also
run fab provider tests there.

This chang consists of:

* migrating to FAB 4.5.3 (includes scrypt algorithm fix)
* monkeypatching flask-session so that it works well with new
  flask-sqlalchey
* caching initial$aization of Session class from flask-session and
  AirflowDatabaseSessionInterface so that DB connection is not
  created multiple times during tests
* fixing imports from tests

Author: potiuk

airflow/api_connexion/security.py

@@ -45,13 +45,14 @@
 
 def check_authentication() -> None:
     """Check that the request has valid authorization information."""
-    for auth in get_airflow_app().api_auth:
+    airflow_app = get_airflow_app()
+    for auth in airflow_app.api_auth:
         response = auth.requires_authentication(Response)()
         if response.status_code == 200:
             return
 
     # Even if the current_user is anonymous, the AUTH_ROLE_PUBLIC might still have permission.
-    appbuilder = get_airflow_app().appbuilder
+    appbuilder = airflow_app.appbuilder
     if appbuilder.get_app.config.get("AUTH_ROLE_PUBLIC", None):
         return
 

airflow/providers/fab/auth_manager/cli_commands/utils.py

@@ -42,7 +42,7 @@ def _return_appbuilder(app: Flask) -> AirflowAppBuilder:
     """Return an appbuilder instance for the given app."""
     init_appbuilder(app)
     init_plugins(app)
-    init_airflow_session_interface(app)
+    init_airflow_session_interface(app, None)
     return app.appbuilder  # type: ignore[attr-defined]
 
 

airflow/providers/fab/auth_manager/fab_auth_manager.py

@@ -95,7 +95,7 @@
     )
     from airflow.providers.common.compat.assets import AssetDetails
     from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
-    from airflow.security.permissions import RESOURCE_ASSET
+    from airflow.security.permissions import RESOURCE_ASSET  # type: ignore[attr-defined]
 else:
     from airflow.providers.common.compat.security.permissions import RESOURCE_ASSET
 
@@ -403,9 +403,7 @@ def get_url_logout(self):
 
     def get_url_user_profile(self) -> str | None:
         """Return the url to a page displaying info about the current user."""
-        if not self.security_manager.user_view or self.appbuilder.get_app.config.get(
-            "AUTH_ROLE_PUBLIC", None
-        ):
+        if not self.security_manager.user_view:
             return None
         return url_for(f"{self.security_manager.user_view.endpoint}.userinfo")
 

airflow/providers/fab/auth_manager/security_manager/override.py

@@ -69,7 +69,6 @@
 from flask_login import LoginManager
 from itsdangerous import want_bytes
 from markupsafe import Markup
-from packaging.version import Version
 from sqlalchemy import and_, func, inspect, literal, or_, select
 from sqlalchemy.exc import MultipleResultsFound
 from sqlalchemy.orm import Session, joinedload
@@ -850,7 +849,8 @@ def _init_config(self):
         app.config.setdefault("AUTH_ROLES_SYNC_AT_LOGIN", False)
         app.config.setdefault("AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS", False)
 
-        # Werkzeug prior to 3.0.0 does not support scrypt
+        from packaging.version import Version
+
         parsed_werkzeug_version = Version(importlib.metadata.version("werkzeug"))
         if parsed_werkzeug_version < Version("3.0.0"):
             app.config.setdefault(
@@ -861,9 +861,10 @@ def _init_config(self):
         else:
             app.config.setdefault(
                 "AUTH_DB_FAKE_PASSWORD_HASH_CHECK",
-                "scrypt:32768:8:1$wiDa0ruWlIPhp9LM$6e409d093e62ad54df2af895d0e125b05ff6cf6414"
-                "8350189ffc4bcc71286edf1b8ad94a442c00f890224bf2b32153d0750c89ee9"
-                "401e62f9dcee5399065e4e5",
+                "scrypt:32768:8:1$wiDa0ruWlIPhp9LM$6e40"
+                "9d093e62ad54df2af895d0e125b05ff6cf6414"
+                "8350189ffc4bcc71286edf1b8ad94a442c00f8"
+                "90224bf2b32153d0750c89ee9401e62f9dcee5399065e4e5",
             )
 
         # LDAP Config

airflow/providers/fab/provider.yaml

@@ -62,7 +62,7 @@ dependencies:
   # Every time we update FAB version here, please make sure that you review the classes and models in
   # `airflow/providers/fab/auth_manager/security_manager/override.py` with their upstream counterparts.
   # In particular, make sure any breaking changes, for example any new methods, are accounted for.
-  - flask-appbuilder==4.5.3
+  - flask-appbuilder==4.5.4
   - google-re2>=1.0
   - jmespath>=0.7.0
 

airflow/www/app.py

@@ -187,7 +187,7 @@ def create_app(config=None, testing=False):
         init_jinja_globals(flask_app)
         init_xframe_protection(flask_app)
         init_cache_control(flask_app)
-        init_airflow_session_interface(flask_app)
+        init_airflow_session_interface(flask_app, db)
         init_check_user_active(flask_app)
     return flask_app
 

airflow/www/extensions/init_session.py

@@ -23,7 +23,52 @@
 from airflow.www.session import AirflowDatabaseSessionInterface, AirflowSecureCookieSessionInterface
 
 
-def init_airflow_session_interface(app):
+# Monkey patch flask-session's create_session_model to fix compatibility with Flask-SQLAlchemy 2.5.1
+# The issue is that dynamically created Session models don't inherit query_class from db.Model,
+# which causes AttributeError when flask-session tries to use .query property.
+# This patch ensures query_class is set on the Session model class.
+def _patch_flask_session_create_session_model():
+    """
+    Patch flask-session's create_session_model to ensure query_class compatibility.
+
+    This fixes the issue where flask-session's Session model doesn't have the query_class
+    attribute required by Flask-SQLAlchemy's _QueryProperty.
+    """
+    try:
+        from flask_session.sqlalchemy import sqlalchemy as flask_session_module
+
+        _original_create_session_model = flask_session_module.create_session_model
+        _session_model = None
+
+        def patched_create_session_model(db, table_name, schema=None, bind_key=None, sequence=None):
+            nonlocal _session_model
+            if _session_model:
+                return _session_model
+
+            # Create new model
+            Session = _original_create_session_model(db, table_name, schema, bind_key, sequence)
+
+            # Ensure query_class is set for compatibility with Flask-SQLAlchemy
+            # Use db.Query which is always available on the SQLAlchemy instance
+            if not hasattr(Session, "query_class"):
+                Session.query_class = getattr(db, "Query", None) or getattr(db.Model, "query_class", None)
+
+            _session_model = Session
+            return Session
+
+        flask_session_module.create_session_model = patched_create_session_model
+    except ImportError:
+        # flask-session not installed, no need to patch
+        pass
+
+
+# Apply the patch immediately when this module is imported
+_patch_flask_session_create_session_model()
+
+_session_interface = None
+
+
+def init_airflow_session_interface(app, sqlalchemy_client):
     """Set airflow session interface."""
     config = app.config.copy()
     selected_backend = conf.get("webserver", "SESSION_BACKEND")
@@ -42,9 +87,13 @@ def make_session_permanent():
 
             app.before_request(make_session_permanent)
     elif selected_backend == "database":
-        app.session_interface = AirflowDatabaseSessionInterface(
+        global _session_interface
+        if _session_interface:
+            app.session_interface = _session_interface
+            return
+        _session_interface = AirflowDatabaseSessionInterface(
             app=app,
-            client=None,
+            client=sqlalchemy_client,
             permanent=permanent_cookie,
             # Typically these would be configurable with Flask-Session,
             # but we will set them explicitly instead as they don't make
@@ -53,6 +102,7 @@ def make_session_permanent():
             key_prefix="",
             use_signer=True,
         )
+        app.session_interface = _session_interface
     else:
         raise AirflowConfigException(
             "Unrecognized session backend specified in "

dev/breeze/src/airflow_breeze/utils/selective_checks.py

@@ -55,7 +55,6 @@
     SelectiveProvidersTestType,
     all_helm_test_packages,
     all_selective_core_test_types,
-    providers_test_type,
 )
 from airflow_breeze.utils.console import get_console
 from airflow_breeze.utils.exclude_from_matrix import excluded_combos
@@ -841,49 +840,51 @@ def _get_core_test_types_to_run(self) -> list[str]:
         return sorted_candidate_test_types
 
     def _get_providers_test_types_to_run(self, split_to_individual_providers: bool = False) -> list[str]:
-        if self._default_branch != "main":
-            return []
-        if self.full_tests_needed:
-            if split_to_individual_providers:
-                return list(providers_test_type())
-            else:
-                return ["Providers"]
-        else:
-            all_providers_source_files = self._matching_files(
-                FileGroupForCi.ALL_PROVIDERS_PYTHON_FILES, CI_FILE_GROUP_MATCHES, CI_FILE_GROUP_EXCLUDES
-            )
-            assets_source_files = self._matching_files(
-                FileGroupForCi.ASSET_FILES, CI_FILE_GROUP_MATCHES, CI_FILE_GROUP_EXCLUDES
-            )
-
-            if (
-                len(all_providers_source_files) == 0
-                and len(assets_source_files) == 0
-                and not self.needs_api_tests
-            ):
-                # IF API tests are needed, that will trigger extra provider checks
-                return []
-            else:
-                affected_providers = self._find_all_providers_affected(
-                    include_docs=False,
-                )
-        candidate_test_types: set[str] = set()
-        if isinstance(affected_providers, AllProvidersSentinel):
-            if split_to_individual_providers:
-                for provider in get_available_packages():
-                    candidate_test_types.add(f"Providers[{provider}]")
-            else:
-                candidate_test_types.add("Providers")
-        elif affected_providers:
-            if split_to_individual_providers:
-                for provider in affected_providers:
-                    candidate_test_types.add(f"Providers[{provider}]")
-            else:
-                candidate_test_types.add(f"Providers[{','.join(sorted(affected_providers))}]")
-        sorted_candidate_test_types = sorted(candidate_test_types)
-        get_console().print("[warning]Selected providers test type candidates to run:[/]")
-        get_console().print(sorted_candidate_test_types)
-        return sorted_candidate_test_types
+        # For v2-11-test branch we always run airflow + fab provider tests
+        return ["Providers[fab]"]
+        # if self._default_branch != "main":
+        #     return []
+        # if self.full_tests_needed:
+        #     if split_to_individual_providers:
+        #         return list(providers_test_type())
+        #     else:
+        #         return ["Providers"]
+        # else:
+        #     all_providers_source_files = self._matching_files(
+        #         FileGroupForCi.ALL_PROVIDERS_PYTHON_FILES, CI_FILE_GROUP_MATCHES, CI_FILE_GROUP_EXCLUDES
+        #     )
+        #     assets_source_files = self._matching_files(
+        #         FileGroupForCi.ASSET_FILES, CI_FILE_GROUP_MATCHES, CI_FILE_GROUP_EXCLUDES
+        #     )
+        #
+        #     if (
+        #         len(all_providers_source_files) == 0
+        #         and len(assets_source_files) == 0
+        #         and not self.needs_api_tests
+        #     ):
+        #         # IF API tests are needed, that will trigger extra provider checks
+        #         return []
+        #     else:
+        #         affected_providers = self._find_all_providers_affected(
+        #             include_docs=False,
+        #         )
+        # candidate_test_types: set[str] = set()
+        # if isinstance(affected_providers, AllProvidersSentinel):
+        #     if split_to_individual_providers:
+        #         for provider in get_available_packages():
+        #             candidate_test_types.add(f"Providers[{provider}]")
+        #     else:
+        #         candidate_test_types.add("Providers")
+        # elif affected_providers:
+        #     if split_to_individual_providers:
+        #         for provider in affected_providers:
+        #             candidate_test_types.add(f"Providers[{provider}]")
+        #     else:
+        #         candidate_test_types.add(f"Providers[{','.join(sorted(affected_providers))}]")
+        # sorted_candidate_test_types = sorted(candidate_test_types)
+        # get_console().print("[warning]Selected providers test type candidates to run:[/]")
+        # get_console().print(sorted_candidate_test_types)
+        # return sorted_candidate_test_types
 
     @staticmethod
     def _extract_long_provider_tests(current_test_types: set[str]):
@@ -930,7 +931,8 @@ def providers_test_types_list_as_string(self) -> str | None:
         if self._default_branch != "main":
             test_types_to_remove: set[str] = set()
             for test_type in current_test_types:
-                if test_type.startswith("Providers"):
+                # For v2-11-test branch we always run airflow + fab provider tests
+                if test_type.startswith("Providers") and not test_type.startswith("Providers[fab]"):
                     get_console().print(
                         f"[warning]Removing {test_type} because the target branch "
                         f"is {self._default_branch} and not main[/]"
@@ -1072,7 +1074,8 @@ def docs_list_as_string(self) -> str | None:
         if not self.docs_build:
             return None
         if self._default_branch != "main":
-            return "apache-airflow docker-stack"
+            # For v2-11-test branch we always run airflow + fab provider tests
+            return "apache-airflow docker-stack apache-airflow-providers-fab"
         if self.full_tests_needed:
             return _ALL_DOCS_LIST
         providers_affected = self._find_all_providers_affected(
@@ -1164,13 +1167,15 @@ def skip_pre_commits(self) -> str:
 
     @cached_property
     def skip_providers_tests(self) -> bool:
-        if self._default_branch != "main":
-            return True
-        if self.full_tests_needed:
-            return False
-        if self._get_providers_test_types_to_run():
-            return False
-        return True
+        # For v2-11-test branch we always run airflow + fab provider tests
+        return False
+        # if self._default_branch != "main":
+        #     return True
+        # if self.full_tests_needed:
+        #     return False
+        # if self._get_providers_test_types_to_run():
+        #     return False
+        # return True
 
     @cached_property
     def test_groups(self):
@@ -1200,18 +1205,21 @@ def helm_test_packages(self) -> str:
 
     @cached_property
     def selected_providers_list_as_string(self) -> str | None:
-        if self._default_branch != "main":
-            return None
-        if self.full_tests_needed:
-            return ""
-        if self._are_all_providers_affected():
-            return ""
-        affected_providers = self._find_all_providers_affected(include_docs=True)
-        if not affected_providers:
-            return None
-        if isinstance(affected_providers, AllProvidersSentinel):
-            return ""
-        return " ".join(sorted(affected_providers))
+        # For v2-11-test branch we always test airflow + fab provider
+        return "fab"
+
+        # if self._default_branch != "main":
+        #     return None
+        # if self.full_tests_needed:
+        #     return ""
+        # if self._are_all_providers_affected():
+        #     return ""
+        # affected_providers = self._find_all_providers_affected(include_docs=True)
+        # if not affected_providers:
+        #     return None
+        # if isinstance(affected_providers, AllProvidersSentinel):
+        #     return ""
+        # return " ".join(sorted(affected_providers))
 
     @cached_property
     def runs_on_as_json_default(self) -> str:

generated/provider_dependencies.json

@@ -550,7 +550,7 @@
     "deps": [
       "apache-airflow-providers-common-compat>=1.2.1",
       "apache-airflow>=2.9.0",
-      "flask-appbuilder==4.5.3",
+      "flask-appbuilder==4.5.4",
       "flask-login>=0.6.3",
       "flask-session>=0.8.0",
       "flask>=2.2,<3",

scripts/ci/pre_commit/check_system_tests.py

@@ -38,21 +38,21 @@
 WATCHER_APPEND_INSTRUCTION_SHORT = " >> watcher()"
 
 PYTEST_FUNCTION = """
-from tests_common.test_utils.system_tests import get_test_run  # noqa: E402
+from tests.test_utils.system_tests import get_test_run  # noqa: E402
 
 # Needed to run the example DAG with pytest (see: tests/system/README.md#run_via_pytest)
 test_run = get_test_run(dag)
 """
 PYTEST_FUNCTION_PATTERN = re.compile(
-    r"from tests_common\.test_utils\.system_tests import get_test_run(?:  # noqa: E402)?\s+"
+    r"from tests\.test_utils\.system_tests import get_test_run(?:  # noqa: E402)?\s+"
     r"(?:# .+\))?\s+"
     r"test_run = get_test_run\(dag\)"
 )
 
 
 def _check_file(file: Path):
     content = file.read_text()
-    if "from tests_common.test_utils.watcher import watcher" in content:
+    if "from tests.test_utils.watcher import watcher" in content:
         index = content.find(WATCHER_APPEND_INSTRUCTION_SHORT)
         if index == -1:
             errors.append(