1141 lines
40 KiB
Python
1141 lines
40 KiB
Python
# -*- 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-RPA-003",
|
||
"POLICY-PACKAGING-001",
|
||
"POLICY-PACKAGING-002",
|
||
"POLICY-DOCS-001",
|
||
"POLICY-LOGGING-001",
|
||
"POLICY-LOGGING-002",
|
||
"POLICY-LOGGING-003",
|
||
"POLICY-LOGGING-004",
|
||
"POLICY-LOGGING-005",
|
||
"POLICY-CONTROL-001",
|
||
"POLICY-CONTROL-002",
|
||
"POLICY-CONTROL-003",
|
||
"POLICY-DATA-PATH-001",
|
||
"POLICY-SKILL-ACTION-001",
|
||
"POLICY-SKILL-ACTION-002",
|
||
"POLICY-SKILL-ACTION-003",
|
||
"POLICY-SKILL-ACTION-004",
|
||
"POLICY-ARCH-CORE-001",
|
||
"POLICY-DATA-SYNC-001",
|
||
)
|
||
|
||
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"
|
||
|
||
RPA_003_SOURCE = (
|
||
"development/RPA.md §5.3; development/TESTING.md §12; development/POLICY_MATRIX.md"
|
||
)
|
||
|
||
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),
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicyRpa003(unittest.TestCase):
|
||
def test_service_layer_retains_video_session_integration(self) -> None:
|
||
skill_root = get_skill_root()
|
||
service_files = _walk_files(skill_root, "scripts/service", suffix=".py")
|
||
self.assertTrue(service_files, msg="scripts/service/*.py not found")
|
||
combined = "\n".join(_read_text(path) for path in service_files)
|
||
missing: list[str] = []
|
||
if "RpaVideoSession" not in combined:
|
||
missing.append("RpaVideoSession")
|
||
if "video.add_step" not in combined and ".add_step(" not in combined:
|
||
missing.append("video.add_step or .add_step(")
|
||
if "merge_video_into_result_summary" not in combined:
|
||
missing.append("merge_video_into_result_summary")
|
||
self.assertEqual(
|
||
missing,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-RPA-003",
|
||
RPA_003_SOURCE,
|
||
"scripts/service/*.py missing: " + ", ".join(missing),
|
||
),
|
||
)
|
||
|
||
|
||
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 TestPolicyControl001(unittest.TestCase):
|
||
def test_no_util_progress_module(self) -> None:
|
||
skill_root = get_skill_root()
|
||
path = os.path.join(skill_root, "scripts", "util", "progress.py")
|
||
self.assertFalse(
|
||
os.path.isfile(path),
|
||
msg=_policy_msg(
|
||
"POLICY-CONTROL-001",
|
||
"development/LOGGING.md §2.5; development/RPA.md §0.1",
|
||
"scripts/util/progress.py must not exist",
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicyControl002(unittest.TestCase):
|
||
def test_task_service_uses_job_context_and_finish(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 "job_context(" not in text:
|
||
missing.append("job_context(")
|
||
if "finish(" not in text:
|
||
missing.append("finish(")
|
||
self.assertEqual(
|
||
missing,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-CONTROL-002",
|
||
"development/LOGGING.md §2.5、§7",
|
||
f"{rel} missing: " + ", ".join(missing),
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicyControl003(unittest.TestCase):
|
||
def test_service_layer_has_no_bare_asyncio_sleep(self) -> None:
|
||
skill_root = get_skill_root()
|
||
offenders: list[str] = []
|
||
for path in _walk_files(skill_root, "scripts/service", 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 "asyncio.sleep" in stripped:
|
||
offenders.append(f"{rel}:{lineno}: {stripped}")
|
||
self.assertEqual(
|
||
offenders,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-CONTROL-003",
|
||
"development/RPA.md §0.1",
|
||
"use interruptible_sleep instead:\n" + "\n".join(offenders),
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicyDataPath001(unittest.TestCase):
|
||
def test_env_example_has_no_cwd_relative_paths(self) -> None:
|
||
path = os.path.join(get_skill_root(), ".env.example")
|
||
offenders: list[str] = []
|
||
with open(path, encoding="utf-8") as f:
|
||
for lineno, line in enumerate(f, 1):
|
||
stripped = line.strip()
|
||
if not stripped or stripped.startswith("#"):
|
||
continue
|
||
active = stripped.split("#", 1)[0].strip()
|
||
if re.search(r"=\s*\./", active):
|
||
offenders.append(f".env.example:{lineno}: {active}")
|
||
self.assertEqual(
|
||
offenders,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-DATA-PATH-001",
|
||
"development/DATA_PATHS.md",
|
||
"\n".join(offenders) or "CWD-relative ./ paths in .env.example",
|
||
),
|
||
)
|
||
|
||
def test_runtime_paths_exposes_resolve_data_path(self) -> None:
|
||
import util.runtime_paths as rp
|
||
|
||
self.assertTrue(callable(getattr(rp, "resolve_data_path", None)))
|
||
self.assertTrue(callable(getattr(rp, "list_resolved_data_paths", None)))
|
||
|
||
def test_business_scripts_do_not_abspath_config_get(self) -> None:
|
||
skill_root = get_skill_root()
|
||
pattern = re.compile(r"os\.path\.abspath\s*\(\s*config\.get\b")
|
||
offenders: list[str] = []
|
||
for rel in _walk_files(skill_root, "scripts", suffix=".py"):
|
||
if rel.replace("\\", "/") == "scripts/util/runtime_paths.py":
|
||
continue
|
||
text = _read_text(os.path.join(skill_root, rel))
|
||
for lineno, line in enumerate(text.splitlines(), 1):
|
||
if line.strip().startswith("#"):
|
||
continue
|
||
if pattern.search(line):
|
||
offenders.append(f"{rel}:{lineno}: {line.strip()}")
|
||
self.assertEqual(
|
||
offenders,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-DATA-PATH-001",
|
||
"development/DATA_PATHS.md",
|
||
"\n".join(offenders),
|
||
),
|
||
)
|
||
|
||
|
||
_TEMPLATE_PLACEHOLDER_SLUG = "your-skill-slug"
|
||
_RPA_SESSION_TOKENS = ("RpaVideoSession",)
|
||
|
||
|
||
def _service_has_rpa_video_session(skill_root: str) -> bool:
|
||
service_dir = os.path.join(skill_root, "scripts", "service")
|
||
if not os.path.isdir(service_dir):
|
||
return False
|
||
for rel in _walk_files(skill_root, "scripts/service", suffix=".py"):
|
||
text = _read_text(os.path.join(skill_root, rel))
|
||
if any(token in text for token in _RPA_SESSION_TOKENS):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _load_actions_json(skill_root: str) -> dict | None:
|
||
path = os.path.join(skill_root, "assets", "actions.json")
|
||
if not os.path.isfile(path):
|
||
return None
|
||
import json
|
||
|
||
with open(path, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
class TestPolicySkillAction001(unittest.TestCase):
|
||
def test_skill_action_runtime_doc_exists(self) -> None:
|
||
path = os.path.join(get_skill_root(), "development", "SKILL_ACTION_RUNTIME.md")
|
||
self.assertTrue(
|
||
os.path.isfile(path),
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-001",
|
||
"development/SKILL_ACTION_RUNTIME.md",
|
||
"file missing",
|
||
),
|
||
)
|
||
text = _read_text(path)
|
||
for marker in ("executionProfile", "async", "任务中心", "run_skill_action"):
|
||
self.assertIn(
|
||
marker,
|
||
text,
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-001",
|
||
"development/SKILL_ACTION_RUNTIME.md",
|
||
f"missing marker {marker!r}",
|
||
),
|
||
)
|
||
|
||
def test_schema_defines_execution_profile(self) -> None:
|
||
import json
|
||
|
||
path = os.path.join(get_skill_root(), "assets", "schemas", "skill-actions.schema.json")
|
||
self.assertTrue(os.path.isfile(path))
|
||
with open(path, encoding="utf-8") as f:
|
||
schema = json.load(f)
|
||
action_props = schema.get("$defs", {}).get("action", {}).get("properties", {})
|
||
profile = action_props.get("executionProfile")
|
||
self.assertIsInstance(
|
||
profile,
|
||
dict,
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-001",
|
||
"assets/schemas/skill-actions.schema.json",
|
||
"action.properties.executionProfile missing",
|
||
),
|
||
)
|
||
self.assertEqual(set(profile.get("enum") or []), {"sync", "async"})
|
||
|
||
def test_actions_md_documents_execution_profile(self) -> None:
|
||
path = os.path.join(get_skill_root(), "references", "ACTIONS.md")
|
||
text = _read_text(path)
|
||
for marker in ("executionProfile", "async", "任务中心", "SKILL_ACTION_RUNTIME"):
|
||
self.assertIn(
|
||
marker,
|
||
text,
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-001",
|
||
"references/ACTIONS.md",
|
||
f"missing marker {marker!r}",
|
||
),
|
||
)
|
||
|
||
def test_actions_json_actions_declare_execution_profile(self) -> None:
|
||
skill_root = get_skill_root()
|
||
manifest = _load_actions_json(skill_root)
|
||
if manifest is None:
|
||
return
|
||
offenders: list[str] = []
|
||
for action in manifest.get("actions") or []:
|
||
action_id = action.get("id", "?")
|
||
profile = action.get("executionProfile")
|
||
if profile not in ("sync", "async"):
|
||
offenders.append(f"{action_id}: executionProfile={profile!r}")
|
||
self.assertEqual(
|
||
offenders,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-001",
|
||
"assets/actions.json",
|
||
"every action must set executionProfile to sync|async:\n" + "\n".join(offenders),
|
||
),
|
||
)
|
||
|
||
def test_business_rpa_skill_has_async_agent_action(self) -> None:
|
||
from util.constants import SKILL_SLUG
|
||
|
||
skill_root = get_skill_root()
|
||
if SKILL_SLUG == _TEMPLATE_PLACEHOLDER_SLUG:
|
||
self.skipTest("template placeholder slug skips business RPA action check")
|
||
if not _service_has_rpa_video_session(skill_root):
|
||
self.skipTest("no RpaVideoSession in scripts/service")
|
||
manifest = _load_actions_json(skill_root)
|
||
self.assertIsNotNone(
|
||
manifest,
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-001",
|
||
"assets/actions.json",
|
||
"RPA skill must provide assets/actions.json",
|
||
),
|
||
)
|
||
found = False
|
||
for action in manifest.get("actions") or []:
|
||
placements = set(action.get("placements") or [])
|
||
if action.get("executionProfile") == "async" and "agent" in placements:
|
||
found = True
|
||
break
|
||
self.assertTrue(
|
||
found,
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-001",
|
||
"assets/actions.json",
|
||
"RPA business skill needs at least one action with executionProfile=async and placements containing agent",
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicySkillAction002(unittest.TestCase):
|
||
def test_schema_requires_execution_profile(self) -> None:
|
||
import json
|
||
|
||
path = os.path.join(get_skill_root(), "assets", "schemas", "skill-actions.schema.json")
|
||
with open(path, encoding="utf-8") as f:
|
||
schema = json.load(f)
|
||
action = schema["$defs"]["action"]
|
||
self.assertIn(
|
||
"executionProfile",
|
||
action.get("required") or [],
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-002",
|
||
"assets/schemas/skill-actions.schema.json",
|
||
"executionProfile must be in action.required",
|
||
),
|
||
)
|
||
self.assertEqual(set(action["properties"]["executionProfile"].get("enum") or []), {"sync", "async"})
|
||
|
||
def test_manifest_actions_declare_execution_profile(self) -> None:
|
||
manifest = _load_actions_json(get_skill_root())
|
||
if manifest is None:
|
||
return
|
||
offenders: list[str] = []
|
||
for action in manifest.get("actions") or []:
|
||
profile = action.get("executionProfile")
|
||
if profile not in ("sync", "async"):
|
||
offenders.append(f"{action.get('id', '?')}: {profile!r}")
|
||
self.assertEqual(
|
||
offenders,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-002",
|
||
"assets/actions.json",
|
||
"every action must set executionProfile to sync|async:\n" + "\n".join(offenders),
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicySkillAction003(unittest.TestCase):
|
||
def test_schema_defines_bind_and_toolbar_requires_it(self) -> None:
|
||
import json
|
||
|
||
path = os.path.join(get_skill_root(), "assets", "schemas", "skill-actions.schema.json")
|
||
with open(path, encoding="utf-8") as f:
|
||
schema = json.load(f)
|
||
self.assertIn(
|
||
"bind",
|
||
schema.get("$defs", {}),
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-003",
|
||
"assets/schemas/skill-actions.schema.json",
|
||
"$defs.bind missing",
|
||
),
|
||
)
|
||
action = schema["$defs"]["action"]
|
||
self.assertIn("bind", action.get("properties", {}))
|
||
then_required = (action.get("then") or {}).get("required") or []
|
||
self.assertIn(
|
||
"bind",
|
||
then_required,
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-003",
|
||
"assets/schemas/skill-actions.schema.json",
|
||
"action if/then must require bind when placements contains toolbar",
|
||
),
|
||
)
|
||
|
||
def test_toolbar_actions_have_bind_tables(self) -> None:
|
||
import re
|
||
|
||
table_re = re.compile(r"^[a-z][a-z0-9_]*$")
|
||
manifest = _load_actions_json(get_skill_root())
|
||
if manifest is None:
|
||
return
|
||
offenders: list[str] = []
|
||
for action in manifest.get("actions") or []:
|
||
if "toolbar" not in (action.get("placements") or []):
|
||
continue
|
||
bind = action.get("bind")
|
||
tables = bind.get("tables") if isinstance(bind, dict) else None
|
||
if not isinstance(tables, list) or not tables:
|
||
offenders.append(f"{action.get('id')}: missing bind.tables")
|
||
continue
|
||
if len(tables) != len(set(tables)):
|
||
offenders.append(f"{action.get('id')}: duplicate bind.tables")
|
||
for table in tables:
|
||
if not isinstance(table, str) or not table_re.fullmatch(table):
|
||
offenders.append(f"{action.get('id')}: invalid table {table!r}")
|
||
self.assertEqual(
|
||
offenders,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-003",
|
||
"assets/actions.json",
|
||
"toolbar actions need legal bind.tables:\n" + "\n".join(offenders),
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicySkillAction004(unittest.TestCase):
|
||
def _forbidden_coupling_phrases(self) -> tuple[str, ...]:
|
||
# 拆开拼接，避免本测试源码或废止说明原文触发误报。
|
||
return (
|
||
"数据管理 toolbar " + "只挂",
|
||
"toolbar " + "只放 async",
|
||
"只挂 **" + "async** action",
|
||
"toolbar " + "只能 async",
|
||
"toolbar only " + "async",
|
||
)
|
||
|
||
def test_docs_do_not_couple_toolbar_to_async(self) -> None:
|
||
skill_root = get_skill_root()
|
||
paths = (
|
||
os.path.join(skill_root, "references", "ACTIONS.md"),
|
||
os.path.join(skill_root, "development", "SKILL_ACTION_RUNTIME.md"),
|
||
os.path.join(skill_root, "SKILL.md"),
|
||
)
|
||
hits: list[str] = []
|
||
for path in paths:
|
||
text = _read_text(path)
|
||
for phrase in self._forbidden_coupling_phrases():
|
||
if phrase in text:
|
||
hits.append(f"{os.path.relpath(path, skill_root)}: {phrase!r}")
|
||
self.assertEqual(
|
||
hits,
|
||
[],
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-004",
|
||
"docs",
|
||
"placements must stay orthogonal to executionProfile:\n" + "\n".join(hits),
|
||
),
|
||
)
|
||
|
||
def test_runtime_doc_states_orthogonality(self) -> None:
|
||
text = _read_text(os.path.join(get_skill_root(), "development", "SKILL_ACTION_RUNTIME.md"))
|
||
for marker in ("正交", "sync|async", "toolbar"):
|
||
self.assertIn(
|
||
marker,
|
||
text,
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-004",
|
||
"development/SKILL_ACTION_RUNTIME.md",
|
||
f"missing marker {marker!r}",
|
||
),
|
||
)
|
||
|
||
def test_schema_then_only_requires_bind_not_execution_profile(self) -> None:
|
||
"""toolbar→bind 的 if/then 不得顺带约束 executionProfile。"""
|
||
import json
|
||
|
||
path = os.path.join(get_skill_root(), "assets", "schemas", "skill-actions.schema.json")
|
||
with open(path, encoding="utf-8") as f:
|
||
schema = json.load(f)
|
||
action = schema["$defs"]["action"]
|
||
then = action.get("then") or {}
|
||
self.assertEqual(
|
||
then.get("required"),
|
||
["bind"],
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-004",
|
||
"assets/schemas/skill-actions.schema.json",
|
||
"action.then must only require bind",
|
||
),
|
||
)
|
||
self.assertNotIn("properties", then)
|
||
self.assertNotIn("else", action)
|
||
matrix_test = os.path.join(get_skill_root(), "tests", "test_actions_manifest.py")
|
||
text = _read_text(matrix_test)
|
||
self.assertIn(
|
||
"test_schema_allows_all_placement_execution_profile_combinations",
|
||
text,
|
||
msg=_policy_msg(
|
||
"POLICY-SKILL-ACTION-004",
|
||
"tests/test_actions_manifest.py",
|
||
"missing Schema 4×2 orthogonality matrix test",
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicyArchCore001(unittest.TestCase):
|
||
def test_docs_declare_single_core_multi_entry(self) -> None:
|
||
skill_root = get_skill_root()
|
||
text = (
|
||
_read_text(os.path.join(skill_root, "development", "DEVELOPMENT.md"))
|
||
+ "\n"
|
||
+ _read_text(os.path.join(skill_root, "development", "SKILL_ACTION_RUNTIME.md"))
|
||
+ "\n"
|
||
+ _read_text(os.path.join(skill_root, "SKILL.md"))
|
||
)
|
||
for marker in ("单业务内核", "多入口", "domain service"):
|
||
self.assertIn(
|
||
marker,
|
||
text,
|
||
msg=_policy_msg(
|
||
"POLICY-ARCH-CORE-001",
|
||
"development docs / SKILL.md",
|
||
f"missing architecture marker {marker!r}",
|
||
),
|
||
)
|
||
sample = os.path.join(skill_root, "tests", "samples", "test_action_core_contract.py.sample")
|
||
self.assertTrue(
|
||
os.path.isfile(sample),
|
||
msg=_policy_msg(
|
||
"POLICY-ARCH-CORE-001",
|
||
"tests/samples/test_action_core_contract.py.sample",
|
||
"sample contract missing",
|
||
),
|
||
)
|
||
|
||
|
||
class TestPolicyDataSync001(unittest.TestCase):
|
||
def test_schema_md_declares_sync_semantics(self) -> None:
|
||
text = _read_text(os.path.join(get_skill_root(), "references", "SCHEMA.md"))
|
||
for marker in ("幂等", "空结果", "事务", "inserted", "导出当前数据表"):
|
||
self.assertIn(
|
||
marker,
|
||
text,
|
||
msg=_policy_msg(
|
||
"POLICY-DATA-SYNC-001",
|
||
"references/SCHEMA.md",
|
||
f"missing sync marker {marker!r}",
|
||
),
|
||
)
|
||
sample = os.path.join(get_skill_root(), "tests", "samples", "test_sync_contract.py.sample")
|
||
self.assertTrue(
|
||
os.path.isfile(sample),
|
||
msg=_policy_msg(
|
||
"POLICY-DATA-SYNC-001",
|
||
"tests/samples/test_sync_contract.py.sample",
|
||
"sample contract missing",
|
||
),
|
||
)
|
||
|
||
|
||
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()
|