Simon
334 lines
12 KiB
Python
334 lines
12 KiB
Python
# =============================================================================
|
||
# 企微IT智能服务台 — Neo4j 客户端单元测试
|
||
# =============================================================================
|
||
# 说明：测试 Neo4jClient 的连接、CRUD、图遍历功能。
|
||
# 优先使用 testcontainers 启动 Docker Neo4j 容器，
|
||
# 不可用时降级为内存 Mock（memory mock）。
|
||
# =============================================================================
|
||
|
||
import os
|
||
import uuid
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
|
||
from app.models.neo4j_schema import ActionNode, IssueNode, RelationEdge
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 尝试导入 testcontainers，不可用时降级为 MemoryMock
|
||
# --------------------------------------------------------------------------
|
||
try:
|
||
from testcontainers.neo4j import Neo4jContainer # type: ignore
|
||
|
||
HAS_TESTCONTAINERS = True
|
||
except ImportError:
|
||
HAS_TESTCONTAINERS = False
|
||
|
||
|
||
# =============================================================================
|
||
# Fixtures
|
||
# =============================================================================
|
||
|
||
|
||
@pytest.fixture(scope="session")
|
||
def neo4j_container():
|
||
"""启动 Neo4j 测试容器（session 级别，所有测试复用）。
|
||
|
||
如果 testcontainers 不可用（CI 环境无 Docker），则跳过并返回 None。
|
||
"""
|
||
if not HAS_TESTCONTAINERS:
|
||
yield None
|
||
return
|
||
|
||
# 检查是否在 CI 环境（无 Docker）中
|
||
if os.environ.get("CI") and not os.environ.get("DOCKER_HOST"):
|
||
yield None
|
||
return
|
||
|
||
try:
|
||
container = Neo4jContainer(
|
||
image="neo4j:5-enterprise",
|
||
username="neo4j",
|
||
password="test1234",
|
||
)
|
||
container.start()
|
||
yield container
|
||
container.stop()
|
||
except Exception:
|
||
# Docker 不可用，降级到 Mock（generator fixture 必须用 yield，不能用 return）
|
||
yield None
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def neo4j_client(neo4j_container):
|
||
"""提供 Neo4jClient 实例（真实容器或 Mock）。
|
||
|
||
优先使用 Docker Neo4j 容器，不可用时降级为内存 Mock。
|
||
"""
|
||
if neo4j_container is not None:
|
||
from app.services.neo4j_client import Neo4jClient
|
||
|
||
bolt_url = neo4j_container.get_connection_url()
|
||
client = Neo4jClient(
|
||
uri=bolt_url,
|
||
user="neo4j",
|
||
password="test1234",
|
||
database="neo4j",
|
||
)
|
||
await client.initialize()
|
||
yield client
|
||
await client.close()
|
||
else:
|
||
# 降级：使用 Mock
|
||
yield _create_mock_neo4j_client()
|
||
|
||
|
||
def _create_mock_neo4j_client():
|
||
"""创建内存 Mock Neo4jClient，用于无 Docker 环境测试。
|
||
|
||
使用内存字典模拟图数据库，实现基本的 CRUD 语义。
|
||
"""
|
||
from app.services.neo4j_client import Neo4jClient
|
||
|
||
client = Neo4jClient.__new__(Neo4jClient)
|
||
client.uri = "mock://memory"
|
||
client.user = "mock"
|
||
client.password = "mock"
|
||
client.database = "mock"
|
||
client._driver = None # 标记为 Mock 模式
|
||
|
||
# 内存存储
|
||
client._nodes: dict = {} # uuid → dict
|
||
client._rels: list = [] # [(from_uuid, to_uuid, rel_type, props)]
|
||
|
||
async def mock_initialize():
|
||
pass
|
||
|
||
async def mock_close():
|
||
pass
|
||
|
||
async def mock_health_check():
|
||
return True
|
||
|
||
async def mock_create_issue_node(issue: IssueNode) -> IssueNode:
|
||
node_uuid = str(uuid.uuid4())
|
||
now = issue.created_at.isoformat() if issue.created_at else "2026-01-01T00:00:00"
|
||
node = {
|
||
"uuid": node_uuid,
|
||
"name": issue.name,
|
||
"category": issue.category,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
"source_suggestion_id": issue.source_suggestion_id,
|
||
}
|
||
client._nodes[node_uuid] = node
|
||
return IssueNode(**node)
|
||
|
||
async def mock_merge_issue(name: str, category: str, props=None) -> IssueNode:
|
||
# 查找已有节点
|
||
for n in client._nodes.values():
|
||
if n.get("name") == name and n.get("__type") == "Issue":
|
||
n["category"] = category
|
||
return IssueNode(**{k: v for k, v in n.items() if not k.startswith("__")})
|
||
# 创建新节点
|
||
node_uuid = str(uuid.uuid4())
|
||
now = "2026-01-01T00:00:00"
|
||
node = {
|
||
"uuid": node_uuid,
|
||
"name": name,
|
||
"category": category,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
"source_suggestion_id": props.get("source_suggestion_id") if props else None,
|
||
"__type": "Issue",
|
||
}
|
||
client._nodes[node_uuid] = node
|
||
return IssueNode(
|
||
uuid=node_uuid, name=name, category=category,
|
||
source_suggestion_id=props.get("source_suggestion_id") if props else None,
|
||
)
|
||
|
||
async def mock_create_action_node(action: ActionNode) -> ActionNode:
|
||
node_uuid = str(uuid.uuid4())
|
||
now = action.created_at.isoformat() if action.created_at else "2026-01-01T00:00:00"
|
||
node = {
|
||
"uuid": node_uuid,
|
||
"name": action.name,
|
||
"description": action.description,
|
||
"created_at": now,
|
||
"source_suggestion_id": action.source_suggestion_id,
|
||
}
|
||
client._nodes[node_uuid] = node
|
||
return ActionNode(**node)
|
||
|
||
async def mock_merge_action(name: str, props=None) -> ActionNode:
|
||
for n in client._nodes.values():
|
||
if n.get("name") == name and n.get("__type") == "Action":
|
||
return ActionNode(**{k: v for k, v in n.items() if not k.startswith("__")})
|
||
node_uuid = str(uuid.uuid4())
|
||
now = "2026-01-01T00:00:00"
|
||
node = {
|
||
"uuid": node_uuid,
|
||
"name": name,
|
||
"description": (props or {}).get("description", ""),
|
||
"created_at": now,
|
||
"source_suggestion_id": (props or {}).get("source_suggestion_id"),
|
||
"__type": "Action",
|
||
}
|
||
client._nodes[node_uuid] = node
|
||
return ActionNode(
|
||
uuid=node_uuid, name=name,
|
||
description=(props or {}).get("description", ""),
|
||
source_suggestion_id=(props or {}).get("source_suggestion_id"),
|
||
)
|
||
|
||
async def mock_create_relation(from_uuid: str, to_uuid: str, rel: RelationEdge) -> bool:
|
||
client._rels.append((from_uuid, to_uuid, str(rel.type), {"order": rel.order, "weight": rel.weight}))
|
||
return True
|
||
|
||
async def mock_find_issue_by_name(name: str):
|
||
for n in client._nodes.values():
|
||
if n.get("name") == name and n.get("__type") == "Issue":
|
||
return IssueNode(**{k: v for k, v in n.items() if not k.startswith("__")})
|
||
return None
|
||
|
||
async def mock_find_related_issues(node_uuid: str, rel_type=None):
|
||
results = []
|
||
for from_u, to_u, rtype, _ in client._rels:
|
||
if from_u == node_uuid:
|
||
if rel_type is None or rtype == rel_type:
|
||
target = client._nodes.get(to_u)
|
||
if target:
|
||
results.append(IssueNode(**{k: v for k, v in target.items() if not k.startswith("__")}))
|
||
return results
|
||
|
||
client.initialize = mock_initialize
|
||
client.close = mock_close
|
||
client.health_check = mock_health_check
|
||
client.create_issue_node = mock_create_issue_node
|
||
client.merge_issue = mock_merge_issue
|
||
client.create_action_node = mock_create_action_node
|
||
client.merge_action = mock_merge_action
|
||
client.create_relation = mock_create_relation
|
||
client.find_issue_by_name = mock_find_issue_by_name
|
||
client.find_related_issues = mock_find_related_issues
|
||
|
||
return client
|
||
|
||
|
||
# =============================================================================
|
||
# 测试用例
|
||
# =============================================================================
|
||
|
||
|
||
class TestNeo4jHealthCheck:
|
||
"""Neo4j 健康检查测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_health_check(self, neo4j_client):
|
||
"""验证健康检查返回 True。"""
|
||
healthy = await neo4j_client.health_check()
|
||
assert healthy is True
|
||
|
||
|
||
class TestNeo4jIssueCRUD:
|
||
"""Issue 节点 CRUD 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_issue_node(self, neo4j_client):
|
||
"""验证创建 Issue 节点。"""
|
||
issue = IssueNode(name="VPN问题", category="网络")
|
||
result = await neo4j_client.create_issue_node(issue)
|
||
assert result.uuid != ""
|
||
assert result.name == "VPN问题"
|
||
assert result.category == "网络"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_merge_issue_idempotent(self, neo4j_client):
|
||
"""验证 MERGE Issue 幂等性。"""
|
||
# 第一次 merge → 创建
|
||
issue1 = await neo4j_client.merge_issue("测试问题", "软件")
|
||
uuid1 = issue1.uuid
|
||
|
||
# 第二次 merge → 返回已有节点
|
||
issue2 = await neo4j_client.merge_issue("测试问题", "软件")
|
||
assert issue2.uuid == uuid1
|
||
assert issue2.name == "测试问题"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_find_issue_by_name(self, neo4j_client):
|
||
"""验证按名称查找 Issue。"""
|
||
await neo4j_client.merge_issue("查找测试问题", "网络")
|
||
found = await neo4j_client.find_issue_by_name("查找测试问题")
|
||
assert found is not None
|
||
assert found.name == "查找测试问题"
|
||
assert found.category == "网络"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_find_issue_by_name_not_found(self, neo4j_client):
|
||
"""验证查找不存在的 Issue 返回 None。"""
|
||
found = await neo4j_client.find_issue_by_name("不存在的问题XYZ123")
|
||
assert found is None
|
||
|
||
|
||
class TestNeo4jActionCRUD:
|
||
"""Action 节点 CRUD 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_action_node(self, neo4j_client):
|
||
"""验证创建 Action 节点。"""
|
||
action = ActionNode(name="个人VPN开通", description="为员工开通个人VPN")
|
||
result = await neo4j_client.create_action_node(action)
|
||
assert result.uuid != ""
|
||
assert result.name == "个人VPN开通"
|
||
assert result.description == "为员工开通个人VPN"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_merge_action_idempotent(self, neo4j_client):
|
||
"""验证 MERGE Action 幂等性。"""
|
||
action1 = await neo4j_client.merge_action("重置密码", {"description": "帮助员工重置域密码"})
|
||
action2 = await neo4j_client.merge_action("重置密码", {"description": "帮助员工重置域密码"})
|
||
assert action1.uuid == action2.uuid
|
||
|
||
|
||
class TestNeo4jRelation:
|
||
"""关系 CRUD 测试。"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_relation(self, neo4j_client):
|
||
"""验证创建关系。"""
|
||
issue = await neo4j_client.merge_issue("关系测试问题", "硬件")
|
||
action = await neo4j_client.merge_action("关系测试动作")
|
||
|
||
rel = RelationEdge(
|
||
from_uuid=issue.uuid,
|
||
to_uuid=action.uuid,
|
||
type="LEADS_TO",
|
||
order=1,
|
||
weight=1.0,
|
||
)
|
||
result = await neo4j_client.create_relation(
|
||
issue.uuid, action.uuid, rel
|
||
)
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_find_related_issues(self, neo4j_client):
|
||
"""验证查找关联 Issue。"""
|
||
issue1 = await neo4j_client.merge_issue("父问题", "网络")
|
||
issue2 = await neo4j_client.merge_issue("子问题", "网络")
|
||
|
||
rel = RelationEdge(
|
||
from_uuid=issue1.uuid,
|
||
to_uuid=issue2.uuid,
|
||
type="LEADS_TO",
|
||
order=1,
|
||
weight=0.8,
|
||
)
|
||
await neo4j_client.create_relation(issue1.uuid, issue2.uuid, rel)
|
||
|
||
related = await neo4j_client.find_related_issues(issue1.uuid, "LEADS_TO")
|
||
assert len(related) >= 1
|
||
assert any(r.name == "子问题" for r in related)
|