"""Offline regression tests. All DWS calls use isolated fake responses.""" import base64 import contextlib import io import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest from unittest.mock import patch import dingtalk_notify as n PROFILE = 'fake-corp:fake-user' GROUP = 'fake-group' PERSON = 'fake-person' OK = {'ok': True, 'identity': 'user', 'tool': 'send_personal_message', 'result': {'success': True, 'errorCode': None, 'result': {'openTaskId': 'fake-task'}}} def reply(data=None, code=0, error=''): return subprocess.CompletedProcess([], code, json.dumps(data or {}, ensure_ascii=False), error) class NotifyTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.path = self.root / '.dingtalk' / 'config.env' self.path.parent.mkdir() self.config() self.calls = [] self.sleeps = [] self.addCleanup(patch.stopall) patch.object(n, 'dws_path', return_value='fake-dws').start() def config(self, **changes): config = {'ENABLED': '1', 'PROFILE': PROFILE, 'TARGET': GROUP, 'PROJECT': '测试项目'} config.update(changes) self.path.write_text(''.join(n.PREFIX + k + '=' + v + '\n' for k, v in config.items()), encoding='utf-8') def fake(self, command, **kwargs): self.calls.append((command, kwargs)) if command[1:3] == ['profile', 'list']: return reply({'profiles': [{'profile': PROFILE, 'userName': '测试人'}]}) if command[1:3] == ['auth', 'status']: return reply({'authenticated': True, 'token_valid': True}) if command[1:3] == ['aisearch', 'person']: return reply({'result': [{'openDingTalkId': PERSON, 'meta': {'name': command[command.index('--query') + 1]}}]}) self.assertEqual(command[1:3], ['chat', '+messages-send']) return reply(OK) def execute(self, args, run=None): out = io.StringIO() with contextlib.redirect_stdout(out): code = n.main(args, root=self.root, run=run or self.fake, sleep=self.sleeps.append) return code, json.loads(out.getvalue()) def test_each_event_personal_markdown(self): for event, label in n.EVENTS.items(): code, result = self.execute(['progress', event, '**进展**\n\n- 已归档']) self.assertEqual(code, 0) self.assertEqual(result['status'], 'accepted') cmd = self.calls[-1][0] self.assertEqual(cmd[cmd.index('--as') + 1], 'user') self.assertEqual(cmd[cmd.index('--group') + 1], GROUP) self.assertEqual(cmd[cmd.index('--profile') + 1], PROFILE) body = cmd[cmd.index('--markdown') + 1] self.assertIn(label, body) self.assertIn('项目进度自动汇报', body) self.assertIn('**进展**\n\n- 已归档', body) self.assertNotIn('--ai-tag=false', cmd) self.assertNotIn('--robot-code', cmd) def test_message_mentions_exact_name_fixed_profile(self): code, data = self.execute(['message', '--title', '提醒', '--text', '请补充资料', '--at', '测试人,测试人']) self.assertEqual(code, 0) self.assertEqual(len(self.calls), 2) for cmd, _ in self.calls: self.assertEqual(cmd[cmd.index('--profile') + 1], PROFILE) cmd = self.calls[-1][0] self.assertEqual(cmd[cmd.index('--at-open-dingtalk-ids') + 1], PERSON) self.assertIn('<@fake-person>', cmd[cmd.index('--markdown') + 1]) self.assertEqual(data['mentions'], ['测试人']) self.assertEqual(data['openTaskId'], 'fake-task') def test_progress_can_include_mention_once(self): code, _ = self.execute(['progress', 'blocked', '缺输入', '--at', '测试人']) self.assertEqual(code, 0) self.assertEqual(sum(c[0][1:3] == ['chat', '+messages-send'] for c in self.calls), 1) def test_lookup_ambiguity_no_match_and_malformed_never_send(self): for rows in [[], [{'openDingTalkId': 'a', 'meta': {'name': '甲'}}, {'openDingTalkId': 'b', 'meta': {'name': '甲'}}], [{'userId': 'wrong-id-kind', 'meta': {'name': '甲'}}], None]: self.calls.clear() def lookup(command, **kwargs): self.calls.append(command) return reply({'result': rows}) code, data = self.execute(['message', '--title', '提醒', '--text', '正文', '--at', '甲'], lookup) self.assertEqual(code, 1) self.assertEqual(data['status'], 'error') self.assertEqual(len(self.calls), 1) def test_check_is_read_only(self): code, data = self.execute(['check', '--at', '测试人']) self.assertEqual(code, 0) self.assertEqual(data['status'], 'checked') self.assertEqual(len(self.calls), 2) self.assertNotIn('fake-person', json.dumps(data)) def test_check_bad_login_stops(self): code, _ = self.execute(['check'], lambda *a, **kw: reply({'authenticated': False, 'token_valid': False})) self.assertEqual(code, 1) def test_disabled_and_missing_config_no_tools(self): self.config(ENABLED='0') self.assertEqual(self.execute(['progress', 'start', '开始'])[1]['status'], 'skipped') self.path.unlink() self.assertEqual(self.execute(['progress', 'start', '开始'])[1]['status'], 'skipped') self.assertFalse(self.calls) def test_preview_without_config_or_dws(self): self.path.unlink() code, result = self.execute(['progress', 'blocked', '说明', '--at', '甲', '--dry-run']) self.assertEqual(code, 0) self.assertEqual(result['status'], 'preview') self.assertFalse(self.calls) def test_legacy_and_invalid_config_fail_without_echo(self): for suffix in ['CLIENT_SECRET=fake-secret', 'MODE=app-bot', 'ENABLED=1', 'UNKNOWN=x']: self.config() with self.path.open('a', encoding='utf-8') as f: f.write(n.PREFIX + suffix + '\n') code, data = self.execute(['progress', 'start', '开始']) self.assertEqual(code, 1) self.assertNotIn('fake-secret', json.dumps(data)) self.assertFalse(self.calls) def test_config_bom_crlf_and_literal_values(self): raw = self.path.read_text(encoding='utf-8').replace('测试项目', '中文 "引号" $()') self.path.write_bytes(b'\xef\xbb\xbf' + raw.replace('\n', '\r\n').encode()) code, data = self.execute(['progress', 'start', '开始', '--dry-run']) self.assertEqual(code, 0) self.assertIn('中文 "引号" $()', data['text']) def test_files_and_invalid_bodies(self): body = self.root / '正文.md' body.write_text('**进展**\n\n- 中文 "引号" $()', encoding='utf-8') for args in [['progress', 'milestone'], ['message', '--title', '说明']]: self.assertEqual(self.execute(args + ['--file', str(body)])[0], 0) before = len(self.calls) for args in [['progress', 'start', ''], ['progress', 'start', 'x' * 12001], ['progress', 'start', '正文', '--file', str(body)], ['message', '--title', '坏\n标题', '--text', '内容'], ['progress', 'start', '--file', str(self.path), '--dry-run']]: self.assertEqual(self.execute(args)[0], 1) self.assertEqual(len(self.calls), before) def test_dns_three_attempts_only(self): calls = [] def failed(*a, **kw): calls.append(a) return reply(code=1, error='lookup api.dingtalk.com: no such host') code, _ = self.execute(['progress', 'start', '开始'], failed) self.assertEqual(code, 1) self.assertEqual(len(calls), 3) self.assertEqual(self.sleeps, [10, 10]) def test_dns_then_success(self): responses = iter([reply(code=1, error='lookup api.dingtalk.com: no such host'), reply(OK)]) self.assertEqual(self.execute(['progress', 'start', '开始'], lambda *a, **kw: next(responses))[0], 0) self.assertEqual(self.sleeps, [10]) def test_failure_and_unknown_never_retry(self): cases = [reply(code=1, error='fake-secret connection reset'), reply(code=1, error=json.dumps({'error': {'category': 'auth', 'message': 'fake-secret'}})), reply(code=1, error=json.dumps({'error': {'retryable': False, 'message': 'lookup api.dingtalk.com: no such host'}})), reply({'success': True}), reply({'ok': False}), reply(OK | {'partial': True})] for result in cases: count = [] def failed(*a, **kw): count.append(1) return result code, data = self.execute(['progress', 'start', '开始'], failed) self.assertEqual(code, 1) self.assertEqual(len(count), 1) self.assertNotIn('fake-secret', json.dumps(data)) self.assertFalse(self.sleeps) def test_receipt_wins_nonzero_exit(self): self.assertEqual(self.execute(['progress', 'start', '开始'], lambda *a, **kw: reply(OK, code=1))[0], 0) def test_wrong_receipts_rejected(self): for bad in [OK | {'identity': 'bot'}, OK | {'tool': 'other'}, OK | {'failures': [1]}, OK | {'result': {'success': False, 'result': {'openTaskId': 'x'}}}, OK | {'result': {'success': True, 'result': {'openTaskId': ''}}}]: self.assertIsNone(n.accepted_task(bad)) def test_timeout_and_launch_error_no_retry(self): for error in [subprocess.TimeoutExpired('fake', 60), OSError('fake-secret')]: calls = [] def fail(*a, **kw): calls.append(1) raise error code, _ = self.execute(['progress', 'start', '开始'], fail) self.assertEqual(code, 1) self.assertEqual(len(calls), 1) def test_no_app_credentials_in_child_environment(self): with patch.dict(os.environ, {'DWS_CLIENT_SECRET': 'fake-secret', 'DWS_CLIENT_ID': 'fake-app', 'DING_SEC': 'fake-key'}): self.execute(['progress', 'start', '开始']) env = self.calls[-1][1]['env'] self.assertNotIn('DWS_CLIENT_SECRET', env) self.assertNotIn('DWS_CLIENT_ID', env) self.assertNotIn('DING_SEC', env) self.assertEqual(self.calls[-1][1]['timeout'], 60) def init_args(self): return ['init', '--profile', PROFILE, '--group', GROUP, '--project', '项目甲'] def test_init_requires_confirmation_and_explicit_replace(self): original = self.path.read_bytes() self.assertEqual(self.execute(self.init_args())[0], 1) self.assertEqual(self.execute(self.init_args() + ['--yes'])[0], 1) self.assertEqual(self.path.read_bytes(), original) self.assertFalse(self.calls) def test_init_creates_minimal_config_no_send(self): self.path.unlink() code, data = self.execute(self.init_args() + ['--yes']) self.assertEqual(code, 0) self.assertEqual(data['status'], 'initialized') self.assertEqual(set(n.read_config(self.root)), n.KEYS) self.assertNotIn(b'\r', self.path.read_bytes()) self.assertEqual(len(self.calls), 2) if os.name != 'nt': self.assertEqual(self.path.stat().st_mode & 0o777, 0o600) def test_init_replaces_legacy_without_retaining_secret(self): self.path.write_text('DINGTALK_PROGRESS_CLIENT_SECRET=fake-secret\n', encoding='utf-8') code, data = self.execute(self.init_args() + ['--yes', '--replace']) self.assertEqual(code, 0) self.assertNotIn('fake-secret', self.path.read_text()) self.assertNotIn('fake-secret', json.dumps(data)) self.assertEqual(len(list(self.path.parent.iterdir())), 1) def test_init_bad_profile_and_line_injection_leave_config(self): before = self.path.read_bytes() for args in [self.init_args() + ['--profile', 'other:user'], self.init_args() + ['--project', 'x\nBAD=1']]: self.assertEqual(self.execute(args + ['--yes', '--replace'])[0], 1) self.assertEqual(self.path.read_bytes(), before) def test_init_failed_auth_leaves_old_config(self): before = self.path.read_bytes() def fake(command, **kw): if command[1] == 'auth': return reply({'authenticated': False}) return self.fake(command, **kw) self.assertEqual(self.execute(self.init_args() + ['--yes', '--replace'], fake)[0], 1) self.assertEqual(self.path.read_bytes(), before) def test_encoded_powershell_arguments(self): args = ['message', '--title', '中文 "标题"', '--text', '**进展**\n\n- 引号 "$()"', '--dry-run'] with patch.dict(os.environ, {n.PREFIX + 'ARGV_B64': base64.b64encode(json.dumps(args).encode()).decode()}): code, data = self.execute(None) self.assertEqual(code, 0) self.assertIn('中文 "标题"', data['text']) self.assertIn('**进展**\n\n- 引号 "$()"', data['text']) def test_bad_encoded_arguments(self): for value in ['!!!', base64.b64encode(b'{}').decode()]: with patch.dict(os.environ, {n.PREFIX + 'ARGV_B64': value}): self.assertEqual(self.execute(None)[0], 1) self.assertFalse(self.calls) def test_hardlinked_config_cannot_be_message_body(self): other = self.root / 'not-a-summary.md' os.link(self.path, other) self.assertEqual(self.execute(['progress', 'start', '--file', str(other), '--dry-run'])[0], 1) def test_real_powershell_launcher_in_isolated_project(self): pwsh = shutil.which('pwsh') if not pwsh or 'windowsapps' in sys.executable.lower(): self.skipTest('需要 PowerShell 和独立 Python，不能用商店版作启动器验收') folder = self.root / 'scripts' folder.mkdir() source = Path(n.__file__).parent for name in ['dingtalk_notify.py', 'dingtalk-python.ps1', 'dingtalk-progress.ps1', 'dingtalk-message.ps1']: shutil.copyfile(source / name, folder / name) env = n.clean_env() env['DINGTALK_PYTHON'] = sys.executable text = '**中文**\n\n- "引号" $() `原样`' for script, args in [('dingtalk-progress.ps1', ['milestone', text]), ('dingtalk-message.ps1', ['--title', '中文 "标题"', '--text', text])]: result = subprocess.run([pwsh, '-NoProfile', '-File', str(folder / script), *args, '--dry-run'], cwd=self.root, env=env, capture_output=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']) result = subprocess.run([pwsh, '-NoProfile', '-File', str(folder / 'dingtalk-python.ps1'), 'init', '--profile', PROFILE, '--group', GROUP, '--project', '甲'], cwd=self.root, env=env, capture_output=True, encoding='utf-8', timeout=30) self.assertEqual(result.returncode, 1) self.assertIn('先向负责人确认', result.stdout) if __name__ == '__main__': unittest.main()