企业官网建设流程全解析

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类。

测试这类客户端有两种层次：

1. 纯单元测试：不依赖网络与后端，只测试纯函数（如parse_s3_object的 URI 解析逻辑），测试文件 wmill_client_test.py 中的TestParseS3Object类即为此类。
2. 集成测试：需要一台真实的 Windmill 后端（Backend，简称 BE）运行在本地，客户端通过BASE_INTERNAL_URL/WM_BASE_URL指向它，再以工作区 Token 发起真实 API 调用。README 给出的方案正是后者，这也是验证客户端与后端契约一致性的最直接手段。

二、前置条件：准备一台本地 Windmill 后端

根据 python-client/tests/README.md，集成测试要求本地有一个 Windmill 后端，监听在localhost:8000。仓库提供了两种常见启动方式：

方式一：通过 cargo 直接运行

在仓库根目录执行：

cargo run

Windmill 后端是 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.24"：HTTP 客户端底层依赖，客户端所有的get/post都是对 httpx 的薄封装。

开发与测试依赖（dependency-groups.dev）则包括pytest>=9.0.2与httpx>=0.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"

各字段含义：

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 / boto3）

test_duckdb_connection_settings验证客户端能把一个 S3 资源转换为 DuckDB 可用的连接 SQL：

settings = 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_style='path'、SET s3_endpoint=...等语句。

test_polars_connection_settings与test_boto3_connection_settings同理，分别验证：

• Polars：返回s3fs_args（endpoint/key/secret/use_ssl/cache_regions/client_kwargs）与polars_cloud_options（aws_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(s3="key")描述目标对象：

# 下载为流式读取（配合文件写入） with wmill.load_s3_file_reader(S3Object(s3="region.csv")) as file_content, open( "region.csv", "wb" ) as output_file: output_file.write(file_content.read()) # 下载为字节内容 file_content = wmill.load_s3_file(S3Object(s3="region.csv")) # 上传（文件流 / 原始字节） with open("region.csv", "rb") as file_content: file_key = wmill.write_s3_file(S3Object(s3="region.csv"), file_content) file_key = wmill.write_s3_file(S3Object(s3="hello-world.txt"), b"Hello 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_file；
• delete_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(s3="dir/file.json", storage=None), ) # 完整 URI 拆分为 storage 与 key self.assertEqual( wmill.parse_s3_object("s3://bucket/dir/f"), S3Object(s3="dir/f", storage="bucket"), ) # 畸形 / 空 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( path="f/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

如果你安装了 pytest（pyproject.toml的 dev 依赖中已包含），也可以直接：

pytest wmill_client_test.py -v

6.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 秒，适合长时间运行的任务（timeout=httpx.Timeout(900.0)）；
• 返回类型：连接设置类均为dict子类，下标与属性两种访问方式等价，测试中可二者任选；
• S3 对象寻址：一律通过S3Object携带 key/storage/presigned，URI 字符串会先经parse_s3_object严格校验再进入 API 调用。

八、常见问题排查

综上，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