Repository: apache/airflow
File: airflow-core/tests/unit/dag_processing/test_processor.py (Python)

Code before:
    assert stat.import_errors == 0


def test_import_error_updates_timestamps(session):
    """last_finish_time should be updated when parsing a dag file results in import errors."""
    finish_time = timezone.utcnow()

    stat = process_parse_results(
        run_duration=1.5,
        finish_time=finish_time,
        run_count=2,
        bundle_name="test-bundle",
        bundle_version="v1",
        parsing_result=None,
        session=session,
        is_callback_only=False,
    )

    assert stat.last_finish_time == finish_time
    assert stat.run_count == 3
    assert stat.import_errors == 1


class TestExecuteCallbacks:
    def test_execute_callbacks_locks_bundle_version(self):
        request = DagCallbackRequest(
            filepath="test.py",
            dag_id="test_dag",
            run_id="test_run",
            bundle_name="testing",
            bundle_version="some_commit_hash",
            is_failure_callback=False,
            msg=None,
        )
        log = structlog.get_logger()
        dagbag = MagicMock()

        with (
            patch("airflow.dag_processing.processor.BundleVersionLock") as mock_lock,
            patch("airflow.dag_processing.processor._execute_dag_callbacks") as mock_execute,
        ):
            _execute_callbacks(dagbag, [request], log)

        mock_lock.assert_called_once_with(bundle_name="testing", bundle_version="some_commit_hash")
        mock_lock.return_value.__enter__.assert_called_once()
        mock_lock.return_value.__exit__.assert_called_once()
        mock_execute.assert_called_once_with(dagbag, request, log)


class TestExecuteDagCallbacks:
    """Test the _execute_dag_callbacks function with context_from_server"""

Review comment (suggestion):
```suggestion
        callbacks = [DagCallbackRequest(
```

Code after:
    assert stat.import_errors == 0


def test_import_error_updates_timestamps(session):
    """last_finish_time should be updated when parsing a dag file results in import errors."""
    finish_time = timezone.utcnow()

    stat = process_parse_results(
        run_duration=1.5,
        finish_time=finish_time,
        run_count=2,
        bundle_name="test-bundle",
        bundle_version="v1",
        parsing_result=None,
        session=session,
        is_callback_only=False,
    )

    assert stat.last_finish_time == finish_time
    assert stat.run_count == 3
    assert stat.import_errors == 1


class TestExecuteCallbacks:
    def test_execute_callbacks_locks_bundle_version(self):
        callbacks = [
            DagCallbackRequest(
                filepath="test.py",
                dag_id="test_dag",
                run_id="test_run",
                bundle_name="testing",
                bundle_version="some_commit_hash",
                is_failure_callback=False,
                msg=None,
            )
        ]
        log = MagicMock(spec=FilteringBoundLogger)
        dagbag = MagicMock(spec=DagBag)

        with (
            patch("airflow.dag_processing.processor.BundleVersionLock") as mock_lock,
            patch("airflow.dag_processing.processor._execute_dag_callbacks") as mock_execute,
        ):
            _execute_callbacks(dagbag, callbacks, log)

        mock_lock.assert_called_once_with(bundle_name="testing", bundle_version="some_commit_hash")
        mock_lock.return_value.__enter__.assert_called_once()
        mock_lock.return_value.__exit__.assert_called_once()
        mock_execute.assert_called_once_with(dagbag, callbacks[0], log)