fix: force DAG last modified time to UTC

[blinkseb]

Hello,

I've found an issue with the handling of the DAG file last modified time which can trigger an infinite parsing loop, when file_parsing_sort_mode is set to modified_time (which seems to be the default value).

os.path.getmtime() returns a timestamp, which is converted to a datetime using datetime.fromtimestamp. According to the documentation:

If optional argument tz is None or not specified, the timestamp is converted to the platform’s local date and time, and the returned datetime object is naive.

The resulting datetime is then made aware using timezone.make_aware using the user configured timezone, which is not guarantee to be the same as the platform's timezone. If that's not the case, we can end up with modified time in the future and trigger an infinite parsing loop.

The fix is to use datetime.utcfromtimestamp to convert the timestamp, to force the resulting datetime in UTC, and explicitly specify the UTC timezone to make_aware.

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[potiuk]

Yes. looks good. The mtime should be in UTC indeed (Except som Windows systems which we do not support anyway it seems) so that loooks good.
@ashb@uranusjr ?

[uranusjr]

Makes sense, except calling make_aware is cumbersome since we control the input. It is better to do datetime.fromtimestamp(ts, tz=timezone.utc) instead. This should be preferred over utcfromtimestamp, see Python documentation.

[blinkseb]

thanks for the feedback! I've updated the PR to use fromtimestamp with the tz parameter

[uranusjr]

Is it possible to have a test for this?

[blinkseb]

I've updated the existing test to trigger the issue by forcing the platform timezone to something different than UTC. Running the test reverting the fix fails (as expected) with this error:

> assert manager._file_path_queue == collections.deque()
E AssertionError: assert deque(['file_1.py']) == deque([])
E Left contains one more item: 'file_1.py'
E Full diff:
E - deque([])
E + deque(['file_1.py'])
tests/dag_processing/test_manager.py:485: AssertionError

Let me know if there's a better way to test that.

[potiuk]

@uranusjr ?

[boring-cyborg]

Awesome work, congrats on your first merged pull request!