refactor(router): 移除废弃的graph兼容性接口并合并测试

将废弃的lightrag兼容性接口从graph_router.py中移除
删除单独的test_graph_router.py测试文件
在test_unified_graph_router.py中添加相关测试
This commit is contained in:
Wenjie Zhang 2025-12-20 15:13:56 +08:00
parent ac848ea165
commit 3f7e70db31
3 changed files with 12 additions and 74 deletions

View File

@ -209,51 +209,6 @@ async def get_graph_stats(
raise HTTPException(status_code=500, detail=f"Failed to get stats: {str(e)}")
# =============================================================================
# === 兼容性接口 (Deprecated/Compatibility) ===
# =============================================================================
@graph.get("/lightrag/subgraph")
async def get_lightrag_subgraph(
db_id: str = Query(..., description="数据库ID"),
node_label: str = Query(..., description="节点标签或实体名称"),
max_depth: int = Query(2, description="最大深度", ge=1, le=5),
max_nodes: int = Query(100, description="最大节点数", ge=1, le=1000),
current_user: User = Depends(get_admin_user),
):
"""(Deprecated) Use /graph/subgraph instead"""
return await get_subgraph(
db_id=db_id, node_label=node_label, max_depth=max_depth, max_nodes=max_nodes, current_user=current_user
)
@graph.get("/lightrag/databases")
async def get_lightrag_databases(current_user: User = Depends(get_admin_user)):
"""(Deprecated) Use /graph/list instead"""
try:
lightrag_databases = knowledge_base.get_lightrag_databases()
return {"success": True, "data": {"databases": lightrag_databases}}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@graph.get("/lightrag/labels")
async def get_lightrag_labels(
db_id: str = Query(..., description="数据库ID"), current_user: User = Depends(get_admin_user)
):
"""(Deprecated) Use /graph/labels instead"""
return await get_graph_labels(db_id=db_id, current_user=current_user)
@graph.get("/lightrag/stats")
async def get_lightrag_stats(
db_id: str = Query(..., description="数据库ID"), current_user: User = Depends(get_admin_user)
):
"""(Deprecated) Use /graph/stats instead"""
return await get_graph_stats(db_id=db_id, current_user=current_user)
@graph.get("/neo4j/nodes")
async def get_neo4j_nodes(
kgdb_name: str = Query(..., description="知识图谱数据库名称"),

View File

@ -1,29 +0,0 @@
"""
Integration tests for graph router endpoints.
"""
from __future__ import annotations
import pytest
pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
async def test_graph_routes_require_auth(test_client):
response = await test_client.get("/api/graph/lightrag/databases")
assert response.status_code == 401
async def test_standard_user_cannot_access_graph_endpoints(test_client, standard_user):
response = await test_client.get("/api/graph/lightrag/databases", headers=standard_user["headers"])
assert response.status_code == 403
async def test_admin_can_list_lightrag_databases(test_client, admin_headers, knowledge_database):
response = await test_client.get("/api/graph/lightrag/databases", headers=admin_headers)
assert response.status_code == 200, response.text
payload = response.json()
assert payload["success"] is True
databases = payload["data"]["databases"]
assert isinstance(databases, list)
assert any(db["db_id"] == knowledge_database["db_id"] for db in databases)

View File

@ -9,6 +9,18 @@ import pytest
pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
async def test_graph_routes_require_auth(test_client):
"""Test that graph endpoints require authentication."""
response = await test_client.get("/api/graph/list")
assert response.status_code == 401
async def test_standard_user_cannot_access_graph_endpoints(test_client, standard_user):
"""Test that standard users cannot access graph endpoints."""
response = await test_client.get("/api/graph/list", headers=standard_user["headers"])
assert response.status_code == 403
async def test_get_graphs_list(test_client, admin_headers):
"""Test retrieving the list of all graphs."""
response = await test_client.get("/api/graph/list", headers=admin_headers)