From 6ed33f3185dfc92e1e4529aef73f8d9e61855b7f Mon Sep 17 00:00:00 2001 From: goulustis Date: Wed, 4 Mar 2026 15:20:52 +0800 Subject: [PATCH] update tests --- tests/test_dashscope_client.py | 44 +++++++++++------------ tests/test_server_dashscope_routes.py | 50 ++++++++++++++------------- 2 files changed, 47 insertions(+), 47 deletions(-) diff --git a/tests/test_dashscope_client.py b/tests/test_dashscope_client.py index d2a58fa..fb1dbbe 100644 --- a/tests/test_dashscope_client.py +++ b/tests/test_dashscope_client.py @@ -8,6 +8,7 @@ Instructions: - Set BASE_URL below to the server base URL you started. - Optionally set environment variables ALI_API_KEY and ALI_APP_ID. """ + import os import json import os.path as osp @@ -31,6 +32,7 @@ except Exception as e: # <<< Paste your running FastAPI base url here >>> BASE_URL = os.getenv("DS_BASE_URL", "http://127.0.0.1:8588/api/") + # Params def _first_non_empty_csv_token(value: str) -> str: parts = [p.strip() for p in (value or "").split(",") if p.strip()] @@ -61,23 +63,24 @@ def _pick_api_key(registry: dict) -> str: return fast_first if ali_key: return ali_key - raise RuntimeError("Missing API key. Set FAST_AUTH_KEYS or ALI_API_KEY in your environment.") + raise RuntimeError( + "Missing API key. Set FAST_AUTH_KEYS or ALI_API_KEY in your environment." + ) def _pick_app_id(api_key: str, registry: dict) -> str: - # Explicit user choice always wins. - explicit = os.getenv("ALI_APP_ID") - if explicit: - return explicit + if api_key: + explicit = (registry.get("api_keys") or {}).get(api_key, {}).get("app_id") + if explicit: + return explicit - routes_obj = registry.get("routes") - if not isinstance(routes_obj, dict): - routes_obj = registry.get("pipelines", {}) - route_ids = [r for r in routes_obj.keys() if isinstance(r, str) and r] + pipelines_obj = registry.get("pipelines") + if not isinstance(pipelines_obj, dict): + pipelines_obj = {} + pipeline_ids = [r for r in pipelines_obj.keys() if isinstance(r, str) and r] - # Prefer an explicitly configured route so test behavior matches registry/routes. - if route_ids: - return route_ids[0] + if pipeline_ids: + return pipeline_ids[0] return "default" @@ -85,12 +88,10 @@ def _warn_if_policy_disallows_app_id(api_key: str, app_id: str, registry: dict) policy = (registry.get("api_keys") or {}).get(api_key, {}) if not isinstance(policy, dict): return - allowed = policy.get("allowed_route_ids") - if allowed is None: - allowed = policy.get("allowed_pipeline_ids") + allowed = policy.get("allowed_pipeline_ids") if isinstance(allowed, list) and allowed and app_id not in allowed: logger.bind(tag=TAG).warning( - f"app_id='{app_id}' is not in allowed_route_ids for current API key; server may return 403." + f"app_id='{app_id}' is not in allowed_pipeline_ids for current API key; server may return 403." ) @@ -102,7 +103,7 @@ SESSION_ID = str(uuid.uuid4()) dialogue = [ {"role": "system", "content": "You are a helpful assistant."}, - {"role": "user", "content": "who are you"}, + {"role": "user", "content": "你叫什么名字"}, ] call_params = { @@ -141,7 +142,7 @@ def main(): continue # SDK流式为增量覆盖，计算差量输出 if len(current_text) >= len(last_text): - delta = current_text[len(last_text):] + delta = current_text[len(last_text) :] else: # 避免偶发回退 delta = current_text @@ -158,7 +159,7 @@ def main(): logger.bind(tag=TAG).error( f"code={responses.status_code}, message={responses.message}, 请参考文档：https://help.aliyun.com/zh/model-studio/developer-reference/error-code" ) - u = "【阿里百练API服务响应异常】" + u = "【阿里百练API服务响应异常】" else: full_text = getattr(getattr(responses, "output", None), "text", "") logger.bind(tag=TAG).info( @@ -168,11 +169,8 @@ def main(): print("from non-stream: ", u) except Exception as e: logger.bind(tag=TAG).error(f"Error: {e}") - u = "【阿里百练API服务响应异常】" - + u = "【阿里百练API服务响应异常】" if __name__ == "__main__": main() - - diff --git a/tests/test_server_dashscope_routes.py b/tests/test_server_dashscope_routes.py index ebfb42b..30cd37f 100644 --- a/tests/test_server_dashscope_routes.py +++ b/tests/test_server_dashscope_routes.py @@ -80,49 +80,53 @@ def dashscope_server(): proc.wait(timeout=10) -def _post_app_response(base_url: str, route_id: str, body: dict, api_key: str): - url = f"{base_url}/api/v1/apps/{route_id}/sessions/test-session/responses" +def _post_app_response(base_url: str, pipeline_id: str, body: dict, api_key: str): + url = f"{base_url}/api/v1/apps/{pipeline_id}/sessions/test-session/responses" headers = {"Authorization": f"Bearer {api_key}"} resp = requests.post(url, json=body, headers=headers, timeout=20) return resp -def test_route_selected_via_route_id_body(dashscope_server): +def test_pipeline_selected_via_pipeline_id_body(dashscope_server): """ - When client specifies `route_id` in the body, it should be used as the selector + When client specifies `pipeline_id` in the body, it should be used as the selector and surfaced back in the JSON response. """ base_url = dashscope_server api_key = _get_service_api_key() if not api_key: - pytest.skip("FAST_AUTH_KEYS is not set; cannot authenticate against server_dashscope") + pytest.skip( + "FAST_AUTH_KEYS is not set; cannot authenticate against server_dashscope" + ) body = { "input": { "prompt": "hello from xiaozhan", "session_id": "sess-1", }, - "route_id": "xiaozhan", + "pipeline_id": "xiaozhan", "stream": False, } - resp = _post_app_response(base_url, route_id="xiaozhan", body=body, api_key=api_key) + resp = _post_app_response( + base_url, pipeline_id="xiaozhan", body=body, api_key=api_key + ) assert resp.status_code == HTTPStatus.OK, resp.text data = resp.json() - assert data.get("route_id") == "xiaozhan" - # Backward-compat alias should still be present. assert data.get("pipeline_id") == "xiaozhan" assert "text" in data.get("output", {}) -def test_route_selected_via_legacy_pipeline_id_body(dashscope_server): +def test_pipeline_selected_via_pipeline_id_body_blueberry(dashscope_server): """ - Backward compatibility: `pipeline_id` in the body should still act as a - route selector, even though route_id is preferred. + When client specifies `pipeline_id` in the body, it should be used as the selector + and surfaced back in the JSON response. """ base_url = dashscope_server api_key = _get_service_api_key() if not api_key: - pytest.skip("FAST_AUTH_KEYS is not set; cannot authenticate against server_dashscope") + pytest.skip( + "FAST_AUTH_KEYS is not set; cannot authenticate against server_dashscope" + ) body = { "input": { "prompt": "hello from blueberry", @@ -132,17 +136,18 @@ def test_route_selected_via_legacy_pipeline_id_body(dashscope_server): "stream": False, } - resp = _post_app_response(base_url, route_id="blueberry", body=body, api_key=api_key) + resp = _post_app_response( + base_url, pipeline_id="blueberry", body=body, api_key=api_key + ) assert resp.status_code == HTTPStatus.OK, resp.text data = resp.json() - assert data.get("route_id") == "blueberry" assert data.get("pipeline_id") == "blueberry" assert "text" in data.get("output", {}) -def test_route_forbidden_for_api_key_when_not_allowed(dashscope_server): +def test_pipeline_forbidden_for_api_key_when_not_allowed(dashscope_server): """ - API key policy in pipeline_registry should prevent a key from using routes + API key policy in pipeline_registry should prevent a key from using pipelines it is not explicitly allowed to access. """ base_url = dashscope_server @@ -151,17 +156,14 @@ def test_route_forbidden_for_api_key_when_not_allowed(dashscope_server): "prompt": "this should be forbidden", "session_id": "sess-3", }, - "route_id": "blueberry", + "pipeline_id": "blueberry", "stream": False, } # Use a guaranteed-wrong API key so we test 401 behavior regardless of registry config. - resp = _post_app_response(base_url, route_id="blueberry", body=body, api_key="invalid-key-for-test") + resp = _post_app_response( + base_url, pipeline_id="blueberry", body=body, api_key="invalid-key-for-test" + ) assert resp.status_code == HTTPStatus.UNAUTHORIZED data = resp.json() assert data.get("detail") == "Invalid API key" - - - - -