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-serviceSDK 与平台交互的 REST API 服务基于 Flask peewee Redis的测试集目前包含三类文件文件职责conftest.py在导入阶段为测试预置REDIS_HOST、ENCRYPTION_KEY、DB_SCHEMA等必需环境变量test_memory_leak_simulation.py内存/资源泄漏仿真测试对照旧有泄漏与新已修复两种代码模式test_auth_middleware.py基于authentication_middleware的 Bearer Token 鉴权单元测试从测试内容看这套测试重点覆盖了两类在生产环境中最常见、也最致命的隐患数据库游标未关闭导致的连接耗尽以及Redis 连接未释放导致的句柄泄漏。它们共同构成 Platform Service 资源管理的回归防线。环境准备与依赖安装测试运行依赖uv与 pytest。根据 pyproject.tomlpytest 被声明在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输出例如泄漏计数的仿真报告追加-suv 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构造可控故障环境。四个测试类的作用与关键实现如下测试类目的核心手段TestConnectionLeakSimulation展示异常发生在close()之前时的游标泄漏FakeCursor类级计数器统计未关闭游标TestMemoryGrowthSimulation用tracemalloc度量真实内存增长快照对比 gc.collect()TestRedisConnectionLeakSimulation对比每次新建连接与连接池复用两种模式统计connections_created与connections_closed差值TestRealWorldScenario模拟 1000 次鉴权请求、30% 失败率下的真实负载旧/新validate_token实现对照FakeCursor可追踪的游标替身FakeCursortest_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.pytracemalloc.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..., port6379)新建连接若set()抛异常则close()被跳过。100 次失败操作后created - closed 100断言泄漏数为 100新模式共享一个FakeRedisPool(max_connections10)客户端从池中取用连接、由池统一管理生命周期因此没有逐请求创建/销毁100 次失败操作后无泄漏断言operations 100证明所有请求仍被处理。这一设计与真实实现一一对应extensions.py 中的get_redis_client()使用惰性单例 create_redis_client(decode_responsesFalse, max_connections10)创建带连接池的 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: 0test_auth_middleware.pyBearer Token 鉴权测试该文件针对 platform.py 的validate_bearer_token编写了 5 个用例覆盖鉴权核心分支测试场景期望test_valid_active_token命中有效且激活的 token(id, token, True)Truetest_rejects_missing_tokentoken 为NoneFalsetest_rejects_unknown_token查询无结果Falsetest_rejects_inactive_tokentoken 存在但is_activeFalseFalsetest_db_error_denies_accesssafe_cursor调用抛RuntimeError(db down)Falsefail-closed实现细节值得借鉴通过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_ctxtest_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模式三资源管理测试的通用原则结合本仓库的实现编写资源类测试时可遵循永远假设正常路径之外还有异常路径所有close()必须位于try/finally或with块内——这也是safe_cursorextensions.py的设计初衷用计数器而非肉眼观察FakeCursor的类级计数让泄漏了多少可量化、可断言新旧对照同时断言旧代码会泄漏与新代码不泄漏避免测试因仿真失效而假阳性内存测试注意 GC清理路径上显式调用gc.collect()减少解释器回收时机带来的抖动注意快照起点tracemalloc.start()之后应尽快取第一个快照且两次快照之间避免无关对象分配保证total_diff反映被测代码的真实增量。常见问题排查现象原因与对策uv sync报 Python 版本不匹配pyproject.toml 要求3.12,3.13使用uv python install 3.12安装后重试导入测试时报缺少环境变量conftest.py 未生效或顺序错误确认在platform-service目录下运行 pytest且 conftest 位于 tests 根目录输出中看不到[OLD CODE] ...等打印未加-s泄漏仿真的可读输出依赖 stdout 透传validate_bearer_token测试报Working outside of application context缺少app_ctxfixture内部依赖flask.current_app记录日志确保测试函数声明该 fixture 参数小结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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
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-serviceSDK 与平台交互的 REST API 服务基于 Flask peewee Redis的测试集目前包含三类文件文件职责conftest.py在导入阶段为测试预置REDIS_HOST、ENCRYPTION_KEY、DB_SCHEMA等必需环境变量test_memory_leak_simulation.py内存/资源泄漏仿真测试对照旧有泄漏与新已修复两种代码模式test_auth_middleware.py基于authentication_middleware的 Bearer Token 鉴权单元测试从测试内容看这套测试重点覆盖了两类在生产环境中最常见、也最致命的隐患数据库游标未关闭导致的连接耗尽以及Redis 连接未释放导致的句柄泄漏。它们共同构成 Platform Service 资源管理的回归防线。环境准备与依赖安装测试运行依赖uv与 pytest。根据 pyproject.tomlpytest 被声明在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输出例如泄漏计数的仿真报告追加-suv 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构造可控故障环境。四个测试类的作用与关键实现如下测试类目的核心手段TestConnectionLeakSimulation展示异常发生在close()之前时的游标泄漏FakeCursor类级计数器统计未关闭游标TestMemoryGrowthSimulation用tracemalloc度量真实内存增长快照对比 gc.collect()TestRedisConnectionLeakSimulation对比每次新建连接与连接池复用两种模式统计connections_created与connections_closed差值TestRealWorldScenario模拟 1000 次鉴权请求、30% 失败率下的真实负载旧/新validate_token实现对照FakeCursor可追踪的游标替身FakeCursortest_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.pytracemalloc.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..., port6379)新建连接若set()抛异常则close()被跳过。100 次失败操作后created - closed 100断言泄漏数为 100新模式共享一个FakeRedisPool(max_connections10)客户端从池中取用连接、由池统一管理生命周期因此没有逐请求创建/销毁100 次失败操作后无泄漏断言operations 100证明所有请求仍被处理。这一设计与真实实现一一对应extensions.py 中的get_redis_client()使用惰性单例 create_redis_client(decode_responsesFalse, max_connections10)创建带连接池的 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: 0test_auth_middleware.pyBearer Token 鉴权测试该文件针对 platform.py 的validate_bearer_token编写了 5 个用例覆盖鉴权核心分支测试场景期望test_valid_active_token命中有效且激活的 token(id, token, True)Truetest_rejects_missing_tokentoken 为NoneFalsetest_rejects_unknown_token查询无结果Falsetest_rejects_inactive_tokentoken 存在但is_activeFalseFalsetest_db_error_denies_accesssafe_cursor调用抛RuntimeError(db down)Falsefail-closed实现细节值得借鉴通过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_ctxtest_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模式三资源管理测试的通用原则结合本仓库的实现编写资源类测试时可遵循永远假设正常路径之外还有异常路径所有close()必须位于try/finally或with块内——这也是safe_cursorextensions.py的设计初衷用计数器而非肉眼观察FakeCursor的类级计数让泄漏了多少可量化、可断言新旧对照同时断言旧代码会泄漏与新代码不泄漏避免测试因仿真失效而假阳性内存测试注意 GC清理路径上显式调用gc.collect()减少解释器回收时机带来的抖动注意快照起点tracemalloc.start()之后应尽快取第一个快照且两次快照之间避免无关对象分配保证total_diff反映被测代码的真实增量。常见问题排查现象原因与对策uv sync报 Python 版本不匹配pyproject.toml 要求3.12,3.13使用uv python install 3.12安装后重试导入测试时报缺少环境变量conftest.py 未生效或顺序错误确认在platform-service目录下运行 pytest且 conftest 位于 tests 根目录输出中看不到[OLD CODE] ...等打印未加-s泄漏仿真的可读输出依赖 stdout 透传validate_bearer_token测试报Working outside of application context缺少app_ctxfixture内部依赖flask.current_app记录日志确保测试函数声明该 fixture 参数小结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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考