DerrorsandClaude Opus 5.5
3700aff343
媒体库整理 - 「二级分类路由」并入「媒体库整理」：转存前选目录、转存后整理，共用开关与设置 - 目录方式：二级分类 / 按媒体类型 / 不分类；按媒体类型的文件夹名可改 - 二级分类规则随插件发布（category.yaml，MoviePilot 式顺序匹配，增加剧集兜底），管理台只读 - 两种目录方式都以流程图展示，按媒体类型的文件夹名在图上修改 插件结构 - 每个插件一个目录；内置插件的默认设置与校验在插件的 settings.py，内核按目录调用 - 插件动作 ui_actions + 通用接口 /api/extensions/<名>/actions/<动作>，插件不依赖 Web 框架 - 前端插件专用界面移到 web/src/plugins/<名>/，按 config_type 登记 - storage_provider 统一校验，内核不再写死媒体库整理 管理台 - 插件页「运行概览」、统计页内部标签、运行状态徽标、标题与副标题「网盘转存助手」 - 设置页两列卡片不留空、下拉框直接选中默认项、统计页 echarts 按需引入（约 1 MB → 490 KB） 清理 - 移除启动迁移与 .env 导入、旧配置写法与 routing、旧凭证文件、旧去重键与旧中断状态 - 数据库结构迁移重置为单一基线；不认识的顶层配置项直接报错 - 删除未使用的代码与组件；修复测试警告；CI 固定 ubuntu-24.04 Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
65 lines
2.7 KiB
Python
65 lines
2.7 KiB
Python
"""examples/plugins 中的示例插件必须能通过外部扩展目录加载并正常工作。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import threading
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from reelay.core import runtime
|
|
from reelay.core.contracts import ParsedResource, PipelineContext, RawMessage, TaskState
|
|
from reelay.core.registry import discover
|
|
|
|
EXAMPLES = Path(__file__).resolve().parents[1] / "examples" / "plugins"
|
|
|
|
|
|
class TestExamplePlugins(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
self.registry = discover(plugin_dirs=[EXAMPLES])
|
|
|
|
def test_examples_load_without_errors(self):
|
|
self.assertEqual(self.registry.load_errors, [])
|
|
self.assertIn("webhook", self.registry.providers["notify"])
|
|
self.assertIn("quality_gate", [h.name for h in self.registry.hooks_for("after_parse", {})])
|
|
|
|
async def test_quality_gate_blocks_cam_releases_only(self):
|
|
gate = next(h for h in self.registry.hooks["after_parse"] if h.name == "quality_gate")
|
|
with runtime.scope({}):
|
|
self.assertFalse(await gate.run(PipelineContext(RawMessage("@a", "1", "新片 HDCAM 中字"))))
|
|
self.assertTrue(await gate.run(PipelineContext(RawMessage("@a", "2", "STS 4K WEB-DL"))))
|
|
|
|
async def test_webhook_posts_task_event_and_notice(self):
|
|
received = []
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
received.append(json.loads(self.rfile.read(int(self.headers["Content-Length"]))))
|
|
self.send_response(204)
|
|
self.end_headers()
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
server = HTTPServer(("127.0.0.1", 0), Handler)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
self.addCleanup(server.server_close) # 清理按后进先出：先停服务，再关监听 socket
|
|
self.addCleanup(server.shutdown)
|
|
webhook = self.registry.get_provider("notify", "webhook")
|
|
ctx = PipelineContext(RawMessage("@a", "1", "x"))
|
|
ctx.parsed = ParsedResource(is_media=True, title="剧", episodes=[3], share_code="secret")
|
|
ctx.state = TaskState.DONE
|
|
with runtime.scope({"plugins": {"webhook": {"url": f"http://127.0.0.1:{server.server_port}/hook"}}}):
|
|
await webhook.notify(ctx, "saved")
|
|
await webhook.notice("日报")
|
|
self.assertEqual([r["event"] for r in received], ["saved", "notice"])
|
|
self.assertEqual(received[0]["title"], "剧")
|
|
self.assertNotIn("secret", json.dumps(received))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|