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?