Unstract Platform Service 测试指南：内存泄漏模拟、Auth 中间件验证与资源管理测试实践

Unstract Platform Service 测试指南：内存泄漏模拟、Auth 中间件验证与资源管理测试实践

【免费下载链接】unstractLLM-Driven Extraction of Unstructured Data — Built for API Deployments & ETL Pipeline Workflows项目地址: https://gitcode.com/GitHub_Trending/un/unstract

导读

本文以 Unstract 仓库中platform-service/tests测试目录为主线，系统讲解 Platform Service（面向 Unstract SDK 的 REST API 服务）的测试组织方式、运行命令与编写规范。你将掌握：如何用uv初始化测试环境并运行全部或指定测试；如何通过test_memory_leak_simulation.py中"旧代码 vs 新代码"对照模式验证游标（Cursor）、内存与 Redis 连接泄漏的修复效果；以及如何借助tracemalloc、MagicMock与safe_cursor等真实源码模式编写可靠的资源管理测试。

Platform Service 测试体系概览

platform-service/tests目录是 platform-service（SDK 与平台交互的 REST API 服务，基于 Flask + peewee + Redis）的测试集，目前包含三类文件：

从测试内容看，这套测试重点覆盖了两类在生产环境中最常见、也最致命的隐患：数据库游标未关闭导致的连接耗尽，以及Redis 连接未释放导致的句柄泄漏。它们共同构成 Platform Service 资源管理的回归防线。

环境准备与依赖安装

测试运行依赖uv与 pytest。根据 pyproject.toml，pytest 被声明在dependency-groups的test分组中，因此需要使用--group test同步：

cd platform-service uv sync --group test

几点说明：

• pyproject.toml要求 Python 版本>=3.12,<3.13，请确保本机解释器版本匹配；
• uv sync会同时安装运行时依赖（flask~=3.1.3、peewee~=3.16、psycopg2-binary~=2.9、redis~=5.2.1等）以及测试依赖；
• unstract-flags、unstract-core、unstract-sdk1通过[tool.uv.sources]以editable = true指向仓库内相对路径，uv 会自动解析，无需单独安装；
• 若希望复现 CI 中最接近生产的环境，可额外安装deploy分组（gunicorn + OpenTelemetry），但运行本测试目录并不需要。

测试执行前，conftest.py 会以setdefault方式注入测试用环境变量：

os.environ.setdefault("REDIS_HOST", "localhost") os.environ.setdefault("ENCRYPTION_KEY", "test-key") os.environ.setdefault("DB_SCHEMA", "unstract")

因为 env.py 在模块导入时会调用EnvManager.raise_for_missing_envs()强制校验必需配置，缺少上述变量会导致导入失败，这也是 conftest 必须先于被测模块生效的原因。

运行测试

运行全部测试

uv run pytest tests/ -v

-v会逐条打印每个用例的执行结果；如需查看print输出（例如泄漏计数的仿真报告），追加-s：

uv run pytest tests/ -v -s

运行内存泄漏仿真测试

# 带详细输出，显示泄漏计数 uv run pytest tests/test_memory_leak_simulation.py -v -s

该文件末尾提供了if __name__ == "__main__": pytest.main([__file__, "-v", "-s"])入口，因此也可直接以脚本方式运行：uv run python tests/test_memory_leak_simulation.py。

运行指定测试类

按类选择，便于单独验证某一类泄漏的修复：

# 数据库游标泄漏仿真 uv run pytest tests/test_memory_leak_simulation.py::TestConnectionLeakSimulation -v -s # 内存增长仿真（tracemalloc） uv run pytest tests/test_memory_leak_simulation.py::TestMemoryGrowthSimulation -v -s # Redis 连接泄漏仿真 uv run pytest tests/test_memory_leak_simulation.py::TestRedisConnectionLeakSimulation -v -s # 真实负载场景（1000 请求、30% 失败率） uv run pytest tests/test_memory_leak_simulation.py::TestRealWorldScenario -v -s

也可继续用::精确定位单个方法，例如tests/test_memory_leak_simulation.py::TestConnectionLeakSimulation::test_new_code_no_leak_on_exception。

运行鉴权中间件测试

uv run pytest tests/test_auth_middleware.py -v

测试文件逐项解析

test_memory_leak_simulation.py：资源泄漏的"旧 vs 新"对照验证

该文件的核心思路是：分别模拟修复前（OLD/leaky）与修复后（NEW/fixed）的代码模式，在相同故障注入条件下比较资源占用，从而以可断言的方式证明修复有效。它不依赖真实数据库或 Redis，而是用自定义的FakeCursor与MagicMock构造可控故障环境。

四个测试类的作用与关键实现如下：

FakeCursor：可追踪的游标替身

FakeCursor（test_memory_leak_simulation.py）用类变量_open_cursors记录所有已创建但未关闭的游标：

• fetchone()可配置为抛出Exception("Simulated database error")，用于模拟查询失败；
• close()会将自身从_open_cursors移除；
• get_open_cursor_count()返回当前泄漏游标数，作为断言依据。

每次测试前通过setup_method调用FakeCursor.reset_tracking()清零计数，保证用例间隔离。

游标泄漏的两条经典路径

路径一：异常打断close()。旧代码在fetchone()之后才调用cursor.close()：

def old_execute_query_buggy(db, query: str) -> str | None: cursor = db.execute_sql(query) result_row = cursor.fetchone() # 若此处抛异常，close() 永远不会执行 cursor.close() ...

在 100 次全部失败的请求后，FakeCursor.get_open_cursor_count()为 100，断言open_cursors == 100通过，直观展示"异常导致游标泄漏"。修复版则用try/finally保证关闭：

def new_execute_query_fixed(db, query: str) -> str | None: cursor = db.execute_sql(query) try: result_row = cursor.fetchone() if not result_row: return None return result_row[0] finally: cursor.close() # 无论如何都会执行

路径二：业务异常（如记录不存在）提前 return/raise。旧代码在raise APIError(...)之前未关闭游标，50 次 "not found" 请求泄漏 50 个游标；修复版在finally中关闭，泄漏数为 0。

值得注意的是，真实仓库中validate_bearer_token与execute_query均通过 extensions.py 中的safe_cursor上下文管理器保证游标关闭：

@contextmanager def safe_cursor(query: str, params: tuple = ()): cursor = db.execute_sql(query, params) try: yield cursor finally: cursor.close()

test_memory_leak_simulation.py中"新代码"的try/finally写法，正是对safe_cursor模式的显式复刻——两者殊途同归，都在强调"无论成功失败，游标必须关闭"。

tracemalloc：以字节为单位的泄漏证据

TestMemoryGrowthSimulation展示了用标准库tracemalloc量化内存增长的范式（test_memory_leak_simulation.py）：

tracemalloc.start() snapshot1 = tracemalloc.take_snapshot() # 运行被测代码…… snapshot2 = tracemalloc.take_snapshot() top_stats = snapshot2.compare_to(snapshot1, "lineno") total_diff = sum(stat.size_diff for stat in top_stats) tracemalloc.stop()

• 泄漏仿真：把 1000 个约 1KB 的"游标样对象"追加进列表并保持引用，total_diff显著增长（示例输出约 1033 KB），断言total_diff > 100_000字节；
• 正确清理：同一批对象创建后立即del obj并gc.collect()，内存增长仅约 0.66 KB，断言total_diff < 50_000字节。

compare_to(..., "lineno")按代码行聚合内存增量，能精确指出内存去向；若断言失败，可进一步用snapshot2.statistics("lineno")定位泄漏行。

Redis：连接池 vs 每次新建

TestRedisConnectionLeakSimulation用两个 Fake 类还原两种典型实现：

• 旧模式：每次请求FakeRedis(host=..., port=6379)新建连接，若set()抛异常则close()被跳过。100 次失败操作后created - closed = 100，断言泄漏数为 100；
• 新模式：共享一个FakeRedisPool(max_connections=10)，客户端从池中取用连接、由池统一管理生命周期，因此"没有逐请求创建/销毁"，100 次失败操作后无泄漏，断言operations == 100证明所有请求仍被处理。

这一设计与真实实现一一对应：extensions.py 中的get_redis_client()使用惰性单例 +create_redis_client(decode_responses=False, max_connections=10)创建带连接池的 Redis 客户端（Sentinel 模式下则由master_for()自行管理池），_redis_client的惰性初始化还避免了导入时阻塞启动。测试中的"池化方案"正是该实现的抽象演练。

真实负载场景

TestRealWorldScenario::test_validate_token_under_load_old_vs_new模拟 API 鉴权端点在 1000 次请求、30% 失败率（call_count % 10 == 3时触发raise_on_fetch）下的表现：

[LOAD TEST] 1000 requests with 30% failure rate: OLD CODE leaked cursors: 100 NEW CODE leaked cursors: 0

旧实现把cursor.close()放在try内、fetchone()之后，异常路径跳过关闭；新实现先cursor = None，再在finally中if cursor is not None: cursor.close()，双保险杜绝泄漏。两条断言old_leaked > 0与new_leaked == 0缺一不可——前者证明仿真确实制造了故障，后者证明修复有效。

典型输出一览（与 README 中记录一致）：

[OLD CODE] Open cursors after 100 failed requests: 100 ❌ LEAKED [NEW CODE] Open cursors after 100 failed requests: 0 ✅ NO LEAK [LEAK SIMULATION] Memory growth with 1000 leaked objects: 1033.37 KB [PROPER CLEANUP] Memory growth with cleanup: 0.66 KB [LOAD TEST] 1000 requests with 30% failure rate: OLD CODE leaked cursors: 100 NEW CODE leaked cursors: 0

test_auth_middleware.py：Bearer Token 鉴权测试

该文件针对 platform.py 的validate_bearer_token编写了 5 个用例，覆盖鉴权核心分支：

实现细节值得借鉴：

• 通过monkeypatch.setattr(platform, "safe_cursor", ...)注入 Fake 游标或抛错函数，完全绕开真实数据库，属于典型的"依赖替换"单测手法；
• safe_cursor的替身用contextmanager包装_Cursor（只实现fetchone()），与被测代码的with safe_cursor(...) as cursor用法完全兼容；
• 由于validate_bearer_token内部通过flask.current_app.logger记录日志，测试用with Flask(__name__).app_context():提供应用上下文（fixtureapp_ctx）；
• test_db_error_denies_access验证的是**失败关闭（fail-closed）**安全语义：数据库不可用时鉴权必须拒绝，而非放行——这是生产安全的关键底线。

编写新测试的推荐模式

README 给出了两类可直接套用的模板，均在tests/test_memory_leak_simulation.py中有完整落地。

模式一：验证游标在异常下被关闭

用MagicMock构造"会失败的游标"，调用被测函数后用assert_called_once()断言close()必然执行：

from unittest.mock import MagicMock def test_cursor_closed_on_exception(self): mock_cursor = MagicMock() mock_cursor.fetchone.side_effect = Exception("DB error") mock_db = MagicMock() mock_db.execute_sql.return_value = mock_cursor with pytest.raises(Exception): your_function(mock_db) # 即使发生异常，游标也必须被关闭 mock_cursor.close.assert_called_once()

模式二：用 tracemalloc 断言无内存泄漏

import tracemalloc def test_no_memory_leak(self): tracemalloc.start() snapshot1 = tracemalloc.take_snapshot() # 在这里运行被测代码 snapshot2 = tracemalloc.take_snapshot() stats = snapshot2.compare_to(snapshot1, "lineno") total_diff = sum(s.size_diff for s in stats) tracemalloc.stop() assert total_diff < threshold, "Memory leak detected"

模式三：资源管理测试的通用原则

结合本仓库的实现，编写资源类测试时可遵循：

1. 永远假设"正常路径之外还有异常路径"：所有close()必须位于try/finally或with块内——这也是safe_cursor（extensions.py）的设计初衷；
2. 用计数器而非肉眼观察：FakeCursor的类级计数让"泄漏了多少"可量化、可断言；
3. 新旧对照：同时断言"旧代码会泄漏"与"新代码不泄漏"，避免测试因仿真失效而假阳性；
4. 内存测试注意 GC：清理路径上显式调用gc.collect()，减少解释器回收时机带来的抖动；
5. 注意快照起点：tracemalloc.start()之后应尽快取第一个快照，且两次快照之间避免无关对象分配，保证total_diff反映被测代码的真实增量。

常见问题排查

小结

platform-service/tests通过"仿真对照 + 精确断言"的方式，为 Platform Service 的数据库游标、内存与 Redis 连接三类资源管理提供了可重复的回归验证：test_memory_leak_simulation.py用FakeCursor、tracemalloc与连接池对比量化泄漏，test_auth_middleware.py用依赖替换守护 Bearer Token 鉴权的安全语义。这套方法论同样适用于仓库中其他基于 Flask/peewee/Redis 的模块（如 backend 各 app、workers 等）——凡是涉及"打开就必须关闭"的资源，都可以复用上述模式编写测试，让资源泄漏在合入前就被拦截。

【免费下载链接】unstractLLM-Driven Extraction of Unstructured Data — Built for API Deployments & ETL Pipeline Workflows项目地址: https://gitcode.com/GitHub_Trending/un/unstract