Files
skill-template/tests/test_development_policy_guard.py
chendelian c67079fb9e
All checks were successful
技能自动化发布 / release (push) Successful in 5s
chore: auto release commit (2026-07-11 15:00:19)
2026-07-11 15:00:20 +08:00

620 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""开发规范 hard policy 守护索引development/POLICY_MATRIX.md"""
from __future__ import annotations
import os
import re
import unittest
from _support import get_skill_root
SKIP_DIR_NAMES = frozenset(
{
".git",
"__pycache__",
".pytest_cache",
"rpa-artifacts",
"videos",
}
)
SKIP_REL_PREFIXES = ("tests/artifacts", "tests/diagnostics")
POLICY_IDS = (
"POLICY-STRUCTURE-001",
"POLICY-RUNTIME-001",
"POLICY-RUNTIME-002",
"POLICY-INSTALL-001",
"POLICY-CONFIG-001",
"POLICY-CONFIG-002",
"POLICY-CONFIG-003",
"POLICY-TESTING-001",
"POLICY-TESTING-002",
"POLICY-RPA-001",
"POLICY-RPA-002",
"POLICY-PACKAGING-001",
"POLICY-PACKAGING-002",
"POLICY-DOCS-001",
"POLICY-LOGGING-001",
"POLICY-LOGGING-002",
"POLICY-LOGGING-003",
"POLICY-LOGGING-004",
"POLICY-LOGGING-005",
)
STRUCTURE_PATHS = (
"scripts/main.py",
"scripts/cli",
"scripts/service",
"scripts/db",
"scripts/util",
)
CONFIG_ENV_BOUNDARY_FILES = frozenset(
{
"scripts/main.py",
"scripts/util/config_bootstrap.py",
"scripts/util/runtime_paths.py",
"scripts/util/logging_config.py",
"scripts/util/constants.py",
}
)
INSTALL_FORBIDDEN_PATTERNS = (
re.compile(r"\bpip\s+install\b", re.IGNORECASE),
re.compile(r"\bpython\s+-m\s+pip\s+install\b", re.IGNORECASE),
re.compile(r"\buv\s+pip\s+install\b", re.IGNORECASE),
re.compile(r"\bplaywright\s+install\b", re.IGNORECASE),
)
CONFIG_ENV_FORBIDDEN_PATTERNS = (
re.compile(r"\bos\.environ\b"),
re.compile(r"\bos\.getenv\b"),
re.compile(r"\bos\.putenv\b"),
re.compile(r"\benviron\.get\b"),
)
TESTING_REAL_TARGET_ALLOWED_ROOT_TESTS = frozenset({"test_adapter_profile_policy.py"})
TESTING_001_FORBIDDEN_SUBSTRINGS = (
"real_" + "api",
"real_" + "rpa",
"OPENCLAW_" + "TEST_TARGET",
)
TESTING_001_SKIP_ROOT_TESTS = frozenset({"test_development_policy_guard.py"})
RPA_001_FORBIDDEN_PATTERNS = (
re.compile(r"\brpa_helpers\b"),
re.compile(r"\binject_account_manager_scripts_path\b"),
re.compile(r"\bget_account_credential\b"),
)
RPA_002_FFMPEG_PATTERNS = (
re.compile(
r"\bsubprocess\.(run|call|Popen|check_output|check_call)\s*\([^)]*ffmpeg",
re.IGNORECASE | re.DOTALL,
),
re.compile(
r"\bsubprocess\.(run|call|Popen|check_output|check_call)\(\s*['\"][^'\"]*ffmpeg",
re.IGNORECASE,
),
re.compile(r"\bos\.system\s*\([^)]*ffmpeg", re.IGNORECASE | re.DOTALL),
re.compile(r"\bos\.popen\s*\([^)]*ffmpeg", re.IGNORECASE | re.DOTALL),
re.compile(
r"\bsubprocess\.(run|call|Popen|check_output|check_call)\([^)]*shell\s*=\s*True[^)]*ffmpeg",
re.IGNORECASE | re.DOTALL,
),
)
RPA_002_EXEMPT_REL = "scripts/service/task_run_support.py"
LOGGING_001_SOURCE = "development/LOGGING.md; development/RUNTIME.md"
LOGGING_002_SOURCE = "development/LOGGING.md; development/DEVELOPMENT.md"
LOGGING_003_SOURCE = "development/LOGGING.md"
LOGGING_004_SOURCE = "development/LOGGING.md; development/RPA.md"
LOGGING_005_SOURCE = "development/LOGGING.md; development/CONFIG.md"
LOGGING_CALL_PATTERN = re.compile(
r"\b(log|logger|get_skill_logger\s*\(\s*\))\s*\.\s*(info|warning|error|debug|exception|critical)\s*\(",
re.IGNORECASE,
)
LOGGING_CALL_GET_LOGGER_PATTERN = re.compile(
r"get_skill_logger\s*\(\s*\)\s*\.\s*(info|warning|error|debug|exception|critical)\s*\(",
re.IGNORECASE,
)
SENSITIVE_LOG_ASSIGNMENT_PATTERN = re.compile(
r"(password|token|cookie|secret|api_key|authorization)\s*=",
re.IGNORECASE,
)
def _policy_msg(policy_id: str, source: str, detail: str) -> str:
return f"{policy_id} [{source}]: {detail}"
def _rel(skill_root: str, path: str) -> str:
return os.path.relpath(path, skill_root).replace("\\", "/")
def _should_skip_dirname(dirname: str) -> bool:
return dirname in SKIP_DIR_NAMES
def _should_skip_rel_path(skill_root: str, abs_path: str) -> bool:
rel = _rel(skill_root, abs_path).replace("\\", "/")
for prefix in SKIP_REL_PREFIXES:
normalized = prefix.rstrip("/")
if rel == normalized or rel.startswith(normalized + "/"):
return True
return False
def _walk_files(skill_root: str, rel_root: str, *, suffix: str | None = None) -> list[str]:
base = os.path.join(skill_root, rel_root)
if not os.path.exists(base):
return []
found: list[str] = []
for root, dirnames, filenames in os.walk(base):
kept: list[str] = []
for d in dirnames:
if _should_skip_dirname(d):
continue
sub = os.path.join(root, d)
if _should_skip_rel_path(skill_root, sub):
continue
kept.append(d)
dirnames[:] = kept
for name in filenames:
full = os.path.join(root, name)
if _should_skip_rel_path(skill_root, full):
continue
if suffix and not name.endswith(suffix):
continue
found.append(full)
return sorted(found)
def _read_text(path: str) -> str:
with open(path, encoding="utf-8") as f:
return f.read()
def _requirement_dependency_lines(req_text: str) -> list[str]:
return [
line.strip()
for line in req_text.splitlines()
if line.strip() and not line.strip().startswith("#")
]
def _scan_install_delivery_files(skill_root: str) -> list[str]:
offenders: list[str] = []
scan_specs: list[tuple[str, str | None]] = [
("scripts", ".py"),
(".", ".ps1"),
(".", ".sh"),
(".", None),
(".github/workflows", ".yml"),
(".github/workflows", ".yaml"),
]
seen: set[str] = set()
for rel_root, suffix in scan_specs:
if rel_root == "." and suffix is None:
req = os.path.join(skill_root, "requirements.txt")
if os.path.isfile(req):
seen.add(req)
text = _read_text(req)
for pattern in INSTALL_FORBIDDEN_PATTERNS:
if pattern.search(text):
offenders.append(
_policy_msg(
"POLICY-INSTALL-001",
"development/RUNTIME.md; development/RPA.md §1.4; development/DEVELOPMENT.md §3.1",
f"{_rel(skill_root, req)} matches {pattern.pattern}",
)
)
continue
for path in _walk_files(skill_root, rel_root, suffix=suffix):
if path in seen:
continue
seen.add(path)
if suffix == ".ps1" and not path.endswith(".ps1"):
continue
if suffix == ".sh" and not path.endswith(".sh"):
continue
text = _read_text(path)
for pattern in INSTALL_FORBIDDEN_PATTERNS:
if pattern.search(text):
offenders.append(
_policy_msg(
"POLICY-INSTALL-001",
"development/RUNTIME.md; development/RPA.md §1.4; development/DEVELOPMENT.md §3.1",
f"{_rel(skill_root, path)} matches {pattern.pattern}",
)
)
return offenders
def _gitignore_has_pattern(lines: list[str], pattern: str) -> bool:
normalized = [ln.strip() for ln in lines if ln.strip() and not ln.strip().startswith("#")]
return pattern in normalized
class TestPolicyStructure001(unittest.TestCase):
def test_standard_scripts_layout_exists(self) -> None:
skill_root = get_skill_root()
missing: list[str] = []
for rel in STRUCTURE_PATHS:
path = os.path.join(skill_root, *rel.split("/"))
if not os.path.exists(path):
missing.append(rel)
self.assertEqual(
missing,
[],
msg=_policy_msg(
"POLICY-STRUCTURE-001",
"development/DEVELOPMENT.md §2; development/RUNTIME.md §目录结构",
"missing: " + ", ".join(missing),
),
)
class TestPolicyRuntime001(unittest.TestCase):
def test_no_vendored_jiangchang_skill_core(self) -> None:
skill_root = get_skill_root()
vendored = os.path.join(skill_root, "scripts", "jiangchang_skill_core")
self.assertFalse(
os.path.isdir(vendored),
msg=_policy_msg(
"POLICY-RUNTIME-001",
"development/RUNTIME.md §共享 Python Runtime",
_rel(skill_root, vendored),
),
)
class TestPolicyRuntime002(unittest.TestCase):
def test_requirements_txt_excludes_shared_runtime_deps(self) -> None:
skill_root = get_skill_root()
req_path = os.path.join(skill_root, "requirements.txt")
self.assertTrue(os.path.isfile(req_path), msg="requirements.txt missing")
text = _read_text(req_path)
dep_lines = _requirement_dependency_lines(text)
offenders: list[str] = []
for line in dep_lines:
lower = line.lower()
if "jiangchang-platform-kit" in lower:
offenders.append(line)
if re.search(r"(?<![\w.-])playwright(?![\w.-])", lower):
offenders.append(line)
self.assertEqual(
offenders,
[],
msg=_policy_msg(
"POLICY-RUNTIME-002",
"development/RUNTIME.md; development/DEVELOPMENT.md §3.1",
"forbidden dependency lines: " + "; ".join(offenders),
),
)
class TestPolicyInstall001(unittest.TestCase):
def test_delivery_files_have_no_auto_install_commands(self) -> None:
offenders = _scan_install_delivery_files(get_skill_root())
self.assertEqual(offenders, [], msg="\n".join(offenders))
class TestPolicyConfig001(unittest.TestCase):
def test_env_example_exists(self) -> None:
skill_root = get_skill_root()
path = os.path.join(skill_root, ".env.example")
self.assertTrue(
os.path.isfile(path),
msg=_policy_msg(
"POLICY-CONFIG-001",
"development/CONFIG.md",
".env.example missing at repo root",
),
)
class TestPolicyConfig002(unittest.TestCase):
def test_gitignore_ignores_env_files(self) -> None:
skill_root = get_skill_root()
path = os.path.join(skill_root, ".gitignore")
self.assertTrue(os.path.isfile(path), msg=".gitignore missing")
lines = _read_text(path).splitlines()
missing: list[str] = []
if not _gitignore_has_pattern(lines, ".env"):
missing.append(".env")
if not _gitignore_has_pattern(lines, "*.env.local"):
missing.append("*.env.local")
self.assertEqual(
missing,
[],
msg=_policy_msg(
"POLICY-CONFIG-002",
"development/CONFIG.md §红线:敏感信息不进 .env",
"missing gitignore entries: " + ", ".join(missing),
),
)
class TestPolicyConfig003(unittest.TestCase):
def test_business_scripts_do_not_read_os_environ_directly(self) -> None:
skill_root = get_skill_root()
offenders: list[str] = []
for path in _walk_files(skill_root, "scripts", suffix=".py"):
rel = _rel(skill_root, path)
if rel.replace("\\", "/") in CONFIG_ENV_BOUNDARY_FILES:
continue
text = _read_text(path)
for lineno, line in enumerate(text.splitlines(), 1):
if line.strip().startswith("#"):
continue
for pattern in CONFIG_ENV_FORBIDDEN_PATTERNS:
if pattern.search(line):
offenders.append(
f"{rel}:{lineno}: {line.strip()} — use jiangchang_skill_core.config.get*"
)
self.assertEqual(
offenders,
[],
msg=_policy_msg(
"POLICY-CONFIG-003",
"development/CONFIG.md; development/ADAPTER.md",
"\n".join(offenders),
),
)
class TestPolicyTesting001(unittest.TestCase):
def test_root_tests_do_not_hardcode_real_targets(self) -> None:
skill_root = get_skill_root()
tests_dir = os.path.join(skill_root, "tests")
offenders: list[str] = []
for name in sorted(os.listdir(tests_dir)):
if not (name.startswith("test_") and name.endswith(".py")):
continue
if name in TESTING_001_SKIP_ROOT_TESTS:
continue
if name in TESTING_REAL_TARGET_ALLOWED_ROOT_TESTS:
continue
path = os.path.join(tests_dir, name)
text = _read_text(path)
for lineno, line in enumerate(text.splitlines(), 1):
stripped = line.strip()
if stripped.startswith("#"):
continue
for token in TESTING_001_FORBIDDEN_SUBSTRINGS:
if token in line:
offenders.append(f"{name}:{lineno}: {stripped}")
break
self.assertEqual(
offenders,
[],
msg=_policy_msg(
"POLICY-TESTING-001",
"development/TESTING.md §默认必跑测试要做什么",
"\n".join(offenders),
),
)
class TestPolicyTesting002(unittest.TestCase):
def test_integration_python_tests_are_sample_only(self) -> None:
skill_root = get_skill_root()
integration_dir = os.path.join(skill_root, "tests", "integration")
if not os.path.isdir(integration_dir):
return
offenders: list[str] = []
for name in sorted(os.listdir(integration_dir)):
if not name.endswith(".py"):
continue
if name.endswith(".sample"):
continue
offenders.append(f"tests/integration/{name}")
self.assertEqual(
offenders,
[],
msg=_policy_msg(
"POLICY-TESTING-002",
"development/TESTING.md §7 真实联调测试的安全约束",
"non-sample integration tests: " + ", ".join(offenders),
),
)
class TestPolicyRpa001(unittest.TestCase):
def test_scripts_have_no_account_manager_internal_imports(self) -> None:
skill_root = get_skill_root()
offenders: list[str] = []
for path in _walk_files(skill_root, "scripts", suffix=".py"):
rel = _rel(skill_root, path)
text = _read_text(path)
for pattern in RPA_001_FORBIDDEN_PATTERNS:
if pattern.search(text):
offenders.append(f"{rel}: {pattern.pattern}")
self.assertEqual(
offenders,
[],
msg=_policy_msg(
"POLICY-RPA-001",
"development/ADAPTER.md §兄弟依赖; development/RPA.md §1.7",
"\n".join(offenders),
),
)
class TestPolicyRpa002(unittest.TestCase):
def test_scripts_do_not_invoke_ffmpeg_directly(self) -> None:
skill_root = get_skill_root()
offenders: list[str] = []
for path in _walk_files(skill_root, "scripts", suffix=".py"):
rel = _rel(skill_root, path)
if rel.replace("\\", "/") == RPA_002_EXEMPT_REL:
continue
text = _read_text(path)
for pattern in RPA_002_FFMPEG_PATTERNS:
if pattern.search(text):
offenders.append(f"{rel}: {pattern.pattern}")
self.assertEqual(
offenders,
[],
msg=_policy_msg(
"POLICY-RPA-002",
"development/RPA.md §5.3 录屏成片标准",
"\n".join(offenders),
),
)
def _scan_sensitive_logging_assignments(skill_root: str) -> list[str]:
offenders: list[str] = []
for path in _walk_files(skill_root, "scripts", suffix=".py"):
rel = _rel(skill_root, path)
for lineno, line in enumerate(_read_text(path).splitlines(), 1):
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
if not LOGGING_CALL_PATTERN.search(line):
continue
if SENSITIVE_LOG_ASSIGNMENT_PATTERN.search(line):
offenders.append(f"{rel}:{lineno}: {stripped}")
return offenders
class TestPolicyLogging001(unittest.TestCase):
def test_cli_app_sets_up_unified_logging(self) -> None:
skill_root = get_skill_root()
path = os.path.join(skill_root, "scripts", "cli", "app.py")
text = _read_text(path)
rel = _rel(skill_root, path)
self.assertIn(
"setup_skill_logging",
text,
msg=_policy_msg(
"POLICY-LOGGING-001",
LOGGING_001_SOURCE,
f"{rel} missing setup_skill_logging",
),
)
has_logger_info = bool(
LOGGING_CALL_GET_LOGGER_PATTERN.search(text)
or re.search(r"\bget_skill_logger\s*\(\s*\)\s*\.\s*info\s*\(", text)
)
self.assertTrue(
has_logger_info,
msg=_policy_msg(
"POLICY-LOGGING-001",
LOGGING_001_SOURCE,
f"{rel} missing get_skill_logger().info (or equivalent)",
),
)
class TestPolicyLogging002(unittest.TestCase):
def test_task_service_uses_get_skill_logger(self) -> None:
skill_root = get_skill_root()
path = os.path.join(skill_root, "scripts", "service", "task_service.py")
text = _read_text(path)
rel = _rel(skill_root, path)
self.assertRegex(
text,
r"get_skill_logger",
msg=_policy_msg(
"POLICY-LOGGING-002",
LOGGING_002_SOURCE,
f"{rel} must import/use get_skill_logger",
),
)
class TestPolicyLogging003(unittest.TestCase):
def test_task_service_logs_start_failure_and_task_logs(self) -> None:
skill_root = get_skill_root()
path = os.path.join(skill_root, "scripts", "service", "task_service.py")
text = _read_text(path)
rel = _rel(skill_root, path)
missing: list[str] = []
if not re.search(r"\b(log|logger)\.info\s*\(", text):
missing.append("log.info or logger.info")
if not re.search(r"\b(log|logger)\.exception\s*\(", text):
missing.append("log.exception or logger.exception")
if "save_task_log" not in text:
missing.append("save_task_log")
self.assertEqual(
missing,
[],
msg=_policy_msg(
"POLICY-LOGGING-003",
LOGGING_003_SOURCE,
f"{rel} missing: " + ", ".join(missing),
),
)
class TestPolicyLogging004(unittest.TestCase):
def test_task_service_emits_progress_for_long_tasks(self) -> None:
skill_root = get_skill_root()
path = os.path.join(skill_root, "scripts", "service", "task_service.py")
text = _read_text(path)
rel = _rel(skill_root, path)
has_progress = bool(
re.search(r"\bemit\s*\(", text)
or re.search(r"\bactivity\.emit\s*\(", text)
or re.search(r"\bvideo\.add_step\s*\(", text)
)
self.assertTrue(
has_progress,
msg=_policy_msg(
"POLICY-LOGGING-004",
LOGGING_004_SOURCE,
f"{rel} must use emit( / activity.emit / video.add_step",
),
)
class TestPolicyLogging005(unittest.TestCase):
def test_scripts_do_not_log_sensitive_assignments(self) -> None:
offenders = _scan_sensitive_logging_assignments(get_skill_root())
self.assertEqual(
offenders,
[],
msg=_policy_msg(
"POLICY-LOGGING-005",
LOGGING_005_SOURCE,
"\n".join(offenders),
),
)
class TestPolicyDocs001(unittest.TestCase):
def test_policy_matrix_exists_and_lists_policy_ids(self) -> None:
skill_root = get_skill_root()
matrix_path = os.path.join(skill_root, "development", "POLICY_MATRIX.md")
self.assertTrue(
os.path.isfile(matrix_path),
msg=_policy_msg(
"POLICY-DOCS-001",
"development/POLICY_MATRIX.md",
"file missing",
),
)
text = _read_text(matrix_path)
missing = [pid for pid in POLICY_IDS if pid not in text]
self.assertEqual(
missing,
[],
msg=_policy_msg(
"POLICY-DOCS-001",
"development/POLICY_MATRIX.md",
"missing policy_id entries: " + ", ".join(missing),
),
)
if __name__ == "__main__":
unittest.main()