瑞达
ab2fba868b
Add hidden compatibility flags and exact high-frequency routes to avoid help-driven recovery loops. Restore granular PAT tests and correct the Runtime Catalog versus curated Schema contract.
461 lines
16 KiB
Python
461 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Regression tests for the Chat Skill's bundled Python scripts."""
|
|
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import re
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
MULTI_ROOT = ROOT / 'skills' / 'multi' / 'dingtalk-chat'
|
|
MONO_ROOT = ROOT / 'skills' / 'mono'
|
|
SCRIPT_NAMES = (
|
|
'chat_export_messages.py',
|
|
'chat_history_with_user.py',
|
|
'bot_broadcast.py',
|
|
)
|
|
|
|
|
|
def load_script(name: str):
|
|
path = MULTI_ROOT / 'scripts' / f'{name}.py'
|
|
spec = importlib.util.spec_from_file_location(f'chat_skill_{name}', path)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f'cannot load {path}')
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class ChatSkillScriptPathTest(unittest.TestCase):
|
|
def test_shortcuts_use_progressive_discovery_in_multi_skill(self):
|
|
text = (MULTI_ROOT / 'SKILL.md').read_text(encoding='utf-8')
|
|
block = text.split('<!-- VISIBLE_SHORTCUTS_START -->', 1)[1].split(
|
|
'<!-- VISIBLE_SHORTCUTS_END -->', 1
|
|
)[0]
|
|
catalog = json.loads(
|
|
(ROOT / 'docs' / 'shortcut-public-catalog.json').read_text(
|
|
encoding='utf-8'
|
|
)
|
|
)
|
|
chat_rows = [
|
|
row for row in catalog['results'] if row.get('service') == 'chat'
|
|
]
|
|
schema_catalog = json.loads(
|
|
(
|
|
ROOT
|
|
/ 'internal'
|
|
/ 'cli'
|
|
/ 'schema_catalog'
|
|
/ 'tools'
|
|
/ 'chat.json'
|
|
).read_text(encoding='utf-8')
|
|
)
|
|
schema_shortcuts = [
|
|
path
|
|
for path in schema_catalog['tools']
|
|
if path.startswith('chat.shortcut_')
|
|
]
|
|
exclusions = json.loads(
|
|
(
|
|
ROOT / 'internal' / 'cli' / 'schema_command_exclusions.json'
|
|
).read_text(encoding='utf-8')
|
|
)
|
|
pending_shortcuts = next(
|
|
group['commands']
|
|
for group in exclusions['groups']
|
|
if group['id'] == 'chat-shortcuts-pending-schema-curation'
|
|
)
|
|
self.assertEqual(97, len(chat_rows))
|
|
self.assertEqual(47, len(schema_shortcuts))
|
|
self.assertEqual(50, len(pending_shortcuts))
|
|
self.assertEqual(
|
|
len(chat_rows), len(schema_shortcuts) + len(pending_shortcuts)
|
|
)
|
|
self.assertIn('## Shortcut 发现（按需）', block)
|
|
self.assertIn('`chat` 当前有 97 条公开 shortcut', block)
|
|
self.assertIn(
|
|
'完整清单保留在 Runtime Shortcut Catalog；已完成 Schema curation '
|
|
'的子集可通过 leaf Schema 查询。',
|
|
block,
|
|
)
|
|
self.assertNotIn('Runtime Catalog 与 Schema', block)
|
|
self.assertIn(
|
|
'dws shortcut list --service chat --compact --format json',
|
|
block,
|
|
)
|
|
self.assertNotRegex(block, r'(?m)^\|\s*`\+[a-z0-9-]+`')
|
|
self.assertNotRegex(block, r'(?m)^\|\s*`dws chat \+[a-z0-9-]+`')
|
|
self.assertNotIn('dws chat +', block)
|
|
for required in (
|
|
'This is the only Shortcut entry point for multi/chat',
|
|
'exact recipe/runnable script > matching public Shortcut > atomic command',
|
|
'dws shortcut list --service chat --compact --format json',
|
|
'never guess names',
|
|
'confirmation=user_required',
|
|
'--idempotency-key',
|
|
'--download-resources',
|
|
'--receiver-open-dingtalk-id',
|
|
):
|
|
self.assertIn(required, text)
|
|
|
|
def test_chat_references_are_atomic_only(self):
|
|
references = list((MULTI_ROOT / 'references').rglob('*.md'))
|
|
self.assertTrue(references)
|
|
for path in references:
|
|
text = path.read_text(encoding='utf-8')
|
|
with self.subTest(path=path.relative_to(MULTI_ROOT)):
|
|
self.assertNotRegex(text, r'(?i)\bshortcut\b')
|
|
self.assertNotRegex(text, r'\+[a-z][a-z0-9-]+')
|
|
|
|
chat_ref = (
|
|
MULTI_ROOT / 'references' / 'chat.md'
|
|
).read_text(encoding='utf-8')
|
|
message_ref = (
|
|
MULTI_ROOT / 'references' / 'chat' / 'chat-message.md'
|
|
).read_text(encoding='utf-8')
|
|
conversation_ref = (
|
|
MULTI_ROOT / 'references' / 'chat' / 'chat-conversation.md'
|
|
).read_text(encoding='utf-8')
|
|
|
|
for required in (
|
|
'message send --group',
|
|
'message search-advanced',
|
|
'message send-by-bot',
|
|
'message send-by-webhook',
|
|
):
|
|
self.assertIn(required, chat_ref)
|
|
for required in (
|
|
'message send-card',
|
|
'message update-card',
|
|
'--flow-status',
|
|
'message download-media',
|
|
'--file-path',
|
|
):
|
|
self.assertIn(required, message_ref)
|
|
for required in (
|
|
'clear-messages',
|
|
'category delete',
|
|
'按 leaf Schema 的确认语义执行',
|
|
):
|
|
self.assertIn(required, conversation_ref)
|
|
|
|
def test_documented_chat_commands_do_not_pass_literal_newline_escapes(self):
|
|
docs = [MULTI_ROOT / 'SKILL.md']
|
|
docs.extend((MULTI_ROOT / 'references').rglob('*.md'))
|
|
offenders = []
|
|
for path in docs:
|
|
for line_number, line in enumerate(
|
|
path.read_text(encoding='utf-8').splitlines(),
|
|
start=1,
|
|
):
|
|
if 'dws chat ' in line and r'\n' in line:
|
|
offenders.append(
|
|
f'{path.relative_to(ROOT)}:{line_number}: {line}'
|
|
)
|
|
self.assertEqual([], offenders)
|
|
|
|
def test_markdown_python_links_resolve(self):
|
|
link_re = re.compile(r'\[[^\]]+\.py\]\(([^)]+\.py)\)')
|
|
docs = list(MULTI_ROOT.rglob('*.md'))
|
|
docs.extend((MONO_ROOT / 'references').rglob('*.md'))
|
|
missing = []
|
|
for doc in docs:
|
|
text = doc.read_text(encoding='utf-8')
|
|
for target in link_re.findall(text):
|
|
resolved = (doc.parent / target).resolve()
|
|
if not resolved.is_file():
|
|
missing.append(f'{doc.relative_to(ROOT)} -> {target}')
|
|
self.assertEqual([], missing)
|
|
|
|
def test_optional_python_invocations_use_python3_and_skill_relative_paths(self):
|
|
docs = {
|
|
MULTI_ROOT: (
|
|
MULTI_ROOT / 'SKILL.md',
|
|
MULTI_ROOT / 'references' / '01-messaging.md',
|
|
MULTI_ROOT / 'references' / 'chat.md',
|
|
MULTI_ROOT / 'references' / 'chat' / 'chat-workflows.md',
|
|
),
|
|
MONO_ROOT: (
|
|
MONO_ROOT / 'references' / 'best_practices' / '01-messaging.md',
|
|
MONO_ROOT / 'references' / 'products' / 'chat.md',
|
|
),
|
|
}
|
|
bad_python = re.compile(r'(?<![A-Za-z0-9_])python\s+')
|
|
invocation = re.compile(r'python3\s+(scripts/[A-Za-z0-9_.-]+\.py)')
|
|
errors = []
|
|
for skill_root, paths in docs.items():
|
|
for doc in paths:
|
|
text = doc.read_text(encoding='utf-8')
|
|
if bad_python.search(text):
|
|
errors.append(f'{doc.relative_to(ROOT)} uses bare python')
|
|
for target in invocation.findall(text):
|
|
if not (skill_root / target).is_file():
|
|
errors.append(
|
|
f'{doc.relative_to(ROOT)} -> {target} is missing'
|
|
)
|
|
self.assertEqual([], errors)
|
|
|
|
def test_chat_workflows_do_not_require_python(self):
|
|
multi_skill = (MULTI_ROOT / 'SKILL.md').read_text(encoding='utf-8')
|
|
multi_recipes = (
|
|
MULTI_ROOT / 'references' / '01-messaging.md'
|
|
).read_text(encoding='utf-8')
|
|
mono_recipes = (
|
|
MONO_ROOT / 'references' / 'best_practices' / '01-messaging.md'
|
|
).read_text(encoding='utf-8')
|
|
|
|
self.assertRegex(multi_skill, r'must work without\s+Python')
|
|
self.assertNotRegex(multi_skill, r'\|\s*“[^”]+”\s*\|\s*`python3 ')
|
|
for recipes in (multi_recipes, mono_recipes):
|
|
self.assertNotRegex(recipes, r'\*\*优先\*\*[^\\n]*python3')
|
|
self.assertIn('chat message list', recipes)
|
|
self.assertIn('chat message send-by-bot', recipes)
|
|
|
|
def test_mono_and_multi_script_copies_match(self):
|
|
bad_python = re.compile(r'(?<![A-Za-z0-9_])python\s+')
|
|
for name in SCRIPT_NAMES:
|
|
multi_script = MULTI_ROOT / 'scripts' / name
|
|
self.assertEqual(
|
|
multi_script.read_bytes(),
|
|
(MONO_ROOT / 'scripts' / name).read_bytes(),
|
|
name,
|
|
)
|
|
self.assertIsNone(
|
|
bad_python.search(multi_script.read_text(encoding='utf-8')),
|
|
name,
|
|
)
|
|
|
|
|
|
class ChatExportMessagesTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.module = load_script('chat_export_messages')
|
|
|
|
def run_script(self, argv, responses):
|
|
calls = []
|
|
|
|
def fake_run_dws(args, dry_run=False):
|
|
self.assertFalse(dry_run)
|
|
calls.append(args)
|
|
response = responses.pop(0)
|
|
if isinstance(response, Exception):
|
|
raise response
|
|
return response
|
|
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
with mock.patch.object(self.module, 'run_dws', fake_run_dws):
|
|
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(
|
|
stderr
|
|
):
|
|
code = self.module.run(argv)
|
|
return code, calls, stdout.getvalue(), stderr.getvalue()
|
|
|
|
def test_paginates_with_last_message_time_and_deduplicates(self):
|
|
responses = [
|
|
{
|
|
'result': {
|
|
'value': [
|
|
{
|
|
'title': '项目群',
|
|
'openConversationId': 'cid-project',
|
|
}
|
|
]
|
|
}
|
|
},
|
|
{
|
|
'result': {
|
|
'messages': [
|
|
{
|
|
'openMessageId': 'm1',
|
|
'createTime': '2026-07-01 10:00:00',
|
|
'content': 'one',
|
|
},
|
|
{
|
|
'openMessageId': 'm2',
|
|
'createTime': '2026-07-01 11:00:00',
|
|
'content': 'two',
|
|
},
|
|
],
|
|
'hasMore': True,
|
|
'nextCursor': 'opaque-not-a-time',
|
|
}
|
|
},
|
|
{
|
|
'result': {
|
|
'messages': [
|
|
{
|
|
'openMessageId': 'm2',
|
|
'createTime': '2026-07-01 11:00:00',
|
|
'content': 'two',
|
|
},
|
|
{
|
|
'openMessageId': 'm3',
|
|
'createTime': '2026-07-01 12:00:00',
|
|
'content': 'three',
|
|
},
|
|
],
|
|
'hasMore': False,
|
|
}
|
|
},
|
|
]
|
|
code, calls, stdout, stderr = self.run_script(
|
|
['--query', '项目群', '--time', '2026-07-01 09:00:00'],
|
|
responses,
|
|
)
|
|
self.assertEqual(0, code, stderr)
|
|
second_page = calls[2]
|
|
self.assertEqual(
|
|
'2026-07-01 11:00:00',
|
|
second_page[second_page.index('--time') + 1],
|
|
)
|
|
self.assertEqual(
|
|
'newer', second_page[second_page.index('--direction') + 1]
|
|
)
|
|
self.assertNotIn('opaque-not-a-time', second_page)
|
|
self.assertIn('合计: 3 条消息', stdout)
|
|
|
|
def test_ambiguous_group_requires_explicit_id(self):
|
|
responses = [
|
|
{
|
|
'result': {
|
|
'value': [
|
|
{'title': '项目一群', 'openConversationId': 'cid-1'},
|
|
{'title': '项目二群', 'openConversationId': 'cid-2'},
|
|
]
|
|
}
|
|
}
|
|
]
|
|
code, calls, _, stderr = self.run_script(
|
|
['--query', '项目', '--time', '2026-07-01 09:00:00'],
|
|
responses,
|
|
)
|
|
self.assertEqual(2, code)
|
|
self.assertEqual(1, len(calls))
|
|
self.assertIn('请指定 --group', stderr)
|
|
|
|
def test_command_failure_is_not_reported_as_empty_result(self):
|
|
error = self.module.ScriptError('upstream failed')
|
|
code, _, stdout, stderr = self.run_script(
|
|
['--group', 'cid-project', '--time', '2026-07-01 09:00:00'],
|
|
[error],
|
|
)
|
|
self.assertEqual(1, code)
|
|
self.assertNotIn('未拉取到消息', stdout)
|
|
self.assertIn('upstream failed', stderr)
|
|
|
|
def test_successful_empty_page_returns_zero(self):
|
|
code, _, stdout, stderr = self.run_script(
|
|
['--group', 'cid-project', '--time', '2026-07-01 09:00:00'],
|
|
[{'result': {'messages': [], 'hasMore': False}}],
|
|
)
|
|
self.assertEqual(0, code, stderr)
|
|
self.assertIn('未拉取到消息', stdout)
|
|
|
|
|
|
class ChatHistoryWithUserTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.module = load_script('chat_history_with_user')
|
|
|
|
def test_ambiguous_user_requires_explicit_id(self):
|
|
calls = []
|
|
|
|
def fake_run_dws(args, dry_run=False):
|
|
calls.append(args)
|
|
return {
|
|
'result': {
|
|
'users': [
|
|
{'name': '张三（研发）', 'userId': 'u-1'},
|
|
{'name': '张三（销售）', 'userId': 'u-2'},
|
|
]
|
|
}
|
|
}
|
|
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
with mock.patch.object(self.module, 'run_dws', fake_run_dws):
|
|
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(
|
|
stderr
|
|
):
|
|
code = self.module.run(
|
|
['--name', '张三', '--time', '2026-07-01 09:00:00']
|
|
)
|
|
self.assertEqual(2, code)
|
|
self.assertEqual(1, len(calls))
|
|
self.assertIn('请指定 --user', stderr.getvalue())
|
|
|
|
def test_no_forward_uses_public_older_direction(self):
|
|
calls = []
|
|
responses = [
|
|
{
|
|
'result': {
|
|
'messages': [
|
|
{
|
|
'openMessageId': 'm1',
|
|
'createTime': '2026-07-01 08:00:00',
|
|
'content': 'hello',
|
|
}
|
|
],
|
|
'hasMore': False,
|
|
}
|
|
}
|
|
]
|
|
|
|
def fake_run_dws(args, dry_run=False):
|
|
calls.append(args)
|
|
return responses.pop(0)
|
|
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(self.module, 'run_dws', fake_run_dws):
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = self.module.run(
|
|
[
|
|
'--user',
|
|
'u-1',
|
|
'--time',
|
|
'2026-07-01 09:00:00',
|
|
'--no-forward',
|
|
]
|
|
)
|
|
self.assertEqual(0, code)
|
|
self.assertEqual(
|
|
'older', calls[0][calls[0].index('--direction') + 1]
|
|
)
|
|
self.assertNotIn('--forward', calls[0])
|
|
|
|
|
|
class BotBroadcastTest(unittest.TestCase):
|
|
def test_business_failure_counts_as_failure(self):
|
|
module = load_script('bot_broadcast')
|
|
completed = types.SimpleNamespace(
|
|
returncode=0,
|
|
stdout='{"success":false,"errorMsg":"robot rejected"}',
|
|
stderr='',
|
|
)
|
|
stdout = io.StringIO()
|
|
with mock.patch.object(module.subprocess, 'run', return_value=completed):
|
|
with contextlib.redirect_stdout(stdout):
|
|
code = module.run(
|
|
[
|
|
'--robot-code',
|
|
'robot',
|
|
'--chats',
|
|
'cid-1',
|
|
'--title',
|
|
'title',
|
|
'--text',
|
|
'text',
|
|
]
|
|
)
|
|
self.assertEqual(1, code)
|
|
self.assertIn('业务调用失败', stdout.getvalue())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|