Files
goauto/tests/test_wiki_docs.py
T

615 lines
24 KiB
Python

from __future__ import annotations
import json
import os
import sys
import tempfile
import unittest
from argparse import Namespace
from io import BytesIO
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from urllib.error import HTTPError
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "dev_scripts"))
import harness # noqa: E402
from harness import ( # noqa: E402
build_archive,
check_agent_efficiency_rules,
check_required_files,
check_task_template,
existing_task_mirrors,
export_task_archives,
main as harness_main,
run_sync,
safe_title,
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()
@patch("wiki_docs.dirty_mirror_paths", return_value=[])
def test_incremental_sync_lists_once_and_skips_unchanged_body(self, _dirty) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "docs" / "README.md"
target.parent.mkdir(parents=True)
target.write_text(render_mirror(self.page), encoding="utf-8")
config = Config(
path=root / "wiki-docs.json",
gitea_url="http://gitea.example",
owner="o",
repository="r",
mappings=(Mapping("Home", "docs/README.md"),),
)
client = Mock()
client.list_pages.return_value = [
{
"title": "Home",
"sub_url": "Home",
"last_commit": {"sha": self.page.revision},
}
]
with patch("wiki_docs.ROOT", root):
messages = sync_all(config, client)
self.assertTrue(messages[0].startswith("无变化:"))
client.list_pages.assert_called_once_with()
client.get_page_from_metadata.assert_not_called()
@patch("wiki_docs.dirty_mirror_paths", return_value=[])
def test_incremental_sync_fetches_changed_page(self, _dirty) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "docs" / "README.md"
target.parent.mkdir(parents=True)
target.write_text(render_mirror(self.page), encoding="utf-8")
changed = WikiPage(
title="Home",
sub_url="Home",
text="# 新首页\n",
revision="b" * 40,
html_url=self.page.html_url,
)
metadata = {
"title": "Home",
"sub_url": "Home",
"last_commit": {"sha": changed.revision},
}
config = Config(
path=root / "wiki-docs.json",
gitea_url="http://gitea.example",
owner="o",
repository="r",
mappings=(Mapping("Home", "docs/README.md"),),
)
client = Mock()
client.list_pages.return_value = [metadata]
client.get_page_from_metadata.return_value = changed
with patch("wiki_docs.ROOT", root):
messages = sync_all(config, client)
_metadata, body = parse_mirror(target.read_text(encoding="utf-8"))
self.assertTrue(messages[0].startswith("已更新:"))
self.assertEqual(body, changed.text)
client.get_page_from_metadata.assert_called_once_with(metadata, "Home")
@patch("wiki_docs.dirty_mirror_paths", return_value=[])
def test_sync_does_not_write_partial_mirrors_when_later_read_fails(
self, _dirty
) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
first_target = root / "docs" / "first.md"
second_target = root / "docs" / "second.md"
first_target.parent.mkdir(parents=True)
first_original = render_mirror(self.page)
first_target.write_text(first_original, encoding="utf-8")
second_target.write_text(first_original, encoding="utf-8")
config = Config(
path=root / "wiki-docs.json",
gitea_url="http://gitea.example",
owner="o",
repository="r",
mappings=(
Mapping("First", "docs/first.md"),
Mapping("Second", "docs/second.md"),
),
)
pages = [
{
"title": name,
"sub_url": name,
"last_commit": {"sha": revision},
}
for name, revision in (("First", "b" * 40), ("Second", "c" * 40))
]
changed = WikiPage(
title="First",
sub_url="First",
text="# Changed\n",
revision="b" * 40,
html_url="http://gitea.example/o/r/wiki/First",
)
client = Mock()
client.list_pages.return_value = pages
client.get_page_from_metadata.side_effect = [
changed,
WikiDocsError("rate limited"),
]
with patch("wiki_docs.ROOT", root):
with self.assertRaises(WikiDocsError):
sync_all(config, client)
self.assertEqual(first_target.read_text(encoding="utf-8"), first_original)
@patch("wiki_docs.dirty_mirror_paths", return_value=[])
def test_sync_fetches_when_remote_revision_is_missing(self, _dirty) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "docs" / "README.md"
target.parent.mkdir(parents=True)
target.write_text(render_mirror(self.page), encoding="utf-8")
metadata = {"title": "Home", "sub_url": "Home"}
config = Config(
path=root / "wiki-docs.json",
gitea_url="http://gitea.example",
owner="o",
repository="r",
mappings=(Mapping("Home", "docs/README.md"),),
)
client = Mock()
client.list_pages.return_value = [metadata]
client.get_page_from_metadata.return_value = self.page
with patch("wiki_docs.ROOT", root):
sync_all(config, client)
client.get_page_from_metadata.assert_called_once_with(metadata, "Home")
def test_deep_check_fetches_unchanged_page_body(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
target = root / "docs" / "README.md"
target.parent.mkdir(parents=True)
target.write_text(render_mirror(self.page), encoding="utf-8")
metadata = {
"title": "Home",
"sub_url": "Home",
"last_commit": {"sha": self.page.revision},
}
config = Config(
path=root / "wiki-docs.json",
gitea_url="http://gitea.example",
owner="o",
repository="r",
mappings=(Mapping("Home", "docs/README.md"),),
)
client = Mock()
client.list_pages.return_value = [metadata]
client.get_page_from_metadata.return_value = self.page
with patch("wiki_docs.ROOT", root):
sync_all(config, client, check=True, deep_check=True)
client.list_pages.assert_called_once_with()
client.get_page_from_metadata.assert_called_once_with(metadata, "Home")
@patch("harness.run_check", return_value=0)
@patch("harness.sync_all", return_value=["已更新:docs/README.md"])
@patch("harness.WikiClient")
@patch("harness.load_config")
def test_verify_reuses_one_deep_sync(
self, load, client_class, sync, check
) -> None:
config = Mock()
load.return_value = config
result = run_sync(
Namespace(verify=True, check=False, deep_check=False, config="config.json")
)
self.assertEqual(result, 0)
sync.assert_called_once_with(
config, client_class.return_value, check=False, deep_check=True
)
check.assert_called_once()
class WikiClientTests(unittest.TestCase):
@patch("wiki_docs.time.sleep")
@patch("wiki_docs.urlopen")
def test_get_retries_a_rate_limited_request(self, urlopen, sleep) -> None:
config = Config(
path=Path("wiki-docs.json"),
gitea_url="http://gitea.example",
owner="o",
repository="r",
mappings=(),
)
limited = HTTPError(
"http://gitea.example/api/v1/repos/o/r/wiki/pages",
429,
"Too Many Requests",
None,
BytesIO(b"limited"),
)
success = MagicMock()
success.__enter__.return_value.read.return_value = b"[]"
urlopen.side_effect = [limited, success]
result = WikiClient(config, token="token")._request(
"GET", "/repos/o/r/wiki/pages"
)
self.assertEqual(result, [])
self.assertEqual(urlopen.call_count, 2)
sleep.assert_called_once_with(1.0)
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("harness.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",
[
"harness.py",
"archive",
"14",
"按需导出",
"--config",
str(config_path),
],
):
result = harness_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("harness.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("harness.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("harness.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)
class GovernancePolicyTests(unittest.TestCase):
def test_repository_rules_enforce_single_task_source_and_optional_archives(self) -> None:
errors: list[str] = []
check_agent_efficiency_rules(errors, ROOT)
check_task_template(errors, ROOT)
self.assertEqual(errors, [])
agents = (ROOT / "AGENTS.md").read_text(encoding="utf-8")
self.assertIn("需要工单的任务以 Gitea 工单作为单次", agents)
self.assertIn("标准流程不创建 Wiki 任务归档", agents)
self.assertIn("Wiki 同步由长期事实变化触发,不由任务完成触发", agents)
def test_current_workflow_requires_new_upstream_governance_sections(self) -> None:
errors: list[str] = []
check_agent_efficiency_rules(errors, ROOT)
self.assertEqual(errors, [])
workflow = (ROOT / "docs" / "01-workflow.md").read_text(encoding="utf-8")
local_development = (
ROOT / "docs" / "04-local-development-and-verification.md"
).read_text(encoding="utf-8")
self.assertIn("## Gitea 交互与工单最小读取", workflow)
self.assertIn("### 线上原型审核与按需导出", workflow)
self.assertIn("## Windows PowerShell 与 UTF-8", local_development)
def test_goauto_uses_lightweight_governance_and_explicit_authorization(self) -> None:
agents = (ROOT / "AGENTS.md").read_text(encoding="utf-8")
profile = (ROOT / "docs" / "00-project-profile.md").read_text(
encoding="utf-8"
)
workflow = (ROOT / "docs" / "01-workflow.md").read_text(encoding="utf-8")
for text in (agents, profile, workflow):
self.assertIn("轻量治理", text)
for text in (agents, workflow):
self.assertIn("明确授权后的执行", text)
self.assertIn("不能单独构成人工授权", text)
self.assertIn("不得仅因操作不可逆而重复询问或拒绝", text)
self.assertIn("平台自身强制的审批", text)
def test_language_and_windows_command_policy_is_documented(self) -> None:
agents = (ROOT / "AGENTS.md").read_text(encoding="utf-8")
workflow = (ROOT / "docs" / "01-workflow.md").read_text(encoding="utf-8")
local_development = (
ROOT / "docs" / "04-local-development-and-verification.md"
).read_text(encoding="utf-8")
for text in (agents, workflow):
self.assertIn("默认使用中文", text)
self.assertIn("日志和错误原文保持原样", text)
self.assertIn("不得默认套用 Bash", agents)
self.assertIn("### PowerShell 语法与外部命令", local_development)
self.assertIn("rg -n -g '*.md'", local_development)
def test_deployment_template_is_required_and_mapped(self) -> None:
path = "docs/templates/deployment.md"
self.assertIn(path, harness.REQUIRED_FILES)
config = load_config()
mappings = {mapping.page: mapping.path for mapping in config.mappings}
self.assertEqual(mappings.get("Deployment-Template"), path)
self.assertNotIn(path, harness.CORE_DOCUMENT_REQUIREMENTS)
def test_missing_deployment_template_is_reported(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
with patch.object(harness, "ROOT", root):
errors: list[str] = []
check_required_files(errors)
self.assertTrue(
any("docs/templates/deployment.md" in error for error in errors)
)
def test_task_template_requires_online_prototype_and_optional_snapshot_fields(
self,
) -> None:
errors: list[str] = []
check_task_template(errors, ROOT)
self.assertEqual(errors, [])
template = (ROOT / ".gitea" / "issue_template" / "task.md").read_text(
encoding="utf-8"
)
self.assertIn("- 本地 HTML 导出:未要求 / 用户明确要求 / 项目规则要求", template)
self.assertIn("## 任务记录与可选快照", template)
if __name__ == "__main__":
unittest.main()