3db196e49d
基于 skill-template 脚手架开发，包含： - payroll_config（费率配置，可编辑）+ payslip_snapshots（解析历史，只读）2 张业务表 - 全国默认 + 北上广 4 套费率（社保/公积金/个税起征点/7 级税率表） - 5 个业务 Action：calculate-payslip-simple / calculate-payslip-full / explain-term / view-config / edit-config - 个税累计预扣法计算，支持多月历史累加 - 20 个薪资名词库（社保/公积金/个税/全勤/加班费等） - 完整市场文档（README/TUTORIAL/DEMO/CHANGELOG） - 296 测试全部通过
667 lines
28 KiB
Python
667 lines
28 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""assets/actions.json 结构与 CLI 映射守护（通用模板，不含具体业务技能名称）。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import unittest
|
||
|
||
from _support import IsolatedDataRoot, get_skill_root
|
||
|
||
from cli.app import build_parser
|
||
from util.constants import SKILL_SLUG
|
||
|
||
_TEMPLATE_PLACEHOLDER_SLUG = "query-payslip-default"
|
||
_ALLOWED_PLACEMENTS = frozenset({"toolbar", "row", "batch", "cron", "agent", "skill-detail"})
|
||
_RESERVED_ACTION_FIELDS = frozenset({"concurrency", "locks"})
|
||
_ALLOWED_EXECUTION_PROFILES = frozenset({"sync", "async"})
|
||
_REQUIRED_MANIFEST_KEYS = frozenset({"schemaVersion", "skill", "actions"})
|
||
_REQUIRED_ACTION_KEYS = frozenset(
|
||
{"id", "label", "description", "placements", "entrypoint", "executionProfile"}
|
||
)
|
||
_TABLE_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]*$")
|
||
_FORBIDDEN_BUSINESS_TERMS = (
|
||
"wechat",
|
||
"微信",
|
||
"contacts",
|
||
"wxid",
|
||
"account_wxid",
|
||
"contact.db",
|
||
"scrape-contacts",
|
||
"douyin",
|
||
"抖音",
|
||
"xiaohongshu",
|
||
"小红书",
|
||
"公众号",
|
||
"logistics",
|
||
"物流",
|
||
)
|
||
|
||
_TEMPLATE_ARG_PATTERN = re.compile(r"\{\{([a-zA-Z_][a-zA-Z0-9_]*)\}\}")
|
||
|
||
|
||
def _load_actions_manifest() -> dict:
|
||
path = os.path.join(get_skill_root(), "assets", "actions.json")
|
||
with open(path, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def _cli_subcommands() -> set[str]:
|
||
parser = build_parser()
|
||
subs = parser._subparsers._group_actions[0].choices # type: ignore[attr-defined]
|
||
return set(subs.keys())
|
||
|
||
|
||
def _collect_template_vars(args: list) -> set[str]:
|
||
found: set[str] = set()
|
||
for item in args:
|
||
if not isinstance(item, str):
|
||
continue
|
||
for match in _TEMPLATE_ARG_PATTERN.finditer(item):
|
||
found.add(match.group(1))
|
||
return found
|
||
|
||
|
||
def _load_actions_schema() -> dict:
|
||
path = os.path.join(get_skill_root(), "assets", "schemas", "skill-actions.schema.json")
|
||
with open(path, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def _validate_bind_tables(action_id: str, bind: object) -> list[str]:
|
||
errors: list[str] = []
|
||
if not isinstance(bind, dict):
|
||
return [f"{action_id}: bind must be object"]
|
||
allowed = {"tables", "inputMapping", "columnMapping", "labelColumn"}
|
||
unknown = set(bind.keys()) - allowed
|
||
if unknown:
|
||
errors.append(f"{action_id}: bind only allows {sorted(allowed)}, got {sorted(bind.keys())}")
|
||
tables = bind.get("tables")
|
||
if not isinstance(tables, list) or len(tables) < 1:
|
||
errors.append(f"{action_id}: bind.tables must be non-empty array")
|
||
return errors
|
||
seen: set[str] = set()
|
||
for table in tables:
|
||
if not isinstance(table, str) or not table.strip():
|
||
errors.append(f"{action_id}: bind.tables entries must be non-empty strings")
|
||
continue
|
||
if table != table.strip() or not _TABLE_NAME_PATTERN.fullmatch(table):
|
||
errors.append(f"{action_id}: bind.tables entry {table!r} must be snake_case")
|
||
if table in seen:
|
||
errors.append(f"{action_id}: bind.tables duplicate {table!r}")
|
||
seen.add(table)
|
||
label_column = bind.get("labelColumn")
|
||
if label_column is not None:
|
||
if not isinstance(label_column, str) or not label_column.strip():
|
||
errors.append(f"{action_id}: bind.labelColumn must be non-empty string")
|
||
elif not _TABLE_NAME_PATTERN.fullmatch(label_column):
|
||
errors.append(f"{action_id}: bind.labelColumn {label_column!r} must be snake_case")
|
||
return errors
|
||
|
||
|
||
def _load_bindable_business_tables() -> set[str]:
|
||
"""在隔离临时数据根下 init_db，返回可绑定到 toolbar 的业务表集合。
|
||
|
||
表必须同时满足：
|
||
- SQLite 物理表存在（排除 sqlite_* 与 _jiangchang_* 元数据表）
|
||
- ``_jiangchang_tables`` 中有对应行且 ``visible=1``
|
||
|
||
模板默认无业务表，允许返回空集。失败时抛出 RuntimeError（带诊断）。
|
||
"""
|
||
from db.connection import init_db
|
||
from db.display_metadata import METADATA_TABLES
|
||
from util.runtime_paths import get_db_path
|
||
|
||
physical: set[str] = set()
|
||
visible_meta: set[str] = set()
|
||
try:
|
||
with IsolatedDataRoot() as data_root:
|
||
init_db()
|
||
db_path = get_db_path()
|
||
if not os.path.isfile(db_path):
|
||
raise RuntimeError(
|
||
f"init_db did not create database file under isolated root "
|
||
f"{data_root!r}: expected {db_path}"
|
||
)
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT name FROM sqlite_master "
|
||
"WHERE type='table' AND name NOT LIKE 'sqlite_%'"
|
||
)
|
||
physical = {
|
||
str(row[0])
|
||
for row in cur.fetchall()
|
||
if str(row[0]) not in METADATA_TABLES
|
||
}
|
||
cur.execute(
|
||
"SELECT table_name FROM _jiangchang_tables WHERE visible = 1"
|
||
)
|
||
visible_meta = {str(row[0]) for row in cur.fetchall()}
|
||
finally:
|
||
conn.close()
|
||
bindable = physical & visible_meta
|
||
return bindable
|
||
except RuntimeError:
|
||
raise
|
||
except Exception as exc:
|
||
raise RuntimeError(
|
||
"failed to load bindable business tables via isolated init_db: "
|
||
f"{type(exc).__name__}: {exc}"
|
||
) from exc
|
||
|
||
|
||
class TestActionsManifest(unittest.TestCase):
|
||
def test_manifest_exists_and_schema(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
self.assertEqual(manifest.get("schemaVersion"), 1)
|
||
skill = manifest.get("skill")
|
||
self.assertIn(skill, (SKILL_SLUG, _TEMPLATE_PLACEHOLDER_SLUG))
|
||
self.assertIsInstance(manifest.get("actions"), list)
|
||
self.assertGreater(len(manifest["actions"]), 0)
|
||
|
||
def test_action_ids_unique_and_cli_valid(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
subs = _cli_subcommands()
|
||
seen: set[str] = set()
|
||
for action in manifest["actions"]:
|
||
action_id = action["id"]
|
||
self.assertNotIn(action_id, seen, msg=f"duplicate action id: {action_id}")
|
||
seen.add(action_id)
|
||
|
||
self.assertTrue(str(action.get("label", "")).strip(), msg=f"{action_id} label empty")
|
||
self.assertTrue(str(action.get("description", "")).strip(), msg=f"{action_id} description empty")
|
||
|
||
entry = action["entrypoint"]
|
||
self.assertEqual(entry["type"], "cli")
|
||
self.assertIn(entry["command"], subs, msg=f"{action_id} command not in CLI")
|
||
|
||
args = entry.get("args")
|
||
self.assertIsInstance(args, list, msg=f"{action_id} args must be array")
|
||
for item in args:
|
||
self.assertIsInstance(
|
||
item,
|
||
(str, int, float, bool),
|
||
msg=f"{action_id} arg item must be scalar",
|
||
)
|
||
self.assertNotIsInstance(item, dict, msg=f"{action_id} arg must not be object")
|
||
self.assertIsNotNone(item, msg=f"{action_id} arg must not be null")
|
||
|
||
placements = action.get("placements")
|
||
self.assertIsInstance(placements, list, msg=f"{action_id} placements must be array")
|
||
replaces = action.get("replaces")
|
||
if replaces is not None:
|
||
self.assertIn(
|
||
replaces,
|
||
{"create", "update", "delete", "import"},
|
||
msg=f"{action_id} invalid replaces {replaces!r}",
|
||
)
|
||
bind = action.get("bind") or {}
|
||
tables = bind.get("tables") or []
|
||
self.assertTrue(
|
||
isinstance(tables, list) and len(tables) > 0,
|
||
msg=f"{action_id} replaces requires non-empty bind.tables",
|
||
)
|
||
else:
|
||
self.assertGreater(len(placements), 0, msg=f"{action_id} placements empty")
|
||
for placement in placements:
|
||
self.assertIn(placement, _ALLOWED_PLACEMENTS, msg=f"{action_id} invalid placement {placement!r}")
|
||
|
||
def test_every_action_declares_execution_profile(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
for action in manifest["actions"]:
|
||
profile = action.get("executionProfile")
|
||
self.assertIn(
|
||
profile,
|
||
_ALLOWED_EXECUTION_PROFILES,
|
||
msg=f"{action['id']} executionProfile must be sync|async, got {profile!r}",
|
||
)
|
||
|
||
def test_schema_requires_execution_profile_and_defines_bind(self) -> None:
|
||
schema = _load_actions_schema()
|
||
action_def = schema["$defs"]["action"]
|
||
self.assertIn("executionProfile", action_def.get("required") or [])
|
||
profile = action_def["properties"]["executionProfile"]
|
||
self.assertEqual(set(profile.get("enum") or []), {"sync", "async"})
|
||
self.assertIn("bind", schema["$defs"])
|
||
self.assertIn("bind", action_def["properties"])
|
||
bind_def = schema["$defs"]["bind"]
|
||
self.assertEqual(bind_def.get("required"), ["tables"])
|
||
self.assertIs(bind_def.get("additionalProperties"), False)
|
||
|
||
def test_schema_defines_interaction_fields(self) -> None:
|
||
schema = _load_actions_schema()
|
||
action_def = schema["$defs"]["action"]
|
||
props = action_def["properties"]
|
||
self.assertIn("interaction", props)
|
||
self.assertEqual(
|
||
set(props["interaction"].get("enum") or []),
|
||
{"auto", "direct", "confirm", "form"},
|
||
)
|
||
self.assertIn("interactionByPlacement", props)
|
||
by_placement = props["interactionByPlacement"]
|
||
self.assertIs(by_placement.get("additionalProperties"), False)
|
||
for key in ("toolbar", "row", "batch", "cron", "agent", "skill-detail", "create", "update", "delete", "import"):
|
||
self.assertIn(key, by_placement.get("properties") or {})
|
||
self.assertEqual(
|
||
set((by_placement["properties"][key].get("enum") or [])),
|
||
{"direct", "confirm", "form"},
|
||
)
|
||
self.assertIn("replaces", props)
|
||
self.assertEqual(props["replaces"].get("$ref"), "#/$defs/mutationKind")
|
||
self.assertEqual(
|
||
set(schema["$defs"]["mutationKind"].get("enum") or []),
|
||
{"create", "update", "delete", "import"},
|
||
)
|
||
|
||
def test_schema_defines_hover_hint(self) -> None:
|
||
schema = _load_actions_schema()
|
||
action_def = schema["$defs"]["action"]
|
||
props = action_def["properties"]
|
||
self.assertIn("hoverHint", props)
|
||
self.assertNotIn("hoverHint", action_def.get("required") or [])
|
||
self.assertEqual(props["hoverHint"].get("type"), "string")
|
||
|
||
def test_schema_defines_visible_when_on_input_property(self) -> None:
|
||
schema = _load_actions_schema()
|
||
prop_def = schema["$defs"]["inputProperty"]
|
||
props = prop_def["properties"]
|
||
self.assertIn("visibleWhen", props)
|
||
self.assertNotIn("visibleWhen", prop_def.get("required") or [])
|
||
# Must live on input property, not action top-level (≠ Policy L3).
|
||
action_props = schema["$defs"]["action"]["properties"]
|
||
self.assertNotIn("visibleWhen", action_props)
|
||
|
||
def test_default_manifest_omits_visible_when(self) -> None:
|
||
for action in _load_actions_manifest()["actions"]:
|
||
properties = ((action.get("inputSchema") or {}).get("properties")) or {}
|
||
for name, prop in properties.items():
|
||
self.assertNotIn(
|
||
"visibleWhen",
|
||
prop if isinstance(prop, dict) else {},
|
||
msg=f"{action['id']}.{name} default template should omit visibleWhen",
|
||
)
|
||
|
||
def test_visible_when_example_exists_and_uses_field_rules(self) -> None:
|
||
path = os.path.join(
|
||
get_skill_root(), "assets", "examples", "visible-when-action.example.json"
|
||
)
|
||
self.assertTrue(os.path.isfile(path), msg="visible-when example must exist")
|
||
with open(path, encoding="utf-8") as f:
|
||
example = json.load(f)
|
||
props = example["actions"][0]["inputSchema"]["properties"]
|
||
self.assertIn("visibleWhen", props["keyword"])
|
||
self.assertIn("visibleWhen", props["authorUrl"])
|
||
self.assertNotIn("visibleWhen", props["collectMode"])
|
||
|
||
def test_default_manifest_omits_hover_hint(self) -> None:
|
||
for action in _load_actions_manifest()["actions"]:
|
||
self.assertNotIn(
|
||
"hoverHint",
|
||
action,
|
||
msg=f"{action['id']} default template action should omit hoverHint",
|
||
)
|
||
|
||
def test_schema_defines_host_effects_and_steps(self) -> None:
|
||
schema = _load_actions_schema()
|
||
action_def = schema["$defs"]["action"]
|
||
props = action_def["properties"]
|
||
self.assertIn("hostEffects", props)
|
||
self.assertEqual(props["hostEffects"].get("$ref"), "#/$defs/hostEffects")
|
||
self.assertIn("steps", props)
|
||
effects = schema["$defs"]["hostEffects"]
|
||
self.assertIs(effects.get("additionalProperties"), False)
|
||
for key in ("onSuccess", "onFailure"):
|
||
self.assertIn(key, effects.get("properties") or {})
|
||
effect_ids = set(schema["$defs"]["hostEffectId"].get("enum") or [])
|
||
self.assertEqual(effect_ids, {"reloadTable", "refreshCatalog", "clearSelection"})
|
||
step = schema["$defs"]["actionStep"]
|
||
self.assertEqual(
|
||
set(step["properties"]["when"].get("enum") or []),
|
||
{"before", "afterSuccess", "afterFailure"},
|
||
)
|
||
self.assertEqual(
|
||
set(step["properties"]["do"].get("enum") or []),
|
||
{"deleteByFk", "deleteRows", "reloadTable", "refreshCatalog", "clearSelection"},
|
||
)
|
||
self.assertEqual(props["steps"].get("maxItems"), 8)
|
||
|
||
def test_manifest_validates_against_json_schema(self) -> None:
|
||
try:
|
||
import jsonschema
|
||
except ImportError:
|
||
self.skipTest("jsonschema not installed")
|
||
schema = _load_actions_schema()
|
||
manifest = _load_actions_manifest()
|
||
jsonschema.validate(instance=manifest, schema=schema)
|
||
|
||
def test_bind_structure_when_present(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
errors: list[str] = []
|
||
for action in manifest["actions"]:
|
||
if "bind" not in action:
|
||
continue
|
||
errors.extend(_validate_bind_tables(action["id"], action["bind"]))
|
||
self.assertEqual(errors, [], msg="\n".join(errors))
|
||
|
||
def test_toolbar_actions_require_bind_tables(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
errors: list[str] = []
|
||
for action in manifest["actions"]:
|
||
placements = action.get("placements") or []
|
||
if "toolbar" not in placements:
|
||
continue
|
||
bind = action.get("bind")
|
||
if bind is None:
|
||
errors.append(f"{action['id']}: toolbar placement requires bind.tables")
|
||
continue
|
||
errors.extend(_validate_bind_tables(action["id"], bind))
|
||
self.assertEqual(errors, [], msg="\n".join(errors))
|
||
|
||
def test_bindable_business_tables_helper_loads_template_db(self) -> None:
|
||
"""helper 本身必须可用：本技能有 2 张业务表，隔离库应可读。"""
|
||
bindable = _load_bindable_business_tables()
|
||
self.assertEqual(bindable, {"payroll_config", "payslip_snapshots"})
|
||
self.assertNotIn("_jiangchang_tables", bindable)
|
||
self.assertNotIn("_jiangchang_columns", bindable)
|
||
|
||
def test_bind_tables_exist_in_physical_and_visible_metadata(self) -> None:
|
||
bindable = _load_bindable_business_tables()
|
||
manifest = _load_actions_manifest()
|
||
missing: list[str] = []
|
||
for action in manifest["actions"]:
|
||
bind = action.get("bind")
|
||
if not isinstance(bind, dict):
|
||
continue
|
||
for table in bind.get("tables") or []:
|
||
if table not in bindable:
|
||
missing.append(
|
||
f"{action['id']}: bind.tables entry {table!r} must exist as "
|
||
f"physical table AND _jiangchang_tables.visible=1 "
|
||
f"(bindable={sorted(bindable)})"
|
||
)
|
||
self.assertEqual(missing, [], msg="\n".join(missing))
|
||
|
||
def test_schema_allows_all_placement_execution_profile_combinations(self) -> None:
|
||
"""POLICY-SKILL-ACTION-004：6 placements × 2 profiles 必须全部通过 Schema。"""
|
||
try:
|
||
import jsonschema
|
||
except ImportError:
|
||
self.fail("jsonschema is required to validate placement×executionProfile orthogonality")
|
||
schema = _load_actions_schema()
|
||
placements = ("toolbar", "row", "batch", "cron", "agent", "skill-detail")
|
||
profiles = ("sync", "async")
|
||
for placement in placements:
|
||
for profile in profiles:
|
||
action: dict = {
|
||
"id": "probe-action",
|
||
"label": "正交探测",
|
||
"description": "验证 placements 与 executionProfile 无条件耦合",
|
||
"placements": [placement],
|
||
"executionProfile": profile,
|
||
"entrypoint": {"type": "cli", "command": "health", "args": []},
|
||
}
|
||
if placement == "toolbar":
|
||
action["bind"] = {"tables": ["demo_items"]}
|
||
manifest = {
|
||
"schemaVersion": 1,
|
||
"skill": "query-payslip-default",
|
||
"actions": [action],
|
||
}
|
||
with self.subTest(placement=placement, executionProfile=profile):
|
||
jsonschema.validate(instance=manifest, schema=schema)
|
||
if placement != "toolbar":
|
||
self.assertNotIn("bind", action)
|
||
|
||
def test_schema_rejects_toolbar_without_bind(self) -> None:
|
||
try:
|
||
import jsonschema
|
||
from jsonschema import ValidationError
|
||
except ImportError:
|
||
self.fail("jsonschema is required")
|
||
schema = _load_actions_schema()
|
||
manifest = {
|
||
"schemaVersion": 1,
|
||
"skill": "query-payslip-default",
|
||
"actions": [
|
||
{
|
||
"id": "probe-toolbar",
|
||
"label": "缺 bind 探测",
|
||
"description": "toolbar 无 bind 应被 Schema 拒绝",
|
||
"placements": ["toolbar"],
|
||
"executionProfile": "sync",
|
||
"entrypoint": {"type": "cli", "command": "health", "args": []},
|
||
}
|
||
],
|
||
}
|
||
with self.assertRaises(ValidationError):
|
||
jsonschema.validate(instance=manifest, schema=schema)
|
||
|
||
def test_schema_allows_bind_label_column(self) -> None:
|
||
try:
|
||
import jsonschema
|
||
except ImportError:
|
||
self.fail("jsonschema is required")
|
||
schema = _load_actions_schema()
|
||
bind_props = schema["$defs"]["bind"]["properties"]
|
||
self.assertIn("labelColumn", bind_props)
|
||
manifest = {
|
||
"schemaVersion": 1,
|
||
"skill": "query-payslip-default",
|
||
"actions": [
|
||
{
|
||
"id": "probe-label-column",
|
||
"label": "处理勾选",
|
||
"description": "验证 bind.labelColumn 可通过 Schema",
|
||
"placements": ["toolbar", "batch"],
|
||
"executionProfile": "sync",
|
||
"interaction": "confirm",
|
||
"bind": {
|
||
"tables": ["demo_items"],
|
||
"inputMapping": {"ids": "$selection.ids"},
|
||
"labelColumn": "display_name",
|
||
},
|
||
"entrypoint": {"type": "cli", "command": "health", "args": []},
|
||
}
|
||
],
|
||
}
|
||
jsonschema.validate(instance=manifest, schema=schema)
|
||
|
||
def test_no_execution_profile_placement_coupling_in_tests(self) -> None:
|
||
"""守护：本文件不得引入入口位置与 sync/async 的错误硬耦合断言。"""
|
||
path = os.path.join(get_skill_root(), "tests", "test_actions_manifest.py")
|
||
with open(path, encoding="utf-8") as f:
|
||
text = f.read()
|
||
# 拆开拼接，避免本测试源码自身触发字符串扫描。
|
||
forbidden_snippets = (
|
||
"toolbar " + "只能 async",
|
||
"toolbar only " + "async",
|
||
'toolbar"' + " and profile != " + '"async"',
|
||
'toolbar")' + " and profile != " + '"async"',
|
||
"toolbar must be " + "async",
|
||
"executionProfile" + " == " + '"async"' + " and " + '"toolbar"',
|
||
)
|
||
for snippet in forbidden_snippets:
|
||
self.assertNotIn(snippet, text)
|
||
|
||
def test_template_variables_have_input_schema(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
for action in manifest["actions"]:
|
||
args = action.get("entrypoint", {}).get("args") or []
|
||
template_vars = _collect_template_vars(args)
|
||
if not template_vars:
|
||
continue
|
||
schema = action.get("inputSchema") or {}
|
||
properties = schema.get("properties") or {}
|
||
for var in template_vars:
|
||
self.assertIn(
|
||
var,
|
||
properties,
|
||
msg=f"{action['id']} uses {{{{{var}}}}} but inputSchema.properties missing it",
|
||
)
|
||
|
||
def test_required_fields_exist_in_properties(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
for action in manifest["actions"]:
|
||
schema = action.get("inputSchema") or {}
|
||
required = schema.get("required") or []
|
||
properties = schema.get("properties") or {}
|
||
for field in required:
|
||
self.assertIn(
|
||
field,
|
||
properties,
|
||
msg=f"{action['id']} required field {field!r} missing from properties",
|
||
)
|
||
|
||
def test_sensitive_params_not_on_toolbar(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
for action in manifest["actions"]:
|
||
props = (action.get("inputSchema") or {}).get("properties") or {}
|
||
has_sensitive = any(p.get("sensitive") for p in props.values())
|
||
if has_sensitive and "toolbar" in (action.get("placements") or []):
|
||
self.fail(f"{action['id']} has sensitive input but is on toolbar")
|
||
|
||
def test_confirmation_message_structure(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
for action in manifest["actions"]:
|
||
confirmation = action.get("confirmation")
|
||
if confirmation is None:
|
||
continue
|
||
self.assertIsInstance(confirmation, dict)
|
||
message = confirmation.get("message")
|
||
self.assertTrue(str(message or "").strip(), msg=f"{action['id']} confirmation.message empty")
|
||
|
||
def test_manifest_has_no_specific_business_skill_names(self) -> None:
|
||
path = os.path.join(get_skill_root(), "assets", "actions.json")
|
||
with open(path, encoding="utf-8") as f:
|
||
raw = f.read().lower()
|
||
for term in _FORBIDDEN_BUSINESS_TERMS:
|
||
self.assertNotIn(term.lower(), raw, msg=f"actions.json must not contain {term!r}")
|
||
|
||
def test_actions_md_exists_and_links_manifest(self) -> None:
|
||
path = os.path.join(get_skill_root(), "references", "ACTIONS.md")
|
||
self.assertTrue(os.path.isfile(path))
|
||
with open(path, encoding="utf-8") as f:
|
||
text = f.read()
|
||
self.assertIn("actions.json", text)
|
||
self.assertIn("schemaVersion", text)
|
||
for term in _FORBIDDEN_BUSINESS_TERMS:
|
||
self.assertNotIn(term.lower(), text.lower(), msg=f"ACTIONS.md must not contain {term!r}")
|
||
|
||
def test_skill_actions_schema_file_exists(self) -> None:
|
||
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)
|
||
self.assertEqual(schema.get("title"), "SkillActionManifest")
|
||
self.assertIn("actions", schema.get("properties", {}))
|
||
|
||
def test_skill_actions_schema_json_parses(self) -> None:
|
||
schema = _load_actions_schema()
|
||
self.assertEqual(schema["properties"]["schemaVersion"]["const"], 1)
|
||
self.assertTrue(schema.get("additionalProperties") is False)
|
||
self.assertIn("action", schema.get("$defs", {}))
|
||
|
||
def test_template_manifest_matches_schema_required_keys(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
schema = _load_actions_schema()
|
||
for key in _REQUIRED_MANIFEST_KEYS:
|
||
self.assertIn(key, manifest)
|
||
self.assertIn(key, schema.get("properties", {}))
|
||
for action in manifest["actions"]:
|
||
for key in _REQUIRED_ACTION_KEYS:
|
||
self.assertIn(key, action)
|
||
|
||
def test_template_example_does_not_use_reserved_action_fields(self) -> None:
|
||
manifest = _load_actions_manifest()
|
||
for action in manifest["actions"]:
|
||
overlap = set(action.keys()) & _RESERVED_ACTION_FIELDS
|
||
self.assertFalse(
|
||
overlap,
|
||
msg=f"{action['id']} uses reserved action fields not supported in template Phase 1: {overlap}",
|
||
)
|
||
|
||
def test_default_manifest_does_not_place_health_on_toolbar(self) -> None:
|
||
"""脚手架默认健康类 Action 不应出现在数据管理 toolbar。"""
|
||
manifest = _load_actions_manifest()
|
||
for action in manifest["actions"]:
|
||
if action["id"] in {"health", "version", "config-path"}:
|
||
self.assertNotIn(
|
||
"toolbar",
|
||
action.get("placements") or [],
|
||
msg=f"{action['id']} must not be auto-placed on toolbar",
|
||
)
|
||
|
||
def test_actions_md_documents_strict_vs_host_boundary(self) -> None:
|
||
path = os.path.join(get_skill_root(), "references", "ACTIONS.md")
|
||
with open(path, encoding="utf-8") as f:
|
||
text = f.read()
|
||
for marker in (
|
||
"严格规范",
|
||
"向后兼容",
|
||
"Phase 1",
|
||
"bind.tables",
|
||
"concurrency",
|
||
"locks",
|
||
"row",
|
||
"batch",
|
||
"additionalProperties",
|
||
"placements",
|
||
"executionProfile",
|
||
"interaction",
|
||
"confirmation",
|
||
"数据管理技能按钮顺序",
|
||
"数组相对顺序",
|
||
"全部直接展示",
|
||
"左右分栏",
|
||
"行级入口",
|
||
"行右键",
|
||
"六职责正交",
|
||
"POLICY.md",
|
||
"hostEffects",
|
||
"steps",
|
||
"hoverHint",
|
||
"visibleWhen",
|
||
"字段条件显隐",
|
||
"reloadTable",
|
||
"afterSuccess",
|
||
"afterFailure",
|
||
"cellPresentation",
|
||
"CELL_PRESENTATION.md",
|
||
"labelColumn",
|
||
"已选择",
|
||
"revealInForm",
|
||
"映射只读",
|
||
):
|
||
self.assertIn(marker, text, msg=f"ACTIONS.md missing {marker!r}")
|
||
self.assertNotIn(
|
||
"其余进「更多操作」",
|
||
text,
|
||
msg="ACTIONS.md must not document removed overflow more-menu truncation",
|
||
)
|
||
self.assertIn(
|
||
"新增 `row-context-menu` 等 placement",
|
||
text,
|
||
msg="ACTIONS.md must forbid inventing a separate row-context-menu placement",
|
||
)
|
||
placement_enum = _load_actions_schema().get("$defs", {}).get("placement", {}).get("enum") or []
|
||
self.assertNotIn(
|
||
"row-context-menu",
|
||
placement_enum,
|
||
msg="skill-actions.schema must not add row-context-menu to placements enum",
|
||
)
|
||
for forbidden in (
|
||
"数据管理 toolbar " + "只挂",
|
||
"toolbar " + "只放 async",
|
||
"toolbar）→ " + "只挂",
|
||
"只挂 **" + "async**",
|
||
):
|
||
self.assertNotIn(forbidden, text, msg=f"ACTIONS.md must not contain deprecated coupling: {forbidden!r}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|