Refactor ti clear next method kwargs tests

[ReadytoRocc]

related: #19120

Refactoring tests introduced in #19183.

[uranusjr]

It's probably better to refactor using paramtrize instead, something like

def _failure():
    raise AirflowException

def _skip():
    raise AirflowSkipException

# Others...

@pytest.mark.parametrize(
    "state, func",
    [
        (State.FAILED, _failure),
        (State.SKIPPED, _skip),
        # Others...
    ],
)
def test_task_wipes_next_fields(self, session, dag_maker, state, func):
    with dag_maker("test_deferred_method_clear"):
        task = PythonOperator(
            ...,
            python_callable=func,
        )
    ...

[ReadytoRocc]

Thanks @uranusjr. We are utilizing @pytest.mark.parametrize on lines L485-L488 for the state var. @jedcunningham suggested refactoring these tests to use a single run() function that takes state as an input.

[jedcunningham]

@ReadytoRocc, I think @uranusjr's suggestion is even better fwiw.

[ReadytoRocc]

Thanks @jedcunningham and @uranusjr. Refactored as suggested.

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

[jedcunningham]

@ReadytoRocc, geez sorry, didn't realize this never got merged! Can you rebase on main when you get the chance?

[ReadytoRocc]

Thanks. @jedcunningham and @uranusjr I needed to slightly modify some of the code due to a mypy check that was not present when this PR was originally opened. Please review again before merging. Thanks!

[dstandish]

if you remove exception_type as a param in _raise_af_exception you can also remove op_kwargs

[ReadytoRocc]

Thanks @dstandish, will do.

[dstandish]

Suggested change

	Test that ensures that tasks wipe their next_method and next_kwargs for the configured states:
	Test that ensures that tasks wipe their next_method and next_kwargs when the TI enters one of the following states:

[ReadytoRocc]

@jedcunningham @uranusjr I needed to slightly modify some of the code due to a mypy check that was not present when this PR was originally opened. Please review again and let me know if any changes are needed before merging. Thanks!

[uranusjr]

Suggested change

	def _raise_af_exception(exception_type):
	def _raise_if_exception(exception_type):

Probably? Or does af stand for Airflow?

[ReadytoRocc]

af stands for Airflow. However, _raise_if_exception better describes the function. I will update this, thanks.

[uranusjr]

This does not correctly cover the AirflowRescheduleException(timezone.utcnow()) case. It’s probably better to change exception_type to exception instead and always instantiate to avoid this subtle issue caused by exception_type may be either a type or an instance.

[ReadytoRocc]

Thanks for the feedback, I will rename exception_type to exception. I will edit as follows:

if state in [State.FAILED, State.UP_FOR_RETRY]:
    with pytest.raises(exception):
        ti.run()
else:
    ti.run()

This ensures that the exception is tied to whatever state is being checked.

The reason I did not include AirflowRescheduleException(timezone.utcnow()) is because that exception is caught during ti.run().