Resolve deprecations in API tests

airflow/api_connexion/endpoints/task_instance_endpoint.py

@@ -405,7 +405,7 @@ def get_task_instances_batch(session: Session = NEW_SESSION) -> APIResponse:
         if not get_auth_manager().batch_is_authorized_dag(requests):
             raise PermissionDenied(detail=f"User not allowed to access some of these DAGs: {list(dag_ids)}")
     else:
-        dag_ids = get_airflow_app().appbuilder.sm.get_accessible_dag_ids(g.user)
+        dag_ids = get_auth_manager().get_permitted_dag_ids(user=g.user)
 
     states = _convert_ti_states(data["state"])
     base_query = select(TI).join(TI.dag_run)

tests/api_connexion/endpoints/test_connection_endpoint.py

@@ -139,7 +139,7 @@ def test_should_respond_200(self, session):
             login="login",
             schema="testschema",
             port=80,
-            extra="{'param': 'value'}",
+            extra='{"param": "value"}',
         )
         session.add(connection_model)
         session.commit()
@@ -157,7 +157,7 @@ def test_should_respond_200(self, session):
             "login": "login",
             "schema": "testschema",
             "port": 80,
-            "extra": "{'param': 'value'}",
+            "extra": '{"param": "value"}',
         }
 
     @pytest.mark.enable_redact
@@ -415,8 +415,8 @@ class TestPatchConnection(TestConnectionEndpoint):
     @pytest.mark.parametrize(
         "payload",
         [
-            {"connection_id": "test-connection-id", "conn_type": "test_type", "extra": "{'key': 'var'}"},
-            {"extra": "{'key': 'var'}"},
+            {"connection_id": "test-connection-id", "conn_type": "test_type", "extra": '{"key": "var"}'},
+            {"extra": '{"key": "var"}'},
         ],
     )
     @provide_session

tests/api_connexion/endpoints/test_dag_run_endpoint.py

@@ -2203,26 +2203,6 @@ def test_should_respond_404(self):
         )
         assert response.status_code == 404
 
-    @conf_vars(
-        {
-            ("api", "auth_backends"): "airflow.api.auth.backend.default",
-        }
-    )
-    def test_should_respond_200_with_anonymous_user(self, dag_maker, session):
-        from airflow.www import app as application
-
-        app = application.create_app(testing=True)
-        app.config["AUTH_ROLE_PUBLIC"] = "Admin"
-        dag_runs = self._create_test_dag_run(DagRunState.SUCCESS)
-        session.add_all(dag_runs)
-        session.commit()
-        created_dr = dag_runs[0]
-        response = app.test_client().patch(
-            f"api/v1/dags/{created_dr.dag_id}/dagRuns/TEST_DAG_RUN_ID_1/setNote",
-            json={"note": "I am setting a note with anonymous user"},
-        )
-        assert response.status_code == 200
-
     @pytest.mark.parametrize(
         "set_auto_role_public, expected_status_code",
         (("Public", 403), ("Admin", 200)),

tests/api_connexion/endpoints/test_extra_link_endpoint.py

@@ -26,7 +26,6 @@
 from airflow.models.dagbag import DagBag
 from airflow.models.xcom import XCom
 from airflow.plugins_manager import AirflowPlugin
-from airflow.providers.google.cloud.operators.bigquery import BigQueryExecuteQueryOperator
 from airflow.security import permissions
 from airflow.timetables.base import DataInterval
 from airflow.utils import timezone
@@ -35,6 +34,7 @@
 from tests.test_utils.api_connexion_utils import create_user, delete_user
 from tests.test_utils.compat import BaseOperatorLink
 from tests.test_utils.db import clear_db_runs, clear_db_xcom
+from tests.test_utils.mock_operators import CustomOperator
 from tests.test_utils.mock_plugins import mock_plugin_manager
 
 pytestmark = pytest.mark.db_test
@@ -96,21 +96,23 @@ def teardown_method(self) -> None:
 
     def _create_dag(self):
         with DAG(dag_id="TEST_DAG_ID", default_args={"start_date": self.default_time}) as dag:
-            BigQueryExecuteQueryOperator(task_id="TEST_SINGLE_QUERY", sql="SELECT 1")
-            BigQueryExecuteQueryOperator(task_id="TEST_MULTIPLE_QUERY", sql=["SELECT 1", "SELECT 2"])
+            CustomOperator(task_id="TEST_SINGLE_LINK", bash_command="TEST_LINK_VALUE")
+            CustomOperator(
+                task_id="TEST_MULTIPLE_LINK", bash_command=["TEST_LINK_VALUE_1", "TEST_LINK_VALUE_2"]
+            )
         return dag
 
     @pytest.mark.parametrize(
         "url, expected_title, expected_detail",
         [
             pytest.param(
-                "/api/v1/dags/INVALID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_QUERY/links",
+                "/api/v1/dags/INVALID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_LINK/links",
                 "DAG not found",
                 'DAG with ID = "INVALID" not found',
                 id="missing_dag",
             ),
             pytest.param(
-                "/api/v1/dags/TEST_DAG_ID/dagRuns/INVALID/taskInstances/TEST_SINGLE_QUERY/links",
+                "/api/v1/dags/TEST_DAG_ID/dagRuns/INVALID/taskInstances/TEST_SINGLE_LINK/links",
                 "DAG Run not found",
                 'DAG Run with ID = "INVALID" not found',
                 id="missing_dag_run",
@@ -136,64 +138,62 @@ def test_should_respond_404(self, url, expected_title, expected_detail):
 
     def test_should_raise_403_forbidden(self):
         response = self.client.get(
-            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_QUERY/links",
+            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_LINK/links",
             environ_overrides={"REMOTE_USER": "test_no_permissions"},
         )
         assert response.status_code == 403
 
     @mock_plugin_manager(plugins=[])
     def test_should_respond_200(self):
         XCom.set(
-            key="job_id_path",
-            value="TEST_JOB_ID",
-            task_id="TEST_SINGLE_QUERY",
+            key="search_query",
+            value="TEST_LINK_VALUE",
+            task_id="TEST_SINGLE_LINK",
             dag_id=self.dag.dag_id,
             run_id="TEST_DAG_RUN_ID",
         )
         response = self.client.get(
-            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_QUERY/links",
+            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_LINK/links",
             environ_overrides={"REMOTE_USER": "test"},
         )
 
         assert 200 == response.status_code, response.data
-        assert {
-            "BigQuery Console": "https://console.cloud.google.com/bigquery?j=TEST_JOB_ID"
-        } == response.json
+        assert {"Google Custom": "http://google.com/custom_base_link?search=TEST_LINK_VALUE"} == response.json
 
     @mock_plugin_manager(plugins=[])
     def test_should_respond_200_missing_xcom(self):
         response = self.client.get(
-            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_QUERY/links",
+            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_LINK/links",
             environ_overrides={"REMOTE_USER": "test"},
         )
 
         assert 200 == response.status_code, response.data
-        assert {"BigQuery Console": None} == response.json
+        assert {"Google Custom": None} == response.json
 
     @mock_plugin_manager(plugins=[])
     def test_should_respond_200_multiple_links(self):
         XCom.set(
-            key="job_id_path",
-            value=["TEST_JOB_ID_1", "TEST_JOB_ID_2"],
-            task_id="TEST_MULTIPLE_QUERY",
+            key="search_query",
+            value=["TEST_LINK_VALUE_1", "TEST_LINK_VALUE_2"],
+            task_id="TEST_MULTIPLE_LINK",
             dag_id=self.dag.dag_id,
             run_id="TEST_DAG_RUN_ID",
         )
         response = self.client.get(
-            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_MULTIPLE_QUERY/links",
+            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_MULTIPLE_LINK/links",
             environ_overrides={"REMOTE_USER": "test"},
         )
 
         assert 200 == response.status_code, response.data
         assert {
-            "BigQuery Console #1": "https://console.cloud.google.com/bigquery?j=TEST_JOB_ID_1",
-            "BigQuery Console #2": "https://console.cloud.google.com/bigquery?j=TEST_JOB_ID_2",
+            "BigQuery Console #1": "https://console.cloud.google.com/bigquery?j=TEST_LINK_VALUE_1",
+            "BigQuery Console #2": "https://console.cloud.google.com/bigquery?j=TEST_LINK_VALUE_2",
         } == response.json
 
     @mock_plugin_manager(plugins=[])
     def test_should_respond_200_multiple_links_missing_xcom(self):
         response = self.client.get(
-            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_MULTIPLE_QUERY/links",
+            "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_MULTIPLE_LINK/links",
             environ_overrides={"REMOTE_USER": "test"},
         )
 
@@ -209,7 +209,7 @@ def get_link(self, operator, dttm):
 
         class S3LogLink(BaseOperatorLink):
             name = "S3"
-            operators = [BigQueryExecuteQueryOperator]
+            operators = [CustomOperator]
 
             def get_link(self, operator, dttm):
                 return (
@@ -228,16 +228,16 @@ class AirflowTestPlugin(AirflowPlugin):
 
         with mock_plugin_manager(plugins=[AirflowTestPlugin]):
             response = self.client.get(
-                "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_QUERY/links",
+                "/api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/taskInstances/TEST_SINGLE_LINK/links",
                 environ_overrides={"REMOTE_USER": "test"},
             )
 
             assert 200 == response.status_code, response.data
             assert {
-                "BigQuery Console": None,
+                "Google Custom": None,
                 "Google": "https://www.google.com",
                 "S3": (
                     "https://s3.amazonaws.com/airflow-logs/"
-                    "TEST_DAG_ID/TEST_SINGLE_QUERY/2020-01-01T00%3A00%3A00%2B00%3A00"
+                    "TEST_DAG_ID/TEST_SINGLE_LINK/2020-01-01T00%3A00%3A00%2B00%3A00"
                 ),
             } == response.json

tests/api_connexion/schemas/test_connection_schema.py

@@ -167,7 +167,7 @@ def test_serialize(self, session):
             schema="testschema",
             port=80,
             password="test-password",
-            extra="{'key':'string'}",
+            extra='{"key": "string"}',
         )
         session.add(connection_model)
         session.commit()
@@ -181,7 +181,7 @@ def test_serialize(self, session):
             "login": "login",
             "schema": "testschema",
             "port": 80,
-            "extra": "{'key':'string'}",
+            "extra": '{"key": "string"}',
         }
 
     def test_deserialize(self):

tests/deprecations_ignore.yml

@@ -16,21 +16,6 @@
 # under the License.
 ---
 
-# API
-- tests/api_connexion/endpoints/test_connection_endpoint.py::TestGetConnection::test_should_respond_200
-- tests/api_connexion/endpoints/test_connection_endpoint.py::TestPatchConnection::test_patch_should_respond_200
-- tests/api_connexion/endpoints/test_dag_run_endpoint.py::TestSetDagRunNote::test_should_respond_200_with_anonymous_user
-- tests/api_connexion/endpoints/test_extra_link_endpoint.py::TestGetExtraLinks::test_should_raise_403_forbidden
-- tests/api_connexion/endpoints/test_extra_link_endpoint.py::TestGetExtraLinks::test_should_respond_200
-- tests/api_connexion/endpoints/test_extra_link_endpoint.py::TestGetExtraLinks::test_should_respond_200_missing_xcom
-- tests/api_connexion/endpoints/test_extra_link_endpoint.py::TestGetExtraLinks::test_should_respond_200_multiple_links
-- tests/api_connexion/endpoints/test_extra_link_endpoint.py::TestGetExtraLinks::test_should_respond_200_multiple_links_missing_xcom
-- tests/api_connexion/endpoints/test_extra_link_endpoint.py::TestGetExtraLinks::test_should_respond_200_support_plugins
-- tests/api_connexion/endpoints/test_extra_link_endpoint.py::TestGetExtraLinks::test_should_respond_404
-- tests/api_connexion/endpoints/test_task_instance_endpoint.py::TestGetTaskInstancesBatch::test_should_respond_200
-- tests/api_connexion/schemas/test_connection_schema.py::TestConnectionSchema::test_serialize
-
-
 # CLI
 # https://github.com/apache/airflow/issues/39199
 - tests/cli/commands/test_kubernetes_command.py::TestGenerateDagYamlCommand::test_generate_dag_yaml

tests/test_utils/mock_operators.py

@@ -139,6 +139,8 @@ def get_link(self, operator, *, ti_key):
         search_query = XCom.get_one(
             task_id=ti_key.task_id, dag_id=ti_key.dag_id, run_id=ti_key.run_id, key="search_query"
         )
+        if not search_query:
+            return None
         return f"http://google.com/custom_base_link?search={search_query}"