From a55375ea63a6bd0d03919e0d0fd53d1b34f296c2 Mon Sep 17 00:00:00 2001 From: Elad Barkay Date: Tue, 6 Oct 2026 13:12:00 +0300 Subject: [PATCH 1/4] test: pin run_started_at when each integration test call starts Tests build their data dates from utcnow(), but Elementary's buckets use dbt's run_started_at, which is set only after the data is seeded. When UTC midnight passed during seeding, the two disagreed by a day and date-based anomaly tests failed. DbtProject.test now passes custom_run_started_at, taken at the start of each call before seeding, unless the test already sets it. Each call gets its own time, so tests that run dbt several times still see time move forward. Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/dbt_project.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/integration_tests/tests/dbt_project.py b/integration_tests/tests/dbt_project.py index af32b9ca5..778806eb6 100644 --- a/integration_tests/tests/dbt_project.py +++ b/integration_tests/tests/dbt_project.py @@ -1,6 +1,7 @@ import json import os from contextlib import contextmanager, nullcontext +from datetime import datetime from pathlib import Path from tempfile import NamedTemporaryFile from typing import Any, Dict, Generator, List, Literal, Optional, Union, overload @@ -257,8 +258,12 @@ def test( ) -> Union[Dict[str, Any], List[Dict[str, Any]]]: if columns and test_column: raise ValueError("You can't specify both 'columns' and 'test_column'.") - test_vars = test_vars or {} + test_vars = dict(test_vars or {}) test_vars["elementary_enabled"] = elementary_enabled + # Pin run_started_at to before seeding, close to when the test generated its + # data, so data dates and Elementary's buckets agree even if UTC midnight + # passes while the data is being seeded. + test_vars.setdefault("custom_run_started_at", datetime.utcnow().isoformat()) test_id = test_id.replace("[", "_").replace("]", "_") if not table_name: From 36047ed35584b75c0edb5886c7d23711e45fb3f0 Mon Sep 17 00:00:00 2001 From: Elad Barkay Date: Tue, 6 Oct 2026 13:15:57 +0300 Subject: [PATCH 2/4] test: shorten run_started_at comment Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/dbt_project.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/integration_tests/tests/dbt_project.py b/integration_tests/tests/dbt_project.py index 778806eb6..020e27db4 100644 --- a/integration_tests/tests/dbt_project.py +++ b/integration_tests/tests/dbt_project.py @@ -260,9 +260,7 @@ def test( raise ValueError("You can't specify both 'columns' and 'test_column'.") test_vars = dict(test_vars or {}) test_vars["elementary_enabled"] = elementary_enabled - # Pin run_started_at to before seeding, close to when the test generated its - # data, so data dates and Elementary's buckets agree even if UTC midnight - # passes while the data is being seeded. + # Set before seeding so buckets match the data dates across UTC midnight. test_vars.setdefault("custom_run_started_at", datetime.utcnow().isoformat()) test_id = test_id.replace("[", "_").replace("]", "_") From 49148eb4c5a62d7cbc56f8d7a201e16aad1ae502 Mon Sep 17 00:00:00 2001 From: Elad Barkay Date: Tue, 6 Oct 2026 18:34:05 +0300 Subject: [PATCH 3/4] test: mock the harness clock to check run_started_at across midnight Co-Authored-By: Claude Opus 5.5 --- .../tests/test_run_started_at.py | 134 ++++++++++++++++++ 1 file changed, 134 insertions(+) create mode 100644 integration_tests/tests/test_run_started_at.py diff --git a/integration_tests/tests/test_run_started_at.py b/integration_tests/tests/test_run_started_at.py new file mode 100644 index 000000000..bec988302 --- /dev/null +++ b/integration_tests/tests/test_run_started_at.py @@ -0,0 +1,134 @@ +from datetime import datetime, time, timedelta +from typing import Any, Dict, List +from unittest import mock + +import pytest +from data_generator import DATE_FORMAT, generate_dates +from dbt_project import DbtProject +from parametrization import Parametrization + +TIMESTAMP_COLUMN = "updated_at" +DBT_TEST_NAME = "elementary.volume_anomalies" +DBT_TEST_ARGS = {"timestamp_column": TIMESTAMP_COLUMN} +DATA = [{TIMESTAMP_COLUMN: "2026-01-01 00:00:00"}] + + +def _freeze_harness_clock(monkeypatch, now: datetime) -> mock.Mock: + """Make DbtProject.test read `now` as the UTC time. dbt keeps the real clock.""" + clock = mock.Mock(wraps=datetime) + clock.utcnow.return_value = now + monkeypatch.setattr("dbt_project.datetime", clock) + return clock + + +@pytest.fixture +def dbt_test_vars(monkeypatch, dbt_project: DbtProject) -> List[Dict[str, Any]]: + """Stub seeding, dbt and result reading, and collect the vars of each dbt call.""" + calls: List[Dict[str, Any]] = [] + + def fake_dbt_test(**kwargs): + calls.append(kwargs["vars"]) + return True + + monkeypatch.setattr(dbt_project, "seed", lambda data, table_name: None) + monkeypatch.setattr(dbt_project.dbt_runner, "test", fake_dbt_test) + monkeypatch.setattr( + dbt_project, "_read_single_test_result", lambda test_id: {"status": "pass"} + ) + return calls + + +def test_run_started_at_is_taken_before_seeding( + test_id: str, dbt_project: DbtProject, monkeypatch, dbt_test_vars +): + before_midnight = datetime(2026, 1, 1, 23, 59, 59) + clock = _freeze_harness_clock(monkeypatch, before_midnight) + + def seed_past_midnight(data, table_name): + clock.utcnow.return_value = before_midnight + timedelta(minutes=5) + + monkeypatch.setattr(dbt_project, "seed", seed_past_midnight) + + dbt_project.test(test_id, DBT_TEST_NAME, DBT_TEST_ARGS, data=DATA) + + assert [v["custom_run_started_at"] for v in dbt_test_vars] == [ + before_midnight.isoformat() + ] + + +def test_run_started_at_from_caller_is_kept( + test_id: str, dbt_project: DbtProject, monkeypatch, dbt_test_vars +): + _freeze_harness_clock(monkeypatch, datetime(2026, 1, 2)) + caller_run_started_at = datetime(2000, 1, 1).isoformat() + + dbt_project.test( + test_id, + DBT_TEST_NAME, + DBT_TEST_ARGS, + data=DATA, + test_vars={"custom_run_started_at": caller_run_started_at}, + ) + + assert [v["custom_run_started_at"] for v in dbt_test_vars] == [ + caller_run_started_at + ] + + +def test_run_started_at_is_taken_per_call( + test_id: str, dbt_project: DbtProject, monkeypatch, dbt_test_vars +): + first_call_at = datetime(2026, 1, 1, 12) + second_call_at = first_call_at + timedelta(hours=1) + clock = _freeze_harness_clock(monkeypatch, first_call_at) + # One dict reused across calls must not freeze the time of the first call. + # Not empty, since `{} or {}` would hand the harness a new dict anyway. + test_vars: Dict[str, Any] = {"debug_logs": True} + + dbt_project.test( + test_id, DBT_TEST_NAME, DBT_TEST_ARGS, data=DATA, test_vars=test_vars + ) + clock.utcnow.return_value = second_call_at + dbt_project.test( + test_id, DBT_TEST_NAME, DBT_TEST_ARGS, data=DATA, test_vars=test_vars + ) + + assert [v["custom_run_started_at"] for v in dbt_test_vars] == [ + first_call_at.isoformat(), + second_call_at.isoformat(), + ] + assert test_vars == {"debug_logs": True} + + +# Runs dbt for real. The harness clock reads one minute before the last UTC +# midnight, while dbt runs on the real clock, a day later: the same as a CI run +# where midnight passes while seeding. The buckets must follow the harness clock. +@Parametrization.autodetect_parameters() +@Parametrization.case( + name="spike_in_unfinished_bucket", spike_days_ago=0, expected_status="pass" +) +@Parametrization.case( + name="spike_in_last_full_bucket", spike_days_ago=1, expected_status="fail" +) +def test_buckets_follow_harness_clock_across_midnight( + test_id: str, + dbt_project: DbtProject, + monkeypatch, + spike_days_ago: int, + expected_status: str, +): + last_midnight = datetime.combine(datetime.utcnow().date(), time.min) + harness_now = last_midnight - timedelta(minutes=1) + _freeze_harness_clock(monkeypatch, harness_now) + + harness_today = harness_now.date() + spike_date = harness_today - timedelta(days=spike_days_ago) + data: List[Dict[str, Any]] = [ + {TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT)} + for cur_date in generate_dates(base_date=harness_today) + ] + data += [{TIMESTAMP_COLUMN: spike_date.strftime(DATE_FORMAT)}] * 10 + + test_args = {**DBT_TEST_ARGS, "backfill_days": 1} + test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data) + assert test_result["status"] == expected_status From 8a790d0ef838b6d66956ae374943b0b045dde80e Mon Sep 17 00:00:00 2001 From: Elad Barkay Date: Wed, 7 Oct 2026 02:19:31 +0300 Subject: [PATCH 4/4] test: shorten midnight test names to fit Postgres's 63-char limit The seeded table is named after the test id, and the old names made it 79 characters, so dbt seed failed on Postgres. Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_run_started_at.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/integration_tests/tests/test_run_started_at.py b/integration_tests/tests/test_run_started_at.py index bec988302..760c9cf21 100644 --- a/integration_tests/tests/test_run_started_at.py +++ b/integration_tests/tests/test_run_started_at.py @@ -103,14 +103,13 @@ def test_run_started_at_is_taken_per_call( # Runs dbt for real. The harness clock reads one minute before the last UTC # midnight, while dbt runs on the real clock, a day later: the same as a CI run # where midnight passes while seeding. The buckets must follow the harness clock. +# The seeded table is named after the test id: Postgres allows 63 characters. @Parametrization.autodetect_parameters() @Parametrization.case( - name="spike_in_unfinished_bucket", spike_days_ago=0, expected_status="pass" + name="unfinished_bucket", spike_days_ago=0, expected_status="pass" ) -@Parametrization.case( - name="spike_in_last_full_bucket", spike_days_ago=1, expected_status="fail" -) -def test_buckets_follow_harness_clock_across_midnight( +@Parametrization.case(name="last_full_bucket", spike_days_ago=1, expected_status="fail") +def test_buckets_follow_harness_clock( test_id: str, dbt_project: DbtProject, monkeypatch,