0

I am trying to mock airflow.operators.python.get_current_context as follows:

@pytest.fixture
def _mock_get_current_context(mocker):
    mocker.patch(
        "airflow.operators.python.get_current_context",
        return_value={},
    )

This pattern works for all other functions I am mocking, for example requests.get, airflow.operators.trigger_dagrun.TriggerDagRunOperator.execute and requests.Response.content. However, when I call get_current_context() in a DAG task, I get the following error:

if not _CURRENT_CONTEXT:
    raise AirflowException(
        "Current context was requested but no context was found! "
        "Are you running within an airflow task?"
        )
E       airflow.exceptions.AirflowException: Current context was requested but no 
context was found! Are you running within an airflow task?

Indicating that the mocking did not work since the source code for get_current_context() looks like this:

    def get_current_context() -> Context:
        if not _CURRENT_CONTEXT:
            raise AirflowException(
                "Current context was requested but no context was found! "
                "Are you running within an airflow task?"
            )
        return _CURRENT_CONTEXT[-1]

Any ideas what can have gone wrong?

Casper Lindberg
  • 600
  • 1
  • 6
  • 21

0 Answers0