Merge branch 'aarthy/freshness' into 'enterprise'

feat(monitors): freshness monitor with gap-based thresholds and schedule inference

See merge request dkinternal/testgen/dataops-testgen!416

Author: ci bot

testgen/__main__.py

@@ -428,7 +428,7 @@ def quick_start(
     click.echo("loading initial data")
     run_quick_start_increment(0)
     now_date = datetime.now(UTC)
-    time_delta = timedelta(days=-30) # 1 month ago
+    time_delta = timedelta(days=-35) # before the first monitor iteration (~34 days back)
     table_group_id = "0ea85e17-acbe-47fe-8394-9970725ad37d"
     test_suite_id = "9df7489d-92b3-49f9-95ca-512160d7896f"
 
@@ -449,16 +449,19 @@ def quick_start(
         run_quick_start_increment(iteration)
         run_test_execution(test_suite_id, run_date=run_date)
 
-    monitor_iterations = 42  # 3 weeks
+    monitor_iterations = 68  # ~5 weeks
     monitor_interval = timedelta(hours=12)
     monitor_test_suite_id = "823a1fef-9b6d-48d5-9d0f-2db9812cc318"
     # Round down to nearest 12-hour mark (12:00 AM or 12:00 PM UTC)
     now = datetime.now(UTC)
     nearest_12h_mark = now.replace(hour=12 if now.hour >= 12 else 0, minute=0, second=0, microsecond=0)
     monitor_run_date = nearest_12h_mark - monitor_interval * (monitor_iterations - 1)
+    weekday_morning_count = 0
     for iteration in range(1, monitor_iterations + 1):
         click.echo(f"Running monitor iteration: {iteration} / {monitor_iterations}")
-        run_monitor_increment(monitor_run_date, iteration)
+        if monitor_run_date.weekday() < 5 and monitor_run_date.hour < 12:
+            weekday_morning_count += 1
+        run_monitor_increment(monitor_run_date, iteration, weekday_morning_count)
         run_test_execution(monitor_test_suite_id, run_date=monitor_run_date)
         monitor_run_date += monitor_interval
 

testgen/commands/queries/execute_tests_query.py

@@ -1,16 +1,26 @@
 import dataclasses
 from collections.abc import Iterable
-from datetime import datetime
+from datetime import date, datetime
 from typing import TypedDict
 from uuid import UUID
 
+import pandas as pd
+
 from testgen.common import read_template_sql_file
 from testgen.common.clean_sql import concat_columns
 from testgen.common.database.database_service import get_flavor_service, get_tg_schema, replace_params
+from testgen.common.freshness_service import (
+    count_excluded_minutes,
+    get_schedule_params,
+    is_excluded_day,
+    resolve_holiday_dates,
+)
 from testgen.common.models.connection import Connection
+from testgen.common.models.scheduler import JobSchedule
 from testgen.common.models.table_group import TableGroup
 from testgen.common.models.test_definition import TestRunType, TestScope
 from testgen.common.models.test_run import TestRun
+from testgen.common.models.test_suite import TestSuite
 from testgen.common.read_file import replace_templated_functions
 from testgen.utils import to_sql_timestamp
 
@@ -49,6 +59,7 @@ class TestExecutionDef(InputParameters):
     skip_errors: int
     history_calculation: str
     custom_query: str
+    prediction: dict | str | None
     run_type: TestRunType
     test_scope: TestScope
     template: str
@@ -88,14 +99,27 @@ class TestExecutionSQL:
         "result_measure",
     )
 
-    def __init__(self, connection: Connection, table_group: TableGroup, test_run: TestRun):
+    def __init__(self, connection: Connection, table_group: TableGroup, test_suite: TestSuite, test_run: TestRun):
         self.connection = connection
         self.table_group = table_group
+        self.test_suite = test_suite
         self.test_run = test_run
         self.run_date = test_run.test_starttime
         self.flavor = connection.sql_flavor
         self.flavor_service = get_flavor_service(self.flavor)
 
+        self._exclude_weekends = bool(self.test_suite.predict_exclude_weekends)
+        self._holiday_dates: set[date] | None = None
+        self._schedule_tz: str | None = None
+        if test_suite.is_monitor:
+            schedule = JobSchedule.get(JobSchedule.kwargs["test_suite_id"].astext == str(test_suite.id))
+            self._schedule_tz = schedule.cron_tz or "UTC" if schedule else None
+            if test_suite.holiday_codes_list:
+                self._holiday_dates = resolve_holiday_dates(
+                    test_suite.holiday_codes_list,
+                    pd.DatetimeIndex([datetime(self.run_date.year - 1, 1, 1), datetime(self.run_date.year + 1, 12, 31)]),
+                )
+
     def _get_input_parameters(self, test_def: TestExecutionDef) -> str:
         return "; ".join(
             f"{field.name}={getattr(test_def, field.name)}"
@@ -135,8 +159,8 @@ def _get_params(self, test_def: TestExecutionDef | None = None) -> dict:
                 "BASELINE_SUM": test_def.baseline_sum,
                 "BASELINE_AVG": test_def.baseline_avg,
                 "BASELINE_SD": test_def.baseline_sd,
-                "LOWER_TOLERANCE": test_def.lower_tolerance or "NULL",
-                "UPPER_TOLERANCE": test_def.upper_tolerance or "NULL",
+                "LOWER_TOLERANCE": "NULL" if test_def.lower_tolerance in (None, "") else test_def.lower_tolerance,
+                "UPPER_TOLERANCE": "NULL" if test_def.upper_tolerance in (None, "") else test_def.upper_tolerance,
                 # SUBSET_CONDITION should be replaced after CUSTOM_QUERY
                 # since the latter may contain the former
                 "SUBSET_CONDITION": test_def.subset_condition or "1=1",
@@ -154,6 +178,32 @@ def _get_params(self, test_def: TestExecutionDef | None = None) -> dict:
                 "COLUMN_TYPE": test_def.column_type,
                 "INPUT_PARAMETERS": self._get_input_parameters(test_def),
             })
+
+            # Freshness exclusion params — computed per test at execution time
+            if test_def.test_type == "Freshness_Trend" and test_def.baseline_sum:
+                sched = get_schedule_params(test_def.prediction)
+                has_exclusions = self._exclude_weekends or sched.excluded_days or sched.window_start is not None
+                if has_exclusions:
+                    last_update = pd.Timestamp(test_def.baseline_sum)
+                    excluded = int(count_excluded_minutes(
+                        last_update, self.run_date, self._exclude_weekends, self._holiday_dates,
+                        tz=self._schedule_tz, excluded_days=sched.excluded_days,
+                        window_start=sched.window_start, window_end=sched.window_end,
+                    ))
+                    is_excl = 1 if is_excluded_day(
+                        pd.Timestamp(self.run_date), self._exclude_weekends, self._holiday_dates,
+                        tz=self._schedule_tz, excluded_days=sched.excluded_days,
+                        window_start=sched.window_start, window_end=sched.window_end,
+                    ) else 0
+                    params["EXCLUDED_MINUTES"] = excluded
+                    params["IS_EXCLUDED_DAY"] = is_excl
+                else:
+                    params["EXCLUDED_MINUTES"] = 0
+                    params["IS_EXCLUDED_DAY"] = 0
+            else:
+                params["EXCLUDED_MINUTES"] = 0
+                params["IS_EXCLUDED_DAY"] = 0
+
         return params
 
     def _get_query(
@@ -266,7 +316,7 @@ def aggregate_cat_tests(
                 td.measure_expression = f"COALESCE(CAST({measure} AS {varchar_type}) {concat_operator} '|', '{self.null_value}|')"
 
                 # For prediction mode, return -1 during training period
-                if td.history_calculation == "PREDICT" and (not td.lower_tolerance or not td.upper_tolerance):
+                if td.history_calculation == "PREDICT" and (td.lower_tolerance in (None, "") or td.upper_tolerance in (None, "")):
                     td.condition_expression = "'-1,'"
                 else:
                     condition = (

testgen/commands/run_quick_start.py

@@ -130,15 +130,18 @@ def _metric_cumulative_shift(iteration: int) -> tuple[float, float]:
     return discount, price
 
 
-def _get_monitor_params_mapping(run_date: datetime, iteration: int = 0) -> dict:
+def _get_monitor_params_mapping(run_date: datetime, iteration: int = 0, weekday_morning_count: int = 0) -> dict:
     # Volume: linear growth with jitter, spike at specific iteration for anomaly
-    if iteration == 37:
+    if iteration == 60:
         new_sales = 100
     else:
-        new_sales = random.randint(8, 12)  # noqa: S311
+        new_sales = random.randint(5, 15)  # noqa: S311
 
-    # Freshness: update every other iteration, late update for anomaly
-    is_update_suppliers_iter = (iteration % 2 == 0 and iteration != 38) or iteration == 39
+    # Freshness: weekday morning updates with 1-day outage after schedule goes active
+    is_weekday = run_date.weekday() < 5
+    is_morning = run_date.hour < 12
+    is_outage = weekday_morning_count == 21
+    is_update_suppliers_iter = is_weekday and is_morning and not is_outage
 
     # Metrics: compute deltas for discount and price shifts
     curr_discount, curr_price = _metric_cumulative_shift(iteration)
@@ -151,11 +154,11 @@ def _get_monitor_params_mapping(run_date: datetime, iteration: int = 0) -> dict:
         "ITERATION_NUMBER": iteration,
         "RUN_DATE": run_date,
         "NEW_SALES": new_sales,
-        "IS_ADD_CUSTOMER_COL_ITER": iteration == 29,
-        "IS_DELETE_CUSTOMER_COL_ITER": iteration == 36,
-        "IS_UPDATE_PRODUCT_ITER": not 14 < iteration < 18,
-        "IS_CREATE_RETURNS_TABLE_ITER": iteration == 32,
-        "IS_DELETE_CUSTOMER_ITER": iteration in (18, 22, 34),
+        "IS_ADD_CUSTOMER_COL_ITER": iteration == 47,
+        "IS_DELETE_CUSTOMER_COL_ITER": iteration == 58,
+        "IS_UPDATE_PRODUCT_ITER": not 24 < iteration < 28,
+        "IS_CREATE_RETURNS_TABLE_ITER": iteration == 52,
+        "IS_DELETE_CUSTOMER_ITER": iteration in (29, 36, 55),
         "IS_UPDATE_SUPPLIERS_ITER": is_update_suppliers_iter,
         "DISCOUNT_DELTA": discount_delta,
         "PRICE_DELTA": price_delta,
@@ -234,8 +237,8 @@ def run_quick_start_increment(iteration):
     setup_cat_tests(iteration)
 
 
-def run_monitor_increment(run_date, iteration):
-    params_mapping = _get_monitor_params_mapping(run_date, iteration)
+def run_monitor_increment(run_date, iteration, weekday_morning_count=0):
+    params_mapping = _get_monitor_params_mapping(run_date, iteration, weekday_morning_count)
     _prepare_connection_to_target_database(params_mapping)
 
     target_db_name = params_mapping["PROJECT_DB"]

testgen/commands/run_test_execution.py

@@ -84,7 +84,7 @@ def run_test_execution(test_suite_id: str | UUID, username: str | None = None, r
         data_chars = run_data_chars_refresh(connection, table_group, test_run.test_starttime)
         test_run.set_progress("data_chars", "Completed")
 
-        sql_generator = TestExecutionSQL(connection, table_group, test_run)
+        sql_generator = TestExecutionSQL(connection, table_group, test_suite, test_run)
 
         if test_suite.is_monitor:
             _sync_monitor_definitions(sql_generator)