test: fill coverage gaps in existing unit tests

Adds 11 tests covering untested logic branches in existing test files:
- _get_table_criteria: table_set-only, no-filters, all-three-combined
- replace_templated_functions: no-template passthrough
- test run notifications: warning-only trigger branch (elif path)
- Job.get_triggering_times: every-5-min, hourly, timezone-aware cron
- profiling notifications: URL encoding of likelihood, is_new flag passthrough

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rboni-dk

tests/unit/commands/queries/test_refresh_data_chars_query.py

@@ -59,3 +59,51 @@ def test_include_empty_include_mask(mask):
     assert r"""AND (
                 (c.table_name LIKE 'important%' ) OR (c.table_name LIKE '%useful[_]%' )
             )""" in query
+
+
+def test_table_set_only():
+    connection = Connection(sql_flavor="postgresql")
+    table_group = TableGroup(
+        table_group_schema="test_schema",
+        profiling_table_set="users, orders, products",
+        profiling_include_mask="",
+        profiling_exclude_mask="",
+    )
+    sql_generator = RefreshDataCharsSQL(connection, table_group)
+    criteria = sql_generator._get_table_criteria()
+
+    assert "IN ('users','orders','products')" in criteria
+    assert "LIKE" not in criteria
+
+
+@pytest.mark.parametrize("include", ("", None))
+@pytest.mark.parametrize("exclude", ("", None))
+def test_no_filters(include, exclude):
+    connection = Connection(sql_flavor="postgresql")
+    table_group = TableGroup(
+        table_group_schema="test_schema",
+        profiling_table_set="",
+        profiling_include_mask=include,
+        profiling_exclude_mask=exclude,
+    )
+    sql_generator = RefreshDataCharsSQL(connection, table_group)
+    criteria = sql_generator._get_table_criteria()
+
+    assert criteria == ""
+
+
+def test_table_set_with_include_exclude():
+    connection = Connection(sql_flavor="postgresql")
+    table_group = TableGroup(
+        table_group_schema="test_schema",
+        profiling_table_set="users, orders",
+        profiling_include_mask="important%",
+        profiling_exclude_mask="temp%",
+    )
+    sql_generator = RefreshDataCharsSQL(connection, table_group)
+    criteria = sql_generator._get_table_criteria()
+
+    assert "IN ('users','orders')" in criteria
+    assert "LIKE 'important%'" in criteria
+    assert "AND NOT" in criteria
+    assert "LIKE 'temp%'" in criteria

tests/unit/common/models/test_custom_types.py

@@ -91,7 +91,7 @@ def test_update_timestamp():
 @patch("testgen.common.encrypt.settings")
 def test_encrypted_bytea_roundtrip(mock_settings):
     mock_settings.APP_ENCRYPTION_SALT = "testsalt12345678"
-    mock_settings.APP_ENCRYPTION_SECRET = "testsecret123456"
+    mock_settings.APP_ENCRYPTION_SECRET = "testsecret123456"  # noqa: S105
 
     t = EncryptedBytea()
     original = "sensitive data"
@@ -115,7 +115,7 @@ def test_encrypted_bytea_none(mock_settings):
 @patch("testgen.common.encrypt.settings")
 def test_encrypted_json_roundtrip(mock_settings):
     mock_settings.APP_ENCRYPTION_SALT = "testsalt12345678"
-    mock_settings.APP_ENCRYPTION_SECRET = "testsecret123456"
+    mock_settings.APP_ENCRYPTION_SECRET = "testsecret123456"  # noqa: S105
 
     t = EncryptedJson()
     original = {"key": "value", "num": 42, "list": [1, 2, 3]}

tests/unit/common/notifications/test_monitor_run_notifications.py

@@ -301,7 +301,7 @@ def test_send_monitor_notifications_url_construction(
 
     context = send_mock.call_args[0][1]
     assert context["view_in_testgen_url"] == (
-        "http://tg-base-url/monitors?project_code=proj-abc&table_group_id=tg-123"
+        "http://tg-base-url/monitors?project_code=proj-abc&table_group_id=tg-123&source=email"
     )
 
     send_mock.reset_mock()
@@ -317,6 +317,6 @@ def test_send_monitor_notifications_url_construction(
 
     context = send_mock.call_args[0][1]
     assert context["view_in_testgen_url"] == (
-        "http://tg-base-url/monitors?project_code=proj-abc&table_group_id=tg-123&table_name_filter=users"
+        "http://tg-base-url/monitors?project_code=proj-abc&table_group_id=tg-123&table_name_filter=users&source=email"
     )
     assert context["summary"]["table_name"] == "users"

tests/unit/common/notifications/test_profiling_run_notifications.py

@@ -1,5 +1,6 @@
 from itertools import count
 from unittest.mock import ANY, Mock, call, patch
+from urllib.parse import quote
 
 import pytest
 
@@ -132,8 +133,8 @@ def test_send_profiling_run_notification(
                 {
                     "profiling_run": {
                         "id": "pr-id",
-                        "issues_url": "http://tg-base-url/profiling-runs:hygiene?run_id=pr-id",
-                        "results_url": "http://tg-base-url/profiling-runs:results?run_id=pr-id",
+                        "issues_url": "http://tg-base-url/profiling-runs:hygiene?run_id=pr-id&source=email",
+                        "results_url": "http://tg-base-url/profiling-runs:results?run_id=pr-id&source=email",
                         "start_time": None,
                         "end_time": None,
                         "status": profiling_run_status,
@@ -164,3 +165,22 @@ def test_send_profiling_run_notification(
     assert all(s.get("label") is not None for s in summary)
     assert all(s.get("priority") in priorities for s in summary)
     assert all(s.get("url") is not None for s in summary)
+
+    # Verify priority-to-likelihood URL mapping and URL encoding
+    expected_likelihoods = {
+        "Definite": "Definite",
+        "Likely": "Likely",
+        "Possible": "Possible",
+        "High": "Potential PII",
+        "Moderate": "Potential PII",
+    }
+    for s in summary:
+        expected_likelihood = expected_likelihoods[s["priority"]]
+        assert f"likelihood={quote(expected_likelihood)}" in s["url"]
+
+    # Verify is_new flags are passed through
+    all_issues = [issue for s in summary for issue in s["issues"]]
+    if not has_prev_run:
+        assert all(issue["is_new"] is True for issue in all_issues)
+    elif new_issue_count == 0:
+        assert all(issue["is_new"] is False for issue in all_issues)

tests/unit/common/notifications/test_score_drop_notifications.py

@@ -171,7 +171,7 @@ def test_send_score_drop_notifications(
                 {
                     "project_name": "Test Project",
                     "definition": score_definition,
-                    "scorecard_url": "http://tg-base-url/quality-dashboard:score-details?definition_id=sd-1",
+                    "scorecard_url": "http://tg-base-url/quality-dashboard:score-details?definition_id=sd-1&source=email",
                     "diff": [
                         {**expected_total_diff, "notify": total_triggers},
                         {**expected_cde_diff, "notify": cde_triggers},

tests/unit/common/notifications/test_test_run_notifications.py

@@ -99,6 +99,7 @@ def select_summary_mock():
     ),
     [
         ("Complete", 0, 0, 0, {}, 0, 0, 0, ["always"]),
+        ("Complete", 0, 5, 0, {}, 0, 5, 0, ["always", "on_warnings"]),
         ("Complete", 1, 1, 1, {}, 1, 1, 1, ["always", "on_failures", "on_warnings"]),
         ("Complete", 50, 50, 50, {"failed": 2, "warning": 3}, 10, 5, 5, [
             "always", "on_failures", "on_warnings", "on_changes",
@@ -173,7 +174,7 @@ def test_send_test_run_notification(
 
     expected_context = {
         "test_run": summary,
-        "test_run_url": "http://tg-base-url/test-runs:results?run_id=tr-id",
+        "test_run_url": "http://tg-base-url/test-runs:results?run_id=tr-id&source=email",
         "test_run_id": "tr-id",
         "test_result_summary": ANY,
     }

tests/unit/common/test_read_file.py

@@ -35,3 +35,8 @@ def test_replace_templated_missing_arg(query):
         match="Templated function call missing required arguments: <%DATEDIFF_YEAR;'{COL_NAME}'::DATE%>",
     ):
         replace_templated_functions(query, "postgresql")
+
+
+def test_replace_templated_functions_no_templates():
+    plain_query = "SELECT col1, col2 FROM my_table WHERE id = 1"
+    assert replace_templated_functions(plain_query, "postgresql") == plain_query

tests/unit/scheduler/test_scheduler_base.py

@@ -55,6 +55,34 @@ def now_func():
         yield now_func
 
 
+def test_get_triggering_times_every_5_min():
+    job = Job(cron_expr="*/5 * * * *", cron_tz="UTC", delayed_policy=DelayedPolicy.ALL)
+    base = datetime(2025, 4, 15, 9, 0, 0, tzinfo=UTC)
+    times = list(islice(job.get_triggering_times(base), 5))
+    minutes = [t.minute for t in times]
+    # cron_converter yields starting at base time, then increments
+    assert minutes == [0, 5, 10, 15, 20]
+    assert all(t.hour == 9 for t in times)
+
+
+def test_get_triggering_times_hourly():
+    job = Job(cron_expr="0 * * * *", cron_tz="UTC", delayed_policy=DelayedPolicy.ALL)
+    base = datetime(2025, 4, 15, 9, 30, 0, tzinfo=UTC)
+    times = list(islice(job.get_triggering_times(base), 3))
+    hours = [t.hour for t in times]
+    assert hours == [10, 11, 12]
+    assert all(t.minute == 0 for t in times)
+
+
+def test_get_triggering_times_timezone():
+    job = Job(cron_expr="0 9 * * *", cron_tz="America/New_York", delayed_policy=DelayedPolicy.ALL)
+    base = datetime(2025, 4, 15, 12, 0, 0, tzinfo=UTC)  # 8 AM ET (EDT)
+    times = list(islice(job.get_triggering_times(base), 2))
+    # 9 AM ET = 13:00 UTC (during EDT)
+    assert times[0].astimezone(UTC).hour == 13
+    assert times[1].astimezone(UTC).hour == 13
+
+
 def test_getting_jobs_wont_crash(scheduler_instance, base_time):
     scheduler_instance.get_jobs.side_effect = Exception
     scheduler_instance.start(base_time)

tests/unit/test_utils.py

@@ -2,10 +2,8 @@
 from datetime import UTC, datetime
 from decimal import Decimal
 from enum import Enum
-from unittest.mock import patch
 from uuid import UUID
 
-import pandas as pd
 import pytest
 
 from testgen.utils import (