data-query-agent/tests/test_timeline_sql.py

import importlib.util
import sys
from pathlib import Path


SCRIPTS = Path(__file__).parents[1] / ".agents" / "skills" / "query-user-behavior-path" / "scripts"
sys.path.insert(0, str(SCRIPTS))
spec = importlib.util.spec_from_file_location("timeline_skill_script", SCRIPTS / "user_timeline.py")
assert spec and spec.loader
timeline = importlib.util.module_from_spec(spec)
spec.loader.exec_module(timeline)


def load_script(name: str):
    module_spec = importlib.util.spec_from_file_location(f"timeline_{name}", SCRIPTS / f"{name}.py")
    assert module_spec and module_spec.loader
    module = importlib.util.module_from_spec(module_spec)
    module_spec.loader.exec_module(module)
    return module


realtime = load_script("user_timeline_realtime")
realtime_batch = load_script("user_timeline_realtime_batch")


def realtime_args(apptype):
    return {
        "mc": "mid_123",
        "mc_list": "'mid_123'",
        "apptype_filter": timeline.build_apptype_filter(apptype),
        "dt_start": "20260806000000",
        "dt_end": "20260806235959",
        "year": "2026",
        "month": "08",
        "day_of_month": "06",
    }


def test_omitted_apptype_adds_no_product_predicate() -> None:
    sql = timeline.SQL_ALL.format(
        day="20260806",
        mc="mid_123",
        apptype_filter=timeline.build_apptype_filter(None),
    )

    assert "AND apptype=" not in sql


def test_explicit_apptype_filters_every_timeline_source() -> None:
    sql = timeline.SQL_ALL.format(
        day="20260806",
        mc="mid_123",
        apptype_filter=timeline.build_apptype_filter("4"),
    )

    assert sql.count("apptype='4'") == 6


def test_realtime_timeline_apptype_is_optional_across_all_sources() -> None:
    unfiltered = realtime.SQL_REALTIME.format(**realtime_args(None))
    filtered = realtime.SQL_REALTIME.format(**realtime_args("4"))

    assert "AND apptype=" not in unfiltered
    assert filtered.count("apptype='4'") == 7


def test_realtime_batch_apptype_is_optional_across_all_sources() -> None:
    unfiltered = realtime_batch.SQL_REALTIME_BATCH.format(**realtime_args(None))
    filtered = realtime_batch.SQL_REALTIME_BATCH.format(**realtime_args("4"))

    assert "AND apptype=" not in unfiltered
    assert filtered.count("apptype='4'") == 7