I want to test operator arguments.

[pyama86]

I want to test if the Operator I am using in my DAG is being called with the given arguments.
I have tried the following and know that it does not work.

    python_operator_mock = mocker.spy(PythonOperator, "__init__")
    print(python_operator_mock.call_args_list)

The error output is as follows.

        # Set upstream task defined by XComArgs passed to template fields of the operator.
        # BUT: only do this _ONCE_, not once for each class in the hierarchy
> if not instantiated_from_mapped and func == self.__init__. __wrapped__: # type: ignore[misc].
E AttributeError: 'function' object has no attribute '__wrapped__'

This error is output here.

airflow/airflow/models/baseoperator.py

Line 438 in 3237cb3

if not instantiated_from_mapped and func == self.__init__.__wrapped__: # type: ignore[misc]

Is it possible to somehow spy an Operator that inherits from BaseOperator?

[pyama86]

I have self-resolved this issue.

def side_effect(*args, **kwargs):
    return EmptyOperator(
        task_id=kwargs["task_id"],
        start_date=DateTime(2023, 6, 13, 1, 0, 0, tzinfo=timezone("UTC")),
    )


def test_create_dag(mocker):
    python_operator_mock = mocker.patch.object(
        common_distribution_report.PythonOperator,
        "__new__",
        side_effect=side_effect,
    )
    print(python_operator_mock.call_args_list)