Is it possible to `freeze` the `create_timestamp` ...
# ask-community
t
Is it possible to
freeze
the
create_timestamp
|
update_timestamp
when creating a run for testing purposes? I want to have it fixed but its always using the current timestamp
Copy code
with instance_for_test() as instance:
    for run in MOCK_RECORDS:
        cast(DagsterInstance, instance).add_run(
            PipelineRun(
                pipeline_name=run.pipeline_run.pipeline_name,
                run_id=run.pipeline_run.run_id,
                status=run.pipeline_run.status,
                )
            )
Copy code
[RunRecord(storage_id=3, pipeline_run=DagsterRun(pipeline_name='job_dependency', run_id='coo', run_config={}, mode=None, asset_selection=None, solid_selection=None, solids_to_execute=None, step_keys_to_execute=None, status=<DagsterRunStatus.SUCCESS: 'SUCCESS'>, tags={}, root_run_id=None, parent_run_id=None, pipeline_snapshot_id=None, execution_plan_snapshot_id=None, external_pipeline_origin=None, pipeline_code_origin=None), create_timestamp=datetime.datetime(2022, 12, 14, 9, 14, 20), update_timestamp=datetime.datetime(2022, 12, 14, 9, 14, 20), start_time=None, end_time=None), ... ]
I want to filter by said dates or is there another way?
j
we use a library called
pendulum
to do things like this. Specifically the
pendulum.test
context manager https://pendulum.eustace.io/docs/#testing here’s an example test file where we use it https://github.com/dagster-io/dagster/blob/master/python_modules/dagster/dagster_tests/daemon_sensor_tests/test_run_status_sensors.py
t
I tried freezing time with the
pendulum.test
approach but it didn’t seem to affect the database timestamps. I finally found something that works for me 🙂
Copy code
with instance_for_test() as instance:
    monkeypatch.setattr(
        instance,
        'get_run_records',
        lambda: [
            RunRecord(
                storage_id=1,
                pipeline_run=PipelineRun(
                    pipeline_name='pipeline',
                    run_id='run_id',
                    status=DagsterRunStatus.SUCCESS,
                ),
                create_timestamp=datetime(2022, 1, 1, 10, 0),
                update_timestamp=datetime(2022, 1, 1, 10, 0),
            )
        ],
    )
But thank you anyways ✌️
Copy code
with pendulum.test(datetime):
    cast(DagsterInstance, instance).add_run(
    pipeline_run=PipelineRun(
        pipeline_name='pipeline',
        run_id='run_id',
        status=DagsterRunStatus.SUCCESS
        )
    )
does not seem to override the
create_timestamp
nor
update_timestamp
j
glad you figured it out! and good to know that pendulum doesn’t affect db times