96 lines
4.8 KiB
Python
96 lines
4.8 KiB
Python
"""Contract coverage for the template-agent lifecycle compatibility surface."""
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _read(relative_path: str) -> str:
|
|
"""Load a repository file as text."""
|
|
return (ROOT / relative_path).read_text(encoding="utf-8")
|
|
|
|
|
|
def test_template_agent_lifecycle_routes_are_registered():
|
|
"""The HM-facing template-agent lifecycle endpoints should exist."""
|
|
app_source = _read("app.py")
|
|
|
|
assert 'class AgentLifecycleResponse(BaseModel):' in app_source
|
|
assert '@app.get("/agents/{agent_name}", response_model=AgentLifecycleResponse)' in app_source
|
|
assert '@app.post("/agents/{agent_name}/stop", response_model=MessageResponse)' in app_source
|
|
assert '"runtime_id": resolved_name' in app_source
|
|
assert '"runtime_status": runtime_status' in app_source
|
|
assert '"subdomain": subdomain' in app_source
|
|
|
|
|
|
def test_create_response_exposes_hm_friendly_alias_fields():
|
|
"""POST /agents should emit the alias fields that HM already knows how to parse."""
|
|
app_source = _read("app.py")
|
|
|
|
assert "runtime_id: Optional[str] = None" in app_source
|
|
assert "agent_id: Optional[str] = None" in app_source
|
|
assert 'result["runtime_id"] = result["name"]' in app_source
|
|
assert 'result["agent_id"] = result["name"]' in app_source
|
|
assert 'result["runtime_status"] = _normalize_agent_runtime_status(result.get("status"))' in app_source
|
|
assert 'result["subdomain"] = access_info.get("domain") or access_info.get("external_ip")' in app_source
|
|
|
|
|
|
def test_delete_endpoint_queries_db_before_openclaw_branch():
|
|
"""DELETE /agents/{id} should not reference db_agent before it is loaded."""
|
|
app_source = _read("app.py")
|
|
delete_start = app_source.index('@app.delete("/agents/{agent_name}", response_model=MessageResponse)')
|
|
delete_section = app_source[delete_start:]
|
|
|
|
assert 'db_agent = db.query(Agent).filter(Agent.name == agent_name).first()' in delete_section
|
|
assert delete_section.index('db_agent = db.query(Agent).filter(Agent.name == agent_name).first()') < delete_section.index(
|
|
"if db_agent and db_agent.agent_framework:"
|
|
)
|
|
|
|
|
|
def test_template_agent_doc_mentions_new_lifecycle_contract():
|
|
"""The coding A2A doc should describe the new lifecycle contract for HM."""
|
|
doc_source = _read("docs/CODING_A2A_AGENT_CREATE_AND_INVOKE.md")
|
|
|
|
assert "## 3.1 生命周期接口" in doc_source
|
|
assert "GET /agents/{agent_name}" in doc_source
|
|
assert "POST /agents/{agent_name}/stop" in doc_source
|
|
assert "`runtime_id` / `agent_id` / `id`" in doc_source
|
|
|
|
|
|
def test_integration_doc_mentions_template_agent_runtime_contract():
|
|
"""The main integration doc should include the /agents lifecycle compatibility APIs."""
|
|
doc_source = _read("docs/HEICODE_API_INTEGRATION.md")
|
|
|
|
assert "### 3.11 模板 Agent Runtime 兼容接口" in doc_source
|
|
assert "| `POST` | `/agents` | 创建模板 Agent；返回 HM 可直接解析的实例标识和访问地址别名字段 |" in doc_source
|
|
assert "| `GET` | `/agents/{agent_name}` | 查询模板 Agent 生命周期状态；返回平铺 `status` / `runtime_status` / `state` |" in doc_source
|
|
assert "| `POST` | `/agents/{agent_name}/stop` | 幂等停止模板 Agent；停止运行 Pod，但保留数据库记录 |" in doc_source
|
|
assert "删除接口当前已修复模板 Agent 场景下的数据库变量引用问题" in doc_source
|
|
|
|
|
|
def test_a2a_servers_enforce_agent_access_token_contract():
|
|
"""A2A servers should support local X-Agent-Access-Token validation."""
|
|
coding_server_source = _read("agent_templates/agents/coding_a2a_agent/a2a_server.py")
|
|
litellm_server_source = _read("agent_templates/agents/a2a_litellm_agent/a2a_server.py")
|
|
|
|
for source in (coding_server_source, litellm_server_source):
|
|
assert 'AGENT_ACCESS_TOKEN = os.getenv("AGENT_ACCESS_TOKEN", "")' in source
|
|
assert 'AGENT_ACCESS_HEADER = "X-Agent-Access-Token"' in source
|
|
assert "secrets.compare_digest" in source
|
|
assert 'status_code=401' in source
|
|
assert 'status_code=403' in source
|
|
assert 'request.headers.get(AGENT_ACCESS_HEADER, "")' in source
|
|
|
|
|
|
def test_a2a_docs_describe_agent_access_token_header():
|
|
"""Template-agent docs should explain the local access-token authentication flow."""
|
|
coding_doc_source = _read("docs/CODING_A2A_AGENT_CREATE_AND_INVOKE.md")
|
|
integration_doc_source = _read("docs/HEICODE_API_INTEGRATION.md")
|
|
|
|
assert "AGENT_ACCESS_TOKEN" in coding_doc_source
|
|
assert "X-Agent-Access-Token" in coding_doc_source
|
|
assert "缺少请求头时返回 `401`" in coding_doc_source
|
|
assert "请求头不匹配时返回 `403`" in coding_doc_source
|
|
assert "客户端直连鉴权" in integration_doc_source
|
|
assert "X-Agent-Access-Token == AGENT_ACCESS_TOKEN" in integration_doc_source
|