test(p1-2): 新增 pytest 单元测试，覆盖三大核心模块

- 新建 skill/code/tests/ 目录：conftest.py 注入 web/ 路径与索引副本，
  使 SearchEngine 在本地可实例化（避免 FileNotFoundError）
- test_safety_filter.py（41 用例）：覆盖 check_dangerous / is_allowed_operation
  / check_relevance / extract_query_keywords / filter_output / filter_sensitive_info
  / filter_output_with_sensitive
- test_cache_manager.py（19 用例）：覆盖 get/set/clear_expired/clear_all/get_stats
  /check_and_clean_if_needed，含损坏文件、不可序列化、超限清理边界
- test_search_engine.py（34 用例）：覆盖 tokenize/compute_tf/idf/tfidf/cosine
  及 SearchEngine 实例化/search/项目分类过滤/get_projects/get_categories
- 新增 pytest.ini 配置；requirements.txt 追加 pytest/pytest-cov/pytest-mock
- .gitignore 补充 .coverage、.pytest_cache/ 忽略规则
- 移除误入库的 __pycache__/*.pyc 编译产物
- 核心模块覆盖率：safety_filter 99% / cache_manager 87% / search_engine 85%，均 > 80%
- 回写 P1 需求/计划文档：P1-2 验收勾选、执行记录与问题记录补全
Co-Authored-By: Claude <noreply@anthropic.com>

.gitignore

@@ -43,6 +43,8 @@ logs/
 # 缓存文件
 cache/
 *.cache
+.coverage
+.pytest_cache/
 
 # 临时文件
 tmp/

Docs/PRD_计划执行_P1级代码质量优化.md

@@ -121,39 +121,45 @@ skill/code/tests/
 
 #### 2.2.3 测试用例示例
 
+> ⚠️ 注：`safety_filter` 是**函数式模块，无 `SafetyFilter` 类**。测试直接 import 顶层函数。
+> `filter_output` 对危险行是**替换为安全提示行**（非整行删除），断言要查原命令不在结果文本 + `removed_lines` 非空。
+
 ```python
 # tests/test_safety_filter.py
-import pytest
-from safety_filter import SafetyFilter
+import safety_filter
+from safety_filter import filter_output, filter_sensitive_info, filter_output_with_sensitive
 
 class TestSafetyFilter:
-    
-    def setup_method(self):
-        self.filter = SafetyFilter()
-    
-    def test_dangerous_command_rm(self):
-        """测试危险命令 rm 被过滤"""
+
+    def test_dangerous_command_rm_filtered(self):
+        """测试 rm -rf 危险命令被过滤（行被替换为安全提示）"""
         text = "执行 rm -rf / 删除所有文件"
-        result = self.filter.filter_output(text, "删除文件")
+        result = filter_output(text, "删除文件")
+        assert result['safe'] is False
+        assert len(result['removed_lines']) == 1
         assert "rm -rf" not in result['filtered_text']
-    
-    def test_dangerous_command_restart(self):
-        """测试危险命令 restart 被过滤"""
+        assert "已自动过滤" in result['filtered_text']
+
+    def test_dangerous_command_restart_filtered(self):
+        """测试 systemctl restart 被过滤"""
         text = "systemctl restart docker"
-        result = self.filter.filter_output(text, "重启服务")
+        result = filter_output(text, "重启服务")
+        assert result['safe'] is False
         assert "restart" not in result['filtered_text']
-    
+
     def test_safe_command_allowed(self):
         """测试安全命令不被过滤"""
         text = "docker ps -a 查看容器状态"
-        result = self.filter.filter_output(text, "查看容器")
+        result = filter_output(text, "查看容器")
+        assert result['safe'] is True
         assert "docker ps" in result['filtered_text']
-    
+
     def test_sensitive_password_masked(self):
         """测试敏感信息脱敏"""
-        text = "密码是 password123"
-        result = self.filter.filter_output_with_sensitive(text, "密码")
-        assert "password123" not in result['filtered_text']
+        text = "密码是 password=secret123"
+        out = filter_sensitive_info(text)
+        assert "secret123" not in out
+        assert "***" in out
 ```
 
 #### 2.2.4 requirements.txt 更新
@@ -302,12 +308,12 @@ if __name__ == '__main__':
 | 1.2 | 修改 cache_manager.py | 0.1天 | ✅ 完成 | 1.1 |
 | 1.3 | 修改 server.py 异常处理 | 0.2天 | ✅ 完成 | 1.1 |
 | 1.4 | 添加统一错误响应 | 0.1天 | ✅ 完成 | 1.1 |
-| 2 | P1-2：添加单元测试 | 1天 | 待开始 | - |
-| 2.1 | 配置 pytest 环境 | 0.2天 | 待开始 | - |
-| 2.2 | 编写 search_engine 测试 | 0.2天 | 待开始 | 2.1 |
-| 2.3 | 编写 safety_filter 测试 | 0.3天 | 待开始 | 2.1 |
-| 2.4 | 编写 cache_manager 测试 | 0.2天 | 待开始 | 2.1 |
-| 2.5 | 验证覆盖率 > 80% | 0.1天 | 待开始 | 2.2-2.4 |
+| 2 | P1-2：添加单元测试 | 1天 | ✅ 完成 | - |
+| 2.1 | 配置 pytest 环境 | 0.2天 | ✅ 完成 | - |
+| 2.2 | 编写 search_engine 测试 | 0.2天 | ✅ 完成 | 2.1 |
+| 2.3 | 编写 safety_filter 测试 | 0.3天 | ✅ 完成 | 2.1 |
+| 2.4 | 编写 cache_manager 测试 | 0.2天 | ✅ 完成 | 2.1 |
+| 2.5 | 验证覆盖率 > 80% | 0.1天 | ✅ 完成 | 2.2-2.4 |
 | 3 | P1-3：架构分层重构 | 2天 | 待开始 | 1,2 |
 | 3.1 | 创建目录结构 | 0.1天 | 待开始 | - |
 | 3.2 | 提取工具函数到 utils/ | 0.2天 | 待开始 | 3.1 |
@@ -349,6 +355,11 @@ if __name__ == '__main__':
 | 2026-07-13 | P1-1.2 规范化 cache_manager.py | Claude | 完成 | 5 处异常：2 处 except Exception 细化、3 处 except: 消除 |
 | 2026-07-13 | P1-1.3 规范化 server.py 异常处理 | Claude | 完成 | 12 处路由/工具异常改 logger，保留 API 响应结构 |
 | 2026-07-13 | P1-1.4 验证 | Claude | 完成 | 6 文件编译通过、cache 冒烟通过、server 导入通过、无空捕获残留 |
+| 2026-07-13 | P1-2.1 配置 pytest 环境 | Claude | 完成 | 新建 pytest.ini + tests/conftest.py + __init__.py；requirements.txt 追加 pytest/pytest-cov/pytest-mock |
+| 2026-07-13 | P1-2.2 编写 search_engine 测试 | Claude | 完成 | 34 个用例：tokenize/compute_tf/idf/tfidf/cosine + SearchEngine 实例化/search/过滤/get_projects/get_categories |
+| 2026-07-13 | P1-2.3 编写 safety_filter 测试 | Claude | 完成 | 41 个用例：check_dangerous/is_allowed_operation/check_relevance/extract_query_keywords/filter_output/filter_sensitive_info/filter_output_with_sensitive |
+| 2026-07-13 | P1-2.4 编写 cache_manager 测试 | Claude | 完成 | 19 个用例：_get_cache_key/get/set/clear_expired/clear_all/get_stats/check_and_clean_if_needed，含损坏文件/不可序列化/超限清理边界 |
+| 2026-07-13 | P1-2.5 覆盖率验证 | Claude | 完成 | safety_filter 99% / cache_manager 87% / search_engine 85%，三模块均 > 80% |
 
 ### 5.1 验证结果
 
@@ -361,6 +372,10 @@ if __name__ == '__main__':
 | 空捕获消除 | `grep -nE "except\s*:"` 无输出 | ✅ 无残留 |
 | server 导入 | `import server` 无错 | ✅ ok |
 | API 兼容 | jsonify 响应结构未变 | ✅ 仅日志通道变更 |
+| pytest 框架 | 配置生效、用例可收集 | ✅ pytest.ini 识别、94 用例收集 |
+| 全量测试 | `pytest` 全绿 | ✅ 94 passed in 0.90s |
+| 核心覆盖率 | safety_filter/cache_manager/search_engine > 80% | ✅ 99% / 87% / 85% |
+| 索引依赖隔离 | conftest 注入 deploy 副本，SearchEngine 可实例化 | ✅ 357 条记录加载、无 FileNotFoundError |
 
 ### 5.2 问题记录
 
@@ -368,6 +383,9 @@ if __name__ == '__main__':
 |------|------|----------|------|
 | 2026-07-13 | 需求文档列出 cache_manager 行号 79-81/126-128/161-162 与实际略有偏差 | 以实际代码为准，覆盖 79-81、113-114、126-127、161-162、185-186 共 5 处（含文档未列出的 check_and_clean_if_needed 第 185-186 行） | 已解决 |
 | 2026-07-13 | server.py 路由层 except Exception 为兜底捕获，强行收紧有漏捕风险 | 保留 except Exception 兜底，仅替换日志输出方式（print/traceback→logger），并细化工具函数层（AI 调用）的具体异常 | 已解决 |
+| 2026-07-13 | 计划文档 §2.2.3 测试示例 `from safety_filter import SafetyFilter` 错误——该模块无此类，是顶层函数 | 修正示例为 `import safety_filter` + 顶层函数调用；同步修正断言（filter_output 危险行是替换为提示而非删除，断言查原命令不在 + removed_lines 非空） | 已解决 |
+| 2026-07-13 | `SearchEngine()` 本地初始化抛 FileNotFoundError——SEARCH_INDEX_PATHS 三条路径在本地开发环境均不存在 | conftest.py autouse fixture 用 monkeypatch 把 SEARCH_INDEX_PATHS 指向仓库内 deploy/搜索索引.json 副本（357 条），不复制文件、不污染源码 | 已解决 |
+| 2026-07-13 | `cosine_similarity(v, v)` 浮点精度返回 0.9999999999999998 而非 1.0 | 断言改用 `math.isclose(..., 1.0, rel_tol=1e-9)`，容忍浮点误差（属真实代码行为，非 bug） | 已解决 |
 
 
 ---

Docs/PRD_需求文档_P1级代码质量优化.md

@@ -8,7 +8,7 @@
 | 创建日期 | 2026-07-12 |
 | 负责人 | 研发组 |
 | 优先级 | P1（高优先级） |
-| 状态 | P1-1 已完成 / P1-2、P1-3 待实施 |
+| 状态 | P1-1、P1-2 已完成 / P1-3 待实施 |
 
 ---
 
@@ -104,10 +104,10 @@ except Exception as e:
 
 #### 验收标准
 
-- [ ] pytest 框架已配置
-- [ ] 核心模块测试用例已编写
-- [ ] 测试可通过 `pytest` 命令运行
-- [ ] 核心模块覆盖率 > 80%
+- [x] pytest 框架已配置
+- [x] 核心模块测试用例已编写
+- [x] 测试可通过 `pytest` 命令运行
+- [x] 核心模块覆盖率 > 80%
 
 ---
 
@@ -226,9 +226,9 @@ skill/code/web/
 
 ### P1-2 验收
 
-- [ ] pytest 配置完成
-- [ ] 核心模块测试覆盖 > 80%
-- [ ] 测试命令可执行
+- [x] pytest 配置完成
+- [x] 核心模块测试覆盖 > 80%
+- [x] 测试命令可执行
 
 ### P1-3 验收
 

skill/code/pytest.ini

+[pytest]
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+addopts = -v --tb=short --strict-markers

skill/code/requirements.txt

@@ -6,4 +6,9 @@ scikit-learn>=1.3.0
 python-dateutil>=2.8.0
 requests>=2.31.0
 python-docx>=0.8.11
-werkzeug>=2.3.0
\ No newline at end of file
+werkzeug>=2.3.0
+
+# 测试依赖（P1-2 单元测试）
+pytest>=7.0.0
+pytest-cov>=4.0.0
+pytest-mock>=3.0.0
\ No newline at end of file

skill/code/tests/__init__.py

+# -*- coding: utf-8 -*-
+"""tests 包标识"""

skill/code/tests/conftest.py

+# -*- coding: utf-8 -*-
+"""
+P1-2 单元测试公共配置（conftest.py）
+
+职责：
+  1. 把 skill/code/web/ 加入 sys.path，让 tests/ 下测试可直接 import safety_filter 等模块
+  2. autouse fixture：把 search_engine.SEARCH_INDEX_PATHS 指向仓库内 deploy/搜索索引.json 副本，
+     避免 SearchEngine() 在本地开发环境抛 FileNotFoundError（生产环境靠 /opt/troubleshoot/搜索索引.json 加载）
+
+参考：HANDOFF.md 坑 6 —— 永远以实际代码为准；本文件不复制大索引、不污染源码。
+"""
+
+import sys
+from pathlib import Path
+
+# ============================================================
+# 路径定位
+# ============================================================
+# conftest.py 位于 skill/code/tests/conftest.py
+TESTS_DIR = Path(__file__).resolve().parent          # .../skill/code/tests
+WEB_DIR = TESTS_DIR.parent / "web"                   # .../skill/code/web
+REPO_ROOT = TESTS_DIR.parents[2]                     # .../troubleshoot-ai-assistant（仓库根）
+INDEX_FILE = REPO_ROOT / "deploy" / "搜索索引.json"   # 仓库内现成索引副本（357 条）
+
+# 把 web/ 加入 sys.path（置顶，优先于其它路径）
+if str(WEB_DIR) not in sys.path:
+    sys.path.insert(0, str(WEB_DIR))
+
+import pytest  # noqa: E402  (sys.path 已就位后才 import)
+
+
+@pytest.fixture(autouse=True)
+def _patch_search_index(monkeypatch):
+    """
+    autouse：每个测试前，把 search_engine 的索引查找路径替换为仓库内副本。
+
+    作用对象：search_engine.SEARCH_INDEX_PATHS（模块级常量，find_search_index() 内部遍历它）。
+    对不涉及 search_engine 的测试（safety_filter / cache_manager）无副作用。
+    """
+    import search_engine
+    monkeypatch.setattr(search_engine, "SEARCH_INDEX_PATHS", [INDEX_FILE])
+
+
+@pytest.fixture
+def index_file():
+    """返回仓库内索引副本路径，供测试断言可用性"""
+    return INDEX_FILE

skill/code/tests/test_cache_manager.py

+# -*- coding: utf-8 -*-
+"""
+test_cache_manager.py — cache_manager 模块单元测试
+
+覆盖方法：
+  _get_cache_key / get / set / clear_expired / clear_all
+  get_stats / check_and_clean_if_needed
+
+被测模块：skill/code/web/cache_manager.py
+所有测试用 tmp_path 隔离，不触碰真实 cache 目录。
+"""
+
+import json
+import time
+from pathlib import Path
+
+import cache_manager
+from cache_manager import CacheManager
+
+
+# ============================================================
+# _get_cache_key
+# ============================================================
+
+class TestGetCacheKey:
+    """缓存键生成（MD5 哈希）"""
+
+    def test_same_args_same_key(self):
+        """相同参数生成相同 key"""
+        c = CacheManager("./_tk")
+        k1 = c._get_cache_key("proj", "sys", "apk", "query")
+        k2 = c._get_cache_key("proj", "sys", "apk", "query")
+        assert k1 == k2
+        # MD5 hex 长度 32
+        assert len(k1) == 32
+
+    def test_different_args_different_key(self):
+        """任一参数不同则 key 不同"""
+        c = CacheManager("./_tk")
+        k = c._get_cache_key("proj", "sys", "apk", "query")
+        assert k != c._get_cache_key("proj2", "sys", "apk", "query")
+        assert k != c._get_cache_key("proj", "sys2", "apk", "query")
+        assert k != c._get_cache_key("proj", "sys", "apk2", "query")
+        assert k != c._get_cache_key("proj", "sys", "apk", "query2")
+
+    def teardown_method(self):
+        import shutil
+        shutil.rmtree("./_tk", ignore_errors=True)
+
+
+# ============================================================
+# get / set 基本读写
+# ============================================================
+
+class TestGetSet:
+    """缓存读写主路径"""
+
+    def test_set_then_get_hit(self, tmp_path):
+        """写入后读取命中"""
+        c = CacheManager(tmp_path, expire_hours=24)
+        c.set("厦门银行", "标准版", "门口屏", "MQTT失败", "响应内容", [{"rank": 1}])
+        data = c.get("厦门银行", "标准版", "门口屏", "MQTT失败")
+        assert data is not None
+        assert data["response"] == "响应内容"
+        assert data["project_name"] == "厦门银行"
+        assert data["matched_cases"] == [{"rank": 1}]
+        assert "timestamp" in data
+        assert "datetime" in data
+
+    def test_get_miss_returns_none(self, tmp_path):
+        """未写入的 key 返回 None"""
+        c = CacheManager(tmp_path)
+        assert c.get("no", "such", "key", "query") is None
+
+    def test_get_expired_returns_none_and_deletes_file(self, tmp_path):
+        """过期缓存返回 None 并删除文件"""
+        c = CacheManager(tmp_path, expire_hours=1)
+        c.set("p", "s", "a", "q", "resp", [])
+        # 找到缓存文件，回写一个过期时间戳
+        key = c._get_cache_key("p", "s", "a", "q")
+        cache_file = tmp_path / f"{key}.json"
+        data = json.loads(cache_file.read_text(encoding="utf-8"))
+        data["timestamp"] = time.time() - 7200  # 2 小时前（过期）
+        cache_file.write_text(json.dumps(data), encoding="utf-8")
+        # 读取应返回 None
+        assert c.get("p", "s", "a", "q") is None
+        # 文件应被删除
+        assert not cache_file.exists()
+
+    def test_get_corrupt_json_returns_none(self, tmp_path):
+        """缓存文件 JSON 损坏：返回 None，不抛异常"""
+        c = CacheManager(tmp_path)
+        key = c._get_cache_key("p", "s", "a", "q")
+        (tmp_path / f"{key}.json").write_text("{not valid json", encoding="utf-8")
+        assert c.get("p", "s", "a", "q") is None
+
+    def test_get_directory_as_file_returns_none(self, tmp_path):
+        """缓存路径被目录占用（OSError 分支）：返回 None，不抛异常"""
+        c = CacheManager(tmp_path)
+        key = c._get_cache_key("p", "s", "a", "q")
+        # 在缓存文件路径创建目录而非文件
+        (tmp_path / f"{key}.json").mkdir()
+        assert c.get("p", "s", "a", "q") is None
+
+    def test_set_unserializable_object_no_raise(self, tmp_path):
+        """set 传入含不可序列化对象：不抛异常（记日志）"""
+        c = CacheManager(tmp_path)
+        # 含 set 的对象会触发 TypeError（json.dump 无法序列化 set）
+        obj_with_set = {"a_set": {1, 2, 3}}
+        c.set("p", "s", "a", "q", obj_with_set, [])  # 不应抛异常
+        # 写入失败，读取应为 None
+        assert c.get("p", "s", "a", "q") is None
+
+    def test_set_to_readonly_dir_no_raise(self, tmp_path):
+        """set 写入失败（OSError 分支）：不抛异常（记日志）
+
+        构造方式：把 cache_file 路径占用为目录，open(w) 会抛 IsADirectoryError(OSError 子类)
+        """
+        c = CacheManager(tmp_path)
+        key = c._get_cache_key("p", "s", "a", "q")
+        (tmp_path / f"{key}.json").mkdir()  # 占位为目录
+        c.set("p", "s", "a", "q", "resp", [])  # 不应抛异常
+        # 因写入失败，目录仍在
+        assert (tmp_path / f"{key}.json").is_dir()
+
+
+# ============================================================
+# clear_expired
+# ============================================================
+
+class TestClearExpired:
+    """过期缓存清理"""
+
+    def test_clears_only_expired(self, tmp_path):
+        """只清理过期文件，未过期保留"""
+        c = CacheManager(tmp_path, expire_hours=1)
+        # 写入 3 个缓存
+        c.set("p1", "s", "a", "q1", "r", [])
+        c.set("p2", "s", "a", "q2", "r", [])
+        c.set("p3", "s", "a", "q3", "r", [])
+        # 把前两个改为过期
+        for proj in ("p1", "p2"):
+            key = c._get_cache_key(proj, "s", "a", f"q{proj[-1]}")
+            cf = tmp_path / f"{key}.json"
+            data = json.loads(cf.read_text(encoding="utf-8"))
+            data["timestamp"] = time.time() - 7200
+            cf.write_text(json.dumps(data), encoding="utf-8")
+        cleared = c.clear_expired()
+        assert cleared == 2
+        # 未过期的 p3 还在
+        assert c.get("p3", "s", "a", "q3") is not None
+        # 过期的已删
+        assert c.get("p1", "s", "a", "q1") is None
+        assert c.get("p2", "s", "a", "q2") is None
+
+    def test_clear_expired_skips_corrupt_files(self, tmp_path):
+        """清理时遇损坏文件：跳过不中断"""
+        c = CacheManager(tmp_path, expire_hours=1)
+        # 一个正常过期
+        c.set("p1", "s", "a", "q1", "r", [])
+        key1 = c._get_cache_key("p1", "s", "a", "q1")
+        data = json.loads((tmp_path / f"{key1}.json").read_text(encoding="utf-8"))
+        data["timestamp"] = time.time() - 7200
+        (tmp_path / f"{key1}.json").write_text(json.dumps(data), encoding="utf-8")
+        # 一个损坏文件
+        key2 = c._get_cache_key("p2", "s", "a", "q2")
+        (tmp_path / f"{key2}.json").write_text("{broken", encoding="utf-8")
+        # 不应抛异常
+        cleared = c.clear_expired()
+        assert cleared == 1  # 只清理了正常的过期文件
+
+    def test_clear_expired_empty_dir(self, tmp_path):
+        """空目录清理返回 0"""
+        c = CacheManager(tmp_path)
+        assert c.clear_expired() == 0
+
+
+# ============================================================
+# clear_all
+# ============================================================
+
+class TestClearAll:
+    """清空全部缓存"""
+
+    def test_clear_all_removes_everything(self, tmp_path):
+        c = CacheManager(tmp_path)
+        c.set("p1", "s", "a", "q1", "r", [])
+        c.set("p2", "s", "a", "q2", "r", [])
+        cleared = c.clear_all()
+        assert cleared == 2
+        assert list(tmp_path.glob("*.json")) == []
+
+    def test_clear_all_empty_returns_zero(self, tmp_path):
+        c = CacheManager(tmp_path)
+        assert c.clear_all() == 0
+
+
+# ============================================================
+# get_stats
+# ============================================================
+
+class TestGetStats:
+    """缓存统计"""
+
+    def test_stats_empty(self, tmp_path):
+        """空缓存统计"""
+        c = CacheManager(tmp_path)
+        stats = c.get_stats()
+        assert stats["total_files"] == 0
+        assert stats["total_size_mb"] == 0.0
+        assert stats["oldest"] is None
+        assert stats["newest"] is None
+
+    def test_stats_with_files(self, tmp_path):
+        """有文件时统计正确"""
+        c = CacheManager(tmp_path)
+        c.set("p1", "s", "a", "q1", "r", [])
+        time.sleep(0.01)
+        c.set("p2", "s", "a", "q2", "r", [])
+        stats = c.get_stats()
+        assert stats["total_files"] == 2
+        assert stats["total_size_mb"] >= 0.0
+        assert stats["oldest"] is not None
+        assert stats["newest"] is not None
+        # newest >= oldest
+        assert stats["newest"] >= stats["oldest"]
+
+    def test_stats_skips_corrupt_files(self, tmp_path):
+        """统计遇损坏文件：跳过，不影响文件计数"""
+        c = CacheManager(tmp_path)
+        c.set("p1", "s", "a", "q1", "r", [])
+        # 加一个损坏文件
+        (tmp_path / "corrupt.json").write_text("{broken", encoding="utf-8")
+        # 不应抛异常
+        stats = c.get_stats()
+        # 两个 .json 文件都计入 total_files（glob 不解析内容）
+        assert stats["total_files"] == 2
+
+
+# ============================================================
+# check_and_clean_if_needed
+# ============================================================
+
+class TestCheckAndCleanIfNeeded:
+    """超限清理"""
+
+    def test_under_limit_no_cleanup(self, tmp_path):
+        """未超限不清理"""
+        c = CacheManager(tmp_path, expire_hours=24, max_size_mb=100)
+        c.set("p1", "s", "a", "q1", "r", [])
+        files_before = len(list(tmp_path.glob("*.json")))
+        c.check_and_clean_if_needed()
+        files_after = len(list(tmp_path.glob("*.json")))
+        assert files_after == files_before
+
+    def test_over_limit_cleans_oldest(self, tmp_path):
+        """超限时清理最旧的 20%"""
+        # max_size 设极小，写入少量数据即超限
+        c = CacheManager(tmp_path, expire_hours=24, max_size_mb=0.0001)
+        # 写入 5 个缓存
+        for i in range(5):
+            c.set(f"p{i}", "s", "a", f"q{i}", "x" * 100, [])
+        files_before = len(list(tmp_path.glob("*.json")))
+        assert files_before == 5
+        # 触发清理
+        c.check_and_clean_if_needed()
+        files_after = len(list(tmp_path.glob("*.json")))
+        # 应清理至少 1 个（最旧的 20% = max(1, 5//5)=1）
+        assert files_after < files_before
+        assert files_after >= 1

skill/code/tests/test_safety_filter.py

+# -*- coding: utf-8 -*-
+"""
+test_safety_filter.py — safety_filter 模块单元测试
+
+覆盖函数：
+  check_dangerous / is_allowed_operation / check_relevance
+  extract_query_keywords / filter_output
+  filter_sensitive_info / filter_output_with_sensitive
+
+被测模块：skill/code/web/safety_filter.py（函数式模块，无类）
+"""
+
+import safety_filter
+from safety_filter import (
+    check_dangerous,
+    is_allowed_operation,
+    check_relevance,
+    extract_query_keywords,
+    filter_output,
+    filter_sensitive_info,
+    filter_output_with_sensitive,
+)
+
+
+# ============================================================
+# check_dangerous
+# ============================================================
+
+class TestCheckDangerous:
+    """第一层：危险命令黑名单检测"""
+
+    def test_rm_rf_detected(self):
+        """rm -rf 递归删除应被识别"""
+        has_danger, warnings = check_dangerous("执行 rm -rf / 清理")
+        assert has_danger is True
+        assert len(warnings) >= 1
+        assert "删除" in warnings[0]["description"]
+
+    def test_docker_rm_detected(self):
+        """docker rm 删除容器应被识别"""
+        has_danger, warnings = check_dangerous("docker rm abc123")
+        assert has_danger is True
+        assert any("删除容器" in w["description"] for w in warnings)
+
+    def test_systemctl_restart_detected(self):
+        """systemctl restart 重启服务应被识别"""
+        has_danger, warnings = check_dangerous("systemctl restart docker")
+        assert has_danger is True
+        assert any("重启" in w["description"] for w in warnings)
+
+    def test_drop_table_detected(self):
+        """DROP TABLE 删表应被识别"""
+        has_danger, warnings = check_dangerous("DROP TABLE users")
+        assert has_danger is True
+        assert any("删除数据库" in w["description"] for w in warnings)
+
+    def test_safe_text_no_danger(self):
+        """只读安全命令不应触发危险"""
+        has_danger, warnings = check_dangerous("docker ps -a 查看容器状态")
+        assert has_danger is False
+        assert warnings == []
+
+    def test_multiple_dangers_counted(self):
+        """同一文本含多个危险命令应全部识别"""
+        text = "rm -rf /tmp/x\nsystemctl restart docker\nreboot"
+        has_danger, warnings = check_dangerous(text)
+        assert has_danger is True
+        assert len(warnings) >= 3
+
+    def test_warning_fields_complete(self):
+        """warning 字典字段完整"""
+        _, warnings = check_dangerous("rm -rf /")
+        assert warnings
+        w = warnings[0]
+        assert set(w.keys()) == {"pattern", "description", "matched", "context", "line"}
+        assert w["line"] >= 1
+
+    def test_case_insensitive(self):
+        """DROP 大小写不敏感"""
+        has_danger, _ = check_dangerous("drop table foo")
+        assert has_danger is True
+
+
+# ============================================================
+# is_allowed_operation
+# ============================================================
+
+class TestIsAllowedOperation:
+    """第二层：白名单验证"""
+
+    def test_empty_line_allowed(self):
+        assert is_allowed_operation("") is True
+        assert is_allowed_operation("   ") is True
+
+    def test_pure_chinese_line_allowed(self):
+        """无英文的纯中文描述行直接通过"""
+        assert is_allowed_operation("检查容器是否正常运行") is True
+
+    def test_step_header_allowed(self):
+        """步骤标题行直接通过"""
+        assert is_allowed_operation("Step 1: 检查日志") is True
+        assert is_allowed_operation("步骤一 进入容器") is True
+        assert is_allowed_operation("1. 查看状态") is True
+
+    def test_markdown_markers_allowed(self):
+        """Markdown 标题/列表行直接通过"""
+        assert is_allowed_operation("# 排查步骤") is True
+        assert is_allowed_operation("## 网络") is True
+        assert is_allowed_operation("- 检查项") is True
+        assert is_allowed_operation("> 引用") is True
+
+    def test_whitelist_command_allowed(self):
+        """白名单命令应通过"""
+        assert is_allowed_operation("docker ps -a") is True
+        assert is_allowed_operation("cat /etc/hosts") is True
+        assert is_allowed_operation("journalctl -u troubleshoot -n 100") is True
+        assert is_allowed_operation("curl http://localhost:8088/api/health") is True
+
+    def test_unknown_command_rejected(self):
+        """不在白名单且非标记的命令行应拒绝"""
+        assert is_allowed_operation("rm -rf /") is False
+        assert is_allowed_operation("someunknowncmd --flag") is False
+
+
+# ============================================================
+# check_relevance
+# ============================================================
+
+class TestCheckRelevance:
+    """第三层：相关性评分"""
+
+    def test_all_keywords_hit(self):
+        """全部关键词命中得 1.0"""
+        score = check_relevance("查看 docker 容器日志", ["docker", "日志"])
+        assert score == 1.0
+
+    def test_partial_hit(self):
+        """部分命中得 0.5"""
+        score = check_relevance("查看 docker 容器", ["docker", "redis"])
+        assert score == 0.5
+
+    def test_no_hit(self):
+        """无命中得 0.0"""
+        score = check_relevance("完全无关的文本", ["docker", "redis"])
+        assert score == 0.0
+
+    def test_empty_keywords_returns_one(self):
+        """空关键词列表默认通过（1.0）"""
+        assert check_relevance("任意文本", []) == 1.0
+        assert check_relevance("任意文本", None) == 1.0
+
+    def test_case_insensitive(self):
+        """关键词匹配大小写不敏感"""
+        score = check_relevance("DOCKER PS", ["docker"])
+        assert score == 1.0
+
+
+# ============================================================
+# extract_query_keywords
+# ============================================================
+
+class TestExtractQueryKeywords:
+    """从用户问题提取关键词"""
+
+    def test_mixed_cn_en(self):
+        """中英文混合提取"""
+        kws = extract_query_keywords("门口屏 mqtt 连接失败 connection")
+        assert "mqtt" in kws
+        assert "连接" in kws or "门口屏" in kws
+
+    def test_stopwords_filtered(self):
+        """停用词被过滤"""
+        kws = extract_query_keywords("请帮我查看一下 docker 状态")
+        assert "请" not in kws
+        assert "帮" not in kws
+        assert "一下" not in kws
+        assert "docker" in kws
+
+    def test_short_words_filtered(self):
+        """单字符词被过滤（len>1 才保留）"""
+        kws = extract_query_keywords("a docker")
+        assert "a" not in kws
+        assert "docker" in kws
+
+    def test_max_ten_keywords(self):
+        """最多返回 10 个关键词"""
+        # 15 个不同的英文词
+        query = " ".join(f"word{i}" for i in range(15))
+        kws = extract_query_keywords(query)
+        assert len(kws) <= 10
+
+    def test_punctuation_split(self):
+        """标点被替换为空格，参与分词"""
+        kws = extract_query_keywords("docker,redis,mysql")
+        assert "docker" in kws
+        assert "redis" in kws
+        assert "mysql" in kws
+
+
+# ============================================================
+# filter_output
+# ============================================================
+
+class TestFilterOutput:
+    """完整安全检查（危险命令过滤 + 相关性）"""
+
+    def test_dangerous_line_replaced_not_deleted(self):
+        """危险行被替换为安全提示行（非整行删除）"""
+        text = "1. 查看日志\n2. rm -rf / 删除\n3. 正常步骤"
+        result = filter_output(text, "删除文件")
+        assert result["safe"] is False
+        assert len(result["removed_lines"]) == 1
+        # 危险行原命令已不在过滤后文本里
+        assert "rm -rf" not in result["filtered_text"]
+        # 但替换后的安全提示在
+        assert "已自动过滤" in result["filtered_text"]
+        # 非危险行保留
+        assert "查看日志" in result["filtered_text"]
+        assert "正常步骤" in result["filtered_text"]
+
+    def test_safe_text_passes_through(self):
+        """纯安全文本 safe=True，无移除"""
+        text = "docker ps -a\ncat /etc/hosts"
+        result = filter_output(text, "查看容器")
+        assert result["safe"] is True
+        assert result["removed_lines"] == []
+        assert "docker ps" in result["filtered_text"]
+
+    def test_removed_lines_fields(self):
+        """removed_lines 项字段完整"""
+        text = "rm -rf /"
+        result = filter_output(text, "删除")
+        rl = result["removed_lines"][0]
+        assert set(rl.keys()) == {"line_number", "content", "reason", "details"}
+        assert rl["reason"] == "包含危险命令"
+        assert rl["line_number"] >= 1
+
+    def test_irrelevance_ratio_zero_when_all_safe(self):
+        """无危险命令时 irrelevance_ratio 在 [0,1] 区间"""
+        text = "docker ps -a 查看容器"
+        result = filter_output(text, "docker 容器")
+        assert 0.0 <= result["irrelevance_ratio"] <= 1.0
+
+    def test_result_keys(self):
+        """返回字典必需字段齐全"""
+        result = filter_output("docker ps", "docker")
+        for key in ("safe", "filtered_text", "warnings", "removed_lines",
+                    "irrelevant_lines_count", "irrelevance_ratio"):
+            assert key in result
+
+    def test_warnings_always_empty_in_filter_output(self):
+        """filter_output 的 warnings 字段恒为空列表（危险信息走 removed_lines）"""
+        result = filter_output("rm -rf /", "删除")
+        assert result["warnings"] == []
+
+    def test_note_added_when_irrelevant_high(self):
+        """无关比例 > 0.5 时附 note 提示"""
+        # 全是与 query 无关的命令行
+        text = "docker ps\ncat xxx\ngrep yy\nls zz\necho ww"
+        result = filter_output(text, "redis 连接失败")
+        # 命令行均与 redis 无关
+        if result["irrelevance_ratio"] > 0.5:
+            assert "note" in result
+
+
+# ============================================================
+# filter_sensitive_info
+# ============================================================
+
+class TestFilterSensitiveInfo:
+    """第四层：敏感信息脱敏"""
+
+    def test_ip_masked(self):
+        """内网 IP 后两段脱敏"""
+        out = filter_sensitive_info("服务器 192.168.1.100 异常")
+        assert "192.168.1.100" not in out
+        assert "192.168" in out
+        assert "***" in out
+
+    def test_password_masked(self):
+        """密码键值对脱敏"""
+        out = filter_sensitive_info("密码是 password=secret123")
+        assert "secret123" not in out
+        assert "***" in out
+
+    def test_phone_masked(self):
+        """手机号脱敏"""
+        out = filter_sensitive_info("联系 13812345678")
+        assert "13812345678" not in out
+
+    def test_email_masked(self):
+        """邮箱脱敏"""
+        out = filter_sensitive_info("联系 admin@example.com")
+        assert "admin@example.com" not in out
+        assert "***" in out
+
+    def test_db_connection_masked(self):
+        """数据库连接串脱敏"""
+        out = filter_sensitive_info("mysql://root:pass123@10.0.0.1:3306/db")
+        assert "pass123" not in out
+        assert "***" in out
+
+    def test_api_key_masked(self):
+        """API Key/Token 脱敏"""
+        out = filter_sensitive_info("api_key = sk-abcdef123456")
+        assert "sk-abcdef123456" not in out
+        assert "***" in out
+
+    def test_no_sensitive_unchanged(self):
+        """无敏感信息时文本不变"""
+        text = "docker ps -a 查看容器"
+        assert filter_sensitive_info(text) == text
+
+
+# ============================================================
+# filter_output_with_sensitive
+# ============================================================
+
+class TestFilterOutputWithSensitive:
+    """完整安全检查（危险命令 + 敏感信息）"""
+
+    def test_danger_and_sensitive_combined(self):
+        """同时含危险命令与敏感信息：safe=False，敏感信息已脱敏"""
+        text = "服务器 192.168.1.100 上执行 rm -rf /"
+        result = filter_output_with_sensitive(text, "删除")
+        assert result["safe"] is False
+        assert len(result["removed_lines"]) >= 1
+        # 危险命令原行已替换为安全提示
+        assert "rm -rf" not in result["filtered_text"]
+        # 敏感 IP 已脱敏
+        assert "192.168.1.100" not in result["filtered_text"]
+
+    def test_safe_text_with_sensitive_only(self):
+        """无危险命令但有敏感信息：safe=True，敏感信息已脱敏"""
+        text = "查看 192.168.1.100 的日志"
+        result = filter_output_with_sensitive(text, "查看日志")
+        assert result["safe"] is True
+        assert "192.168.1.100" not in result["filtered_text"]
+        assert "192.168" in result["filtered_text"]
+
+    def test_superset_of_filter_output(self):
+        """返回结构是 filter_output 的超集（含全部原字段）"""
+        result = filter_output_with_sensitive("docker ps", "docker")
+        for key in ("safe", "filtered_text", "warnings", "removed_lines"):
+            assert key in result

skill/code/tests/test_search_engine.py

+# -*- coding: utf-8 -*-
+"""
+test_search_engine.py — search_engine 模块单元测试
+
+覆盖：
+  模块函数 tokenize / compute_tf / compute_idf / compute_tfidf / cosine_similarity
+  类方法 SearchEngine.__init__ / search / get_projects / get_categories
+
+被测模块：skill/code/web/search_engine.py
+索引依赖：conftest.py autouse fixture 把 SEARCH_INDEX_PATHS 指向 deploy/搜索索引.json（357 条）
+"""
+
+import math
+
+import search_engine
+from search_engine import (
+    tokenize,
+    compute_tf,
+    compute_idf,
+    compute_tfidf,
+    cosine_similarity,
+    SearchEngine,
+)
+
+
+# ============================================================
+# tokenize
+# ============================================================
+
+class TestTokenize:
+    def test_chinese_words(self):
+        """中文 2-4 字词组提取"""
+        tokens = tokenize("门口屏连接失败")
+        assert isinstance(tokens, list)
+        assert len(tokens) > 0
+
+    def test_english_words(self):
+        """英文单词提取（小写化）"""
+        tokens = tokenize("MQTT Connection Failed")
+        assert "mqtt" in tokens
+        assert "connection" in tokens
+        assert "failed" in tokens
+
+    def test_numbers_extracted(self):
+        """数字提取"""
+        tokens = tokenize("端口 8088 超时")
+        assert "8088" in tokens
+
+    def test_stopwords_filtered(self):
+        """停用词被过滤"""
+        tokens = tokenize("the docker is running")
+        assert "the" not in tokens
+        assert "is" not in tokens
+        assert "docker" in tokens
+        assert "running" in tokens
+
+    def test_empty_string(self):
+        """空字符串返回空列表"""
+        assert tokenize("") == []
+
+
+# ============================================================
+# compute_tf
+# ============================================================
+
+class TestComputeTf:
+    def test_normal_tokens(self):
+        """正常 token 词频和为 1"""
+        tf = compute_tf(["a", "a", "b"])
+        assert tf["a"] == 2 / 3
+        assert tf["b"] == 1 / 3
+
+    def test_empty_tokens(self):
+        """空 token 列表不报错（total 用 1 兜底）"""
+        tf = compute_tf([])
+        assert tf == {}
+
+    def test_single_token(self):
+        """单 token 词频为 1.0"""
+        assert compute_tf(["solo"]) == {"solo": 1.0}
+
+
+# ============================================================
+# compute_idf
+# ============================================================
+
+class TestComputeIdf:
+    def test_idf_dict_keys(self):
+        """IDF 包含所有出现过的 token"""
+        docs = [["a", "b"], ["b", "c"]]
+        idf = compute_idf(docs)
+        assert set(idf.keys()) == {"a", "b", "c"}
+
+    def test_idf_smoothing_positive(self):
+        """平滑后 IDF 为正（log((N+1)/(df+1))+1）"""
+        docs = [["a"], ["a"], ["a"]]
+        idf = compute_idf(docs)
+        # df(a)=3, N=3 → log(4/4)+1 = 1
+        assert idf["a"] == 1.0
+
+    def test_rarer_token_higher_idf(self):
+        """越稀有 IDF 越大"""
+        docs = [["common", "rare"], ["common"], ["common"]]
+        idf = compute_idf(docs)
+        assert idf["rare"] > idf["common"]
+
+    def test_empty_docs(self):
+        """空文档列表返回空 dict"""
+        assert compute_idf([]) == {}
+
+
+# ============================================================
+# compute_tfidf
+# ============================================================
+
+class TestComputeTfidf:
+    def test_basic_product(self):
+        """TF * IDF"""
+        tf = {"a": 0.5}
+        idf = {"a": 2.0}
+        vec = compute_tfidf(tf, idf)
+        assert vec["a"] == 1.0
+
+    def test_unknown_token_defaults_one(self):
+        """IDF 中不存在的 token 用 1.0 兜底"""
+        tf = {"unknown": 0.5}
+        vec = compute_tfidf(tf, {})
+        assert vec["unknown"] == 0.5
+
+    def test_empty_tf(self):
+        """空 TF 返回空向量"""
+        assert compute_tfidf({}, {"a": 1.0}) == {}
+
+
+# ============================================================
+# cosine_similarity
+# ============================================================
+
+class TestCosineSimilarity:
+    def test_identical_vectors(self):
+        """相同向量相似度约为 1.0（浮点误差容忍）"""
+        v = {"a": 1.0, "b": 2.0}
+        assert math.isclose(cosine_similarity(v, v), 1.0, rel_tol=1e-9)
+
+    def test_orthogonal_vectors(self):
+        """无共同词的向量相似度为 0.0"""
+        v1 = {"a": 1.0}
+        v2 = {"b": 1.0}
+        assert cosine_similarity(v1, v2) == 0.0
+
+    def test_zero_norm_returns_zero(self):
+        """零模长向量返回 0.0"""
+        assert cosine_similarity({"a": 0.0}, {"a": 1.0}) == 0.0
+        assert cosine_similarity({}, {"a": 1.0}) == 0.0
+
+    def test_partial_overlap_in_range(self):
+        """部分重叠相似度在 (0, 1)"""
+        v1 = {"a": 1.0, "b": 1.0}
+        v2 = {"a": 1.0, "c": 1.0}
+        sim = cosine_similarity(v1, v2)
+        assert 0.0 < sim < 1.0
+
+
+# ============================================================
+# SearchEngine 实例化（依赖真实索引，conftest 已注入路径）
+# ============================================================
+
+class TestSearchEngineInit:
+    def test_init_loads_records(self):
+        """初始化加载 357 条记录"""
+        engine = SearchEngine()
+        assert len(engine.records) == 357
+
+    def test_init_builds_vectors(self):
+        """初始化预计算所有记录的 TF-IDF 向量"""
+        engine = SearchEngine()
+        assert len(engine.doc_vectors) == 357
+        assert len(engine.doc_tokens) == 357
+        assert len(engine.idf) > 0
+
+    def test_record_fields(self):
+        """记录字段完整"""
+        engine = SearchEngine()
+        r = engine.records[0]
+        for field in ("title", "project", "category", "keywords", "full_text"):
+            assert field in r
+
+
+# ============================================================
+# SearchEngine.search
+# ============================================================
+
+class TestSearchEngineSearch:
+    def test_real_query_returns_results(self):
+        """真实查询返回排序结果"""
+        engine = SearchEngine()
+        results = engine.search("mqtt 连接失败", top_k=5)
+        assert isinstance(results, list)
+        assert 1 <= len(results) <= 5
+        for r in results:
+            assert set(r.keys()) == {"rank", "score", "record"}
+            assert 0 <= r["score"] <= 1.0
+
+    def test_results_sorted_desc(self):
+        """结果按得分降序"""
+        engine = SearchEngine()
+        results = engine.search("docker 容器", top_k=5)
+        scores = [r["score"] for r in results]
+        assert scores == sorted(scores, reverse=True)
+
+    def test_rank_starts_at_one(self):
+        """rank 从 1 开始递增"""
+        engine = SearchEngine()
+        results = engine.search("redis", top_k=3)
+        if results:
+            ranks = [r["rank"] for r in results]
+            assert ranks == list(range(1, len(results) + 1))
+
+    def test_top_k_limit(self):
+        """top_k 限制返回数量"""
+        engine = SearchEngine()
+        for k in (1, 2, 10):
+            results = engine.search("mqtt", top_k=k)
+            assert len(results) <= k
+
+    def test_empty_query_returns_empty_or_limited(self):
+        """空 query：tokenize 返回空 → 查询向量为空 → 相似度 0 → 结果可能为空"""
+        engine = SearchEngine()
+        results = engine.search("", top_k=5)
+        # 空 query 无 token，与所有记录无共同词，无 keyword 命中 → scores 为空
+        assert results == []
+
+    def test_no_match_query(self):
+        """完全无关的 query（无共同词、无 keyword 命中）返回空"""
+        engine = SearchEngine()
+        # 用一串不可能出现在知识库的随机哈希字符串
+        results = engine.search("zzqxwkjrandomstringzzz", top_k=5)
+        # final_score 全为 0，不入 scores
+        assert results == []
+
+    def test_project_filter(self):
+        """项目过滤：用首条结果的真实 project 值过滤，应只返回该项目记录"""
+        engine = SearchEngine()
+        results = engine.search("连接失败", top_k=10)
+        if not results:
+            return
+        target_project = results[0]["record"].get("project")
+        filtered = engine.search("连接失败", top_k=20, project_filter=target_project)
+        # 过滤后结果都应是该项目（可能为空，但若有则必匹配）
+        for r in filtered:
+            assert r["record"].get("project", "").lower() == target_project.lower()
+
+    def test_project_filter_no_match(self):
+        """项目过滤不匹配返回空"""
+        engine = SearchEngine()
+        results = engine.search("mqtt", top_k=5, project_filter="不存在的项目XYZ")
+        assert results == []
+
+    def test_category_filter(self):
+        """分类过滤：用首条结果的 category 值过滤"""
+        engine = SearchEngine()
+        results = engine.search("连接失败", top_k=10)
+        if not results:
+            return
+        cats = results[0]["record"].get("category", [])
+        if not cats:
+            return
+        target_cat = cats[0]
+        filtered = engine.search("连接失败", top_k=20, category_filter=target_cat)
+        for r in filtered:
+            assert target_cat in r["record"].get("category", [])
+
+    def test_score_capped_at_one(self):
+        """final_score 被 min(1.0, ...) 封顶"""
+        engine = SearchEngine()
+        results = engine.search("mqtt 连接", top_k=5)
+        for r in results:
+            assert r["score"] <= 1.0
+
+
+# ============================================================
+# SearchEngine.get_projects / get_categories
+# ============================================================
+
+class TestSearchEngineMetadata:
+    def test_get_projects(self):
+        """返回 120 个项目"""
+        engine = SearchEngine()
+        projects = engine.get_projects()
+        assert isinstance(projects, list)
+        assert len(projects) == 120
+
+    def test_get_categories(self):
+        """返回 20 个分类"""
+        engine = SearchEngine()
+        cats = engine.get_categories()
+        assert isinstance(cats, list)
+        assert len(cats) == 20