Merge remote-tracking branch 'origin/enterprise' into feat/TG-1047-deduplicate-run-tables

Author: rboni-dk

testgen/commands/test_thresholds_prediction.py

@@ -89,7 +89,7 @@ def run(self) -> None:
 
             # Freshness update events are fetched as secondary data only when the suite
             # is a monitor — Volume_Trend / Metric_Trend in monitor suites couple to the
-            # Freshness_Trend signal to avoid stairstep false positives. 
+            # Freshness_Trend signal to avoid stairstep false positives.
             freshness_updates_by_table: dict[tuple[str, str], list[str]] = (
                 self._fetch_freshness_updates_by_table() if self.test_suite.is_monitor else {}
             )
@@ -268,10 +268,16 @@ def compute_freshness_threshold(
                     window_end=schedule.window_end if has_window else None,
                 )
                 lower, upper = result.lower, result.upper
-                staleness = result.staleness
             except NotEnoughData:
                 pass  # Keep first-pass thresholds
 
+        # An active schedule certifies the cadence is regular, so the tight staleness
+        # threshold (median * factor) applies. Full-week schedules skip the recompute
+        # above and use the first-pass result — with no excluded days, those gaps are
+        # already in business minutes. Non-active stages keep staleness = None so the
+        # prediction SQL falls back to upper_tolerance (avoids false weekend anomalies).
+        staleness = result.staleness
+
         # Override upper threshold with schedule-based deadline (daily/weekly only)
         if schedule.frequency != "sub_daily":
             holiday_dates = resolve_holiday_dates(holiday_codes, history.index) if holiday_codes else None
@@ -344,7 +350,7 @@ def compute_volume_or_metric_threshold(
     This avoids the "stairstep" false-positive shape where inter-change plateaus collapse the SE estimate.
     The returned prediction JSON is augmented with `freshness_gated` and `baseline_value` so
     that test execution can apply dual-branch evaluation.
-    
+
     If the filtered fit fails for any reason, falls back to fit SARIMAX on
     the raw value series and emits a prediction JSON without the freshness-gating markers.
 
@@ -381,5 +387,5 @@ def compute_volume_or_metric_threshold(
         exclude_weekends=exclude_weekends,
         holiday_codes=holiday_codes,
         schedule_tz=schedule_tz,
-    )        
+    )
     return lower, upper, None, prediction

testgen/template/dbupgrade/0194_incremental_upgrade.sql

@@ -0,0 +1,34 @@
+SET SEARCH_PATH TO {SCHEMA_NAME};
+
+-- Schema-name matching in the SQL Server DDF query is now case-sensitive,
+-- consistent with TestGen's case-sensitive joins and the other flavors. 
+-- A case-insensitive source previously accepted a table
+-- group whose configured schema name differed in case from the database;
+-- such a group would now match no tables on its next profiling run.
+--
+-- data_table_chars.schema_name holds the case the source database actually
+-- reported (it is populated from the DDF's c.table_schema, not the entered
+-- value), so realign table_group_schema to that case. Guards:
+--   * only mssql connections -- the DDF change was made only for that flavor,
+--   * only when the catalog reports a single unambiguous schema for the group,
+--   * only a case-only difference (LOWER() match), never a different schema,
+--   * '<>' is case-sensitive in PostgreSQL, so correct rows are left untouched.
+-- After this, the group's next profiling run stamps profile_results with the
+-- corrected case, realigning the case-sensitive downstream joins.
+UPDATE table_groups tg
+SET table_group_schema = actual.schema_name
+FROM (
+    SELECT table_groups_id, MIN(schema_name) AS schema_name
+    FROM data_table_chars
+    WHERE drop_date IS NULL
+    GROUP BY table_groups_id
+    HAVING COUNT(DISTINCT schema_name) = 1
+) actual
+WHERE actual.table_groups_id = tg.id
+  AND tg.table_group_schema <> actual.schema_name
+  AND LOWER(tg.table_group_schema) = LOWER(actual.schema_name)
+  AND EXISTS (
+      SELECT 1 FROM connections cn
+      WHERE cn.connection_id = tg.connection_id
+        AND cn.sql_flavor = 'mssql'
+  );

testgen/template/flavors/mssql/data_chars/get_schema_ddf.sql

@@ -54,5 +54,5 @@ SELECT
 FROM information_schema.columns c
     LEFT JOIN approx_cts a ON c.table_schema = a.schema_name AND c.table_name = a.table_name
     LEFT JOIN information_schema.tables it ON c.table_schema = it.table_schema AND c.table_name = it.table_name
-WHERE c.table_schema = '{DATA_SCHEMA}' {TABLE_CRITERIA}
+WHERE c.table_schema = '{DATA_SCHEMA}' COLLATE Latin1_General_BIN {TABLE_CRITERIA}
 ORDER BY c.table_schema, c.table_name, c.ordinal_position;

tests/unit/common/conftest.py

@@ -294,6 +294,24 @@ def _gen_subdaily_gap_schedule_phase() -> list[tuple[pd.Timestamp, float]]:
     return _to_csv_rows(_make_observations(start, end, 2, updates))
 
 
+def _gen_twice_daily_outage() -> list[tuple[pd.Timestamp, float]]:
+    """Updates every 12h (00:00/12:00 UTC), 7 days a week, 5 weeks, then updates stop.
+
+    The 12h cadence classifies as "daily" frequency with a full-week active schedule,
+    so no excluded days or sub-daily window apply — staleness must come from the
+    first-pass gap thresholds.
+    """
+    start = datetime(2025, 10, 6, 0, 0)
+    end = datetime(2025, 11, 11, 12, 0)
+    outage_start = datetime(2025, 11, 10, 0, 0)
+    updates: set[datetime] = set()
+    d = start
+    while d < outage_start:
+        updates.add(d)
+        d += timedelta(hours=12)
+    return _to_csv_rows(_make_observations(start, end, 12, updates))
+
+
 def _gen_weekly_early() -> list[tuple[pd.Timestamp, float]]:
     start = datetime(2025, 8, 7, 10, 0)
     end = datetime(2025, 11, 6, 22, 0)

tests/unit/common/test_freshness_scenarios.py

@@ -25,6 +25,7 @@
     _gen_subdaily_gap_schedule_phase,
     _gen_subdaily_regular,
     _gen_training_only,
+    _gen_twice_daily_outage,
     _gen_weekly_early,
     _run_scenario,
 )
@@ -472,3 +473,48 @@ def test_recovery_passes(self, results: list[ScenarioPoint]) -> None:
         assert post_recovery[0].result_code == 0
         # Second update after recovery should pass
         assert post_recovery[1].result_code == 1
+
+
+# ─── Scenario 9: Twice-Daily Outage (full-week schedule) ────────────
+
+
+class Test_TwiceDailyOutage:
+    """Updates every 12h around the clock for 5 weeks, then updates stop.
+
+    Full-week "daily"-frequency schedule: no excluded days and no sub-daily
+    window, so the staleness threshold must come from the first-pass gap
+    computation. With a 720-min median gap, staleness = 720 * 0.85 = 612 min,
+    so the first missed check (gap 720 min) flags Late.
+    """
+
+    @pytest.fixture(scope="class")
+    def results(self) -> list[ScenarioPoint]:
+        rows = _gen_twice_daily_outage()
+        return _run_scenario(rows, PredictSensitivity.medium, exclude_weekends=False, tz="UTC")
+
+    def test_schedule_goes_active_all_days(self, results: list[ScenarioPoint]) -> None:
+        sched = _schedule(_updates(results)[-1])
+        assert sched is not None
+        assert sched.get("schedule_stage") == "active"
+        assert sched.get("active_days") == [0, 1, 2, 3, 4, 5, 6]
+
+    def test_staleness_stored_once_active(self, results: list[ScenarioPoint]) -> None:
+        last_update = _updates(results)[-1]
+        assert last_update.staleness == pytest.approx(720 * 0.85)
+
+    def test_no_anomalies_before_outage(self, results: list[ScenarioPoint]) -> None:
+        outage_start = pd.Timestamp("2025-11-10")
+        assert all(p.timestamp >= outage_start for p in _anomalies(results))
+
+    def test_first_missed_check_flags_late(self, results: list[ScenarioPoint]) -> None:
+        anomalies = _anomalies(results)
+        assert len(anomalies) > 0
+        first = anomalies[0]
+        assert first.timestamp == pd.Timestamp("2025-11-10 00:00")
+        assert first.value == 720
+
+    def test_all_missed_checks_flag(self, results: list[ScenarioPoint]) -> None:
+        outage_start = pd.Timestamp("2025-11-10")
+        missed_checks = [p for p in results if p.timestamp >= outage_start]
+        assert len(missed_checks) == 4
+        assert all(p.result_code == 0 for p in missed_checks)