Windmill Python 客户端wmill测试指南从本地后端搭建到 S3 与 API 集成测试实战【免费下载链接】windmillOpen-source developer platform to power your entire infra and turn scripts into webhooks, workflows and UIs. Fastest workflow engine (13x vs Airflow). Open-source alternative to Retool and Temporal.项目地址: https://gitcode.com/GitHub_Trending/wi/windmill本指南聚焦 Windmill 开源仓库中 python-client/tests/README.md 所描述的测试方案完整讲解如何在本地启动 Windmill 后端、安装wmillPython 客户端并基于 wmill_client_test.py 编写和运行针对客户端各核心功能的集成测试与纯单元测试。阅读完成后你将掌握一套可复用的本地测试流程并能独立为变量、资源、脚本执行、S3 文件读写、DuckDB/Polars/boto3 连接配置等能力补充自己的测试用例。一、测试方案概览为什么需要本地后端 真实客户端wmill是 Windmill 平台的官方 Python 客户端项目位于 python-client/wmill包名为wmill它通过 HTTP 调用 Windmill 后端 API 完成变量读写、资源访问、脚本/流程执行、S3 文件操作、OIDC 令牌获取等能力。其核心实现见 client.py 中的Windmill类。测试这类客户端有两种层次纯单元测试不依赖网络与后端只测试纯函数如parse_s3_object的 URI 解析逻辑测试文件 wmill_client_test.py 中的TestParseS3Object类即为此类。集成测试需要一台真实的 Windmill 后端Backend简称 BE运行在本地客户端通过BASE_INTERNAL_URL/WM_BASE_URL指向它再以工作区 Token 发起真实 API 调用。README 给出的方案正是后者这也是验证客户端与后端契约一致性的最直接手段。二、前置条件准备一台本地 Windmill 后端根据 python-client/tests/README.md集成测试要求本地有一个 Windmill 后端监听在localhost:8000。仓库提供了两种常见启动方式方式一通过 cargo 直接运行在仓库根目录执行cargo runWindmill 后端是 Rust 实现的cargo run会编译并启动完整服务默认监听 8000 端口。方式二通过 docker compose 启动仓库根目录提供了 docker-compose.yml其中包含后端及其依赖PostgreSQL、MinIO 等的容器编排docker compose up -d无论采用哪种方式都需要保证后端进程健康运行且 API 可访问端口8000未被占用或映射正确工作区与用户账号已初始化且你拥有一个可用的访问 Token可在 Windmill 界面的 User Settings 中生成。从源码结构看客户端构造时会把base_url与/api拼接为 API 基地址见 client.py因此测试中的_host http://localhost:8000实际指向的是http://localhost:8000/api。三、安装本地包到虚拟环境README 给出的安装方式是直接在本地产物上执行pip install .这样可以确保测试使用的是当前仓库的代码而不是 PyPI 上已发布的旧版本cd ./wmill pip3 install .即进入 python-client/wmill 目录该目录是包的根包含pyproject.toml执行安装。安装前建议先激活你的虚拟环境venv / conda / uv 均可。关于依赖可以补充说明pyproject.toml 声明了该包的核心依赖python ^3.7支持 Python 3.7 及以上版本httpx 0.24HTTP 客户端底层依赖客户端所有的get/post都是对 httpx 的薄封装。开发与测试依赖dependency-groups.dev则包括pytest9.0.2与httpx0.28.1如果你更习惯 pytest 运行测试可以直接使用。四、配置测试Token、Workspace 与 Host安装完成后打开测试文件 wmill_client_test.py其中TestStringMethods类头部定义了三个关键配置项class TestStringMethods(unittest.TestCase): _token WM_TOKEN _workspace storage _host http://localhost:8000 _resource_path u/admin/docker_minio各字段含义字段默认示例说明_tokenWM_TOKEN后端访问令牌必须替换为真实 Token否则鉴权失败_workspacestorage目标工作区 ID必须是后端中真实存在的工作区_hosthttp://localhost:8000本地后端地址与 README 要求一致_resource_pathu/admin/docker_minio测试用 S3 资源在 Windmill 中的路径需提前在对应工作区创建setUp方法在每条用例执行前把这三个值注入环境变量这正是客户端读取配置的机制def setUp(self): os.environ[WM_WORKSPACE] self._workspace os.environ[WM_TOKEN] self._token os.environ[BASE_INTERNAL_URL] self._host对照 client.py 的Windmill.__init__可以看到这些环境变量的真实作用BASE_INTERNAL_URL或WM_BASE_URL决定 API 基地址无显式base_url时WM_TOKEN默认认证令牌注入Authorization: Bearer token请求头WM_WORKSPACE默认工作区 ID缺少时客户端会直接assert失败其余如WM_JOB_ID、WM_ROOT_FLOW_JOB_ID、WM_STATE_PATH等则服务于脚本内运行场景父作业追踪、状态路径等测试脚本中一般不设置。五、深入测试文件三类可复用的测试样板wmill_client_test.py 实际上提供了三类现成样板覆盖了 wmill 客户端最常用的能力面。5.1 S3 连接设置生成测试DuckDB / Polars / boto3test_duckdb_connection_settings验证客户端能把一个 S3 资源转换为 DuckDB 可用的连接 SQLsettings wmill.duckdb_connection_settings(self._resource_path) self.assertIsNotNone(settings) self.assertEqual(settings[connection_settings_str], expected_settings_str) self.assertEqual(settings.connection_settings_str, expected_settings_str)值得注意返回的DuckDbConnectionSettings定义在 s3_types.py同时支持字典下标settings[connection_settings_str]与属性访问settings.connection_settings_str两种方式因为它是dict子类并实现了__getattr__。断言中还验证了connection_settings_str生成的 SQL 包含INSTALL httpfs、SET s3_url_stylepath、SET s3_endpoint...等语句。test_polars_connection_settings与test_boto3_connection_settings同理分别验证Polars返回s3fs_argsendpoint/key/secret/use_ssl/cache_regions/client_kwargs与polars_cloud_optionsaws_endpoint_url/aws_access_key_id/aws_secret_access_key/aws_region/aws_allow_http两组参数boto3返回endpoint_url、region_name、use_ssl、aws_access_key_id、aws_secret_access_key若 S3 资源带token还会追加aws_session_token见 client.py。在客户端源码中这三个能力分别由get_duckdb_connection_settings、get_polars_connection_settings、get_boto3_connection_settings实现均通过POST /w/{workspace}/job_helpers/v2/...系列端点从后端获取 S3 资源信息。5.2 S3 文件读写与删除测试测试文件给出了一套完整的 S3 文件生命周期用例全部以wmill.S3Object(s3key)描述目标对象# 下载为流式读取配合文件写入 with wmill.load_s3_file_reader(S3Object(s3region.csv)) as file_content, open( region.csv, wb ) as output_file: output_file.write(file_content.read()) # 下载为字节内容 file_content wmill.load_s3_file(S3Object(s3region.csv)) # 上传文件流 / 原始字节 with open(region.csv, rb) as file_content: file_key wmill.write_s3_file(S3Object(s3region.csv), file_content) file_key wmill.write_s3_file(S3Object(s3hello-world.txt), bHello Windmill!) # 删除并验证 wmill.delete_s3_object(s3_obj) with self.assertRaises(Exception): wmill.load_s3_file(s3_obj)对应客户端实现client.py中有几个值得测试关注的细节load_s3_file内部复用load_s3_file_reader返回S3BufferedReader流式读取write_s3_file接受BufferedReader或bytes内部把 BufferedReader 转成字节生成器后以application/octet-stream直传job_helpers/upload_s3_filedelete_s3_object通过client.delete调用job_helpers/delete_s3_file。5.3 parse_s3_object 纯单元测试无需后端TestParseS3Object是测试文件中唯一默认启用未加unittest.skip的用例类因为它不依赖网络与环境变量可随时运行。它验证wmill.parse_s3_object的 URI 解析规则# 裸 key 被拒绝错误信息会提示使用 s3:/// 写法 with self.assertRaisesRegex(ValueError, s3:///dir/file.json): wmill.parse_s3_object(dir/file.json) # 三斜杠 URI 表示默认存储 self.assertEqual( wmill.parse_s3_object(s3:///dir/file.json), S3Object(s3dir/file.json, storageNone), ) # 完整 URI 拆分为 storage 与 key self.assertEqual( wmill.parse_s3_object(s3://bucket/dir/f), S3Object(s3dir/f, storagebucket), ) # 畸形 / 空 key / 空字符串一律报错 with self.assertRaises(ValueError): wmill.parse_s3_object(s3://broken) with self.assertRaises(ValueError): wmill.parse_s3_object(s3:///) with self.assertRaises(ValueError): wmill.parse_s3_object()这套解析规则的意义在于宁可在客户端解析阶段快速失败也不要把文件静默写入错误的位置或使用自动生成的 key。S3Object直接透传、可同时承载s3文件 key、storage存储桶标识与presigned预签名令牌三个字段见 s3_types.py。六、编写与运行你自己的测试6.1 基于现有样板扩展README 明确建议You can then implement your own test calling any function in the wmill client and test its output. 即你可以仿照现有用例在TestStringMethods中新增方法测试客户端暴露的任何能力。wmill 包通过init.py 把client与s3_types的全部符号导出到顶层因此import wmill后即可直接使用wmill.get_variable、wmill.run_script、wmill.get_resource、wmill.set_state、wmill.get_state等常用函数见 python-client/README.md 的 Basic Usage 示例。例如验证变量读写def test_variable_set_and_get(self): path u/admin/test_variable wmill.set_variable(path, hello-windmill) self.assertEqual(wmill.get_variable(path), hello-windmill)验证脚本同步执行def test_run_script_by_path_sync(self): result wmill.run_script_by_path_sync( pathf/admin/my_script, args{arg1: value1}, ) self.assertIsNotNone(result)6.2 运行测试使用 Python 标准库unittest直接运行整个测试文件python -m unittest wmill_client_test -v或单独运行某一用例类/方法跳过类外的__main__判断python -m unittest wmill_client_test.TestParseS3Object -v python -m unittest wmill_client_test.TestStringMethods.test_upload_s3_raw_bytes -v如果你安装了 pytestpyproject.toml的 dev 依赖中已包含也可以直接pytest wmill_client_test.py -v6.3 注意默认跳过标记默认情况下TestStringMethods中所有需要后端的用例都带有unittest.skip(skipping)运行时会直接跳过只有TestParseS3Object会真正执行。这是仓库故意为之避免在没有后端的环境下误跑集成测试。要启用某个用例删除其上的unittest.skip装饰器即可同时务必保证_token等配置真实有效。七、测试链路背后的客户端机制小结透过这些测试可以总结出wmill客户端的核心行为约定它们也是你编写更多测试时的事实依据实现均可回溯至 client.py配置即环境变量WM_TOKEN、WM_WORKSPACE、BASE_INTERNAL_URL/WM_BASE_URL三个变量决定了客户端连哪里、以谁的身份、在哪个工作区鉴权方式请求头固定为Authorization: Bearer token与Content-Type: application/json一起在构造时生成超时策略httpx 客户端超时默认 900 秒适合长时间运行的任务timeouthttpx.Timeout(900.0)返回类型连接设置类均为dict子类下标与属性两种访问方式等价测试中可二者任选S3 对象寻址一律通过S3Object携带 key/storage/presignedURI 字符串会先经parse_s3_object严格校验再进入 API 调用。八、常见问题排查现象可能原因处理建议所有集成测试报 401/403_token仍是占位符或已过期在 Windmill 用户设置中重新生成 Token 并更新_token报 workspace 相关断言失败WM_WORKSPACE对应工作区不存在改用后端实际存在的工作区 ID并确认_workspace正确连接被拒绝Connection refused后端未启动或端口不是 8000确认cargo run或docker compose up -d已就绪核对_hostS3 用例失败S3 资源如u/admin/docker_minio未创建或凭据错误在对应工作区创建 S3 资源并确保资源路径与_resource_path一致parse_s3_object相关用例报错对 URI 语义理解偏差记住规则s3:///key默认存储s3://storage/key指定存储裸 key 与空 key 均非法综上python-client/tests/README.md 提供的是一套轻量而完整的本地后端 真实客户端测试方法论一条安装命令、三个环境变量、一个可复用的测试文件即可覆盖 wmill 客户端从 API 调用到 S3 数据面的主要能力。配合TestParseS3Object这类零依赖的纯单元测试你可以在任何 CI 环境中快速回归客户端自身逻辑同时保留本地集成验证的完整链路。【免费下载链接】windmillOpen-source developer platform to power your entire infra and turn scripts into webhooks, workflows and UIs. Fastest workflow engine (13x vs Airflow). Open-source alternative to Retool and Temporal.项目地址: https://gitcode.com/GitHub_Trending/wi/windmill创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考