skills upload function test_web.py

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"