437 lines
23 KiB
Python
437 lines
23 KiB
Python
"""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 in n.EVENTS:
|
|
code, result = self.execute(['progress', event, '--stage', '测试验证', '**进展**\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.assertEqual(body.splitlines()[0], '**测试项目｜测试验证**')
|
|
self.assertIn('项目进度自动汇报', body)
|
|
self.assertIn('**进展**\n\n- 已归档', body)
|
|
self.assertNotIn('--ai-tag=false', cmd)
|
|
self.assertNotIn('--robot-code', cmd)
|
|
|
|
def test_progress_requires_stage_before_any_dws_call(self):
|
|
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as error:
|
|
self.execute(['progress', 'milestone', '已更新'])
|
|
self.assertEqual(error.exception.code, 2)
|
|
self.assertFalse(self.calls)
|
|
|
|
def test_invalid_stage_never_sends(self):
|
|
for stage in ['', ' ', '测试\n验证', '测试验证\r', '测试\t验证', '测试\0验证', '验' * 201]:
|
|
with self.subTest(stage=stage):
|
|
code, result = self.execute(['progress', 'milestone', '已更新', '--stage', stage])
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result['status'], 'error')
|
|
self.assertFalse(self.calls)
|
|
|
|
def test_stage_changes_per_notification_without_config_change(self):
|
|
original = self.path.read_bytes()
|
|
text = '**待办事项**\n\n- 由 Hub 请项目负责人确认测试结果。\n\n**遗留**\n\n- 无。'
|
|
for stage in [' 阶段六：测试验证 ', '阶段七：平台/客户验收']:
|
|
code, result = self.execute(['progress', 'milestone', text, '--stage', stage, '--dry-run'])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result['text'].splitlines()[0], '**测试项目｜' + stage.strip() + '**')
|
|
self.assertTrue(result['text'].endswith(text))
|
|
self.assertEqual(self.path.read_bytes(), original)
|
|
self.assertFalse(self.calls)
|
|
|
|
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_approved_closing_template_preserves_conditions_and_scope(self):
|
|
body = '''**已确认的进展**
|
|
|
|
- 相关负责人已确认：产品验收材料已在线下完成，业务已确认客户验收通过。
|
|
- 项目已进入资料归档与复盘结项阶段，目前开始整理结项资料，项目尚未关闭。
|
|
- 已通知相关负责人进行阶段交接，不新增审批。
|
|
|
|
**待办事项**
|
|
|
|
- 结项资料整理：由 Hub 整理四类结项工作稿。
|
|
- 资料补充：由产品 Agent、业务 Agent 补充现有线下成果及确认索引，不重做验收。
|
|
- 旧事项停止：由 Hub 停止此前的验收准备催办。
|
|
- 结项确认：四类成果具备后，由 Hub 请 @{测试人} 确认结项，确认后再锁定成果。
|
|
- 本轮工作范围：不新增应用修复、设备测试或发布验证。
|
|
|
|
**遗留**
|
|
|
|
- 原测试不通过的结果如实保留，不因进入结项阶段改为通过。
|
|
- 已接受延期的问题如实保留，后续沿用 Bug 系统中的记录和已有处理安排继续跟进。'''
|
|
path = self.root / '通知正文.md'
|
|
path.write_bytes(body.replace('\n', '\r\n').encode('utf-8'))
|
|
before = self.path.read_bytes()
|
|
args = ['progress', 'milestone', '--stage', '阶段八：复盘结项', '--file', str(path), '--at', '测试人']
|
|
code, preview = self.execute(args + ['--dry-run'])
|
|
self.assertEqual(code, 0)
|
|
self.assertFalse(self.calls)
|
|
self.assertEqual(preview['text'].splitlines()[0], '**测试项目｜阶段八：复盘结项**')
|
|
self.assertEqual(preview['text'].split('\n\n', 2)[2], body.replace('@{测试人}', '@测试人'))
|
|
code, result = self.execute(args)
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result['status'], 'accepted')
|
|
self.assertEqual(len(self.calls), 2)
|
|
command = self.calls[-1][0]
|
|
markdown = command[command.index('--markdown') + 1]
|
|
self.assertEqual(markdown.split('\n\n', 2)[2], body.replace('@{测试人}', '<@fake-person>'))
|
|
self.assertEqual(command[command.index('--at-open-dingtalk-ids') + 1], PERSON)
|
|
self.assertEqual(self.path.read_bytes(), before)
|
|
|
|
def test_progress_can_include_mention_once(self):
|
|
code, _ = self.execute(['progress', 'blocked', '--stage', '测试验证', '缺输入', '--at', '测试人'])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(sum(c[0][1:3] == ['chat', '+messages-send'] for c in self.calls), 1)
|
|
|
|
def test_inline_mentions_match_each_item_and_deduplicate_recipients(self):
|
|
ids = {'甲': 'fake-a', '乙': 'fake-b', '丙': 'fake-c'}
|
|
def lookup(command, **kwargs):
|
|
if command[1:3] == ['aisearch', 'person']:
|
|
self.calls.append((command, kwargs))
|
|
name = command[command.index('--query') + 1]
|
|
return reply({'result': [{'openDingTalkId': ids[name], 'meta': {'name': name}}]})
|
|
return self.fake(command, **kwargs)
|
|
body = '**待办事项**\n\n- 由 Hub 请 @{甲} 确认测试结果。\n- 请 @{乙} 补充说明。\n\n**遗留**\n\n- @{乙} 按已定计划跟进未解决问题。'
|
|
code, data = self.execute(['progress', 'milestone', '--stage', '阶段六：测试验证',
|
|
body, '--at', '甲,乙,甲', '--at', '丙'], lookup)
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(data['mentions'], ['甲', '乙', '丙'])
|
|
self.assertEqual(len(self.calls), 4)
|
|
cmd = self.calls[-1][0]
|
|
actual = cmd[cmd.index('--markdown') + 1]
|
|
expected = body.replace('@{甲}', '<@fake-a>').replace('@{乙}', '<@fake-b>')
|
|
self.assertTrue(actual.endswith(expected + '\n\n<@fake-c>'))
|
|
self.assertEqual(cmd[cmd.index('--at-open-dingtalk-ids') + 1], 'fake-a,fake-b,fake-c')
|
|
for command, _ in self.calls:
|
|
self.assertEqual(command[command.index('--profile') + 1], PROFILE)
|
|
|
|
def test_inline_mentions_in_message_file(self):
|
|
path = self.root / '正文.md'
|
|
path.write_text('**待办事项**\n\n- 请 @{测试人} 补充资料。', encoding='utf-8')
|
|
code, _ = self.execute(['message', '--title', '阶段三：方案设计', '--file', str(path), '--at', '测试人'])
|
|
self.assertEqual(code, 0)
|
|
cmd = self.calls[-1][0]
|
|
text = cmd[cmd.index('--markdown') + 1]
|
|
self.assertTrue(text.endswith('- 请 <@fake-person> 补充资料。'))
|
|
self.assertEqual(text.count('<@fake-person>'), 1)
|
|
|
|
def test_inline_preview_shows_names_without_lookup(self):
|
|
code, data = self.execute(['progress', 'blocked', '--stage', '阶段三：方案设计',
|
|
'- 请 @{甲} 补充接口说明。', '--at', '甲，甲', '--at', '乙', '--dry-run'])
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(data['text'].endswith('- 请 @甲 补充接口说明。\n\n@乙'))
|
|
self.assertEqual(data['mentions'], ['甲', '乙'])
|
|
self.assertFalse(self.calls)
|
|
|
|
def test_invalid_inline_mentions_stop_before_lookup_or_send(self):
|
|
for body in ['请 @{乙} 确认', '请 @{} 确认', '请 @{甲 确认', '请 @{甲\n} 确认']:
|
|
for options in [[], ['--dry-run']]:
|
|
with self.subTest(body=body, options=options):
|
|
code, result = self.execute(['message', '--title', '提醒', '--text', body, '--at', '甲'] + options)
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result['status'], 'error')
|
|
self.assertFalse(self.calls)
|
|
|
|
def test_inline_mentions_with_conflicting_lookup_results_never_send(self):
|
|
code, result = self.execute(['message', '--title', '提醒', '--text', '- @{甲} 和 @{乙} 请核对。', '--at', '甲,乙'])
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result['status'], 'error')
|
|
self.assertEqual(len(self.calls), 2)
|
|
self.assertTrue(all(c[0][1:3] == ['aisearch', 'person'] for c in self.calls))
|
|
|
|
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', '--stage', '测试验证', '开始'])[1]['status'], 'skipped')
|
|
self.path.unlink()
|
|
self.assertEqual(self.execute(['progress', 'start', '--stage', '测试验证', '开始'])[1]['status'], 'skipped')
|
|
self.assertFalse(self.calls)
|
|
|
|
def test_preview_without_config_or_dws(self):
|
|
self.path.unlink()
|
|
code, result = self.execute(['progress', 'blocked', '--stage', '测试验证', '说明', '--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', '--stage', '测试验证', '开始'])
|
|
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', '--stage', '测试验证', '开始', '--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', '--stage', '测试验证'], ['message', '--title', '说明']]:
|
|
self.assertEqual(self.execute(args + ['--file', str(body)])[0], 0)
|
|
before = len(self.calls)
|
|
for args in [['progress', 'start', '--stage', '测试验证', ''], ['progress', 'start', '--stage', '测试验证', 'x' * 12001],
|
|
['progress', 'start', '--stage', '测试验证', '正文', '--file', str(body)],
|
|
['message', '--title', '坏\n标题', '--text', '内容'],
|
|
['progress', 'start', '--stage', '测试验证', '--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', '--stage', '测试验证', '开始'], 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', '--stage', '测试验证', '开始'], 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', '--stage', '测试验证', '开始'], 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', '--stage', '测试验证', '开始'], 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', '--stage', '测试验证', '开始'], 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', '--stage', '测试验证', '开始'])
|
|
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', '--stage', '测试验证', '--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, '--stage', '测试验证']),
|
|
('dingtalk-message.ps1', ['--title', '中文 "标题"', '--text', text])]:
|
|
result = subprocess.run([pwsh, '-NoProfile', '-File', str(folder / script), *args, '--at', '测试人', '--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.replace('@{测试人}', '@测试人'), data['text'])
|
|
if script == 'dingtalk-progress.ps1':
|
|
self.assertEqual(data['text'].splitlines()[0], '**测试项目｜测试验证**')
|
|
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()
|