fix: Type mismatch for DateInterval in latest only operator

[leakec]

The call to DateInterval in latest_only_opeartor is currently passing in a datetime where it should be passing in a DateTime. This PR fixes that issue.

closes: #50911

[leakec]

I've tested the fix out, and it resolves the issue I've run into. However, I'm having issues reproducing it in a unit test. I've made an attempt in the PR, but need some help from someone more familiar with airflow unit testing. I've kept the fix and unit test on separate commits to hopefully make things simpler.

Tagging @gopidesupavan for feedback on the PR itself and the unit test.

[gopidesupavan]

@leakec LGTM, can you please fix the static checks and compat tests?

[gopidesupavan]

please look at quickstart for reference https://github.com/apache/airflow/blob/main/contributing-docs/03_contributors_quick_start.rst

[leakec]

@leakec LGTM, can you please fix the static checks and compat tests?

@gopidesupavan Sure thing. I've added a few more commits to patch the issues, and now it passes all the tests. I do still need some help with the unit test though. The one I added passes, but the problem is it doesn't fail before the fix I implemented, i.e., it passes with both versions, so it doesn't really replicate the problem I saw when running this DAG in a normal airflow instance:

from airflow import DAG
from datetime import datetime
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.latest_only import LatestOnlyOperator

with DAG(
    "test_job",
    description="Job to test latest only operator.",
    schedule="* * * * *",
    start_date=datetime(2024, 12, 11),
    catchup=False,
):

    # latest only operator
    task1 = LatestOnlyOperator(task_id="latest_only")
    task2 = BashOperator(bash_command="echo test", task_id="task2")

    task1 >> task2

I think there is likely a mistake I'm making when setting up or running tasks in the unit test, but I'm not familiar enough with the airflow testing system to know where. I'll continue reading up on it some more in my free time, but if you see anything obviously amiss, please let me know.

[gopidesupavan]

@leakec LGTM, can you please fix the static checks and compat tests?

@gopidesupavan Sure thing. I've added a few more commits to patch the issues, and now it passes all the tests. I do still need some help with the unit test though. The one I added passes, but the problem is it doesn't fail before the fix I implemented, i.e., it passes with both versions, so it doesn't really replicate the problem I saw when running this DAG in a normal airflow instance:

from airflow import DAG
from datetime import datetime
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.latest_only import LatestOnlyOperator

with DAG(
    "test_job",
    description="Job to test latest only operator.",
    schedule="* * * * *",
    start_date=datetime(2024, 12, 11),
    catchup=False,
):

    # latest only operator
    task1 = LatestOnlyOperator(task_id="latest_only")
    task2 = BashOperator(bash_command="echo test", task_id="task2")

    task1 >> task2

I think there is likely a mistake I'm making when setting up or running tasks in the unit test, but I'm not familiar enough with the airflow testing system to know where. I'll continue reading up on it some more in my free time, but if you see anything obviously amiss, please let me know.

the test is fine i just triggered same one , it looks fine, it failed in my case.

[gopidesupavan]

cool , @leakec good work thanks for fixing this.

[gopidesupavan]

Failures are not related merging it now.

[leakec]

Of course, thanks for all the help with the review and PR!