data-query-agent/tests/test_product_efficiency_skill_contract.py

import json
import subprocess
import sys
from pathlib import Path


PROJECT = Path(__file__).parents[1]
SKILL_DIR = PROJECT / ".agents" / "skills" / "odps-product-efficiency-report"


def test_return_attribution_fields_are_explicit() -> None:
    contract = "\n".join(
        [
            (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8"),
            (SKILL_DIR / "references" / "metrics.md").read_text(encoding="utf-8"),
            (SKILL_DIR / "references" / "raw-output-contract.md").read_text(encoding="utf-8"),
        ]
    )

    assert "topic='share'" in contract
    assert "topic='click'" in contract
    assert "type='share'" in contract
    assert "Never use `type='share'`" in contract


def test_generic_catalog_uses_topic_for_return_chain() -> None:
    catalog = (
        PROJECT / ".agents" / "skills" / "query-odps-data" / "references" / "data-catalog.md"
    ).read_text(encoding="utf-8")

    assert "源分享使用 `topic='share'`" in catalog
    assert "点击使用 `topic='click'`" in catalog
    assert "禁止用 `type='share'` 判断源分享" in catalog


def test_realtime_total_only_contract_is_explicit() -> None:
    skill = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
    raw_contract = (SKILL_DIR / "references" / "raw-output-contract.md").read_text(
        encoding="utf-8"
    )

    assert "仅全量" in skill
    assert "`dt LIKE 'yyyyMMdd%'`" in skill
    assert "Overall-only variant" in raw_contract
    assert "do not run the 16-bucket formatter" in raw_contract


def test_realtime_video_sources_and_partition_formats_are_explicit() -> None:
    contract = "\n".join(
        [
            (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8"),
            (SKILL_DIR / "references" / "metrics.md").read_text(encoding="utf-8"),
            (SKILL_DIR / "references" / "raw-output-contract.md").read_text(
                encoding="utf-8"
            ),
        ]
    )

    assert "video_action_log_per5min" in contract
    assert "`dt LIKE 'yyyyMMdd%'`" in contract
    assert "`dt='DD'`" in contract
    assert "`year='yyyy'`" in contract
    assert "`month='MM'`" in contract
    assert "全版本" in contract and "优先" in contract


def test_realtime_normalizer_selects_video_source_by_version(tmp_path) -> None:
    script = SKILL_DIR / "scripts" / "normalize_request.py"
    request = {
        "app_type": "0",
        "date_from": "20260812",
        "date_to": "20260812",
        "data_mode": "realtime",
        "bucket_position_from_end": 3,
        "experiment_buckets": ["0", "1"],
        "version": "all",
    }
    request_path = tmp_path / "request.json"

    request_path.write_text(json.dumps(request), encoding="utf-8")
    all_version = subprocess.run(
        [sys.executable, str(script), str(request_path)],
        check=True,
        capture_output=True,
        text=True,
    )
    assert json.loads(all_version.stdout)["tables"]["video"].endswith(
        "video_action_log_per5min"
    )

    request["version"] = "1578"
    request_path.write_text(json.dumps(request), encoding="utf-8")
    specified_version = subprocess.run(
        [sys.executable, str(script), str(request_path)],
        check=True,
        capture_output=True,
        text=True,
    )
    assert json.loads(specified_version.stdout)["tables"]["video"].endswith(
        "video_action_log_flow"
    )