From fea4f2b5126fe93aebfc77d474ee2b91e1f1103f Mon Sep 17 00:00:00 2001 From: kuangji <819823900@qq.com> Date: Tue, 26 May 2026 10:34:22 +0800 Subject: [PATCH] skills upload function test_web.py --- tests/test_web.py | 98 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/tests/test_web.py b/tests/test_web.py index b3c6664..58fe74d 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -1,10 +1,40 @@ +import asyncio from pathlib import Path +import zipfile from docx import Document +import app.main as main from app.main import OUTPUT_DIR, ROOT_DIR, analyze_saved_docx, app +class FakeUploadFile: + def __init__(self, filename: str, content: bytes) -> None: + self.filename = filename + self._content = content + + async def read(self) -> bytes: + return self._content + + +def _write_skill_collection_zip(path: Path) -> None: + with zipfile.ZipFile(path, "w") as archive: + archive.writestr( + "index.md", + "| Skill | Description | Use When |\n" + "| --- | --- | --- |\n" + "| [demo-skill](demo-skill/SKILL.md) | 示例技能 | 上传合集测试 |\n", + ) + archive.writestr( + "demo-skill/SKILL.md", + "---\n" + "name: demo-skill\n" + "description: 示例技能\n" + "---\n" + "# Demo Skill\n", + ) + + def test_index_template_contains_upload_ui() -> None: html = (ROOT_DIR / "app" / "templates" / "index.html").read_text(encoding="utf-8") js = (ROOT_DIR / "app" / "static" / "app.js").read_text(encoding="utf-8") @@ -17,10 +47,78 @@ def test_index_template_contains_upload_ui() -> None: assert "download-md" in js assert "pollTask" in js assert "skill_collection" in html + assert "skill-upload-form" in html + assert "/skill-collections/upload" in js assert "预留后续版本:单个技能集合内的 skill 筛选功能" in html assert not any(route.path == "/skills" for route in app.routes) +def test_skill_collection_options_discover_added_directory(tmp_path: Path, monkeypatch) -> None: + skills_root = tmp_path / "skills" + collection = skills_root / "interesting_physics_skills" + (collection / "demo-skill").mkdir(parents=True) + (collection / "index.md").write_text( + "| Skill | Description | Use When |\n" + "| --- | --- | --- |\n" + "| [demo-skill](demo-skill/SKILL.md) | 示例技能 | 后台新增合集 |\n", + encoding="utf-8", + ) + (collection / "demo-skill" / "SKILL.md").write_text( + "---\nname: demo-skill\ndescription: 示例技能\n---\n# Demo\n", + encoding="utf-8", + ) + monkeypatch.setattr(main, "SKILL_ROOT", skills_root) + + options = main._skill_collection_options() + + assert [option["slug"] for option in options] == ["interesting_physics_skills"] + assert options[0]["skill_count"] == 1 + + +def test_upload_skill_collection_zip_extracts_and_lists(tmp_path: Path, monkeypatch) -> None: + skills_root = tmp_path / "skills" + monkeypatch.setattr(main, "SKILL_ROOT", skills_root) + archive_path = tmp_path / "uploaded_skills.zip" + _write_skill_collection_zip(archive_path) + + upload = FakeUploadFile("uploaded_skills.zip", archive_path.read_bytes()) + payload = asyncio.run(main.upload_skill_collection(upload)) + + assert payload["collection"]["slug"] == "uploaded_skills" + assert payload["collection"]["skill_count"] == 1 + assert (skills_root / "uploaded_skills" / "index.md").exists() + assert any(collection["slug"] == "uploaded_skills" for collection in payload["collections"]) + + +def test_upload_skill_collection_rejects_non_zip(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(main, "SKILL_ROOT", tmp_path / "skills") + + upload = FakeUploadFile("uploaded_skills.txt", b"not zip") + try: + asyncio.run(main.upload_skill_collection(upload)) + except main.HTTPException as exc: + assert exc.status_code == 400 + assert "zip" in exc.detail + else: + raise AssertionError("non-zip upload should fail") + + +def test_install_skill_collection_zip_rejects_unsafe_paths(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(main, "SKILL_ROOT", tmp_path / "skills") + archive_path = tmp_path / "unsafe.zip" + with zipfile.ZipFile(archive_path, "w") as archive: + archive.writestr("../index.md", "bad") + + upload = FakeUploadFile("unsafe.zip", archive_path.read_bytes()) + try: + asyncio.run(main.upload_skill_collection(upload)) + except main.HTTPException as exc: + assert exc.status_code == 400 + assert "非法路径" in exc.detail + else: + raise AssertionError("unsafe zip should fail") + + def test_analyze_saved_docx_reports_progress(tmp_path: Path) -> None: updates: list[tuple[int, str]] = [] docx_path = tmp_path / "progress.docx"