mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-28 19:06:36 +08:00
448 lines
16 KiB
Python
448 lines
16 KiB
Python
"""
|
||
feedback-issue skill 内部脚本的单元测试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import importlib
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
from urllib.parse import quote
|
||
|
||
import pytest
|
||
|
||
from app.agent.tools.factory import MoviePilotToolFactory
|
||
from app.agent.tools.impl.ask_user_choice import AskUserChoiceInput
|
||
from app.runtime.config import settings
|
||
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parents[1] / "skills" / "feedback-issue" / "scripts"
|
||
# 只在导入脚本时添加搜索路径，避免改变后续测试的模块解析顺序。
|
||
with patch.object(sys, "path", [str(SCRIPT_DIR), *sys.path]):
|
||
collect_script = importlib.import_module("collect_feedback_diagnostics")
|
||
common = importlib.import_module("feedback_issue_common")
|
||
prepare_script = importlib.import_module("prepare_feedback_issue")
|
||
submit_script = importlib.import_module("submit_feedback_issue")
|
||
|
||
|
||
class _FakeResponse:
|
||
"""
|
||
``requests.Response`` 的最小替身，覆盖提交脚本使用的属性和方法。
|
||
"""
|
||
|
||
def __init__(self, status_code, payload=None, headers=None, text=""):
|
||
"""
|
||
保存响应状态、JSON 数据、响应头和文本。
|
||
"""
|
||
self.status_code = status_code
|
||
self._payload = payload
|
||
self.headers = headers or {}
|
||
self.text = text
|
||
|
||
def json(self):
|
||
"""
|
||
返回预设 JSON；没有 JSON 时模拟解析失败。
|
||
"""
|
||
if self._payload is None:
|
||
raise ValueError("no json body")
|
||
return self._payload
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _isolate_feedback_config(tmp_path, monkeypatch):
|
||
"""
|
||
为反馈脚本提供独立配置目录和空令牌，用例结束后恢复原始设置。
|
||
|
||
:param tmp_path: pytest 提供的独立临时目录
|
||
:param monkeypatch: 负责恢复全局配置属性的补丁夹具
|
||
"""
|
||
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
|
||
monkeypatch.setattr(settings, "GITHUB_TOKEN", None)
|
||
settings.LOG_PATH.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def _write_log(text: str) -> Path:
|
||
"""
|
||
写入临时 moviepilot.log 并返回路径。
|
||
"""
|
||
log_path = settings.LOG_PATH / "moviepilot.log"
|
||
log_path.parent.mkdir(parents=True, exist_ok=True)
|
||
log_path.write_text(text, encoding="utf-8")
|
||
return log_path
|
||
|
||
|
||
def _valid_draft(diagnostics_file: str) -> dict:
|
||
"""
|
||
构造一份可通过质量校验的 Issue 草稿。
|
||
"""
|
||
return {
|
||
"title": "[错误报告]: 订阅刷新接口返回 500 错误码",
|
||
"version": "v2.12.2",
|
||
"environment": "Docker",
|
||
"issue_type": "主程序运行问题",
|
||
"original_user_request": "订阅刷新接口返回 500，帮我提交上游 Issue",
|
||
"diagnostics_file": diagnostics_file,
|
||
"description": (
|
||
"## 现象\n"
|
||
"- 订阅刷新接口持续返回 500，调用 /api/v1/subscribe/refresh 后失败。\n\n"
|
||
"## 复现步骤\n"
|
||
"1. 在 WebUI 触发刷新订阅。\n"
|
||
"2. 后端日志出现 RecognizeError。\n"
|
||
"3. 前端弹出 500。\n\n"
|
||
"## 期望行为\n"
|
||
"- 正常完成订阅刷新流程，无 500 错误。\n\n"
|
||
"## 已定位 / 推测\n"
|
||
"- 仅为推测：订阅刷新链路的识别异常未被正确处理。\n\n"
|
||
"## 已尝试的处理\n"
|
||
"- 重启后仍可复现。"
|
||
),
|
||
}
|
||
|
||
|
||
def _create_diagnostics_file(logs: str = "ERROR demo") -> Path:
|
||
"""
|
||
创建脚本运行时诊断文件并返回路径。
|
||
"""
|
||
diagnostics_file = common.runtime_file("diagnostics", ".json")
|
||
common.write_json_file(
|
||
diagnostics_file,
|
||
{
|
||
"original_user_request": "订阅刷新接口返回 500，帮我提交上游 Issue",
|
||
"found": bool(logs),
|
||
"logs": logs,
|
||
"doctor": {
|
||
"success": True,
|
||
"report": {
|
||
"status": "degraded",
|
||
"summary": {"total": 2, "error": 1, "warn": 1, "fixed": 0},
|
||
"environment": {"runtime": "Docker"},
|
||
"findings": [
|
||
{
|
||
"severity": "error",
|
||
"title": "后端端口被占用",
|
||
"recommendation": "修改 PORT 或停止占用进程",
|
||
}
|
||
],
|
||
},
|
||
},
|
||
"source_files": [str(settings.LOG_PATH / "moviepilot.log")],
|
||
"log_selection": {
|
||
"strategy": "time_window_and_keyword_block_match",
|
||
"time_window_minutes": 30,
|
||
"window_start": datetime.now().isoformat(timespec="seconds"),
|
||
"keywords": ["RecognizeError"],
|
||
"max_lines_per_file": 80,
|
||
"matched_files": [
|
||
{
|
||
"path": str(settings.LOG_PATH / "moviepilot.log"),
|
||
"matched_keywords": ["RecognizeError"],
|
||
"line_count": 1,
|
||
}
|
||
],
|
||
"warning": "",
|
||
},
|
||
},
|
||
)
|
||
return diagnostics_file
|
||
|
||
|
||
def test_redact_logs_strips_common_secrets() -> None:
|
||
"""
|
||
日志脱敏应覆盖 token、Cookie、PII 和本机用户路径。
|
||
"""
|
||
sample = (
|
||
"Cookie: session=foo; passkey=secret123\n"
|
||
"Authorization: Bearer ghp_abcdefghijklmnopqrstuvwx\n"
|
||
"api_key=mysecret\n"
|
||
"password: hunter2\n"
|
||
"user@example.com\n"
|
||
"/Users/alice/Library"
|
||
)
|
||
out = common.redact_logs(sample)
|
||
for secret in (
|
||
"secret123",
|
||
"ghp_abcdefghijklmnopqrstuvwx",
|
||
"mysecret",
|
||
"hunter2",
|
||
"user@example.com",
|
||
"/Users/alice/",
|
||
):
|
||
assert secret not in out
|
||
assert "<REDACTED>" in out
|
||
|
||
|
||
def test_build_prefill_url_encodes_and_redacts() -> None:
|
||
"""
|
||
预填 URL 应正确编码中文并脱敏日志。
|
||
"""
|
||
url = common.build_prefill_url(
|
||
title="[错误报告]: 版本测试",
|
||
version="v2.12.2",
|
||
environment="Docker",
|
||
issue_type="主程序运行问题",
|
||
description="line1\nline2",
|
||
logs="Cookie: leak_me",
|
||
)
|
||
assert "%E7%89%88" in url
|
||
assert "%0A" in url
|
||
assert "template=bug_report.yml" in url
|
||
assert quote("leak_me", safe="") not in url
|
||
|
||
|
||
def test_check_content_quality_rejects_test_intent() -> None:
|
||
"""
|
||
原始请求暴露测试链路意图时必须拒绝。
|
||
"""
|
||
error = common.check_content_quality(
|
||
title="[错误报告]: TMDB识别错误，将动画识别为其他作品",
|
||
original_user_request="我是开发者，为我反馈一个测试 ISSUE，看能否跑通",
|
||
description=(
|
||
"## 现象\nTMDB识别错误。\n\n## 复现步骤\n1. 搜索动画。\n2. 识别结果错误。\n\n## 期望行为\n正确识别。"
|
||
),
|
||
logs="ERROR demo",
|
||
)
|
||
assert error is not None
|
||
assert "测试 issue" in error.lower()
|
||
|
||
|
||
def test_factory_no_longer_registers_feedback_issue_tools() -> None:
|
||
"""
|
||
Agent 工厂不应再注册 feedback-issue 专用工具。
|
||
"""
|
||
with patch(
|
||
"app.agent.tools.factory._get_plugin_agent_tools",
|
||
return_value=[],
|
||
):
|
||
tools = MoviePilotToolFactory.create_tools(
|
||
session_id="feedback-issue-session",
|
||
user_id="10001",
|
||
)
|
||
tool_names = {tool.name for tool in tools}
|
||
assert "collect_feedback_diagnostics" not in tool_names
|
||
assert "prepare_feedback_issue" not in tool_names
|
||
assert "submit_feedback_issue" not in tool_names
|
||
|
||
|
||
def test_normalize_keywords_drops_vague_terms() -> None:
|
||
"""
|
||
关键词过滤应丢弃错误、异常等泛词。
|
||
"""
|
||
out = collect_script.normalize_keywords(["TMDB", "错误", "异常", "scrape_metadata", "x"])
|
||
assert out == ["TMDB", "scrape_metadata"]
|
||
|
||
|
||
def test_has_explicit_feedback_intent() -> None:
|
||
"""
|
||
入口意图门只放行明确提 Issue 的请求。
|
||
"""
|
||
assert collect_script.has_explicit_feedback_intent("TMDB 出错了，帮我提 issue")
|
||
assert collect_script.has_explicit_feedback_intent("希望增加一个能力，帮我提需求")
|
||
assert not collect_script.has_explicit_feedback_intent("TMDB 一直在报错")
|
||
|
||
|
||
def test_filter_lines_drops_history_and_meta_noise() -> None:
|
||
"""
|
||
筛选日志时应丢掉历史行和 Agent 自身噪音。
|
||
"""
|
||
now = datetime.now()
|
||
old = now - timedelta(hours=3)
|
||
recent = now - timedelta(minutes=5)
|
||
text = "\n".join(
|
||
[
|
||
f"【INFO】{old.strftime('%Y-%m-%d %H:%M:%S')},123 - tmdb - TMDB failed 历史",
|
||
f"【DEBUG】{recent.strftime('%Y-%m-%d %H:%M:%S')},100 - base.py - Executing tool",
|
||
f"【ERROR】{recent.strftime('%Y-%m-%d %H:%M:%S')},123 - tmdb - TMDB failed 当前",
|
||
" Traceback (most recent call last):",
|
||
]
|
||
)
|
||
out, matched_keywords = collect_script.filter_lines(
|
||
text,
|
||
keywords=["TMDB"],
|
||
max_lines=80,
|
||
window_start=now - timedelta(minutes=30),
|
||
)
|
||
joined = "\n".join(out)
|
||
assert "当前" in joined
|
||
assert "Traceback" in joined
|
||
assert "历史" not in joined
|
||
assert "Executing tool" not in joined
|
||
assert matched_keywords == ["TMDB"]
|
||
|
||
|
||
def test_filter_lines_requires_specific_keyword_match() -> None:
|
||
"""
|
||
没有具体关键词时不应回退采集近期无关日志。
|
||
"""
|
||
now = datetime.now()
|
||
recent = now - timedelta(minutes=5)
|
||
text = f"【ERROR】{recent.strftime('%Y-%m-%d %H:%M:%S')},123 - tmdb - unrelated error"
|
||
|
||
out, matched_keywords = collect_script.filter_lines(
|
||
text,
|
||
keywords=[],
|
||
max_lines=80,
|
||
window_start=now - timedelta(minutes=30),
|
||
)
|
||
|
||
assert out == []
|
||
assert matched_keywords == []
|
||
|
||
|
||
def test_collect_writes_diagnostics_file_without_returning_logs() -> None:
|
||
"""
|
||
collect 脚本结果应返回文件句柄和统计，不直接返回日志正文。
|
||
"""
|
||
recent = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
_write_log(f"【ERROR】{recent},000 - tmdb - TMDB lookup failed Cookie: secret")
|
||
result = collect_script.collect_diagnostics(
|
||
original_user_request="TMDB 报错，帮我反馈 issue",
|
||
keywords=["TMDB"],
|
||
max_lines=80,
|
||
time_window_minutes=30,
|
||
)
|
||
assert result["success"]
|
||
assert "diagnostics_file" in result
|
||
assert "logs" not in result
|
||
diagnostics = common.read_json_file(result["diagnostics_file"])
|
||
assert "TMDB lookup failed" in diagnostics["logs"]
|
||
assert "Cookie: <REDACTED>" in diagnostics["logs"]
|
||
assert "secret" not in diagnostics["logs"]
|
||
assert "doctor" in diagnostics
|
||
assert "log_selection" in diagnostics
|
||
assert diagnostics["log_selection"]["matched_files"][0]["matched_keywords"] == ["TMDB"]
|
||
|
||
|
||
def test_collect_without_keywords_records_selection_but_no_logs() -> None:
|
||
"""
|
||
无有效关键词时只记录筛选依据，不采集近期无关日志正文。
|
||
"""
|
||
recent = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
_write_log(f"【ERROR】{recent},000 - tmdb - TMDB lookup failed")
|
||
|
||
result = collect_script.collect_diagnostics(
|
||
original_user_request="TMDB 报错，帮我反馈 issue",
|
||
keywords=["错误"],
|
||
max_lines=80,
|
||
time_window_minutes=30,
|
||
)
|
||
|
||
assert result["success"]
|
||
assert not result["found"]
|
||
assert "未提供具体关键词" in result["log_selection_summary"]
|
||
diagnostics = common.read_json_file(result["diagnostics_file"])
|
||
assert diagnostics["logs"] == ""
|
||
assert diagnostics["log_selection"]["matched_files"] == []
|
||
|
||
|
||
def test_prepare_generates_payload_and_preview_files() -> None:
|
||
"""
|
||
预览保留脱敏证据，并提供可直接传给按钮工具的确认、修改、取消选项。
|
||
"""
|
||
diagnostics_file = _create_diagnostics_file("ERROR demo Cookie: secret")
|
||
draft_file = common.runtime_file("draft", ".json")
|
||
common.write_json_file(draft_file, _valid_draft(str(diagnostics_file)))
|
||
|
||
result = prepare_script.prepare_issue(draft_file)
|
||
|
||
assert result["success"]
|
||
assert Path(result["payload_file"]).exists()
|
||
preview = Path(result["preview_file"]).read_text(encoding="utf-8")
|
||
assert "请确认是否提交以下问题反馈" in preview
|
||
assert "Doctor 摘要" in preview
|
||
assert "日志筛选依据" in preview
|
||
assert "后端端口被占用" in preview
|
||
assert "Cookie: <REDACTED>" in preview
|
||
assert "secret" not in preview
|
||
assert "回复「确认」" not in preview
|
||
choice = AskUserChoiceInput(message=preview, options=result["confirmation_options"])
|
||
assert [option.label for option in choice.options] == ["确认提交", "修改内容", "取消"]
|
||
assert len({option.value for option in choice.options}) == 3
|
||
assert "ask_user_choice" in result["message"]
|
||
assert "否则使用文字确认" in result["message"]
|
||
|
||
|
||
def test_prepare_rejects_invalid_draft() -> None:
|
||
"""
|
||
prepare 脚本应拒绝缺少结构信息的草稿。
|
||
"""
|
||
diagnostics_file = _create_diagnostics_file()
|
||
draft = _valid_draft(str(diagnostics_file))
|
||
draft["description"] = (
|
||
"用户反馈下载任务完成后无法移动文件，系统看起来没有按照配置执行"
|
||
"媒体库转移，请协助排查下载器联动和转移模块之间是否存在后端异常。"
|
||
)
|
||
draft_file = common.runtime_file("draft", ".json")
|
||
common.write_json_file(draft_file, draft)
|
||
|
||
result = prepare_script.prepare_issue(draft_file)
|
||
|
||
assert not result["success"]
|
||
assert result["reason"] == "invalid_draft"
|
||
assert "结构信息" in result["message"]
|
||
|
||
|
||
def test_submit_returns_prefill_url_without_token() -> None:
|
||
"""
|
||
未配置 GITHUB_TOKEN 时 submit 脚本应返回预填 URL。
|
||
"""
|
||
diagnostics_file = _create_diagnostics_file("ERROR demo")
|
||
draft_file = common.runtime_file("draft", ".json")
|
||
common.write_json_file(draft_file, _valid_draft(str(diagnostics_file)))
|
||
prepared = prepare_script.prepare_issue(draft_file)
|
||
|
||
result = submit_script.submit_issue(prepared["payload_file"], username="admin")
|
||
|
||
assert not result["success"]
|
||
assert result["reason"] == "no_token"
|
||
assert "https://github.com/jxxghp/MoviePilot/issues/new" in result["prefill_url"]
|
||
|
||
|
||
def test_submit_success_with_github_token() -> None:
|
||
"""
|
||
配置 GITHUB_TOKEN 且 API 返回 201 时 submit 脚本应报告成功。
|
||
"""
|
||
settings.GITHUB_TOKEN = "ghp_test_token"
|
||
diagnostics_file = _create_diagnostics_file("ERROR demo")
|
||
draft_file = common.runtime_file("draft", ".json")
|
||
common.write_json_file(draft_file, _valid_draft(str(diagnostics_file)))
|
||
prepared = prepare_script.prepare_issue(draft_file)
|
||
|
||
with patch(
|
||
"submit_feedback_issue.RequestUtils.post",
|
||
return_value=_FakeResponse(
|
||
201,
|
||
payload={
|
||
"number": 9999,
|
||
"html_url": "https://github.com/jxxghp/MoviePilot/issues/9999",
|
||
},
|
||
),
|
||
):
|
||
result = submit_script.submit_issue(prepared["payload_file"], username="admin")
|
||
|
||
assert result["success"]
|
||
assert result["issue_number"] == 9999
|
||
assert "/9999" in result["issue_url"]
|
||
|
||
|
||
def test_submit_user_rate_limit() -> None:
|
||
"""
|
||
同一管理员连续提交应被脚本级冷却限制挡住。
|
||
"""
|
||
state = common.load_submission_state()
|
||
state["user_submissions"] = {"admin": [time.time()]}
|
||
common.save_submission_state(state)
|
||
diagnostics_file = _create_diagnostics_file("ERROR demo")
|
||
draft_file = common.runtime_file("draft", ".json")
|
||
draft = _valid_draft(str(diagnostics_file))
|
||
draft["title"] = "[错误报告]: 另一个完全不同的后端报错"
|
||
common.write_json_file(draft_file, draft)
|
||
prepared = prepare_script.prepare_issue(draft_file)
|
||
|
||
result = submit_script.submit_issue(prepared["payload_file"], username="admin")
|
||
|
||
assert result["reason"] == "rate_limited_user"
|
||
assert "30 分钟" in result["message"]
|