93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.infrastructure.neo4j_profile_projection import Neo4jProfileProjection
|
|
|
|
|
|
class FakeDriver:
|
|
def __init__(self, *, applied: bool = True) -> None:
|
|
self.applied = applied
|
|
self.calls: list[tuple[str, dict[str, object]]] = []
|
|
|
|
async def execute_query(self, query: str, **parameters: object) -> SimpleNamespace:
|
|
self.calls.append((query, parameters))
|
|
return SimpleNamespace(records=[{"applied": True}] if self.applied else [])
|
|
|
|
|
|
def payload() -> dict[str, object]:
|
|
return {
|
|
"customer_id": 7,
|
|
"profile_uuid": "profile-7-v1",
|
|
"profile_version": 1,
|
|
"memory_sources": [
|
|
{
|
|
"memory_uuid": "memory-1",
|
|
"memory_key": "preference:risk_level",
|
|
"content": "稳健型",
|
|
"memory_type": "preference",
|
|
"confidence": 0.9,
|
|
"version": 2,
|
|
"valid_until": None,
|
|
},
|
|
{
|
|
"memory_uuid": "memory-2",
|
|
"memory_key": "goal:liquidity",
|
|
"content": "保持流动性",
|
|
"memory_type": "goal",
|
|
"confidence": 0.8,
|
|
"version": 1,
|
|
"valid_until": None,
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_projects_only_fixed_preference_and_goal_queries() -> None:
|
|
driver = FakeDriver()
|
|
result = await Neo4jProfileProjection(driver).upsert(payload())
|
|
|
|
assert result.applied is True
|
|
assert len(driver.calls) == 3
|
|
assert "MERGE (c:Customer" in driver.calls[0][0]
|
|
assert "PREFERS" in driver.calls[1][0]
|
|
assert "HAS_GOAL" in driver.calls[2][0]
|
|
assert driver.calls[1][1]["items"][0]["memory_uuid"] == "memory-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lower_profile_version_is_skipped_without_writes() -> None:
|
|
driver = FakeDriver(applied=False)
|
|
result = await Neo4jProfileProjection(driver).upsert(payload())
|
|
|
|
assert result.applied is False
|
|
assert result.reason == "newer_profile_version_exists"
|
|
assert len(driver.calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sensitive_memory_content_is_redacted_before_projection() -> None:
|
|
data = payload()
|
|
source = data["memory_sources"][0]
|
|
assert isinstance(source, dict)
|
|
source["content"] = "我的密码是123456，手机号13800138000"
|
|
driver = FakeDriver()
|
|
|
|
await Neo4jProfileProjection(driver).upsert(data)
|
|
|
|
projected = driver.calls[1][1]["items"][0]["content"]
|
|
assert "123456" not in projected
|
|
assert "13800138000" not in projected
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_memory_key_is_rejected() -> None:
|
|
data = payload()
|
|
source = data["memory_sources"][0]
|
|
assert isinstance(source, dict)
|
|
source["memory_key"] = "account:balance"
|
|
|
|
with pytest.raises(ValueError, match="not projectable"):
|
|
await Neo4jProfileProjection(FakeDriver()).upsert(data)
|