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"


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

    async def unexpected_run(*args, **kwargs):
        raise AssertionError("total-only report must not invoke the bucket formatter")

    monkeypatch.setattr(executor, "_run", unexpected_run)
    frame = pd.DataFrame(
        {
            "stat_date": ["20260812"],
            "app_type": ["0"],
            "version_code": ["all"],
            "dau": [100],
            "head_exposure_pv": [50],
            "recommend_return_uv_per_dau": [0.2],
            "all_rov": [0.1],
        }
    )
    artifact = await executor.format_report(
        "odps-product-efficiency-report",
        parameters(
            app_type="0",
            date_from="20260812",
            date_to="20260812",
            data_mode="realtime",
            bucket_position_from_end=None,
            experiment_buckets=None,
            version="all",
        ),
        tmp_path,
        frame,
        "SELECT 1",
        {"ODPS instance_id": "i-total"},
    )

    assert list(artifact.dataframe.columns) == [
        "日期",
        "产品类型",
        "版本号",
        "DAU",
        "头部曝光PV",
        "推荐回流UV/DAU",
        "全部流量ROV（回流UV/曝光PV）",
    ]
    assert artifact.xlsx_path.is_file()
    assert artifact.instance_id == "i-total"