"""文档中心（md 单一来源）端点测试 —— 任务 yanxue-20260924-cma-docs-center-poc 覆盖： 1. 列表：元数据/分类视图/排序/默认不带正文、`with_content=true` 带正文 2. 单篇：200 / 未知 slug 404 / 非法 slug 400（含路径穿越） 3. 鉴权：无 token → 401（不得成为 B 类缺口）；持 token → 200 4. 健壮性：缺 front-matter / 缺必填字段 → skipped 且不影响其它文档；目录缺失 → available=false 不 500 5. **核心判据**：改 md 后同一进程再次请求即返回新内容（运行期读盘、无缓存、无 build-time 打包） 6. 生产内容目录金丝雀（目录不存在时跳过） """ import json import os import pytest from fastapi.testclient import TestClient from tests.conftest import create_test_user, get_token_for_user, auth_header DOCS_ENV = "CMA_DOCS_DIR" PROD_CONTENT = "/srv/hermes/shared/cma-content" def write_doc(root, slug, body="#### 标题\n\n正文段落。\n", title="标题", category="term", summary="摘要", category_name="术语解释", icon="📖", order=10, extra=""): path = root / "docs" / f"{slug}.md" path.write_text( "---\n" f'title: "{title}"\n' f"category: {category}\n" f"category_name: {category_name}\n" f'icon: "{icon}"\n' f'summary: "{summary}"\n' f"order: {order}\n" f"{extra}" "---\n" + body, encoding="utf-8", ) return path @pytest.fixture def content_dir(tmp_path, monkeypatch): """隔离的临时内容目录（不动生产目录）""" root = tmp_path / "cma-content" (root / "docs").mkdir(parents=True) (root / "categories.json").write_text(json.dumps({"categories": [ {"key": "faq", "name": "常见问题", "icon": "❓", "order": 40}, {"key": "term", "name": "术语解释", "icon": "📖", "order": 10}, ]}, ensure_ascii=False), encoding="utf-8") monkeypatch.setenv(DOCS_ENV, str(root)) return root @pytest.fixture def token(client: TestClient, db) -> str: create_test_user(db) return get_token_for_user(client) class TestListDocs: def test_meta_only_and_category_order(self, client, content_dir, token): write_doc(content_dir, "term-b", title="B术语", order=20) write_doc(content_dir, "term-a", title="A术语", order=10) write_doc(content_dir, "faq-x", title="常见问题X", category="faq", category_name="常见问题", icon="❓", order=10) r = client.get("/api/cma/docs", headers=auth_header(token)) assert r.status_code == 200, r.text body = r.json() assert body["count"] == 3 and body["available"] is True and body["format"] == "markdown" assert body["source"].endswith("/docs") # 分类顺序 = categories.json 的 order（term 10 在 faq 40 之前，与文件写入顺序无关） assert [c["key"] for c in body["categories"]] == ["term", "faq"] assert [c["count"] for c in body["categories"]] == [2, 1] assert body["categories"][0]["name"] == "术语解释" and body["categories"][0]["icon"] == "📖" # 文档顺序 = (分类 order, 篇内 order, slug) assert [d["slug"] for d in body["data"]] == ["term-a", "term-b", "faq-x"] assert "content" not in body["data"][0], "默认不得带正文" assert set(body["data"][0]) >= {"slug", "title", "summary", "category", "category_name", "icon", "order", "bytes", "updated_at"} assert body["skipped"] == [] def test_with_content(self, client, content_dir, token): write_doc(content_dir, "term-a", body="#### H\n\n**加粗** 正文\n") r = client.get("/api/cma/docs?with_content=true", headers=auth_header(token)) assert r.status_code == 200 assert r.json()["data"][0]["content"] == "#### H\n\n**加粗** 正文\n" def test_filter_by_category(self, client, content_dir, token): write_doc(content_dir, "term-a") write_doc(content_dir, "faq-x", category="faq", category_name="常见问题") r = client.get("/api/cma/docs?category=faq", headers=auth_header(token)) body = r.json() assert [d["slug"] for d in body["data"]] == ["faq-x"] # 分类视图按全量统计，不受过滤影响 assert [c["count"] for c in body["categories"]] == [1, 1] def test_unregistered_category_still_listed(self, client, content_dir, token): write_doc(content_dir, "other-a", category="其他类", category_name="其他类") r = client.get("/api/cma/docs", headers=auth_header(token)) body = r.json() assert body["count"] == 1 assert [c["key"] for c in body["categories"]] == ["term", "faq", "其他类"] class TestGetDoc: def test_detail(self, client, content_dir, token): write_doc(content_dir, "term-a", body="#### H\n\n正文x\n") r = client.get("/api/cma/docs/term-a", headers=auth_header(token)) assert r.status_code == 200 d = r.json() assert d["slug"] == "term-a" and d["title"] == "标题" and d["content"] == "#### H\n\n正文x\n" def test_unknown_slug_404(self, client, content_dir, token): r = client.get("/api/cma/docs/no-such-doc", headers=auth_header(token)) assert r.status_code == 404 @pytest.mark.parametrize("slug", ["Upper", "with space", "a@b", "-lead", "x" * 65]) def test_invalid_slug_400(self, client, content_dir, token, slug): r = client.get(f"/api/cma/docs/{slug}", headers=auth_header(token)) assert r.status_code in (400, 404), f"非法 slug 不得进入文件系统查找: {slug} → {r.status_code}" if r.status_code == 400: assert "无效的文档标识" in r.text def test_path_traversal_blocked(self, content_dir): """越权取文件：`../`、绝对路径、URL 编码一律被 slug 白名单拦下（无 HTTP 层绕行）""" from app.api import doc_center from fastapi import HTTPException for bad in ["../categories", "..%2F..%2Fetc%2Fpasswd", "/etc/passwd", "%2e%2e%2fsecret", "....//x", "a/b"]: with pytest.raises(HTTPException) as ei: doc_center.get_doc(bad) assert ei.value.status_code == 400 # 正对照：合法 slug 走查文件路径（不存在 → 404，不是 400） with pytest.raises(HTTPException) as ei: doc_center.get_doc("term-a") assert ei.value.status_code == 404 class TestAuth: def test_no_token_401(self, client, content_dir): assert client.get("/api/cma/docs").status_code == 401 assert client.get("/api/cma/docs/term-a").status_code == 401 def test_with_token_200(self, client, content_dir, token): write_doc(content_dir, "term-a") assert client.get("/api/cma/docs", headers=auth_header(token)).status_code == 200 class TestRobustness: def test_bad_md_skipped_not_fatal(self, client, content_dir, token): (content_dir / "docs" / "no-front.md").write_text("没有 front-matter\n", encoding="utf-8") write_doc(content_dir, "missing-title", title="", summary="s") (content_dir / "docs" / "BadSlug.md").write_text( '---\ntitle: "T"\ncategory: term\nsummary: "S"\n---\nX\n', encoding="utf-8") write_doc(content_dir, "term-good") r = client.get("/api/cma/docs", headers=auth_header(token)) assert r.status_code == 200 body = r.json() assert [d["slug"] for d in body["data"]] == ["term-good"] files = {s["file"]: s["reason"] for s in body["skipped"]} assert "no-front.md" in files and "front-matter" in files["no-front.md"] assert "missing-title.md" in files and "title" in files["missing-title.md"] assert "BadSlug.md" in files and "slug" in files["BadSlug.md"] def test_missing_dir_not_500(self, client, tmp_path, monkeypatch, token): monkeypatch.setenv(DOCS_ENV, str(tmp_path / "not-exist")) r = client.get("/api/cma/docs", headers=auth_header(token)) assert r.status_code == 200 body = r.json() assert body["count"] == 0 and body["available"] is False and body["data"] == [] def test_front_matter_multi_colon_and_quotes(self, content_dir): from app.api import doc_center p = write_doc(content_dir, "term-q", title='带:冒号"引号"的标题', summary="摘要: 含冒号") doc, err = doc_center._read_doc(p) assert err is None assert doc["title"] == '带:冒号"引号"的标题' and doc["summary"] == "摘要: 含冒号" class TestRuntimeSource: """核心判据：内容运行期读盘 —— 改 md 即生效，无需改前端 / rebuild / 重启""" def test_edit_md_visible_on_next_request(self, client, content_dir, token): p = write_doc(content_dir, "term-a", body="#### 旧内容\n") r1 = client.get("/api/cma/docs/term-a", headers=auth_header(token)).json() assert r1["content"] == "#### 旧内容\n" # 只改 md 文件（不改码、不重启、不构建） p.write_text(p.read_text(encoding="utf-8").replace("旧内容", "新内容"), encoding="utf-8") r2 = client.get("/api/cma/docs/term-a", headers=auth_header(token)).json() assert r2["content"] == "#### 新内容\n" def test_add_md_visible_without_restart(self, client, content_dir, token): assert client.get("/api/cma/docs", headers=auth_header(token)).json()["count"] == 0 write_doc(content_dir, "term-new", title="新增篇目") body = client.get("/api/cma/docs?with_content=true", headers=auth_header(token)).json() assert body["count"] == 1 and body["data"][0]["title"] == "新增篇目" @pytest.mark.skipif(not os.path.isdir(PROD_CONTENT), reason="生产内容目录不存在（非本机环境）") class TestProdContentCanary: """生产内容目录金丝雀：真实目录必须可读、无 skipped、分类齐备（防内容被写坏）""" def test_prod_content_readable(self, client, token, monkeypatch): monkeypatch.setenv(DOCS_ENV, PROD_CONTENT) r = client.get("/api/cma/docs", headers=auth_header(token)) assert r.status_code == 200 body = r.json() assert body["available"] is True and body["skipped"] == [], body["skipped"] assert body["count"] >= 20, f"迁移应有 20 篇，实际 {body['count']}" # 口径（2026-10-12 失鲜收口）：分类集合随内容扩展（当日新增 manual/模块手册，order 5）， # 故只断言「必备分类在场 + 相对顺序正确」，不再硬编码全集，避免下次扩类再留假红。 keys = [c["key"] for c in body["categories"]] required = ["term", "formula", "practice", "faq"] assert set(required) <= set(keys), \ f"生产内容缺必备分类：{sorted(set(required) - set(keys))}；实际 {keys}" assert [k for k in keys if k in required] == required, f"必备分类相对顺序异常：{keys}" for slug in ("term-kpi", "formula-achievement-rate", "practice-threshold-tips", "faq-data-sync"): d = client.get(f"/api/cma/docs/{slug}", headers=auth_header(token)).json() assert d["title"] and d["content"].strip()