dev_harness/tests/test_wiki_docs.py

from __future__ import annotations

import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch


ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "dev_scripts"))

from new_task_archive import (  # noqa: E402
    build_archive,
    main as new_archive_main,
    safe_title,
)
from export_task_archives import (  # noqa: E402
    existing_task_mirrors,
    export_task_archives,
    task_target,
)
from wiki_docs import (  # noqa: E402
    Config,
    Mapping,
    WikiClient,
    WikiDocsError,
    WikiPage,
    dirty_mirror_paths,
    load_config,
    parse_mirror,
    render_mirror,
    sync_all,
    validate_mappings,
)


class MappingTests(unittest.TestCase):
    def test_rejects_path_outside_docs(self) -> None:
        with self.assertRaisesRegex(WikiDocsError, "docs/"):
            validate_mappings([{"page": "Home", "path": "README.md"}])

    def test_rejects_duplicate_page(self) -> None:
        with self.assertRaisesRegex(WikiDocsError, "重复映射"):
            validate_mappings(
                [
                    {"page": "Home", "path": "docs/README.md"},
                    {"page": "Home", "path": "docs/other.md"},
                ]
            )

    def test_normalizes_api_suffix_from_environment(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "wiki-docs.json"
            path.write_text(
                json.dumps(
                    {
                        "schema_version": 1,
                        "gitea_url": "http://configured.example",
                        "owner": "owner",
                        "repository": "repo",
                        "mappings": [
                            {"page": "Home", "path": "docs/README.md"}
                        ],
                    }
                ),
                encoding="utf-8",
            )
            with patch.dict(
                os.environ, {"GITEA_URL": "http://gitea.example/api/v1"}, clear=False
            ):
                config = load_config(path)
        self.assertEqual(config.gitea_url, "http://gitea.example")


class MirrorTests(unittest.TestCase):
    def setUp(self) -> None:
        self.page = WikiPage(
            title="Home",
            sub_url="Home",
            text="# 首页\n",
            revision="a" * 40,
            html_url="http://gitea.example/o/r/wiki/Home",
        )

    def test_render_includes_traceable_metadata(self) -> None:
        rendered = render_mirror(self.page)
        metadata, body = parse_mirror(rendered)
        self.assertEqual(metadata["wiki_page"], "Home")
        self.assertEqual(metadata["wiki_revision"], "a" * 40)
        self.assertTrue(metadata["synchronized_at"].endswith("Z"))
        self.assertEqual(body, "# 首页\n")

    def test_unchanged_revision_preserves_sync_time(self) -> None:
        first = render_mirror(self.page)
        second = render_mirror(self.page, first)
        self.assertEqual(first, second)

    @patch("wiki_docs.subprocess.run")
    def test_dirty_mirror_paths_are_reported(self, run) -> None:
        run.return_value.stdout = " M docs/README.md\n"
        config = Config(
            path=Path("wiki-docs.json"),
            gitea_url="http://gitea.example",
            owner="o",
            repository="r",
            mappings=(Mapping("Home", "docs/README.md"),),
        )
        self.assertEqual(dirty_mirror_paths(config), [" M docs/README.md"])

    @patch("wiki_docs.dirty_mirror_paths", return_value=[" M docs/README.md"])
    def test_sync_stops_before_reading_wiki_when_mirror_is_dirty(self, _dirty) -> None:
        config = Config(
            path=Path("wiki-docs.json"),
            gitea_url="http://gitea.example",
            owner="o",
            repository="r",
            mappings=(Mapping("Home", "docs/README.md"),),
        )
        client = Mock()
        with self.assertRaisesRegex(WikiDocsError, "未提交改动"):
            sync_all(config, client)
        client.get_page.assert_not_called()


class WikiClientTests(unittest.TestCase):
    def test_encoded_unicode_sub_url_is_not_double_encoded(self) -> None:
        config = Config(
            path=Path("wiki-docs.json"),
            gitea_url="http://gitea.example",
            owner="o",
            repository="r",
            mappings=(Mapping("中文", "docs/chinese.md"),),
        )
        client = WikiClient(config, token="")
        client.list_pages = Mock(
            return_value=[{"title": "中文", "sub_url": "%E4%B8%AD%E6%96%87.-"}]
        )
        encoded = __import__("base64").b64encode("# 中文\n".encode()).decode()
        with patch.object(
            client,
            "_request",
            return_value={
                "title": "中文",
                "content_base64": encoded,
                "last_commit": {"sha": "b" * 40},
            },
        ) as request:
            page = client.get_page("中文")
        api_path = request.call_args.args[1]
        self.assertIn("%E4%B8%AD%E6%96%87.-", api_path)
        self.assertNotIn("%25E4", api_path)
        self.assertTrue(page.html_url.endswith("/%E4%B8%AD%E6%96%87.-"))


class ArchiveTests(unittest.TestCase):
    def test_safe_title_handles_windows_characters(self) -> None:
        self.assertEqual(safe_title(' 修复:"登录" / 超时 '), "修复-登录-超时")

    def test_build_archive_replaces_known_fields(self) -> None:
        template = "# <工单号> <标题>\nYYYY-MM-DD\n<链接>\n<页面名>\n"
        result = build_archive(template, "12", "修复登录", "Task-12-login", "http://i/12")
        self.assertIn("# 12 修复登录", result)
        self.assertIn("http://i/12", result)
        self.assertIn("Task-12-login", result)
        self.assertNotIn("YYYY-MM-DD", result)

    @patch("new_task_archive.WikiClient")
    def test_create_archive_does_not_change_core_mapping(self, client_class) -> None:
        with tempfile.TemporaryDirectory() as directory:
            config_path = Path(directory) / "wiki-docs.json"
            original = json.dumps(
                {
                    "schema_version": 1,
                    "gitea_url": "http://gitea.example",
                    "owner": "o",
                    "repository": "r",
                    "mappings": [
                        {"page": "Home", "path": "docs/README.md"}
                    ],
                }
            )
            config_path.write_text(original, encoding="utf-8")
            client = client_class.return_value
            client.list_pages.return_value = []
            client.get_page.return_value = WikiPage(
                title="Task-Archive-Template",
                sub_url="Task-Archive-Template.-",
                text="# <工单号> <标题>\nYYYY-MM-DD\n<链接>\n<页面名>\n",
                revision="a" * 40,
                html_url="http://gitea.example/wiki/template",
            )
            client.create_page.return_value = WikiPage(
                title="Task-14-按需导出",
                sub_url="Task-14.-",
                text="# 14 按需导出\n",
                revision="b" * 40,
                html_url="http://gitea.example/wiki/task-14",
            )
            with patch.object(
                sys,
                "argv",
                [
                    "new_task_archive.py",
                    "14",
                    "按需导出",
                    "--config",
                    str(config_path),
                ],
            ):
                result = new_archive_main()
            self.assertEqual(config_path.read_text(encoding="utf-8"), original)
        self.assertEqual(result, 0)
        client.create_page.assert_called_once()

    def test_task_target_uses_stable_safe_name(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            target = task_target("Task-14-修复:导出", Path(directory))
        self.assertEqual(target.name, "14-修复-导出.md")

    def test_existing_mirror_keeps_historical_custom_filename(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            path = root / "docs" / "task" / "2-初级维护者文档体系.md"
            path.parent.mkdir(parents=True)
            page = WikiPage(
                title="Task-2-Junior-Maintainer-Docs",
                sub_url="Task-2-Junior-Maintainer-Docs.-",
                text="# 2 文档\n",
                revision="c" * 40,
                html_url="http://gitea.example/wiki/task-2",
            )
            path.write_text(render_mirror(page), encoding="utf-8")
            mirrors = existing_task_mirrors(root)
        self.assertEqual(
            mirrors["Task-2-Junior-Maintainer-Docs"].name,
            "2-初级维护者文档体系.md",
        )

    @patch("export_task_archives.dirty_paths", return_value=[])
    def test_incremental_export_skips_same_revision(self, _dirty) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            path = root / "docs" / "task" / "14-按需导出.md"
            path.parent.mkdir(parents=True)
            page = WikiPage(
                title="Task-14-按需导出",
                sub_url="Task-14.-",
                text="# 14 按需导出\n",
                revision="d" * 40,
                html_url="http://gitea.example/wiki/task-14",
            )
            path.write_text(render_mirror(page), encoding="utf-8")
            client = Mock()
            client.list_pages.return_value = [
                {
                    "title": page.title,
                    "sub_url": page.sub_url,
                    "last_commit": {"sha": page.revision},
                }
            ]
            messages = export_task_archives(client, root=root)
        self.assertTrue(messages[0].startswith("跳过："))
        client.get_page_from_metadata.assert_not_called()

    @patch(
        "export_task_archives.dirty_paths",
        return_value=[" M docs/task/14-按需导出.md"],
    )
    def test_export_stops_before_wiki_read_when_task_mirror_is_dirty(
        self, _dirty
    ) -> None:
        client = Mock()
        with self.assertRaisesRegex(WikiDocsError, "未提交改动"):
            export_task_archives(client)
        client.list_pages.assert_not_called()

    @patch("export_task_archives.dirty_paths", return_value=[])
    def test_full_export_reads_all_and_never_deletes_extra_file(self, _dirty) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            task_dir = root / "docs" / "task"
            task_dir.mkdir(parents=True)
            extra = task_dir / "99-历史快照.md"
            extra_page = WikiPage(
                title="Task-99-历史快照",
                sub_url="Task-99.-",
                text="# 99 历史快照\n",
                revision="e" * 40,
                html_url="http://gitea.example/wiki/task-99",
            )
            extra.write_text(render_mirror(extra_page), encoding="utf-8")
            page = WikiPage(
                title="Task-14-按需导出",
                sub_url="Task-14.-",
                text="# 14 按需导出\n",
                revision="f" * 40,
                html_url="http://gitea.example/wiki/task-14",
            )
            client = Mock()
            metadata = {
                "title": page.title,
                "sub_url": page.sub_url,
                "last_commit": {"sha": page.revision},
            }
            client.list_pages.return_value = [metadata]
            client.get_page_from_metadata.return_value = page
            messages = export_task_archives(client, export_all=True, root=root)
            exported = root / "docs" / "task" / "14-按需导出.md"
            self.assertTrue(exported.is_file())
            self.assertTrue(extra.is_file())
        self.assertTrue(messages[0].startswith("已导出："))
        client.get_page_from_metadata.assert_called_once_with(metadata, page.title)


if __name__ == "__main__":
    unittest.main()