wehub-resource-sync
234 lines
9.2 KiB
Python
234 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import asyncpg
|
|
import httpx
|
|
import pytest
|
|
|
|
pytestmark = [pytest.mark.asyncio, pytest.mark.e2e, pytest.mark.slow]
|
|
|
|
POLL_INTERVAL_SECONDS = float(os.getenv("E2E_RUN_POLL_INTERVAL_SECONDS", "2"))
|
|
RUN_TIMEOUT_SECONDS = int(os.getenv("E2E_RUN_TIMEOUT_SECONDS", "240"))
|
|
EVAL_EXPECTED_OUTPUT = "AGENT_EVAL_E2E_OK"
|
|
CALL_EXPECTED_OUTPUT = "AGENT_CALL_E2E_OK"
|
|
|
|
|
|
def _postgres_dsn() -> str:
|
|
return os.getenv("POSTGRES_URL", "postgresql+asyncpg://postgres:postgres@postgres:5432/yuxi").replace(
|
|
"+asyncpg", ""
|
|
)
|
|
|
|
|
|
def _json_dict(value: Any) -> dict[str, Any]:
|
|
if isinstance(value, dict):
|
|
return value
|
|
if isinstance(value, str) and value:
|
|
parsed = json.loads(value)
|
|
assert isinstance(parsed, dict), parsed
|
|
return parsed
|
|
return {}
|
|
|
|
|
|
async def _create_agent(client: httpx.AsyncClient, headers: dict[str, str], uid: str) -> str:
|
|
default_response = await client.get("/api/agent/default", headers=headers)
|
|
assert default_response.status_code == 200, default_response.text
|
|
default_context = ((default_response.json().get("agent") or {}).get("config_json") or {}).get("context") or {}
|
|
|
|
slug = f"e2e-agent-call-{uuid.uuid4().hex[:8]}"
|
|
context: dict[str, Any] = {
|
|
"system_prompt": (
|
|
"你是端到端测试专用智能体。不要调用任何工具。如果用户要求输出一个 AGENT_*_E2E_OK 标记，只输出该标记本身。"
|
|
),
|
|
"tools": [],
|
|
"knowledges": [],
|
|
"mcps": [],
|
|
"skills": [],
|
|
"subagents": [],
|
|
}
|
|
if default_context.get("model"):
|
|
context["model"] = default_context["model"]
|
|
|
|
response = await client.post(
|
|
"/api/agent",
|
|
json={
|
|
"name": f"E2E Agent Call {slug[-8:]}",
|
|
"slug": slug,
|
|
"backend_id": "ChatbotAgent",
|
|
"description": "真实 Agent Call/Eval E2E 临时智能体",
|
|
"config_json": {"context": context},
|
|
"share_config": {"access_level": "user", "department_ids": [], "user_uids": [uid]},
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
assert (response.json().get("agent") or {}).get("slug") == slug
|
|
return slug
|
|
|
|
|
|
async def _delete_agent(client: httpx.AsyncClient, headers: dict[str, str], slug: str) -> None:
|
|
response = await client.delete(f"/api/agent/{slug}", headers=headers)
|
|
assert response.status_code in {200, 404}, response.text
|
|
|
|
|
|
async def _cancel_run(client: httpx.AsyncClient, headers: dict[str, str], run_id: str | None) -> None:
|
|
if not run_id:
|
|
return
|
|
response = await client.post(f"/api/agent/runs/{run_id}/cancel", headers=headers)
|
|
assert response.status_code < 500, response.text
|
|
|
|
|
|
async def _wait_agent_call_result(
|
|
client: httpx.AsyncClient,
|
|
headers: dict[str, str],
|
|
*,
|
|
run_id: str,
|
|
agent_slug: str,
|
|
) -> dict[str, Any]:
|
|
deadline = asyncio.get_running_loop().time() + RUN_TIMEOUT_SECONDS
|
|
last_payload: dict[str, Any] | None = None
|
|
|
|
while asyncio.get_running_loop().time() < deadline:
|
|
response = await client.post(
|
|
"/api/agent-invocation/agent-call/runs/result",
|
|
json={"run_id": run_id, "agent_slug": agent_slug},
|
|
headers=headers,
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
last_payload = response.json()
|
|
if last_payload.get("status") in {"completed", "failed", "cancelled", "interrupted"}:
|
|
return last_payload
|
|
await asyncio.sleep(POLL_INTERVAL_SECONDS)
|
|
|
|
pytest.fail("Agent Call run timed out: " + json.dumps(last_payload or {}, ensure_ascii=False))
|
|
|
|
|
|
async def _load_run_metadata(run_id: str) -> dict[str, Any]:
|
|
conn = await asyncpg.connect(_postgres_dsn())
|
|
try:
|
|
row = await conn.fetchrow(
|
|
"""
|
|
SELECT
|
|
ar.id,
|
|
ar.request_id,
|
|
ar.status,
|
|
ar.run_type,
|
|
ar.agent_slug,
|
|
conv.extra_metadata AS conversation_metadata,
|
|
input_msg.extra_metadata AS input_metadata,
|
|
input_msg.content AS input_content
|
|
FROM agent_runs ar
|
|
JOIN conversations conv ON conv.id = ar.conversation_id
|
|
LEFT JOIN messages input_msg ON input_msg.id = ar.input_message_id
|
|
WHERE ar.id = $1
|
|
""",
|
|
run_id,
|
|
)
|
|
assert row, f"agent run row missing for {run_id}"
|
|
return {
|
|
"request_id": row["request_id"],
|
|
"status": row["status"],
|
|
"run_type": row["run_type"],
|
|
"agent_slug": row["agent_slug"],
|
|
"conversation_metadata": _json_dict(row["conversation_metadata"]),
|
|
"input_metadata": _json_dict(row["input_metadata"]),
|
|
"input_content": row["input_content"],
|
|
}
|
|
finally:
|
|
await conn.close()
|
|
|
|
|
|
async def test_agent_eval_and_agent_call_entrypoints_share_run_invocation_flow(
|
|
e2e_client: httpx.AsyncClient,
|
|
e2e_headers: dict[str, str],
|
|
e2e_agent_context: dict[str, str],
|
|
):
|
|
uid = e2e_agent_context["uid"]
|
|
agent_slug = await _create_agent(e2e_client, e2e_headers, uid)
|
|
agent_call_run_id: str | None = None
|
|
agent_call_completed = False
|
|
|
|
try:
|
|
eval_request_id = f"agent-eval-e2e-{uuid.uuid4()}"
|
|
eval_metadata = {
|
|
"dataset_name": "agent-entrypoint-e2e",
|
|
"dataset_item_id": f"item-{uuid.uuid4().hex[:8]}",
|
|
"experiment_name": "agent-entrypoint-e2e",
|
|
}
|
|
eval_response = await e2e_client.post(
|
|
"/api/agent-invocation/eval/runs",
|
|
json={
|
|
"query": f"请只输出 {EVAL_EXPECTED_OUTPUT}，不要添加任何解释。",
|
|
"agent_slug": agent_slug,
|
|
"evaluation": eval_metadata,
|
|
"meta": {"request_id": eval_request_id},
|
|
},
|
|
headers=e2e_headers,
|
|
)
|
|
assert eval_response.status_code == 200, eval_response.text
|
|
eval_payload = eval_response.json()
|
|
assert eval_payload.get("status") == "completed", eval_payload
|
|
assert eval_payload.get("request_id") == eval_request_id
|
|
assert EVAL_EXPECTED_OUTPUT in str(eval_payload.get("output") or ""), eval_payload
|
|
|
|
eval_run_id = eval_payload.get("agent_run_id")
|
|
assert eval_run_id, eval_payload
|
|
eval_run = await _load_run_metadata(str(eval_run_id))
|
|
assert eval_run["status"] == "completed"
|
|
assert eval_run["run_type"] == "chat"
|
|
assert eval_run["conversation_metadata"]["source"] == "agent_evaluation"
|
|
assert eval_run["conversation_metadata"]["agent_invocation_meta"] == {"evaluation": eval_metadata}
|
|
assert eval_run["input_metadata"]["source"] == "agent_evaluation"
|
|
assert eval_run["input_metadata"]["agent_invocation_meta"] == {"evaluation": eval_metadata}
|
|
assert "evaluation" not in eval_run["input_metadata"]
|
|
|
|
agent_call_request_id = f"agent-call-e2e-{uuid.uuid4()}"
|
|
create_response = await e2e_client.post(
|
|
"/api/agent-invocation/agent-call/runs",
|
|
json={
|
|
"agent_slug": agent_slug,
|
|
"messages": [{"role": "user", "content": f"请只输出 {CALL_EXPECTED_OUTPUT}，不要添加任何解释。"}],
|
|
"request_id": agent_call_request_id,
|
|
"async_mode": True,
|
|
},
|
|
headers=e2e_headers,
|
|
)
|
|
assert create_response.status_code == 200, create_response.text
|
|
create_payload = create_response.json()
|
|
agent_call_run_id = create_payload.get("run_id")
|
|
assert agent_call_run_id, create_payload
|
|
assert create_payload.get("request_id") == agent_call_request_id
|
|
assert create_payload.get("status") == "pending"
|
|
|
|
call_payload = await _wait_agent_call_result(
|
|
e2e_client,
|
|
e2e_headers,
|
|
run_id=str(agent_call_run_id),
|
|
agent_slug=agent_slug,
|
|
)
|
|
assert call_payload.get("status") == "completed", call_payload
|
|
assert call_payload.get("request_id") == agent_call_request_id
|
|
assert CALL_EXPECTED_OUTPUT in str(call_payload.get("output") or ""), call_payload
|
|
assert call_payload["choices"][0]["messages"] == [
|
|
{"role": "assistant", "content": call_payload.get("output") or ""}
|
|
]
|
|
assert call_payload["choices"][0]["finish_reason"] == "stop"
|
|
agent_call_completed = True
|
|
|
|
agent_call_run = await _load_run_metadata(str(agent_call_run_id))
|
|
assert agent_call_run["status"] == "completed"
|
|
assert agent_call_run["run_type"] == "chat"
|
|
assert agent_call_run["conversation_metadata"]["source"] == "agent_call"
|
|
assert "agent_invocation_meta" not in agent_call_run["conversation_metadata"]
|
|
assert agent_call_run["input_metadata"]["source"] == "agent_call"
|
|
assert "agent_invocation_meta" not in agent_call_run["input_metadata"]
|
|
assert "custom_variables" not in agent_call_run["input_metadata"]
|
|
finally:
|
|
if not agent_call_completed:
|
|
await _cancel_run(e2e_client, e2e_headers, agent_call_run_id)
|
|
await _delete_agent(e2e_client, e2e_headers, agent_slug)
|