""" 透传优化回归测试（依赖隔离：桩掉 chongming 包链、auth 与 chongming_cache） 覆盖: A. 回复路径 bytes 透传: 1. 普通端点返回 worker 原始字节（byte-for-byte，零再序列化）+ application/json 2. 框架错误（call_raw 抛 RuntimeError）-> HTTP 503 3. 业务错误（{"code":1,...} 普通回复）-> HTTP 200 透传（网关不解析） B. validate=false 请求透传: 4. POST 信封字节拼接精确断言（网关零解析/零转义） 5. 空 body 兜底 {"meta":...,"data":{}} 6. validate=true 端点仍走 Pydantic 校验（422）；validate=false 跳过校验（透传） 7. GET 端点不受 validate 影响（validate 仅对 body 端点生效） C. OpenAPI: 8. validate=false 端点 requestBody $ref 注入 + components schema 完整 D. WorkerManager 单测（FakeSession）: 9. call_raw 零解析返回原始字节；call() 仍解析为 dict（MQTT 语义保留） 10. cacheable: 缓存原始 bytes、二次命中不再发查询、业务错误不缓存 11. reply.err -> 重试 -> RuntimeError；空回复兜底 b"{}" 运行: uv run python test_passthrough_optimization.py """ import asyncio import importlib.util import logging import sys import types from pathlib import Path from types import SimpleNamespace import orjson from fastapi import FastAPI from fastapi.testclient import TestClient REPO_ROOT = Path(__file__).resolve().parent sys.path.insert(0, str(REPO_ROOT / "src")) sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src")) # ---- 依赖隔离：桩掉 chongming 包链 / auth / chongming_cache ---- _pkg = types.ModuleType("chongming") _pkg.__path__ = [] _app = types.ModuleType("chongming.app") _app.__path__ = [] _core = types.ModuleType("chongming.app.core") _core.__path__ = [] _logger_stub = types.ModuleType("chongming.app.core.logger") _logger_stub.get_logger = lambda name: logging.getLogger(name) _auth_stub = types.ModuleType("chongming.app.core.auth") _auth_stub.store_user_to_request = lambda x: x # 仅用于 Depends，桩掉鉴权 _cache_stub = types.ModuleType("chongming_cache") _cache_stub.Cache = object for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub): sys.modules[mod.__name__] = mod # 加载真实 worker_manager（route_generator 依赖其 StreamIdleTimeoutError） _wm_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "worker_manager.py" _wm_spec = importlib.util.spec_from_file_location( "chongming.app.core.worker_manager", _wm_path ) _wm_mod = importlib.util.module_from_spec(_wm_spec) sys.modules["chongming.app.core.worker_manager"] = _wm_mod _wm_spec.loader.exec_module(_wm_mod) # type: ignore[union-attr] # 加载真实 call_stats（route_generator 依赖其 spawn/record_call） _cs_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "call_stats.py" _cs_spec = importlib.util.spec_from_file_location( "chongming.app.core.call_stats", _cs_path ) _cs_mod = importlib.util.module_from_spec(_cs_spec) sys.modules["chongming.app.core.call_stats"] = _cs_mod _cs_spec.loader.exec_module(_cs_mod) # type: ignore[union-attr] # 加载真实 route_generator _rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py" _rg_spec = importlib.util.spec_from_file_location( "chongming.app.core.route_generator", _rg_path ) _rg_mod = importlib.util.module_from_spec(_rg_spec) sys.modules["chongming.app.core.route_generator"] = _rg_mod _rg_spec.loader.exec_module(_rg_mod) # type: ignore[union-attr] register_route_group = _rg_mod.register_route_group _inject_schemas_to_openapi = _rg_mod._inject_schemas_to_openapi WorkerManager = _wm_mod.WorkerManager failures = [] def check(name, cond, detail=""): status = "PASS" if cond else "FAIL" print(f"[{status}] {name}" + (f" — {detail}" if detail else "")) if not cond: failures.append(name) class FakeWorkerManager: """模拟 WorkerManager：call_raw 返回原始字节；error="no_worker" 时抛 RuntimeError""" def __init__(self, reply=b'{"result": 5}', error=None): self.reply = reply self.error = error self.call_raw_payloads = [] # 捕获的预序列化请求字节 async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False): self.call_raw_payloads.append(payload) if self.error == "no_worker": raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。") return self.reply ROUTES = [ # validate=true 对照（默认） { "subject": "demo.echo", "path": "/echo", "method": "POST", "params": ["msg: str"], "summary": "echo", "timeout": 3.0, "auth_required": False, }, { "subject": "demo.echo2", "path": "/echo2", "method": "POST", "params": ["n: int"], "summary": "typed echo", "timeout": 3.0, "auth_required": False, }, { "subject": "demo.add", "path": "/add", "method": "GET", "params": ["a: int", "b: int"], "summary": "add", "timeout": 3.0, "auth_required": False, }, # validate=false：请求体原样透传 { "subject": "demo.raw", "path": "/raw/echo", "method": "POST", "params": ["msg: str"], "summary": "raw echo", "validate": False, "timeout": 3.0, "auth_required": False, }, # validate=false + GET 对照（validate 仅对 body 端点生效） { "subject": "demo.rawget", "path": "/raw/get", "method": "GET", "params": ["a: int"], "summary": "raw get", "validate": False, "timeout": 3.0, "auth_required": False, }, ] def build_app(fwm): app = FastAPI() extra_schemas = {} raw_body_models = {} register_route_group( app, fwm, "demo", ROUTES, prefix="/api/v1", extra_schemas=extra_schemas, raw_body_models=raw_body_models, ) if extra_schemas: _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models) return TestClient(app) # ---- A. 回复路径 bytes 透传 ---- def test_reply_bytes_passthrough(): fwm = FakeWorkerManager(reply=b'{"result": 5}') client = build_app(fwm) r = client.get("/api/v1/add?a=1&b=2") check("GET 状态码 200", r.status_code == 200, str(r.status_code)) check( "回复为 worker 原始字节（byte-for-byte 零再序列化）", r.content == b'{"result": 5}', repr(r.content), ) check( "Content-Type application/json", r.headers["content-type"].startswith("application/json"), r.headers.get("content-type", ""), ) data = orjson.loads(fwm.call_raw_payloads[0])["data"] check("query 参数透传到信封 data", data == {"a": 1, "b": 2}, str(data)) def test_framework_error_503(): fwm = FakeWorkerManager(error="no_worker") client = build_app(fwm) r = client.get("/api/v1/add?a=1&b=2") check("框架错误（无 worker）-> 503", r.status_code == 503, str(r.status_code)) def test_business_error_passthrough_200(): fwm = FakeWorkerManager(reply=b'{"code": 1, "error": "nope"}') client = build_app(fwm) r = client.post("/api/v1/echo", json={"msg": "hi"}) check("业务错误（普通回复）-> 200 透传", r.status_code == 200, str(r.status_code)) check("业务错误体原样透传", r.content == b'{"code": 1, "error": "nope"}', repr(r.content)) # ---- B. validate=false 请求透传 ---- def test_validate_false_envelope_splice(): fwm = FakeWorkerManager() client = build_app(fwm) r = client.post("/api/v1/raw/echo", json={"msg": "hello"}) check("validate=false POST 状态码 200", r.status_code == 200, str(r.status_code)) expected = b'{"meta":{"method":"POST"},"data":{"msg":"hello"}}' check( "信封字节拼接（网关零解析/零转义）", fwm.call_raw_payloads[0] == expected, repr(fwm.call_raw_payloads[0]), ) check("响应透传", r.content == b'{"result": 5}', repr(r.content)) def test_validate_false_empty_body(): fwm = FakeWorkerManager() client = build_app(fwm) r = client.post("/api/v1/raw/echo", content=b"") check("空 body 状态码 200", r.status_code == 200, str(r.status_code)) check( "空 body 兜底 data:{}", fwm.call_raw_payloads[0] == b'{"meta":{"method":"POST"},"data":{}}', repr(fwm.call_raw_payloads[0]), ) def test_validate_true_still_422(): fwm = FakeWorkerManager() client = build_app(fwm) r = client.post("/api/v1/echo2", json={"n": "not-a-number"}) check("validate=true 仍 Pydantic 校验（422）", r.status_code == 422, str(r.status_code)) check("校验失败不发起调用", len(fwm.call_raw_payloads) == 0) def test_validate_false_skips_validation(): fwm = FakeWorkerManager() client = build_app(fwm) r = client.post("/api/v1/raw/echo", json={"n": "not-a-number"}) check("validate=false 跳过校验（200）", r.status_code == 200, str(r.status_code)) check( "原始 body 透传（含类型错误输入）", fwm.call_raw_payloads[0] == b'{"meta":{"method":"POST"},"data":{"n":"not-a-number"}}', repr(fwm.call_raw_payloads[0]), ) def test_validate_false_get_unaffected(): fwm = FakeWorkerManager() client = build_app(fwm) r = client.get("/api/v1/raw/get?a=5") check("validate=false GET 不受影响（200）", r.status_code == 200, str(r.status_code)) data = orjson.loads(fwm.call_raw_payloads[0])["data"] check("GET 参数正常透传", data == {"a": 5}, str(data)) # ---- C. OpenAPI ---- def test_openapi_request_body_injected(): fwm = FakeWorkerManager() client = build_app(fwm) schema = client.app.openapi() # validate=false 端点：requestBody $ref 注入 raw_op = schema["paths"]["/api/v1/raw/echo"]["post"] check( "validate=false 端点 requestBody 已注入", raw_op.get("requestBody", {}).get("content", {}).get("application/json", {}).get("schema", {}).get("$ref") == "#/components/schemas/demo_raw_echo_POST_Request", str(raw_op.get("requestBody")), ) check( "请求模型注册到 components/schemas", "demo_raw_echo_POST_Request" in schema["components"]["schemas"], ) # validate=true 端点：FastAPI 原生 requestBody（不受影响） ok_op = schema["paths"]["/api/v1/echo"]["post"] check( "validate=true 端点原生 requestBody 保留", "requestBody" in ok_op, str(ok_op.get("requestBody")), ) # GET 端点无 requestBody check( "GET 端点无 requestBody", "requestBody" not in schema["paths"]["/api/v1/raw/get"]["get"], ) # ---- D. WorkerManager 单测（FakeSession）---- class FakeSession: """模拟 zenoh Session：get() 返回单回复迭代器""" def __init__(self, payload=b'{"result": 5}', err_payload=None): self.payload = payload self.err_payload = err_payload self.get_calls = 0 def get(self, *args, **kwargs): self.get_calls += 1 if self.err_payload is not None: reply = SimpleNamespace(ok=None, err=SimpleNamespace(payload=self.err_payload)) else: reply = SimpleNamespace(ok=SimpleNamespace(payload=self.payload), err=None) return iter([reply]) def make_wm(session, cache=None): wm = WorkerManager(session, cache=cache if cache is not None else {}) wm._workers_cache = {"demo": ["worker/demo/w1"]} return wm async def test_wm_call_raw_no_parse(): session = FakeSession(payload=b'{"result": 5}') wm = make_wm(session) result = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}') check( "call_raw 返回原始字节（零解析）", isinstance(result, bytes) and result == b'{"result": 5}', repr(result), ) check("session.get 仅调用 1 次", session.get_calls == 1, str(session.get_calls)) async def test_wm_call_dict_semantics(): session = FakeSession(payload=b'{"result": 5}') wm = make_wm(session) result = await wm.call("demo", "calc.add", {"meta": {}, "data": {}}) check("call() 仍解析为 dict（MQTT 语义保留）", result == {"result": 5}, repr(result)) async def test_wm_cache_bytes(): cache = {} session = FakeSession(payload=b'{"result": 5}') wm = make_wm(session, cache) r1 = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True) r2 = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True) check("缓存命中返回原始 bytes", r1 == r2 == b'{"result": 5}', repr(r1)) check("二次调用命中缓存（不再发查询）", session.get_calls == 1, str(session.get_calls)) check("缓存值为原始 bytes", all(isinstance(v, bytes) for v in cache.values())) # 业务错误（code!=0）不缓存 session2 = FakeSession(payload=b'{"code": 1, "error": "x"}') wm2 = make_wm(session2, {}) await wm2.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True) check("业务错误不写入缓存", len(wm2.cache) == 0, str(wm2.cache)) async def test_wm_reply_err_retry(): session = FakeSession(err_payload=b'{"error": "boom"}') wm = make_wm(session) try: await wm.call_raw("demo", "calc.add", b'{}') check("reply.err 应抛 RuntimeError", False) except RuntimeError as e: check("reply.err 重试后抛 RuntimeError", "boom" in str(e), str(e)) check("reply.err 触发重试（2 次查询）", session.get_calls == 2, str(session.get_calls)) async def test_wm_empty_reply_fallback(): session = FakeSession(payload=b"") wm = make_wm(session) check("call_raw 空回复兜底 b\"{}\"", await wm.call_raw("demo", "calc.add", b"{}") == b"{}") check("call() 空回复 → {}", await wm.call("demo", "calc.add", {"meta": {}, "data": {}}) == {}) def run(): # HTTP 层（TestClient） test_reply_bytes_passthrough() test_framework_error_503() test_business_error_passthrough_200() test_validate_false_envelope_splice() test_validate_false_empty_body() test_validate_true_still_422() test_validate_false_skips_validation() test_validate_false_get_unaffected() test_openapi_request_body_injected() # WorkerManager 层（FakeSession） asyncio.run(test_wm_call_raw_no_parse()) asyncio.run(test_wm_call_dict_semantics()) asyncio.run(test_wm_cache_bytes()) asyncio.run(test_wm_reply_err_retry()) asyncio.run(test_wm_empty_reply_fallback()) print(f"\n结果: {len(failures)} 个失败") return 1 if failures else 0 if __name__ == "__main__": sys.exit(run())