execute_in_process with FakeDatetimes do not work

[derHeinzer]

Dagster version

1.7.4

What's the issue?

I am trying to test a sensor, that checks if a job has run successfully today. The definition of "today" may depend on timezones. Hence I want to fake a situation where a job runs close to a day switch and in one tz has already run but not in the other. However the creation_timestamp of the RunRecord wont be faked correctly.

What did you expect to happen?

The first assertion in test_job_run_today_with_tz_gap should pass, but it does not. The creation_timestamp of the run record will be the current datetime e.g. FakeDatetime(2024, 5, 16, 14, 47, 34)

How to reproduce?

# standard library imports
import pytz
from datetime import datetime
from unittest import TestCase

# dagster imports
from dagster import DagsterInstance, job, op, RunsFilter, DagsterRunStatus
from freezegun import freeze_time


# standard library imports
from typing import Optional


def midnight_today(timezone: str = "UTC") -> datetime:
    return datetime.now(tz=pytz.timezone(timezone)).replace(
        hour=0, minute=0, second=0, microsecond=0
    )


def has_job_succeeded_today(
        job_name: str,
        instance: DagsterInstance,
        timezone: Optional[str] = "UTC",
):
    run = instance.get_run_records(
        RunsFilter(
            job_name=job_name,
            statuses=[DagsterRunStatus.SUCCESS],
            created_after=midnight_today(timezone),
        ),
        limit=1,
    )
    return len(run) > 0


class HasJobSucceededToday(TestCase):
    def setUp(self):
        self.instance = DagsterInstance.ephemeral()

        @op()
        def do_nothing():
            return None

        @job(
            name="do_nothing_job",
        )
        def do_nothing_job():
            do_nothing()

        self.do_nothing_job = do_nothing_job

    def test_job_never_run(self):
        self.assertFalse(has_job_succeeded_today("do_nothing_job", self.instance))

    @freeze_time(datetime(1983, 10, 21, 12, 0, 0, tzinfo=pytz.utc))
    def test_job_run_today(self):
        self.do_nothing_job.execute_in_process(instance=self.instance)
        self.assertTrue(has_job_succeeded_today("do_nothing_job", self.instance))

    #@freeze_time(datetime(1983, 10, 21, 0, 30, 0, tzinfo=pytz.timezone("Europe/Berlin")))
    @freeze_time("1983-10-21 00:30:00", tz_offset=2)
    def test_job_run_today_with_tz_gap(self):
        self.do_nothing_job.execute_in_process(instance=self.instance)
        self.assertFalse(has_job_succeeded_today("do_nothing_job", self.instance))
        self.assertTrue(has_job_succeeded_today("do_nothing_job", self.instance, timezone="Europe/Berlin"))

Deployment type

None

Deployment details

No response

Additional information

No response

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization.