data-query-agent/tests/test_skill_executor.py

from dataclasses import replace

import pandas as pd
import pytest

from data_query_agent.config import Settings
from data_query_agent.models import SkillParameters
from data_query_agent.skill_executor import SkillExecutor


def parameters(**overrides) -> SkillParameters:
    values = {
        "user_id": None,
        "date": None,
        "apptype": None,
        "realtime": None,
        "app_type": None,
        "date_from": None,
        "date_to": None,
        "data_mode": None,
        "bucket_position_from_end": None,
        "experiment_buckets": None,
        "control_buckets": None,
        "version": None,
        "first_layer_rule": None,
        "exclude_qywx": None,
    }
    values.update(overrides)
    return SkillParameters(**values)


@pytest.mark.asyncio
async def test_user_timeline_without_apptype_omits_product_argument(tmp_path, monkeypatch) -> None:
    executor = SkillExecutor(replace(Settings.load(), runtime_dir=tmp_path))

    async def fake_run(args, *, env=None):
        assert args[1].endswith("query-user-behavior-path/scripts/user_timeline.py")
        assert args[2:4] == ["mid_123", "20260806"]
        assert args[4:] == ["--output-dir", str(tmp_path)]
        assert env and "ODPS_ACCESS_ID" in env and "FEISHU_APP_SECRET" not in env
        pd.DataFrame({"北京时间": ["2026-08-06 10:00:00"], "来源": ["video"]}).to_excel(
            tmp_path / "timeline_test.xlsx", sheet_name="行为路径", index=False
        )
        return "[ODPS] InstanceId: i-test\n[XLSX] done"

    monkeypatch.setattr(executor, "_run", fake_run)
    artifact = await executor.run_user_timeline(
        parameters(user_id="mid_123", date="20260806", apptype=None, realtime=False),
        tmp_path,
    )

    assert artifact.instance_id == "i-test"
    assert len(artifact.dataframe) == 1
    assert artifact.xlsx_path.name == "timeline_test.xlsx"


@pytest.mark.asyncio
async def test_user_timeline_with_explicit_apptype_passes_exact_filter(tmp_path, monkeypatch) -> None:
    executor = SkillExecutor(replace(Settings.load(), runtime_dir=tmp_path))

    async def fake_run(args, *, env=None):
        assert args[2:5] == ["mid_123", "20260806", "4"]
        pd.DataFrame({"来源": ["video"]}).to_excel(
            tmp_path / "timeline_filtered.xlsx", sheet_name="行为路径", index=False
        )
        return "[ODPS] InstanceId: i-filtered"

    monkeypatch.setattr(executor, "_run", fake_run)
    artifact = await executor.run_user_timeline(
        parameters(user_id="mid_123", date="20260806", apptype="4", realtime=False),
        tmp_path,
    )

    assert artifact.instance_id == "i-filtered"


@pytest.mark.asyncio
async def test_user_timeline_rejects_invalid_host_parameters(tmp_path) -> None:
    executor = SkillExecutor(replace(Settings.load(), runtime_dir=tmp_path))
    with pytest.raises(ValueError, match="mid/machinecode"):
        await executor.run_user_timeline(
            parameters(user_id="bad value", date="20260806", apptype="0", realtime=False),
            tmp_path,
        )


@pytest.mark.asyncio
async def test_generic_sql_result_uses_existing_workbook_path(tmp_path) -> None:
    executor = SkillExecutor(replace(Settings.load(), runtime_dir=tmp_path))
    artifact = await executor.format_report(
        "query-odps-data",
        parameters(),
        tmp_path,
        pd.DataFrame({"dau": [10]}),
        "SELECT 10 AS dau",
        {"ODPS instance_id": "i-generic"},
    )

    assert artifact.xlsx_path.is_file()
    assert artifact.instance_id == "i-generic"