Tangledup-ai/lang-agent
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update tests

Browse Source
This commit is contained in:
goulustis
2026-03-04 15:20:52 +08:00
parent 9284e8f928
commit 6ed33f3185
2 changed files with 47 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
tests/test_dashscope_client.py
View File

@@ -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()

50
tests/test_server_dashscope_routes.py
View File

@@ -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"