#!/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()