Remove dead code from airflow-core and task-sdk unit tests (non-db pass)

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_import_error.py

@@ -52,21 +52,6 @@
 BUNDLE_NAME = "testing"
 
 
-@pytest.fixture
-@provide_session
-def permitted_dag_model(testing_dag_bundle, session: Session = NEW_SESSION) -> DagModel:
-    dag_model = DagModel(
-        fileloc=FILENAME1,
-        relative_fileloc=FILENAME1,
-        dag_id="dag_id1",
-        is_paused=False,
-        bundle_name=BUNDLE_NAME,
-    )
-    session.add(dag_model)
-    session.commit()
-    return dag_model
-
-
 @pytest.fixture
 @provide_session
 def permitted_dag_model_all(testing_dag_bundle, session: Session = NEW_SESSION) -> set[str]:

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -757,20 +757,6 @@ def one_task_with_mapped_tis(self, dag_maker, session):
             },
         )
 
-    @pytest.fixture
-    def one_task_with_single_mapped_ti(self, dag_maker, session):
-        self.create_dag_runs_with_mapped_tasks(
-            dag_maker,
-            session,
-            dags={
-                "mapped_tis": {
-                    "success": 1,
-                    "failed": 0,
-                    "running": 0,
-                },
-            },
-        )
-
     @pytest.fixture
     def one_task_with_many_mapped_tis(self, dag_maker, session):
         self.create_dag_runs_with_mapped_tasks(

airflow-core/tests/unit/assets/test_manager.py

@@ -18,7 +18,6 @@
 from __future__ import annotations
 
 import concurrent.futures
-import itertools
 import logging
 from collections import Counter
 from typing import TYPE_CHECKING
@@ -39,7 +38,7 @@
     DagScheduleAssetAliasReference,
     DagScheduleAssetReference,
 )
-from airflow.models.dag import DAG, DagModel
+from airflow.models.dag import DagModel
 from airflow.sdk.definitions.asset import Asset
 
 from tests_common.test_utils.config import conf_vars
@@ -66,13 +65,6 @@ def mock_task_instance():
     return None
 
 
-def create_mock_dag():
-    for dag_id in itertools.count(1):
-        mock_dag = mock.Mock(spec=DAG)
-        mock_dag.dag_id = dag_id
-        yield mock_dag
-
-
 class TestAssetManager:
     def test_register_asset_change_asset_doesnt_exist(self, mock_task_instance):
         mock_task_instance = mock.Mock()

airflow-core/tests/unit/cli/commands/test_info_command.py

@@ -23,7 +23,6 @@
 
 import httpx
 import pytest
-from rich.console import Console
 
 from airflow.cli import cli_parser
 from airflow.cli.commands import info_command
@@ -34,13 +33,6 @@
 from tests_common.test_utils.config import conf_vars
 
 
-def capture_show_output(instance):
-    console = Console()
-    with console.capture() as capture:
-        instance.info(console)
-    return capture.get()
-
-
 class TestPiiAnonymizer:
     def setup_method(self) -> None:
         self.instance = info_command.PiiAnonymizer()

airflow-core/tests/unit/cli/commands/test_task_command.py

@@ -22,9 +22,8 @@
 import json
 import logging
 import os
-import shutil
 from argparse import ArgumentParser
-from contextlib import contextmanager, redirect_stdout
+from contextlib import redirect_stdout
 from pathlib import Path
 from typing import TYPE_CHECKING
 from unittest import mock
@@ -45,7 +44,7 @@
 from airflow.providers.standard.operators.bash import BashOperator
 from airflow.serialization.serialized_objects import DagSerialization, LazyDeserializedDAG
 from airflow.utils.session import create_session
-from airflow.utils.state import State, TaskInstanceState
+from airflow.utils.state import State
 from airflow.utils.types import DagRunTriggeredByType, DagRunType
 
 from tests_common.test_utils.config import conf_vars
@@ -70,13 +69,6 @@ def reset(dag_id):
         session.execute(delete(SerializedDagModel).where(SerializedDagModel.dag_id == dag_id))
 
 
-@contextmanager
-def move_back(old_path, new_path):
-    shutil.move(old_path, new_path)
-    yield
-    shutil.move(new_path, old_path)
-
-
 class TestCliTasks:
     run_id = "TEST_RUN_ID"
     dag_id = "example_python_operator"
@@ -542,12 +534,6 @@ def test_task_render_for_multi_line_properties(self, dag_maker):
         assert "# property: bash_command" in output.split("\n")
 
 
-def _set_state_and_try_num(ti, session):
-    ti.state = TaskInstanceState.QUEUED
-    ti.try_number += 1
-    session.commit()
-
-
 class TestLogsfromTaskRunCommand:
     def setup_method(self) -> None:
         self.dag_id = "test_logging_dag"

airflow-core/tests/unit/cli/conftest.py

@@ -28,10 +28,8 @@
 
 from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.stream_capture_manager import (
-    CombinedCaptureManager,
     StderrCaptureManager,
     StdoutCaptureManager,
-    StreamCaptureManager,
 )
 
 # Create custom executors here because conftest is imported first
@@ -82,21 +80,3 @@ def stderr_capture(request):
     """Fixture that captures stderr only."""
     request.getfixturevalue("caplog")
     return StderrCaptureManager()
-
-
-@pytest.fixture
-def stream_capture(request):
-    """Fixture that returns a configurable stream capture manager."""
-
-    def _capture(stdout=True, stderr=False):
-        request.getfixturevalue("caplog")
-        return StreamCaptureManager(capture_stdout=stdout, capture_stderr=stderr)
-
-    return _capture
-
-
-@pytest.fixture
-def combined_capture(request):
-    """Fixture that captures both stdout and stderr."""
-    request.getfixturevalue("caplog")
-    return CombinedCaptureManager()

airflow-core/tests/unit/plugins/test_plugins_manager.py

@@ -17,7 +17,6 @@
 # under the License.
 from __future__ import annotations
 
-import contextlib
 import importlib
 import inspect
 import logging
@@ -56,21 +55,6 @@ def _clean_listeners():
     get_listener_manager().clear()
 
 
-@pytest.fixture
-def mock_metadata_distribution(mocker):
-    @contextlib.contextmanager
-    def wrapper(*args, **kwargs):
-        if sys.version_info < (3, 12):
-            patch_fq = "importlib_metadata.distributions"
-        else:
-            patch_fq = "importlib.metadata.distributions"
-
-        with mock.patch(patch_fq, *args, **kwargs) as m:
-            yield m
-
-    return wrapper
-
-
 class TestPluginsManager:
     @pytest.fixture(autouse=True)
     def clean_plugins(self):

scripts/ci/prek/known_provide_session_positional.txt

@@ -54,7 +54,7 @@ airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py::1
 airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_tags.py::1
 airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py::1
 airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_event_logs.py::1
-airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_import_error.py::8
+airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_import_error.py::7
 airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py::1
 airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_monitor.py::2
 airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_pools.py::2