gyxx-flow/tests/test_product_runtime_data_boundaries.py

from __future__ import annotations

import importlib.util
import sys
from pathlib import Path

import pytest

PROJECT_ROOT = Path(__file__).resolve().parents[1]
MODULE_ROOT = (
    PROJECT_ROOT
    / "src"
    / "gyxx_flow"
    / "modules"
    / "product_commerce"
)


def _load_runtime_paths(monkeypatch: pytest.MonkeyPatch, data_root: Path):
    monkeypatch.setenv("GYXX_DATA_ROOT", str(data_root))
    monkeypatch.setenv("GYXX_MODULE_ROOT", str(MODULE_ROOT))
    module_path = MODULE_ROOT / "runtime_paths.py"
    name = "_gyxx_product_runtime_paths_boundaries"
    spec = importlib.util.spec_from_file_location(name, module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    try:
        spec.loader.exec_module(module)
    finally:
        sys.modules.pop(name, None)
    return module


def test_managed_data_path_rejects_writes_outside_data_home(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    paths = _load_runtime_paths(monkeypatch, tmp_path)

    assert paths.managed_data_path("data/raw/product_commerce/run.json") == (
        tmp_path / "data" / "raw" / "product_commerce" / "run.json"
    )
    assert paths.managed_data_path(paths.TMP_ROOT / "batch.json") == (
        tmp_path / "tmp" / "product_commerce" / "batch.json"
    )
    with pytest.raises(ValueError, match="outside GYXX_DATA_ROOT"):
        paths.managed_data_path(tmp_path.parent / "escape.json")
    with pytest.raises(ValueError, match="outside GYXX_DATA_ROOT"):
        paths.managed_data_path("../escape.json")


def test_artifact_relative_path_is_relative_to_data_home(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    paths = _load_runtime_paths(monkeypatch, tmp_path)

    artifact = paths.RAW_DATA_ROOT / "jd" / "record.json"
    assert paths.artifact_relative_path(artifact) == (
        "data/raw/product_commerce/jd/record.json"
    )
    with pytest.raises(ValueError, match="outside GYXX_DATA_ROOT"):
        paths.artifact_relative_path(MODULE_ROOT / "record.json")


@pytest.mark.parametrize(
    ("relative", "forbidden"),
    [
        ("collect_retry_utils.py", 'Path("data/logs")'),
        ("rebuild_market_rank_documents.py", 'PROJECT_ROOT / "data'),
        ("scripts/insert_main_image_records.py", 'PROJECT_ROOT / "_tmp_'),
        ("jd_product_data_collector.py", 'os.path.join(PROJECT_ROOT, "debug"'),
        ("vendors/jd-data-flow/jd_data_collector.py", "os.getcwd()"),
        (
            "vendors/jd-data-flow/jd_peer_product_data_collector.py",
            "os.path.dirname(os.path.abspath(__file__)), \"data\"",
        ),
    ],
)
def test_runtime_writers_do_not_fall_back_to_cwd_or_source_tree(
    relative: str, forbidden: str
) -> None:
    text = (MODULE_ROOT / relative).read_text(encoding="utf-8-sig")
    assert forbidden not in text


@pytest.mark.parametrize(
    "relative",
    [
        "collect_jd_market_rank.py",
        "collect_sycm_market_rank.py",
        "scripts/insert_jd_main_image_records.py",
        "scripts/insert_main_image_records.py",
    ],
)
def test_runtime_artifact_references_do_not_assume_source_tree_parent(
    relative: str,
) -> None:
    text = (MODULE_ROOT / relative).read_text(encoding="utf-8-sig")
    assert ".relative_to(PROJECT_ROOT)" not in text


def test_checkpoints_and_operational_json_use_state_and_log_roots() -> None:
    retry_text = (MODULE_ROOT / "collect_retry_utils.py").read_text(
        encoding="utf-8-sig"
    )
    assert "LOG_ROOT / \"failures\"" in retry_text

    for relative in (
        "collect_dy_market_rank.py",
        "collect_jd_market_rank.py",
        "collect_sycm_market_rank.py",
    ):
        text = (MODULE_ROOT / relative).read_text(encoding="utf-8-sig")
        assert "STATE_ROOT" in text
        assert "CHECKPOINT_PATH = STATE_ROOT" in text

    decline = (MODULE_ROOT / "check_nine_day_decline.py").read_text(
        encoding="utf-8-sig"
    )
    insert = (MODULE_ROOT / "insert_bitable_records.py").read_text(
        encoding="utf-8-sig"
    )
    assert "LOG_PATH = LOG_ROOT" in decline
    assert "failures_path = LOG_ROOT" in insert


@pytest.mark.parametrize(
    "relative",
    [
        "aggregate_daily_final.py",
        "collect_erp_yesterday_metrics.py",
        "check_nine_day_decline.py",
        "jd_main_image_collector.py",
        "jd_self_inventory_sales_collector.py",
        "jd_product_data_collector.py",
        "taobao_dmp_item_crowd_insight_screenshots.py",
        "taobao_wanxiang_ai_creative_report.py",
        "scripts/insert_jd_main_image_records.py",
        "scripts/insert_main_image_records.py",
        "vendors/jd-data-flow/jd_data_collector.py",
        "vendors/jd-data-flow/jd_product_data_collector.py",
    ],
)
def test_cli_output_overrides_use_the_data_home_guard(relative: str) -> None:
    text = (MODULE_ROOT / relative).read_text(encoding="utf-8-sig")
    assert "managed_data_path" in text