324 lines
17 KiB
Python
324 lines
17 KiB
Python
"""Offline tests: fake configs, fake credentials, fake DWS. Never use live config."""
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import dingtalk_notify as notify
|
|
|
|
DNS = 'lookup api.dingtalk.com on 127.0.0.1:53: i/o timeout'
|
|
OK = '{"processQueryKey":"test-receipt"}'
|
|
|
|
|
|
def response(code=0, stdout=OK, stderr=''):
|
|
return subprocess.CompletedProcess(['fake-dws'], code, stdout, stderr)
|
|
|
|
|
|
class NotifyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='通知测试 with spaces ')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
(self.root / '.dingtalk').mkdir()
|
|
self.config = self.root / '.dingtalk/config.env'
|
|
self.configure()
|
|
self.environment = mock.patch.dict(os.environ, {'DING_SEC': 'ignored-env-secret',
|
|
'DWS_CLIENT_SECRET': 'ignored-dws-secret'}, clear=True)
|
|
self.environment.start()
|
|
self.addCleanup(self.environment.stop)
|
|
self.locator = mock.patch.object(notify, 'dws_path', return_value='fake-dws')
|
|
self.locator.start()
|
|
self.addCleanup(self.locator.stop)
|
|
|
|
def configure(self, **updates):
|
|
values = dict(ENABLED='1', DRY_RUN='0', MODE='app-bot', CLIENT_ID='fake-client',
|
|
TARGET='fake-group', ROBOT_CODE='fake-robot', PROJECT='离线测试',
|
|
PROFILE='fake-corp:fake-user', CLIENT_SECRET='fake-secret')
|
|
values.update(updates)
|
|
self.config.write_text('\n'.join(notify.PREFIX + k + '=' + v for k, v in values.items()),
|
|
encoding='utf-8')
|
|
self.config.chmod(0o600)
|
|
|
|
def call(self, args, responses=None):
|
|
run = mock.Mock(side_effect=responses or [response()])
|
|
sleep = mock.Mock()
|
|
output, errors = io.StringIO(), io.StringIO()
|
|
with contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors):
|
|
code = notify.main(args, root=self.root, run=run, sleep=sleep)
|
|
return code, json.loads(output.getvalue()), run, sleep, errors.getvalue()
|
|
|
|
def entries(self):
|
|
return [['progress', 'milestone', '**完成**\n\n- 内容'],
|
|
['message', '--title', '负责人提醒', '--text', '**请确认**\n\n- 内容']]
|
|
|
|
def test_events(self):
|
|
for event, label in notify.EVENTS.items():
|
|
with self.subTest(event=event):
|
|
code, data, run, _, _ = self.call(['progress', event, '中文摘要'])
|
|
self.assertEqual((code, data['status']), (0, 'accepted'))
|
|
payload = json.loads(run.call_args.kwargs['input'])
|
|
self.assertIn(label, json.loads(payload['msgParam'])['title'])
|
|
|
|
def test_payload_and_credentials_not_on_argv(self):
|
|
text = '**结果**\n\n- "中文" 😀 `代码` $变量 & <内容> \\路径'
|
|
code, data, run, _, _ = self.call(['message', '--title', '摘要', '--text', text])
|
|
self.assertEqual(code, 0)
|
|
command = run.call_args.args[0]
|
|
kwargs = run.call_args.kwargs
|
|
payload = json.loads(kwargs['input'])
|
|
self.assertEqual(json.loads(payload['msgParam'])['text'], text)
|
|
self.assertEqual(payload['openConversationId'], 'fake-group')
|
|
self.assertEqual(payload['msgKey'], 'sampleMarkdown')
|
|
self.assertEqual(command[command.index('--data') + 1], '-')
|
|
self.assertNotIn('fake-secret', str(command))
|
|
self.assertNotIn('fake-group', str(command))
|
|
self.assertEqual(kwargs['env']['DWS_CLIENT_SECRET'], 'fake-secret')
|
|
self.assertNotIn('DING_SEC', kwargs['env'])
|
|
self.assertNotIn('fake-secret', json.dumps(data))
|
|
|
|
def test_dns_retries_both_entries(self):
|
|
for args in self.entries():
|
|
with self.subTest(args=args):
|
|
code, data, run, sleep, _ = self.call(args, [response(1, '', DNS)] * 3)
|
|
self.assertEqual((code, data['status'], run.call_count), (1, 'error', 3))
|
|
self.assertEqual(sleep.call_args_list, [mock.call(10), mock.call(10)])
|
|
|
|
def test_transient_dns_success_stops(self):
|
|
for args in self.entries():
|
|
code, data, run, sleep, _ = self.call(args, [response(1, '', DNS), response()])
|
|
self.assertEqual((code, data['status'], run.call_count, sleep.call_count), (0, 'accepted', 2, 1))
|
|
|
|
def test_ambiguous_and_auth_errors_never_replayed_or_echoed(self):
|
|
for args in self.entries():
|
|
for error in ['i/o timeout', 'EOF', 'HTTP 503', 'permission denied', 'fake-secret App Token']:
|
|
with self.subTest(error=error):
|
|
code, data, run, sleep, stderr = self.call(args, [response(1, '', error)])
|
|
self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
|
|
self.assertNotIn(error, json.dumps(data, ensure_ascii=False) + stderr)
|
|
|
|
def test_receipt_wins_even_nonzero(self):
|
|
for args in self.entries():
|
|
code, data, run, sleep, _ = self.call(args, [response(1, OK, DNS)])
|
|
self.assertEqual((code, data['status'], run.call_count, sleep.call_count), (0, 'accepted', 1, 0))
|
|
|
|
def test_bad_receipts_do_not_replay(self):
|
|
for raw in ['{}', '{"processQueryKey":" "}', '{"processQueryKey":123}', 'not json', '[]']:
|
|
code, _, run, sleep, _ = self.call(self.entries()[0], [response(0, raw)])
|
|
self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
|
|
|
|
def test_timeout_no_replay(self):
|
|
code, data, run, sleep, _ = self.call(self.entries()[0], [subprocess.TimeoutExpired('fake', 60)])
|
|
self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
|
|
self.assertIn('结果未知', data['message'])
|
|
self.assertEqual(run.call_args.kwargs['timeout'], 60)
|
|
|
|
def test_startup_failure_no_replay(self):
|
|
code, _, run, sleep, _ = self.call(self.entries()[0], [OSError('private machine detail')])
|
|
self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
|
|
|
|
def test_missing_credentials_no_commands(self):
|
|
self.configure(CLIENT_SECRET='')
|
|
code, data, run, _, _ = self.call(self.entries()[0])
|
|
self.assertEqual((code, run.call_count), (1, 0))
|
|
self.assertIn('凭据', data['message'])
|
|
|
|
def test_skipped_no_tools(self):
|
|
self.configure(ENABLED='0')
|
|
for args in self.entries():
|
|
code, data, run, _, _ = self.call(args)
|
|
self.assertEqual((code, data['status'], run.call_count), (0, 'skipped', 0))
|
|
self.config.unlink()
|
|
self.assertEqual(self.call(self.entries()[0])[1]['status'], 'skipped')
|
|
|
|
def test_preview_never_uses_or_prints_secret_or_resolves_people(self):
|
|
for configured in [False, True]:
|
|
self.configure(DRY_RUN='1' if configured else '0')
|
|
for args in self.entries():
|
|
if args[0] == 'message':
|
|
args += ['--at', '完整姓名']
|
|
if not configured:
|
|
args += ['--dry-run']
|
|
with mock.patch.object(notify, 'secret_for', side_effect=AssertionError('secret accessed')):
|
|
code, data, run, _, _ = self.call(args)
|
|
self.assertEqual((code, data['status'], run.call_count), (0, 'preview', 0))
|
|
self.assertNotIn('fake-group', str(data))
|
|
self.assertNotIn('fake-secret', str(data))
|
|
|
|
def test_preview_without_secret(self):
|
|
self.configure(CLIENT_SECRET='')
|
|
code, data, run, _, _ = self.call(self.entries()[0] + ['--dry-run'])
|
|
self.assertEqual((code, data['status'], run.call_count), (0, 'preview', 0))
|
|
|
|
def test_config_cannot_be_used_as_body_even_in_preview(self):
|
|
for args in [['progress', 'milestone'], ['message', '--title', '摘要']]:
|
|
for flags in [[], ['--dry-run']]:
|
|
code, data, run, _, _ = self.call(args + ['--file', str(self.config)] + flags)
|
|
self.assertEqual((code, run.call_count), (1, 0))
|
|
self.assertNotIn('fake-secret', str(data))
|
|
|
|
def test_preview_without_config(self):
|
|
self.config.unlink()
|
|
self.assertEqual(self.call(self.entries()[0] + ['--dry-run'])[1]['status'], 'preview')
|
|
|
|
def test_utf8_bom_crlf_literal_config(self):
|
|
raw = self.config.read_text(encoding='utf-8') + '\nDINGTALK_PROGRESS_TITLE_PREFIX=$(not-executed)'
|
|
self.config.write_bytes(('\ufeff' + raw.replace('\n', '\r\n')).encode('utf-8'))
|
|
self.assertEqual(notify.read_config(self.root)['TITLE_PREFIX'], '$(not-executed)')
|
|
|
|
def test_invalid_config_safe_error(self):
|
|
for raw in ['DING_SEC=fake-secret', 'invalid fake-secret',
|
|
'DINGTALK_PROGRESS_ENABLED=1\nDINGTALK_PROGRESS_ENABLED=0',
|
|
'DINGTALK_PROGRESS_ENABLED=yes']:
|
|
self.config.write_text(raw, encoding='utf-8')
|
|
code, data, run, _, _ = self.call(self.entries()[0])
|
|
self.assertEqual((code, run.call_count), (1, 0))
|
|
self.assertNotIn('fake-secret', str(data))
|
|
|
|
def test_body_files_both_entries(self):
|
|
body = self.root / '中文 summary.md'
|
|
body.write_bytes('\ufeff**结果**\r\n\r\n- 已确认 😀'.encode('utf-8'))
|
|
for args in [['progress', 'milestone'], ['message', '--title', '摘要']]:
|
|
code, _, run, _, _ = self.call(args + ['--file', str(body)])
|
|
self.assertEqual(code, 0)
|
|
text = json.loads(json.loads(run.call_args.kwargs['input'])['msgParam'])['text']
|
|
self.assertIn('**结果**\n\n- 已确认 😀', text)
|
|
|
|
def test_empty_or_large_body_fails_before_tools(self):
|
|
for text in ['', ' ', '字' * 5000]:
|
|
code, _, run, _, _ = self.call(['message', '--title', '摘要', '--text', text])
|
|
self.assertEqual((code, run.call_count), (1, 0))
|
|
|
|
def test_exact_name_and_dedup_same_profile(self):
|
|
found = json.dumps({'result': [{'meta': {'name': '完整姓名'}, 'userId': 'fake-person'}]})
|
|
args = self.entries()[1] + ['--at', '完整姓名，完整姓名', '--at-id', '完整姓名=fake-person']
|
|
code, _, run, _, _ = self.call(args, [response(0, found), response()])
|
|
self.assertEqual((code, run.call_count), (0, 2))
|
|
for call in run.call_args_list:
|
|
command = call.args[0]
|
|
self.assertEqual(command[command.index('--profile') + 1], 'fake-corp:fake-user')
|
|
self.assertNotIn('DWS_CLIENT_SECRET', run.call_args_list[0].kwargs['env'])
|
|
message = json.loads(json.loads(run.call_args.kwargs['input'])['msgParam'])
|
|
self.assertEqual(message['atUserIds'], ['fake-person'])
|
|
self.assertTrue(message['text'].endswith('@fake-person'))
|
|
|
|
def test_names_zero_multiple_malformed_stop_before_send(self):
|
|
for rows in [[], [{'title': '完整姓名', 'userId': 'a'}, {'title': '完整姓名', 'userId': 'b'}],
|
|
[{'title': '别的姓名', 'userId': 'a'}], {}, [None]]:
|
|
code, _, run, sleep, _ = self.call(self.entries()[1] + ['--at', '完整姓名'],
|
|
[response(0, json.dumps({'result': rows}))])
|
|
self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
|
|
|
|
def test_name_search_requires_explicit_profile(self):
|
|
self.configure(PROFILE='')
|
|
code, _, run, _, _ = self.call(self.entries()[1] + ['--at', '完整姓名'])
|
|
self.assertEqual((code, run.call_count), (1, 0))
|
|
|
|
def test_known_mapping_and_group_compatibility(self):
|
|
code, _, run, _, _ = self.call(self.entries()[1] + ['--at-id', '姓名=fake-id', '--group', 'fake-other'])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(json.loads(run.call_args.kwargs['input'])['openConversationId'], 'fake-other')
|
|
|
|
def test_config_secret_is_only_source_on_all_platforms(self):
|
|
for platform in ['darwin', 'win32', 'linux']:
|
|
with self.subTest(platform=platform), mock.patch.object(sys, 'platform', platform):
|
|
self.assertEqual(notify.secret_for({'CLIENT_SECRET': 'fake-secret'}), 'fake-secret')
|
|
with self.assertRaises(notify.NoticeError):
|
|
notify.secret_for({})
|
|
|
|
def test_secret_special_characters_are_literal(self):
|
|
secret = 'fake=secret#literal$not-expanded'
|
|
self.configure(CLIENT_SECRET=secret)
|
|
code, data, run, _, _ = self.call(self.entries()[0])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(run.call_args.kwargs['env']['DWS_CLIENT_SECRET'], secret)
|
|
self.assertNotIn(secret, str(data))
|
|
self.assertNotIn(secret, str(run.call_args.args))
|
|
self.assertNotIn(secret, run.call_args.kwargs['input'])
|
|
|
|
|
|
class LauncherTests(unittest.TestCase):
|
|
def test_real_subprocess_stdin_without_live_dws(self):
|
|
with tempfile.TemporaryDirectory(prefix='通知 fake transport ') as folder:
|
|
root = Path(folder)
|
|
(root / '.dingtalk').mkdir()
|
|
config = root / '.dingtalk/config.env'
|
|
config.write_text('DINGTALK_PROGRESS_ENABLED=1\nDINGTALK_PROGRESS_DRY_RUN=0\n'
|
|
'DINGTALK_PROGRESS_CLIENT_ID=fake-client\n'
|
|
'DINGTALK_PROGRESS_CLIENT_SECRET=fake-secret\n'
|
|
'DINGTALK_PROGRESS_TARGET=fake-group\n'
|
|
'DINGTALK_PROGRESS_ROBOT_CODE=fake-robot\n', encoding='utf-8')
|
|
config.chmod(0o600)
|
|
fake = root / 'fake dws.py'
|
|
fake.write_text(
|
|
'import json, os, sys\n'
|
|
'assert sys.argv[1:] == ["api", "POST", "/v1.0/robot/groupMessages/send", '
|
|
'"--data", "-", "--yes", "--format", "json"]\n'
|
|
'payload = json.loads(sys.stdin.buffer.read().decode("utf-8"))\n'
|
|
'assert payload["openConversationId"] == "fake-group"\n'
|
|
'assert json.loads(payload["msgParam"])["text"] == "中文\\n\\n- \\\"引号\\\" 😀"\n'
|
|
'assert os.environ["DWS_CLIENT_SECRET"] == "fake-secret"\n'
|
|
'assert "DING_SEC" not in os.environ\n'
|
|
'print(json.dumps({"processQueryKey": "fake-transport-receipt"}))\n', encoding='utf-8')
|
|
|
|
def transport(command, **kwargs):
|
|
return subprocess.run([sys.executable, str(fake), *command[1:]], **kwargs)
|
|
|
|
out = io.StringIO()
|
|
with mock.patch.object(notify, 'dws_path', return_value='fake-dws'), \
|
|
mock.patch.dict(os.environ, {'DING_SEC': 'ignored-env-secret'}), \
|
|
contextlib.redirect_stdout(out):
|
|
code = notify.main(['message', '--title', '摘要', '--text', '中文\n\n- "引号" 😀'],
|
|
root=root, run=transport)
|
|
self.assertEqual(code, 0, out.getvalue())
|
|
self.assertEqual(json.loads(out.getvalue())['status'], 'accepted')
|
|
|
|
def test_real_launchers_preview_in_isolated_copy(self):
|
|
"""Exercise OS processes/quoting, but copy no real config and never run dws."""
|
|
with tempfile.TemporaryDirectory(prefix='通知 launch with spaces ') as folder:
|
|
scripts = Path(folder) / 'scripts'
|
|
scripts.mkdir()
|
|
for name in ['dingtalk_notify.py', 'dingtalk-progress', 'dingtalk-bot-message',
|
|
'dingtalk-progress.ps1', 'dingtalk-bot-message.ps1']:
|
|
shutil.copy2(Path(__file__).parent / name, scripts / name)
|
|
text = '**结果**\n\n- "中文" 😀 `代码` $变量 & 内容'
|
|
body = Path(folder) / '中文正文.md'
|
|
body.write_text(text, encoding='utf-8')
|
|
env = {k: v for k, v in os.environ.items()
|
|
if not k.startswith(('DING', 'DWS_'))}
|
|
# Run from outside the script project: --file is cwd-relative.
|
|
commands = [[sys.executable, str(scripts / 'dingtalk_notify.py'), 'progress', 'milestone',
|
|
'--file', body.name, '--dry-run']]
|
|
pwsh = shutil.which('pwsh')
|
|
if os.name == 'nt':
|
|
self.assertIsNotNone(pwsh, 'Windows launcher test needs PowerShell 7.3+')
|
|
if pwsh:
|
|
commands += [[pwsh, '-NoProfile', '-File', str(scripts / 'dingtalk-progress.ps1'),
|
|
'milestone', text, '--dry-run'],
|
|
[pwsh, '-NoProfile', '-File', str(scripts / 'dingtalk-bot-message.ps1'),
|
|
'--title', '中文摘要', '--text', text, '--dry-run']]
|
|
bash = Path('C:/Program Files/Git/bin/bash.exe') if os.name == 'nt' else shutil.which('bash')
|
|
if bash and Path(bash).exists():
|
|
for name, args in [('dingtalk-progress', ['milestone']),
|
|
('dingtalk-bot-message', ['--title', '中文摘要'])]:
|
|
commands.append([str(bash), str(scripts / name), *args,
|
|
'--file', body.name, '--dry-run'])
|
|
for command in commands:
|
|
with self.subTest(entry=command[:3]):
|
|
result = subprocess.run(command, cwd=folder, env=env, capture_output=True,
|
|
text=True, encoding='utf-8', timeout=30)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
data = json.loads(result.stdout)
|
|
self.assertEqual(data['status'], 'preview')
|
|
self.assertIn(text, data['text'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|