gyxx-flow/tests/test_shop_supply_source_snapshot.py

from __future__ import annotations

import hashlib
import json
import re
from pathlib import Path, PurePosixPath

import pytest

from gyxx_flow.security.scanner import scan_repository

PROJECT_ROOT = Path(__file__).resolve().parents[1]
MANIFEST_ROOT = PROJECT_ROOT / "config" / "source-manifests"
SHA256 = re.compile(r"^[0-9a-f]{64}$")
OLD_ROOTS = (
    "d:\\shop-data-flow",
    "d:/shop-data-flow",
    "e:\\auto-flow",
    "e:/auto-flow",
)
NON_PORTABLE_RUNTIME_PATTERNS = (
    re.compile(r"e:[\\/]+auto-flow", re.IGNORECASE),
    re.compile(r"d:[\\/]+erp", re.IGNORECASE),
    re.compile(r"path\s*\(\s*['\"]d:[\\/]+['\"]\s*\)\.iterdir", re.IGNORECASE),
    re.compile(r"sys\.path\.insert\s*\(", re.IGNORECASE),
)
EXPECTED = {
    "shop_intelligence": {
        "minimum_files": 22,
        "required": {
            "adaptive_selectors.py",
            "collectors/jd_data_collector.py",
            "collectors/jd_peer_store_data_collector.py",
            "collectors/jd_self_operated_brand_daily.py",
            "collectors/dy_store_competitor_store_scraping.py",
            "collectors/taobao_sycm.py",
            "runners/run_shop.py",
            "runners/run_peer_store.py",
            "writers/shop_base_writer.py",
            "writers/peer_store_writer.py",
            "db/db.py",
            "db/schema.sql",
        },
        "classifications": {"scheduled", "manual", "internal"},
    },
    "supply_chain": {
        "minimum_files": 37,
        "required": {
            "run.py",
            "orchestrator/mcp_workflow.py",
            "orchestrator/monitor.py",
            "orchestrator/pg_writer.py",
            "orchestrator/scripts/PurchaseConfirmation.py",
            "orchestrator/scripts/ProductReplenishment.py",
            "orchestrator/scripts/PurchaseOrderUpdate.py",
            "orchestrator/scripts/batch_process.py",
            "orchestrator/scripts/trigger_purchase_order_update.py",
            "orchestrator/scripts/insert_replenishment_bitable.py",
            "orchestrator/scripts/send_card_notification.py",
            "orchestrator/sql/backfill_history.py",
            "skills/collector/SKILL.md",
            "skills/analyzer/SKILL.md",
        },
        "classifications": {
            "scheduled",
            "manual",
            "event",
            "maintenance",
            "internal",
        },
    },
}


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


@pytest.mark.parametrize("module", sorted(EXPECTED))
def test_source_manifest_covers_runtime_sources_and_entrypoints(module: str) -> None:
    expected = EXPECTED[module]
    manifest_path = MANIFEST_ROOT / f"{module}.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))

    assert manifest["schema_version"] == 1
    assert manifest["module"] == module
    files = manifest["files"]
    assert len(files) >= expected["minimum_files"]
    assert expected["required"] <= {
        item["source_relative_path"] for item in files
    }

    target_paths = [item["target_relative_path"] for item in files]
    assert len(target_paths) == len(set(target_paths))
    for item in files:
        target = PurePosixPath(item["target_relative_path"])
        assert not target.is_absolute()
        assert ".." not in target.parts
        if item["category"] == "test":
            assert target.parts[:3] == ("tests", "modules", module)
        elif target.parts[0] == "src":
            assert target.parts[:4] == ("src", "gyxx_flow", "modules", module)
        else:
            assert target.parts[:2] == ("docs", "history")
        assert SHA256.fullmatch(item["source_sha256"])
        assert SHA256.fullmatch(item["target_sha256"])
        assert item["transformed"] is (
            item["source_sha256"] != item["target_sha256"]
        )

    entrypoints = manifest["entrypoints"]
    assert {item["classification"] for item in entrypoints} == expected[
        "classifications"
    ]
    assert all(item["source_relative_path"] for item in entrypoints)
    assert all(item["target_relative_path"] for item in entrypoints)


@pytest.mark.parametrize("module", sorted(EXPECTED))
def test_runtime_files_match_manifest_and_are_decoupled(module: str) -> None:
    manifest_path = MANIFEST_ROOT / f"{module}.json"
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    runtime_root = PROJECT_ROOT / "src" / "gyxx_flow" / "modules" / module

    for item in manifest["files"]:
        target = PROJECT_ROOT.joinpath(
            *PurePosixPath(item["target_relative_path"]).parts
        )
        assert target.is_file(), item["target_relative_path"]
        assert _sha256(target) == item["target_sha256"]

    for path in runtime_root.rglob("*"):
        if not path.is_file() or b"\x00" in path.read_bytes():
            continue
        text = path.read_text(encoding="utf-8-sig", errors="strict").casefold()
        assert not any(root in text for root in OLD_ROOTS), path

    manifest_text = manifest_path.read_text(encoding="utf-8").casefold()
    assert not any(root in manifest_text for root in OLD_ROOTS)
    assert scan_repository(runtime_root) == []


@pytest.mark.parametrize("module", sorted(EXPECTED))
def test_source_manifest_explicitly_excludes_secrets_and_runtime_state(
    module: str,
) -> None:
    manifest = json.loads(
        (MANIFEST_ROOT / f"{module}.json").read_text(encoding="utf-8")
    )
    excluded = {item["pattern"] for item in manifest["intentionally_excluded"]}

    assert {
        ".git/**",
        ".venv/**",
        "**/__pycache__/**",
        ".env",
        "**/*cookies*",
        "**/*profile*/**",
        "**/data/**",
        "**/logs/**",
    } <= excluded
    if module == "shop_intelligence":
        assert {
            "scripts/setup_scheduler.ps1",
            "scripts/remove_scheduler.ps1",
        } <= excluded


@pytest.mark.parametrize("module", sorted(EXPECTED))
def test_runtime_has_no_legacy_absolute_paths_or_sys_path_bootstrap(
    module: str,
) -> None:
    runtime_root = PROJECT_ROOT / "src" / "gyxx_flow" / "modules" / module
    for path in runtime_root.rglob("*"):
        if not path.is_file() or b"\x00" in path.read_bytes():
            continue
        text = path.read_text(encoding="utf-8-sig", errors="strict")
        assert not any(
            pattern.search(text) for pattern in NON_PORTABLE_RUNTIME_PATTERNS
        ), path


def test_supply_backfill_reads_only_from_portable_shared_data_root() -> None:
    backfill = (
        PROJECT_ROOT
        / "src"
        / "gyxx_flow"
        / "modules"
        / "supply_chain"
        / "orchestrator"
        / "sql"
        / "backfill_history.py"
    ).read_text(encoding="utf-8-sig")

    assert "from gyxx_flow.modules.supply_chain.paths import SHARED_ROOT" in backfill
    assert "ROOT = SHARED_ROOT" in backfill