from __future__ import annotations import hashlib import json import re from pathlib import Path, PurePosixPath from gyxx_flow.security.scanner import scan_repository PROJECT_ROOT = Path(__file__).parents[1] RUNTIME_ROOT = ( PROJECT_ROOT / "src" / "gyxx_flow" / "modules" / "content_marketing" / "runtime" ) MANIFEST_PATH = ( PROJECT_ROOT / "config" / "source-manifests" / "content_marketing.json" ) SHA256 = re.compile(r"^[0-9a-f]{64}$") OLD_PROJECT_ROOTS = ( "d:\\yingxiaoyunying", "d:/yingxiaoyunying", "d:\\shop-data-flow", "d:/shop-data-flow", "d:\\product-collector-analyze-flow", "d:/product-collector-analyze-flow", "e:\\auto-flow", "e:/auto-flow", ) FIXED_USER_PATHS = ( "c:\\users\\administrator", "c:/users/administrator", ) EXPECTED_CATEGORIES = { "config", "dependency", "launcher", "source", "sql", "test", "tool", } REQUIRED_SOURCES = { "run_all.py", "bilibili_scraper.py", "pgy_xhs_scraper_v2.py", "xingtu_scraper_v2.py", "chanmama_scraper.py", "daily_marketing_report.py", "data/tools/daily_run.bat", "data/tools/friday_relogin_parallel.py", "data/tools/schema_gyxx_super_data.sql", "data/tools/migrations/005_style_product_profile.sql", "data/config/db.env.example", "data/config/款式_多维表格_对照.json", "tests/test_collection_completeness.py", } 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() def _manifest() -> dict[str, object]: return json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) def test_content_source_snapshot_manifest_is_complete_and_well_formed() -> None: manifest = _manifest() assert manifest["schema_version"] == 1 assert manifest["module"] == "content_marketing" files = manifest["files"] assert isinstance(files, list) assert len(files) == 90 sources = [entry["source_relative_path"] for entry in files] targets = [entry["target_relative_path"] for entry in files] assert sources == sorted(sources, key=str.casefold) assert len(sources) == len(set(sources)) assert len(targets) == len(set(targets)) assert REQUIRED_SOURCES <= set(sources) assert {entry["category"] for entry in files} == EXPECTED_CATEGORIES for entry in files: relative = PurePosixPath(entry["target_relative_path"]) assert not relative.is_absolute() assert ".." not in relative.parts assert relative.parts[:5] == ( "src", "gyxx_flow", "modules", "content_marketing", "runtime", ) assert SHA256.fullmatch(entry["source_sha256"]) assert SHA256.fullmatch(entry["target_sha256"]) assert entry["transformed"] is ( entry["source_sha256"] != entry["target_sha256"] ) def test_content_source_snapshot_targets_exist_and_match_manifest_hashes() -> None: for entry in _manifest()["files"]: target = PROJECT_ROOT.joinpath(*PurePosixPath(entry["target_relative_path"]).parts) assert target.is_file(), entry["target_relative_path"] assert _sha256(target) == entry["target_sha256"] generated = _manifest()["generated_files"] assert [entry["target_relative_path"] for entry in generated] == [ "src/gyxx_flow/modules/content_marketing/runtime/runtime_paths.py" ] for entry in generated: target = PROJECT_ROOT.joinpath(*PurePosixPath(entry["target_relative_path"]).parts) assert target.is_file(), entry["target_relative_path"] assert SHA256.fullmatch(entry["target_sha256"]) assert _sha256(target) == entry["target_sha256"] def test_content_source_snapshot_exclusions_cover_runtime_and_sensitive_state() -> None: excluded = {entry["pattern"] for entry in _manifest()["intentionally_excluded"]} assert { ".git/**", "**/__pycache__/**", "data/config/*.env", "data/**/*cookies*", ".*chrome_profile*/**", "data/logs/**", "data/v2_results/**", "data/notes/**", "data/reports/**", "data/tmp/**", } <= excluded def test_content_source_snapshot_has_no_legacy_roots_or_plaintext_credentials() -> None: manifest_text = MANIFEST_PATH.read_text(encoding="utf-8").casefold() assert not any(root in manifest_text for root in OLD_PROJECT_ROOTS) 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_PROJECT_ROOTS), path assert not any(root in text for root in FIXED_USER_PATHS), path assert scan_repository(RUNTIME_ROOT) == [] def test_chanmama_credentials_are_externalized_in_the_copied_source() -> None: source = (RUNTIME_ROOT / "chanmama_scraper.py").read_text(encoding="utf-8-sig") assert 'os.getenv("CHANMAMA_ACCOUNT", "")' in source assert 'os.getenv("CHANMAMA_PASSWORD", "")' in source assert not re.search(r'^PASSWORD\s*=\s*["\'][^"\']+["\']', source, re.MULTILINE) def test_content_runtime_has_one_portable_path_boundary() -> None: python_sources = list(RUNTIME_ROOT.rglob("*.py")) path_mutations: list[Path] = [] forbidden_cross_module_imports: list[Path] = [] for path in python_sources: source = path.read_text(encoding="utf-8-sig") if re.search(r"sys\.path\.(?:insert|append)\(", source): path_mutations.append(path.relative_to(RUNTIME_ROOT)) if "lark_cli_runtime" in source: forbidden_cross_module_imports.append(path.relative_to(RUNTIME_ROOT)) assert path_mutations == [] assert forbidden_cross_module_imports == [] runtime_paths = ( RUNTIME_ROOT / "runtime_paths.py" ).read_text(encoding="utf-8-sig") assert "GYXX_DATA_ROOT" in runtime_paths assert "GYXX_MODULE_ROOT" in runtime_paths def test_content_runtime_uses_portable_python_launchers_and_data_paths() -> None: batch_sources = list(RUNTIME_ROOT.rglob("*.bat")) assert batch_sources for path in batch_sources: source = path.read_text(encoding="utf-8-sig").casefold() assert "%gyxx_python%" in source, path assert "hermes-agent" not in source, path assert "appdata" not in source, path path_constants = (RUNTIME_ROOT / "runtime_paths.py").read_text( encoding="utf-8-sig" ) assert 'DataLayout(data_root).for_module("content_marketing")' in path_constants assert "raw_root = _DATA_PATHS.raw_root" in path_constants assert "normalized_root = _DATA_PATHS.normalized_root" in path_constants assert "curated_root = _DATA_PATHS.curated_root" in path_constants assert "exports_root = _DATA_PATHS.exports_root" in path_constants