ahdoawhfo/netdisk-fast-download
1
0
Fork 0
mirror of https://github.com/qaiu/netdisk-fast-download.git synced 2026-01-12 17:34:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

docs: 更新文档导航和解析器指南

Browse Source 
- 添加演练场（Playground）文档导航区到主 README
- 新增 Python 解析器文档链接（开发指南、测试报告、LSP集成）
- 更新前端版本号至 0.1.9b19p
- 补充 Python 解析器 requests 库使用章节和官方文档链接
- 添加 JavaScript 和 Python 解析器的语言版本和官方文档
- 优化文档结构，分类为项目文档和外部资源
This commit is contained in:
q
2026-01-11 22:35:45 +08:00
parent b8eee2b8a7
commit 2fcf9cfab1
60 changed files with 10132 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
web-service/src/main/java/cn/qaiu/lz/AppMain.java
View File

@@ -63,6 +63,9 @@ public class AppMain {
System.out.println(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"));
System.out.println("数据库连接成功");
// 初始化示例解析器
initExampleParsers();
// 加载演练场解析器
loadPlaygroundParsers();
@@ -105,6 +108,17 @@ public class AppMain {
PlaygroundConfig.loadFromJson(jsonObject);
}
/**
* 初始化示例解析器（JS和Python）
*/
private static void initExampleParsers() {
DbService dbService = AsyncServiceUtil.getAsyncServiceInstance(DbService.class);
dbService.initExampleParsers()
.onSuccess(v -> log.info("示例解析器初始化检查完成"))
.onFailure(e -> log.error("示例解析器初始化失败", e));
}
/**
* 在启动时加载所有已发布的演练场解析器
*/

364
web-service/src/main/java/cn/qaiu/lz/web/controller/PylspWebSocketHandler.java Normal file
View File

@@ -0,0 +1,364 @@
package cn.qaiu.lz.web.controller;
import cn.qaiu.lz.web.config.PlaygroundConfig;
import cn.qaiu.vx.core.annotaions.RouteHandler;
import cn.qaiu.vx.core.annotaions.SockRouteMapper;
import io.vertx.core.buffer.Buffer;
import io.vertx.ext.web.handler.sockjs.SockJSSocket;
import lombok.extern.slf4j.Slf4j;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Python LSP (pylsp/jedi) WebSocket 桥接处理器
*
* 通过 WebSocket 将前端 LSP 请求转发到 pylsp 子进程，
* 实现实时代码检查、自动完成、悬停提示等功能。
*
* 使用 jedi 的 python-lsp-server (pylsp)，需要预先安装:
* pip install python-lsp-server[all]
*
* @author <a href="https://qaiu.top">QAIU</a>
*/
@RouteHandler(value = "/v2/ws")
@Slf4j
public class PylspWebSocketHandler {
// 存储每个 WebSocket 连接对应的 pylsp 进程
private static final ConcurrentHashMap<String, PylspSession> sessions = new ConcurrentHashMap<>();
/**
* WebSocket LSP 端点
* 前端通过此端点连接，发送 LSP JSON-RPC 消息
*/
@SockRouteMapper("/pylsp")
public void handlePylsp(SockJSSocket socket) {
String sessionId = socket.writeHandlerID();
log.info("========================================");
log.info("[PYLSP] WebSocket Handler 被调用!");
log.info("[PYLSP] Session ID: {}", sessionId);
log.info("[PYLSP] Remote Address: {}", socket.remoteAddress());
log.info("========================================");
// 检查 Playground 是否启用
PlaygroundConfig config = PlaygroundConfig.getInstance();
log.info("[PYLSP] Playground enabled: {}", config.isEnabled());
log.info("[PYLSP] Playground public: {}", config.isPublic());
if (!config.isEnabled()) {
log.error("[PYLSP] Playground功能已禁用! 请检查配置文件中 playground.enabled 设置");
log.error("[PYLSP] 当前配置: enabled={}, public={}", config.isEnabled(), config.isPublic());
socket.write(Buffer.buffer("{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32603,\"message\":\"Playground功能已禁用，请联系管理员\"},\"id\":null}"));
socket.close();
return;
}
// 创建 pylsp 会话
PylspSession session = new PylspSession(socket, sessionId);
sessions.put(sessionId, session);
// 启动 pylsp 进程
if (!session.start()) {
socket.write(Buffer.buffer("{\"jsonrpc\":\"2.0\",\"error\":{\"code\":-32603,\"message\":\"无法启动pylsp服务\"},\"id\":null}"));
socket.close();
sessions.remove(sessionId);
return;
}
// 处理来自前端的消息
socket.handler(buffer -> {
String message = buffer.toString(StandardCharsets.UTF_8);
log.debug("收到 LSP 请求: {}", message);
session.sendToLsp(message);
});
// 处理连接关闭
socket.endHandler(v -> {
log.info("pylsp WebSocket 连接关闭: {}", sessionId);
session.stop();
sessions.remove(sessionId);
});
// 处理异常
socket.exceptionHandler(e -> {
log.error("pylsp WebSocket 异常: {}", sessionId, e);
session.stop();
sessions.remove(sessionId);
});
}
/**
* 查找 graalpy-packages 目录路径
* 支持多种运行环境：开发环境、IDE 运行、jar 包运行
*/
private static String findGraalPyPackagesPath(String userDir) {
// 按优先级尝试多个可能的路径
String[] possiblePaths = {
// 开发环境 - IDE 直接运行
userDir + "/parser/src/main/resources/graalpy-packages",
// Maven 编译后路径
userDir + "/parser/target/classes/graalpy-packages",
// jar 包同级目录
userDir + "/graalpy-packages",
// jar 包运行时的 resources 目录
userDir + "/resources/graalpy-packages",
// 相对于 web-service 模块
userDir + "/../parser/src/main/resources/graalpy-packages",
// 从 web-service/target/package 向上查找
userDir + "/../../parser/src/main/resources/graalpy-packages",
userDir + "/../../../parser/src/main/resources/graalpy-packages",
};
for (String path : possiblePaths) {
File dir = new File(path);
if (dir.exists() && dir.isDirectory()) {
File pylspModule = new File(dir, "pylsp");
if (pylspModule.exists()) {
try {
String canonicalPath = dir.getCanonicalPath();
log.info("[PYLSP] 找到 graalpy-packages: {}", canonicalPath);
return canonicalPath;
} catch (IOException e) {
log.warn("[PYLSP] 获取规范路径失败: {}", path);
return dir.getAbsolutePath();
}
}
}
}
// 打印尝试的所有路径用于调试
log.error("[PYLSP] 尝试的路径:");
for (String path : possiblePaths) {
File dir = new File(path);
log.error("[PYLSP] {} (exists={})", path, dir.exists());
}
return null;
}
/**
* pylsp 会话管理类
* 管理单个 pylsp 子进程和对应的 WebSocket 连接
*/
private static class PylspSession {
private final SockJSSocket socket;
private final String sessionId;
private Process process;
private BufferedWriter processWriter;
private Thread readerThread;
private final AtomicBoolean running = new AtomicBoolean(false);
public PylspSession(SockJSSocket socket, String sessionId) {
this.socket = socket;
this.sessionId = sessionId;
}
/**
* 启动 pylsp 子进程
*
* 使用 GraalPy 和打包在 jar 中的 python-lsp-server。
* graalpy-packages 中包含完整的 pylsp 依赖。
*/
public boolean start() {
try {
// 检测运行环境（开发环境 vs jar 包）
String userDir = System.getProperty("user.dir");
String graalPyPackagesPath = findGraalPyPackagesPath(userDir);
if (graalPyPackagesPath == null) {
log.error("[PYLSP] 找不到 graalpy-packages 目录!");
log.error("[PYLSP] 已尝试的路径: {}", userDir);
log.error("[PYLSP] 请运行: parser/setup-graalpy-packages.sh");
return false;
}
// 检查 pylsp 是否存在
File pylspModule = new File(graalPyPackagesPath + "/pylsp");
if (!pylspModule.exists()) {
log.error("[PYLSP] pylsp 模块不存在: {}", pylspModule.getAbsolutePath());
log.error("[PYLSP] 请运行: parser/setup-graalpy-packages.sh");
return false;
}
// 使用系统 Python (因为 GraalPy 不支持作为独立进程运行 pylsp)
// 但通过 PYTHONPATH 使用打包的 pylsp
ProcessBuilder pb = new ProcessBuilder(
"python3", "-m", "pylsp",
"-v" // 详细日志
);
// 设置环境变量
var env = pb.environment();
env.put("PYTHONPATH", graalPyPackagesPath);
log.info("[PYLSP] PYTHONPATH: {}", graalPyPackagesPath);
pb.redirectErrorStream(false);
process = pb.start();
processWriter = new BufferedWriter(
new OutputStreamWriter(process.getOutputStream(), StandardCharsets.UTF_8)
);
running.set(true);
// 启动读取线程，将 pylsp 的输出转发到 WebSocket
readerThread = new Thread(() -> readLspOutput(), "pylsp-reader-" + sessionId);
readerThread.setDaemon(true);
readerThread.start();
// 启动错误读取线程
Thread errorThread = new Thread(() -> readLspError(), "pylsp-error-" + sessionId);
errorThread.setDaemon(true);
errorThread.start();
log.info("[PYLSP] pylsp 进程已启动 (Session: {})", sessionId);
log.info("[PYLSP] 进程 PID: {}", process.pid());
return true;
} catch (Exception e) {
log.error("[PYLSP] 启动 pylsp 进程失败", e);
log.error("[PYLSP] 错误详情: {}", e.getMessage());
log.error("[PYLSP] 请确保:");
log.error("[PYLSP] 1. 已运行 parser/setup-graalpy-packages.sh");
log.error("[PYLSP] 2. 系统已安装 python3");
log.error("[PYLSP] 3. graalpy-packages 中包含 pylsp 模块");
return false;
}
}
/**
* 发送消息到 pylsp 进程
*/
public void sendToLsp(String message) {
if (!running.get() || processWriter == null) {
return;
}
try {
// LSP 协议: Content-Length: xxx\r\n\r\n{json}
byte[] contentBytes = message.getBytes(StandardCharsets.UTF_8);
String header = "Content-Length: " + contentBytes.length + "\r\n\r\n";
processWriter.write(header);
processWriter.write(message);
processWriter.flush();
log.debug("发送到 pylsp: {}", message);
} catch (IOException e) {
log.error("发送消息到 pylsp 失败", e);
}
}
/**
* 读取 pylsp 输出并转发到 WebSocket
*/
private void readLspOutput() {
try {
InputStream inputStream = process.getInputStream();
BufferedReader reader = new BufferedReader(
new InputStreamReader(inputStream, StandardCharsets.UTF_8)
);
while (running.get()) {
// 读取 LSP 头部
String line = reader.readLine();
if (line == null) {
break;
}
// 解析 Content-Length
int contentLength = -1;
while (line != null && !line.isEmpty()) {
if (line.startsWith("Content-Length:")) {
contentLength = Integer.parseInt(line.substring(15).trim());
}
line = reader.readLine();
}
if (contentLength > 0) {
// 读取 JSON 内容
char[] content = new char[contentLength];
int read = 0;
while (read < contentLength) {
int r = reader.read(content, read, contentLength - read);
if (r == -1) break;
read += r;
}
String jsonContent = new String(content);
log.debug("pylsp 响应: {}", jsonContent);
// 发送到 WebSocket
if (socket != null && running.get()) {
socket.write(Buffer.buffer(jsonContent));
}
}
}
} catch (Exception e) {
if (running.get()) {
log.error("读取 pylsp 输出失败", e);
}
}
}
/**
* 读取 pylsp 错误输出
*/
private void readLspError() {
try {
InputStream errorStream = process.getErrorStream();
BufferedReader reader = new BufferedReader(
new InputStreamReader(errorStream, StandardCharsets.UTF_8)
);
String line;
while (running.get() && (line = reader.readLine()) != null) {
log.debug("pylsp stderr: {}", line);
}
} catch (Exception e) {
if (running.get()) {
log.error("读取 pylsp 错误输出失败", e);
}
}
}
/**
* 停止 pylsp 会话
*/
public void stop() {
running.set(false);
try {
if (processWriter != null) {
processWriter.close();
}
} catch (IOException e) {
// ignore
}
if (process != null && process.isAlive()) {
process.destroy();
try {
// 等待进程结束
if (!process.waitFor(5, java.util.concurrent.TimeUnit.SECONDS)) {
process.destroyForcibly();
}
} catch (InterruptedException e) {
process.destroyForcibly();
}
}
log.info("pylsp 会话已停止: {}", sessionId);
}
}
/**
* 获取当前活跃的 pylsp 会话数
*/
public static int getActiveSessionCount() {
return sessions.size();
}
}

4
web-service/src/main/java/cn/qaiu/lz/web/controller/ServerApi.java
View File

@@ -48,7 +48,7 @@ public class ServerApi {
return cacheService.getCachedByShareUrlAndPwd(url, pwd, JsonObject.of("UA",request.headers().get("user-agent")));
}
@RouteMapping(value = "/json/:type/:key", method = RouteMethod.GET)
@RouteMapping(value = "/json/:type/:key", method = RouteMethod.GET, order = 1000)
public Future<CacheLinkInfo> parseKeyJson(HttpServerRequest request, String type, String key) {
String pwd = "";
if (key.contains("@")) {
@@ -59,7 +59,7 @@ public class ServerApi {
return cacheService.getCachedByShareKeyAndPwd(type, key, pwd, JsonObject.of("UA",request.headers().get("user-agent")));
}
@RouteMapping(value = "/:type/:key", method = RouteMethod.GET)
@RouteMapping(value = "/:type/:key", method = RouteMethod.GET, order = 1000)
public Future<Void> parseKey(HttpServerResponse response, HttpServerRequest request, String type, String key) {
Promise<Void> promise = Promise.promise();
String pwd = "";

8
web-service/src/main/java/cn/qaiu/lz/web/model/PlaygroundParser.java
View File

@@ -2,6 +2,7 @@ package cn.qaiu.lz.web.model;
import cn.qaiu.db.ddl.Constraint;
import cn.qaiu.db.ddl.Length;
import cn.qaiu.db.ddl.NewField;
import cn.qaiu.db.ddl.Table;
import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
@@ -46,7 +47,12 @@ public class PlaygroundParser {
@Length(varcharSize = 65535)
@Constraint(notNull = true)
private String jsCode; // JavaScript代码
private String jsCode; // JavaScript/Python代码
@NewField("脚本语言类型")
@Length(varcharSize = 32)
@Constraint(defaultValue = "javascript")
private String language; // 脚本语言: javascript 或 python
@Length(varcharSize = 64)
private String ip; // 创建者IP

10
web-service/src/main/java/cn/qaiu/lz/web/service/DbService.java
View File

@@ -50,4 +50,14 @@ public interface DbService extends BaseAsyncService {
*/
Future<JsonObject> getPlaygroundParserById(Long id);
/**
* 根据type查询解析器是否存在
*/
Future<Boolean> existsPlaygroundParserByType(String type);
/**
* 初始化示例解析器（JS和Python）
*/
Future<Void> initExampleParsers();
}

185
web-service/src/main/java/cn/qaiu/lz/web/service/impl/DbServiceImpl.java
View File

@@ -90,6 +90,7 @@ public class DbServiceImpl implements DbService {
parser.put("version", row.getString("version"));
parser.put("matchPattern", row.getString("match_pattern"));
parser.put("jsCode", row.getString("js_code"));
parser.put("language", row.getString("language") != null ? row.getString("language") : "javascript");
parser.put("ip", row.getString("ip"));
// 将LocalDateTime转换为字符串格式，避免序列化为数组
var createTime = row.getLocalDateTime("create_time");
@@ -119,8 +120,8 @@ public class DbServiceImpl implements DbService {
String sql = """
INSERT INTO playground_parser
(name, type, display_name, description, author, version, match_pattern, js_code, ip, create_time, enabled)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NOW(), ?)
(name, type, display_name, description, author, version, match_pattern, js_code, language, ip, create_time, enabled)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW(), ?)
""";
client.preparedQuery(sql)
@@ -133,6 +134,7 @@ public class DbServiceImpl implements DbService {
parser.getString("version"),
parser.getString("matchPattern"),
parser.getString("jsCode"),
parser.getString("language", "javascript"),
parser.getString("ip"),
parser.getBoolean("enabled", true)
))
@@ -242,6 +244,7 @@ public class DbServiceImpl implements DbService {
parser.put("version", row.getString("version"));
parser.put("matchPattern", row.getString("match_pattern"));
parser.put("jsCode", row.getString("js_code"));
parser.put("language", row.getString("language") != null ? row.getString("language") : "javascript");
parser.put("ip", row.getString("ip"));
// 将LocalDateTime转换为字符串格式，避免序列化为数组
var createTime = row.getLocalDateTime("create_time");
@@ -265,4 +268,182 @@ public class DbServiceImpl implements DbService {
return promise.future();
}
@Override
public Future<Boolean> existsPlaygroundParserByType(String type) {
JDBCPool client = JDBCPoolInit.instance().getPool();
Promise<Boolean> promise = Promise.promise();
String sql = "SELECT COUNT(*) as count FROM playground_parser WHERE type = ?";
client.preparedQuery(sql)
.execute(Tuple.of(type))
.onSuccess(rows -> {
Integer count = rows.iterator().next().getInteger("count");
promise.complete(count > 0);
})
.onFailure(e -> {
log.error("existsPlaygroundParserByType failed", e);
promise.fail(e);
});
return promise.future();
}
@Override
public Future<Void> initExampleParsers() {
Promise<Void> promise = Promise.promise();
// JS 示例解析器代码
String jsExampleCode = """
// ==UserScript==
// @name 示例JS解析器
// @description 演示如何编写JavaScript解析器，访问 https://httpbin.org/html 获取HTML内容
// @type example-js
// @displayName JS示例
// @version 1.0.0
// @author System
// @matchPattern ^https?://httpbin\\.org/.*$
// ==/UserScript==
/**
* 解析入口函数
* @param {string} url 分享链接URL
* @param {string} pwd 提取码（可选）
* @returns {object} 包含下载链接的结果对象
*/
function parse(url, pwd) {
log.info("开始解析: " + url);
// 使用内置HTTP客户端发送GET请求
var response = http.get("https://httpbin.org/html");
if (response.statusCode === 200) {
var body = response.body;
log.info("获取到HTML内容，长度: " + body.length);
// 提取标题
var titleMatch = body.match(/<title>([^<]+)<\\/title>/i);
var title = titleMatch ? titleMatch[1] : "未知标题";
// 返回结果
return {
downloadUrl: "https://httpbin.org/html",
fileName: title + ".html",
fileSize: body.length,
extra: {
title: title,
contentType: "text/html"
}
};
} else {
log.error("请求失败，状态码: " + response.statusCode);
throw new Error("请求失败: " + response.statusCode);
}
}
""";
// Python 示例解析器代码
String pyExampleCode = """
# ==UserScript==
# @name 示例Python解析器
# @description 演示如何编写Python解析器，访问 https://httpbin.org/json 获取JSON数据
# @type example-py
# @displayName Python示例
# @version 1.0.0
# @author System
# @matchPattern ^https?://httpbin\\.org/.*$
# ==/UserScript==
def parse(url: str, pwd: str = None) -> dict:
\"\"\"
解析入口函数
Args:
url: 分享链接URL
pwd: 提取码（可选）
Returns:
包含下载链接的结果字典
\"\"\"
log.info(f"开始解析: {url}")
# 使用内置HTTP客户端发送GET请求
response = http.get("https://httpbin.org/json")
if response['statusCode'] == 200:
body = response['body']
log.info(f"获取到JSON内容，长度: {len(body)}")
# 解析JSON
import json
data = json.loads(body)
# 返回结果
return {
"downloadUrl": "https://httpbin.org/json",
"fileName": "data.json",
"fileSize": len(body),
"extra": {
"title": data.get("slideshow", {}).get("title", "未知"),
"contentType": "application/json"
}
}
else:
log.error(f"请求失败，状态码: {response['statusCode']}")
raise Exception(f"请求失败: {response['statusCode']}")
""";
// 先检查JS示例是否存在
existsPlaygroundParserByType("example-js").compose(jsExists -> {
if (jsExists) {
log.info("JS示例解析器已存在，跳过初始化");
return Future.succeededFuture();
}
// 插入JS示例解析器
JsonObject jsParser = new JsonObject()
.put("name", "示例JS解析器")
.put("type", "example-js")
.put("displayName", "JS示例")
.put("description", "演示如何编写JavaScript解析器")
.put("author", "System")
.put("version", "1.0.0")
.put("matchPattern", "^https?://httpbin\\.org/.*$")
.put("jsCode", jsExampleCode)
.put("language", "javascript")
.put("ip", "127.0.0.1")
.put("enabled", false); // 默认禁用，避免干扰正常解析
return savePlaygroundParser(jsParser);
}).compose(v -> {
// 检查Python示例是否存在
return existsPlaygroundParserByType("example-py");
}).compose(pyExists -> {
if (pyExists) {
log.info("Python示例解析器已存在，跳过初始化");
return Future.succeededFuture();
}
// 插入Python示例解析器
JsonObject pyParser = new JsonObject()
.put("name", "示例Python解析器")
.put("type", "example-py")
.put("displayName", "Python示例")
.put("description", "演示如何编写Python解析器")
.put("author", "System")
.put("version", "1.0.0")
.put("matchPattern", "^https?://httpbin\\.org/.*$")
.put("jsCode", pyExampleCode)
.put("language", "python")
.put("ip", "127.0.0.1")
.put("enabled", false); // 默认禁用，避免干扰正常解析
return savePlaygroundParser(pyParser);
}).onSuccess(v -> {
log.info("示例解析器初始化完成");
promise.complete();
}).onFailure(e -> {
log.error("初始化示例解析器失败", e);
promise.fail(e);
});
return promise.future();
}
}

304
web-service/src/test/java/cn/qaiu/lz/web/playground/PyPlaygroundTest.java Normal file
View File

@@ -0,0 +1,304 @@
package cn.qaiu.lz.web.playground;
import cn.qaiu.entity.ShareLinkInfo;
import cn.qaiu.parser.ParserCreate;
import cn.qaiu.parser.custompy.PyContextPool;
import cn.qaiu.parser.custompy.PyPlaygroundExecutor;
import cn.qaiu.parser.custompy.PyPlaygroundLogger;
import org.graalvm.polyglot.Context;
import org.graalvm.polyglot.Value;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.*;
/**
* Python 演练场单元测试
* 测试 GraalPy 环境和代码执行
*/
public class PyPlaygroundTest {
private static final Logger log = LoggerFactory.getLogger(PyPlaygroundTest.class);
@BeforeClass
public static void setup() {
log.info("初始化 PyContextPool...");
// 预热 Context Pool
PyContextPool.getInstance();
}
/**
* 测试基础的 Context 创建和 Python 代码执行
*/
@Test
public void testBasicPythonExecution() {
log.info("=== 测试基础 Python 执行 ===");
PyContextPool pool = PyContextPool.getInstance();
try (Context context = pool.createFreshContext()) {
// 测试简单的 Python 表达式
Value result = context.eval("python", "1 + 2");
assertEquals(3, result.asInt());
log.info("✓ 基础 Python 表达式执行成功: 1 + 2 = {}", result.asInt());
// 测试字符串操作
Value strResult = context.eval("python", "'hello'.upper()");
assertEquals("HELLO", strResult.asString());
log.info("✓ 字符串操作成功: 'hello'.upper() = {}", strResult.asString());
}
}
/**
* 测试 requests 库导入
*/
@Test
public void testRequestsImport() {
log.info("=== 测试 requests 库导入 ===");
PyContextPool pool = PyContextPool.getInstance();
try (Context context = pool.createFreshContext()) {
// 测试 requests 导入
context.eval("python", "import requests");
log.info("✓ requests 导入成功");
// 验证 requests 版本
Value version = context.eval("python", "requests.__version__");
log.info("✓ requests 版本: {}", version.asString());
assertNotNull(version.asString());
}
}
/**
* 测试标准库导入
*/
@Test
public void testStandardLibraries() {
log.info("=== 测试标准库导入 ===");
PyContextPool pool = PyContextPool.getInstance();
try (Context context = pool.createFreshContext()) {
// 测试 json
context.eval("python", "import json");
Value jsonResult = context.eval("python", "json.dumps({'a': 1})");
assertEquals("{\"a\": 1}", jsonResult.asString());
log.info("✓ json 库工作正常");
// 测试 re
context.eval("python", "import re");
Value reResult = context.eval("python", "bool(re.match(r'\\d+', '123'))");
assertTrue(reResult.asBoolean());
log.info("✓ re 库工作正常");
// 测试 base64
context.eval("python", "import base64");
Value b64Result = context.eval("python", "base64.b64encode(b'hello').decode()");
assertEquals("aGVsbG8=", b64Result.asString());
log.info("✓ base64 库工作正常");
}
}
/**
* 测试简单的 parse 函数执行
*/
@Test
public void testSimpleParseFunction() {
log.info("=== 测试简单 parse 函数 ===");
String pyCode = """
def parse(share_link_info, http, logger):
logger.info("测试开始")
return "https://example.com/download/test.zip"
""";
PyContextPool pool = PyContextPool.getInstance();
try (Context context = pool.createFreshContext()) {
// 创建必要的对象
ShareLinkInfo shareLinkInfo = ShareLinkInfo.newBuilder()
.shareUrl("https://example.com/s/abc")
.build();
PyPlaygroundLogger logger = new PyPlaygroundLogger();
// 注入对象
Value bindings = context.getBindings("python");
bindings.putMember("logger", logger);
// 执行代码定义函数
context.eval("python", pyCode);
// 获取并调用 parse 函数
Value parseFunc = bindings.getMember("parse");
assertNotNull("parse 函数应该存在", parseFunc);
assertTrue("parse 应该是可执行的", parseFunc.canExecute());
// 执行函数
Value result = parseFunc.execute(null, null, logger);
assertEquals("https://example.com/download/test.zip", result.asString());
log.info("✓ parse 函数执行成功，返回: {}", result.asString());
// 检查日志
assertFalse("应该有日志记录", logger.getLogs().isEmpty());
log.info("✓ 日志记录数: {}", logger.getLogs().size());
}
}
/**
* 测试带 requests 的 parse 函数
*/
@Test
public void testParseWithRequests() {
log.info("=== 测试带 requests 的 parse 函数 ===");
// 使用一个简单的模板，不实际发起网络请求
String pyCode = """
import requests
import json
def parse(share_link_info, http, logger):
logger.info("开始解析")
# 验证 requests 可用
logger.info(f"requests 版本: {requests.__version__}")
# 返回测试结果
return "https://example.com/download/file.zip"
""";
PyContextPool pool = PyContextPool.getInstance();
try (Context context = pool.createFreshContext()) {
PyPlaygroundLogger logger = new PyPlaygroundLogger();
Value bindings = context.getBindings("python");
bindings.putMember("logger", logger);
// 执行代码
context.eval("python", pyCode);
// 调用 parse
Value parseFunc = bindings.getMember("parse");
assertNotNull(parseFunc);
Value result = parseFunc.execute(null, null, logger);
assertEquals("https://example.com/download/file.zip", result.asString());
log.info("✓ 带 requests 的 parse 函数执行成功");
// 打印日志
for (PyPlaygroundLogger.LogEntry entry : logger.getLogs()) {
log.info(" [{}] {}", entry.getLevel(), entry.getMessage());
}
}
}
/**
* 测试完整的 PyPlaygroundExecutor
*/
@Test
public void testPyPlaygroundExecutor() throws Exception {
log.info("=== 测试 PyPlaygroundExecutor ===");
String pyCode = """
import json
def parse(share_link_info, http, logger):
url = share_link_info.get_share_url()
logger.info(f"解析链接: {url}")
return "https://example.com/download/test.zip"
""";
// 创建 ShareLinkInfo
ParserCreate parserCreate = ParserCreate.fromShareUrl("https://example.com/s/abc");
ShareLinkInfo shareLinkInfo = parserCreate.getShareLinkInfo();
// 创建执行器
PyPlaygroundExecutor executor = new PyPlaygroundExecutor(shareLinkInfo, pyCode);
// 异步执行
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> resultRef = new AtomicReference<>();
AtomicReference<Throwable> errorRef = new AtomicReference<>();
executor.executeParseAsync()
.onSuccess(result -> {
resultRef.set(result);
latch.countDown();
})
.onFailure(e -> {
errorRef.set(e);
latch.countDown();
});
// 等待结果
assertTrue("执行应该在 30 秒内完成", latch.await(30, TimeUnit.SECONDS));
// 检查结果
if (errorRef.get() != null) {
log.error("执行失败", errorRef.get());
fail("执行失败: " + errorRef.get().getMessage());
}
assertEquals("https://example.com/download/test.zip", resultRef.get());
log.info("✓ PyPlaygroundExecutor 执行成功，返回: {}", resultRef.get());
// 检查日志
log.info("✓ 执行日志:");
for (PyPlaygroundLogger.LogEntry entry : executor.getLogs()) {
log.info(" [{}] {}", entry.getLevel(), entry.getMessage());
}
}
/**
* 测试安全检查器拦截危险代码
*/
@Test
public void testSecurityCheckerBlocks() throws Exception {
log.info("=== 测试安全检查器拦截 ===");
String dangerousCode = """
import subprocess
def parse(share_link_info, http, logger):
result = subprocess.run(['ls'], capture_output=True)
return result.stdout.decode()
""";
ParserCreate parserCreate = ParserCreate.fromShareUrl("https://example.com/s/abc");
ShareLinkInfo shareLinkInfo = parserCreate.getShareLinkInfo();
PyPlaygroundExecutor executor = new PyPlaygroundExecutor(shareLinkInfo, dangerousCode);
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<Throwable> errorRef = new AtomicReference<>();
executor.executeParseAsync()
.onSuccess(result -> {
latch.countDown();
})
.onFailure(e -> {
errorRef.set(e);
latch.countDown();
});
assertTrue("执行应该在 30 秒内完成", latch.await(30, TimeUnit.SECONDS));
// 应该被安全检查器拦截
assertNotNull("应该抛出异常", errorRef.get());
assertTrue("应该是安全检查失败",
errorRef.get().getMessage().contains("安全检查") ||
errorRef.get().getMessage().contains("subprocess"));
log.info("✓ 安全检查器正确拦截了危险代码: {}", errorRef.get().getMessage());
}
}

415
web-service/src/test/java/cn/qaiu/lz/web/playground/RequestsIntegrationTest.java Normal file
View File

@@ -0,0 +1,415 @@
package cn.qaiu.lz.web.playground;
import cn.qaiu.entity.ShareLinkInfo;
import cn.qaiu.parser.ParserCreate;
import cn.qaiu.parser.custompy.PyContextPool;
import cn.qaiu.parser.custompy.PyPlaygroundExecutor;
import cn.qaiu.parser.custompy.PyPlaygroundLogger;
import org.graalvm.polyglot.Context;
import org.graalvm.polyglot.Value;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.*;
/**
* requests 库集成测试
*
* 测试 Python 代码在 API 场景下使用 requests 库的功能
* 验证 GraalPy 环境中 requests 库的可用性
*/
public class RequestsIntegrationTest {
private static final Logger log = LoggerFactory.getLogger(RequestsIntegrationTest.class);
@BeforeClass
public static void setup() {
log.info("初始化 PyContextPool...");
PyContextPool.getInstance();
}
/**
* 测试1: 基础 requests 导入
* 验证 requests 库可以在顶层导入
*/
@Test
public void testRequestsBasicImport() throws Exception {
log.info("=== 测试1: 基础 requests 导入 ===");
String pyCode = """
import requests
def parse(share_link_info, http, logger):
logger.info(f"requests 版本: {requests.__version__}")
return "https://example.com/download.zip"
""";
executeAndVerify(pyCode, "https://example.com/download.zip", "requests 顶层导入");
}
/**
* 测试2: requests.Session 创建
* 验证可以创建和使用 Session
*/
@Test
public void testRequestsSession() throws Exception {
log.info("=== 测试2: requests.Session 创建 ===");
String pyCode = """
import requests
def parse(share_link_info, http, logger):
session = requests.Session()
session.headers.update({
'User-Agent': 'TestBot/1.0',
'Accept': 'application/json'
})
logger.info("Session 创建成功")
logger.info(f"Headers: {dict(session.headers)}")
return "https://example.com/session.zip"
""";
executeAndVerify(pyCode, "https://example.com/session.zip", "Session 创建");
}
/**
* 测试3: requests GET 请求（模拟）
* 不发起真实网络请求，验证请求构建逻辑
*/
@Test
public void testRequestsGetPrepare() throws Exception {
log.info("=== 测试3: requests GET 请求准备 ===");
String pyCode = """
import requests
def parse(share_link_info, http, logger):
# 准备请求，但不发送
req = requests.Request('GET', 'https://api.example.com/data',
headers={'Authorization': 'Bearer test'},
params={'id': '123'}
)
prepared = req.prepare()
logger.info(f"请求 URL: {prepared.url}")
logger.info(f"请求方法: {prepared.method}")
return "https://example.com/prepared.zip"
""";
executeAndVerify(pyCode, "https://example.com/prepared.zip", "GET 请求准备");
}
/**
* 测试4: requests POST 请求（模拟）
*/
@Test
public void testRequestsPostPrepare() throws Exception {
log.info("=== 测试4: requests POST 请求准备 ===");
String pyCode = """
import requests
import json
def parse(share_link_info, http, logger):
data = {'username': 'test', 'password': 'secret'}
req = requests.Request('POST', 'https://api.example.com/login',
json=data,
headers={'Content-Type': 'application/json'}
)
prepared = req.prepare()
logger.info(f"请求 URL: {prepared.url}")
logger.info(f"请求体: {prepared.body}")
return "https://example.com/post.zip"
""";
executeAndVerify(pyCode, "https://example.com/post.zip", "POST 请求准备");
}
/**
* 测试5: 完整的解析脚本模板
* 模拟真实的网盘解析脚本结构
*/
@Test
public void testFullParserTemplate() throws Exception {
log.info("=== 测试5: 完整解析脚本模板 ===");
String pyCode = """
import requests
import re
import json
def parse(share_link_info, http, logger):
\"\"\"
解析单个文件
@match https://example\\.com/s/.*
@name ExampleParser
@version 1.0.0
\"\"\"
share_url = share_link_info.get_share_url()
logger.info(f"开始解析: {share_url}")
# 创建会话
session = requests.Session()
session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
'Accept': 'text/html,application/json',
'Accept-Language': 'zh-CN,zh;q=0.9'
})
# 模拟从URL提取文件ID
match = re.search(r'/s/([a-zA-Z0-9]+)', share_url)
if not match:
raise Exception("无法提取文件ID")
file_id = match.group(1)
logger.info(f"提取文件ID: {file_id}")
# 模拟构建API请求
api_url = f"https://api.example.com/file/{file_id}"
logger.info(f"API URL: {api_url}")
# 返回模拟的下载链接
download_url = f"https://download.example.com/{file_id}/file.zip"
logger.info(f"下载链接: {download_url}")
return download_url
""";
ParserCreate parserCreate = ParserCreate.fromShareUrl("https://example.com/s/abc123def");
ShareLinkInfo shareLinkInfo = parserCreate.getShareLinkInfo();
PyPlaygroundExecutor executor = new PyPlaygroundExecutor(shareLinkInfo, pyCode);
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> resultRef = new AtomicReference<>();
AtomicReference<Throwable> errorRef = new AtomicReference<>();
executor.executeParseAsync()
.onSuccess(result -> {
resultRef.set(result);
latch.countDown();
})
.onFailure(e -> {
errorRef.set(e);
latch.countDown();
});
assertTrue("执行应在30秒内完成", latch.await(30, TimeUnit.SECONDS));
if (errorRef.get() != null) {
log.error("执行失败", errorRef.get());
fail("执行失败: " + errorRef.get().getMessage());
}
String result = resultRef.get();
assertNotNull("结果不应为空", result);
assertTrue("结果应包含文件ID", result.contains("abc123def"));
log.info("✓ 完整解析脚本执行成功: {}", result);
// 打印日志
log.info(" 执行日志:");
for (PyPlaygroundLogger.LogEntry entry : executor.getLogs()) {
log.info(" [{}] {}", entry.getLevel(), entry.getMessage());
}
}
/**
* 测试6: 多次 requests 操作
*/
@Test
public void testMultipleRequestsOperations() throws Exception {
log.info("=== 测试6: 多次 requests 操作 ===");
String pyCode = """
import requests
import json
def parse(share_link_info, http, logger):
# 创建多个请求
urls = [
"https://api1.example.com/data",
"https://api2.example.com/info",
"https://api3.example.com/file"
]
results = []
for url in urls:
req = requests.Request('GET', url)
prepared = req.prepare()
results.append(prepared.url)
logger.info(f"准备请求: {prepared.url}")
logger.info(f"共准备 {len(results)} 个请求")
return "https://example.com/multi.zip"
""";
executeAndVerify(pyCode, "https://example.com/multi.zip", "多次 requests 操作");
}
/**
* 测试7: requests 异常处理
*/
@Test
public void testRequestsExceptionHandling() throws Exception {
log.info("=== 测试7: requests 异常处理 ===");
String pyCode = """
import requests
def parse(share_link_info, http, logger):
try:
# 尝试创建无效请求
req = requests.Request('INVALID_METHOD', 'not_a_url')
logger.info("创建了请求")
except Exception as e:
logger.warn(f"预期的异常: {type(e).__name__}")
return "https://example.com/exception.zip"
""";
executeAndVerify(pyCode, "https://example.com/exception.zip", "异常处理");
}
/**
* 测试8: ShareLinkInfo 与 requests 结合使用
*/
@Test
public void testShareLinkInfoWithRequests() throws Exception {
log.info("=== 测试8: ShareLinkInfo 与 requests 结合 ===");
String pyCode = """
import requests
import json
def parse(share_link_info, http, logger):
share_url = share_link_info.get_share_url()
share_key = share_link_info.get_share_key() or "default_key"
logger.info(f"分享链接: {share_url}")
logger.info(f"分享密钥: {share_key}")
# 使用 share_url 构建请求
session = requests.Session()
# 模拟提取信息
if 'example.com' in share_url:
return "https://download.example.com/file.zip"
return None
""";
ParserCreate parserCreate = ParserCreate.fromShareUrl("https://example.com/s/test");
ShareLinkInfo shareLinkInfo = parserCreate.getShareLinkInfo();
PyPlaygroundExecutor executor = new PyPlaygroundExecutor(shareLinkInfo, pyCode);
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> resultRef = new AtomicReference<>();
AtomicReference<Throwable> errorRef = new AtomicReference<>();
executor.executeParseAsync()
.onSuccess(result -> {
resultRef.set(result);
latch.countDown();
})
.onFailure(e -> {
errorRef.set(e);
latch.countDown();
});
assertTrue(latch.await(30, TimeUnit.SECONDS));
if (errorRef.get() != null) {
fail("执行失败: " + errorRef.get().getMessage());
}
assertEquals("https://download.example.com/file.zip", resultRef.get());
log.info("✓ ShareLinkInfo 与 requests 结合使用成功");
}
// ========== 辅助方法 ==========
/**
* 执行代码并验证结果
*/
private void executeAndVerify(String pyCode, String expectedResult, String testName) throws Exception {
ParserCreate parserCreate = ParserCreate.fromShareUrl("https://example.com/s/test123");
ShareLinkInfo shareLinkInfo = parserCreate.getShareLinkInfo();
PyPlaygroundExecutor executor = new PyPlaygroundExecutor(shareLinkInfo, pyCode);
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> resultRef = new AtomicReference<>();
AtomicReference<Throwable> errorRef = new AtomicReference<>();
executor.executeParseAsync()
.onSuccess(result -> {
resultRef.set(result);
latch.countDown();
})
.onFailure(e -> {
errorRef.set(e);
latch.countDown();
});
assertTrue("执行应在30秒内完成", latch.await(30, TimeUnit.SECONDS));
if (errorRef.get() != null) {
Throwable error = errorRef.get();
String errorMsg = error.getMessage();
// 检查是否是已知的 GraalPy 限制
if (errorMsg != null && (errorMsg.contains("unicodedata") || errorMsg.contains("LLVM"))) {
log.warn("⚠️ GraalPy unicodedata/LLVM 限制，跳过测试: {}", testName);
log.warn(" 错误: {}", errorMsg);
return; // 跳过此测试
}
log.error("执行失败", error);
fail("执行失败: " + errorMsg);
}
assertEquals(expectedResult, resultRef.get());
log.info("✓ {} 测试通过: {}", testName, resultRef.get());
// 打印日志
for (PyPlaygroundLogger.LogEntry entry : executor.getLogs()) {
log.info(" [{}] {}", entry.getLevel(), entry.getMessage());
}
}
// ========== main 方法 ==========
public static void main(String[] args) {
log.info("======================================");
log.info(" requests 集成测试套件");
log.info("======================================");
org.junit.runner.Result result = org.junit.runner.JUnitCore.runClasses(RequestsIntegrationTest.class);
log.info("\n======================================");
log.info(" 测试结果");
log.info("======================================");
log.info("运行测试数: {}", result.getRunCount());
log.info("失败测试数: {}", result.getFailureCount());
log.info("忽略测试数: {}", result.getIgnoreCount());
log.info("运行时间: {} ms", result.getRunTime());
if (result.wasSuccessful()) {
log.info("\n✅ 所有 {} 个测试通过!", result.getRunCount());
} else {
log.error("\n❌ {} 个测试失败:", result.getFailureCount());
for (org.junit.runner.notification.Failure failure : result.getFailures()) {
log.error(" - {}", failure.getTestHeader());
log.error(" 错误: {}", failure.getMessage());
}
}
System.exit(result.wasSuccessful() ? 0 : 1);
}
}

50
web-service/src/test/java/cn/qaiu/lz/web/playground/RunPlaygroundTests.java Normal file
View File

@@ -0,0 +1,50 @@
package cn.qaiu.lz.web.playground;
import org.junit.runner.JUnitCore;
import org.junit.runner.Result;
import org.junit.runner.notification.Failure;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* 手动运行 Playground 测试
* 绕过 maven surefire 的 skipTests 配置
*/
public class RunPlaygroundTests {
private static final Logger log = LoggerFactory.getLogger(RunPlaygroundTests.class);
public static void main(String[] args) {
log.info("======================================");
log.info(" Python Playground 测试套件");
log.info("======================================");
// 运行 PyPlaygroundTest
log.info("\n>>> 运行 PyPlaygroundTest...\n");
Result result = JUnitCore.runClasses(PyPlaygroundTest.class);
// 输出结果
log.info("\n======================================");
log.info(" 测试结果");
log.info("======================================");
log.info("运行测试数: {}", result.getRunCount());
log.info("失败测试数: {}", result.getFailureCount());
log.info("忽略测试数: {}", result.getIgnoreCount());
log.info("运行时间: {} ms", result.getRunTime());
if (result.wasSuccessful()) {
log.info("\n✅ 所有测试通过!");
} else {
log.error("\n❌ 部分测试失败:");
for (Failure failure : result.getFailures()) {
log.error(" - {}: {}", failure.getTestHeader(), failure.getMessage());
if (failure.getTrace() != null) {
log.error(" 堆栈: {}", failure.getTrace().substring(0, Math.min(500, failure.getTrace().length())));
}
}
}
// 退出码
System.exit(result.wasSuccessful() ? 0 : 1);
}
}

451
web-service/src/test/python/test_playground_api.py Normal file
View File

@@ -0,0 +1,451 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Playground API 测试脚本 (使用 pytest)
用于测试 /v2/playground/* 接口的功能，特别是 Python 脚本执行。
需要后端服务运行在 http://localhost:8080
安装依赖:
pip install pytest requests
运行测试:
pytest test_playground_api.py -v
或者运行特定测试:
pytest test_playground_api.py::test_status_api -v
"""
import pytest
import requests
import json
import time
# 配置
BASE_URL = "http://localhost:8080"
PLAYGROUND_BASE = f"{BASE_URL}/v2/playground"
# 测试用的分享链接
TEST_SHARE_URL = "https://www.123684.com/s/test123"
class TestPlaygroundAPI:
"""Playground API 测试类"""
@pytest.fixture(autouse=True)
def setup(self):
"""测试前置：检查服务是否可用"""
try:
resp = requests.get(f"{PLAYGROUND_BASE}/status", timeout=5)
if resp.status_code != 200:
pytest.skip("后端服务不可用")
except requests.exceptions.ConnectionError:
pytest.skip("无法连接到后端服务")
def test_status_api(self):
"""测试状态查询 API"""
resp = requests.get(f"{PLAYGROUND_BASE}/status")
assert resp.status_code == 200
data = resp.json()
assert "data" in data
assert "enabled" in data["data"]
print(f"状态响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
def test_python_simple_code(self):
"""测试简单 Python 代码执行"""
code = '''
def parse(share_link_info, http, logger):
logger.info("简单测试开始")
return "https://example.com/download/test.zip"
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
# 检查执行结果
assert data.get("success") == True, f"执行失败: {data.get('error')}"
assert data.get("result") == "https://example.com/download/test.zip"
def test_python_with_json_library(self):
"""测试使用 json 库的 Python 代码"""
code = '''
import json
def parse(share_link_info, http, logger):
data = {"url": "https://example.com/file.zip", "size": 1024}
logger.info(f"数据: {json.dumps(data)}")
return data["url"]
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
assert data.get("success") == True, f"执行失败: {data.get('error')}"
assert "example.com" in data.get("result", "")
def test_python_with_requests_import(self):
"""测试导入 requests 库（不发起实际请求）"""
code = '''
import requests
def parse(share_link_info, http, logger):
logger.info(f"requests 版本: {requests.__version__}")
# 只测试导入，不发起实际网络请求
return "https://example.com/download/file.zip"
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
# 注意: 由于 GraalPy 限制，此测试可能失败
if not data.get("success"):
print(f"⚠ requests 导入可能失败 (GraalPy 限制): {data.get('error')}")
pytest.skip("GraalPy requests 导入限制")
assert data.get("result") is not None
def test_python_with_requests_get(self):
"""测试使用 requests 发起 GET 请求"""
code = '''
import requests
def parse(share_link_info, http, logger):
logger.info("开始 HTTP 请求测试")
# 发起简单的 GET 请求
try:
resp = requests.get("https://httpbin.org/get", timeout=10)
logger.info(f"响应状态码: {resp.status_code}")
if resp.status_code == 200:
return "https://example.com/success.zip"
else:
return None
except Exception as e:
logger.error(f"请求失败: {str(e)}")
return None
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload, timeout=60)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
# 检查日志
if "logs" in data:
for log_entry in data["logs"]:
print(f" [{log_entry.get('level')}] {log_entry.get('message')}")
# 如果由于 GraalPy 限制失败，跳过测试
if not data.get("success"):
error = data.get("error", "")
if "unicodedata" in error or "LLVM" in error:
pytest.skip("GraalPy requests 限制")
pytest.fail(f"执行失败: {error}")
def test_python_security_block_subprocess(self):
"""测试安全检查器拦截 subprocess"""
code = '''
import subprocess
def parse(share_link_info, http, logger):
result = subprocess.run(['ls'], capture_output=True)
return result.stdout.decode()
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
# 应该被安全检查器拦截
assert data.get("success") == False
assert "subprocess" in data.get("error", "").lower() or \
"安全" in data.get("error", "")
def test_python_security_block_os_system(self):
"""测试安全检查器拦截 os.system"""
code = '''
import os
def parse(share_link_info, http, logger):
os.system("ls")
return "test"
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
# 应该被安全检查器拦截
assert data.get("success") == False
def test_python_with_logger(self):
"""测试日志记录功能"""
code = '''
def parse(share_link_info, http, logger):
logger.debug("这是 debug 消息")
logger.info("这是 info 消息")
logger.warn("这是 warn 消息")
logger.error("这是 error 消息")
return "https://example.com/logged.zip"
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
assert data.get("success") == True
assert "logs" in data
assert len(data["logs"]) >= 4, "应该有至少 4 条日志"
# 检查日志级别
log_levels = [log["level"] for log in data["logs"]]
assert "DEBUG" in log_levels or "debug" in log_levels
assert "INFO" in log_levels or "info" in log_levels
def test_empty_code_validation(self):
"""测试空代码验证"""
payload = {
"code": "",
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
assert data.get("success") == False
assert "空" in data.get("error", "") or "empty" in data.get("error", "").lower()
def test_invalid_language(self):
"""测试无效语言类型"""
payload = {
"code": "print('test')",
"shareUrl": TEST_SHARE_URL,
"language": "rust", # 不支持的语言
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
assert data.get("success") == False
assert "不支持" in data.get("error", "") or "language" in data.get("error", "").lower()
def test_javascript_code(self):
"""测试 JavaScript 代码执行"""
code = '''
function parse(shareLinkInfo, http, logger) {
logger.info("JavaScript 测试");
return "https://example.com/js-result.zip";
}
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "javascript",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
assert resp.status_code == 200
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
assert data.get("success") == True
assert "js-result" in data.get("result", "")
class TestRequestsIntegration:
"""requests 库集成测试"""
@pytest.fixture(autouse=True)
def setup(self):
"""测试前置：检查服务是否可用"""
try:
resp = requests.get(f"{PLAYGROUND_BASE}/status", timeout=5)
if resp.status_code != 200:
pytest.skip("后端服务不可用")
except requests.exceptions.ConnectionError:
pytest.skip("无法连接到后端服务")
def test_requests_session(self):
"""测试 requests.Session"""
code = '''
import requests
def parse(share_link_info, http, logger):
session = requests.Session()
session.headers.update({"User-Agent": "TestBot/1.0"})
logger.info("Session 创建成功")
return "https://example.com/session.zip"
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload)
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
if not data.get("success"):
error = data.get("error", "")
if "unicodedata" in error or "LLVM" in error:
pytest.skip("GraalPy requests 限制")
pytest.fail(f"执行失败: {error}")
def test_requests_post_json(self):
"""测试 requests POST JSON"""
code = '''
import requests
import json
def parse(share_link_info, http, logger):
data = {"test": "value"}
logger.info(f"准备 POST 数据: {json.dumps(data)}")
try:
resp = requests.post(
"https://httpbin.org/post",
json=data,
timeout=10
)
logger.info(f"响应状态: {resp.status_code}")
return "https://example.com/post-success.zip"
except Exception as e:
logger.error(f"POST 请求失败: {str(e)}")
return None
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload, timeout=60)
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
if not data.get("success"):
error = data.get("error", "")
if "unicodedata" in error or "LLVM" in error:
pytest.skip("GraalPy requests 限制")
def test_requests_with_headers(self):
"""测试 requests 自定义 headers"""
code = '''
import requests
def parse(share_link_info, http, logger):
headers = {
"User-Agent": "CustomBot/2.0",
"Accept": "application/json",
"X-Custom-Header": "TestValue"
}
logger.info("准备发送带自定义 headers 的请求")
try:
resp = requests.get(
"https://httpbin.org/headers",
headers=headers,
timeout=10
)
logger.info(f"响应: {resp.status_code}")
return "https://example.com/headers-success.zip"
except Exception as e:
logger.error(f"请求失败: {str(e)}")
return None
'''
payload = {
"code": code,
"shareUrl": TEST_SHARE_URL,
"language": "python",
"method": "parse"
}
resp = requests.post(f"{PLAYGROUND_BASE}/test", json=payload, timeout=60)
data = resp.json()
print(f"响应: {json.dumps(data, ensure_ascii=False, indent=2)}")
if not data.get("success"):
error = data.get("error", "")
if "unicodedata" in error or "LLVM" in error:
pytest.skip("GraalPy requests 限制")
if __name__ == "__main__":
# 直接运行测试
pytest.main([__file__, "-v", "--tb=short"])