89 lines
3.9 KiB
Python
89 lines
3.9 KiB
Python
|
"""客户画像候选批准到画像快照的真实 MySQL 集成验证。"""
|
||
|
|
|
||
|
|
from datetime import UTC, datetime
|
||
|
|
from uuid import uuid4
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from sqlalchemy import delete, select
|
||
|
|
|
||
|
|
from app.infrastructure.db import SessionFactory
|
||
|
|
from app.model.memory import MemoryConflict, MemorySyncOutbox, MemoryUnit, ProfileSnapshot
|
||
|
|
from app.service.customer_profile_candidate_service import CustomerProfileCandidateService
|
||
|
|
|
||
|
|
pytestmark = pytest.mark.integration
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_approved_candidate_creates_current_snapshot_and_outbox(
|
||
|
|
monkeypatch: pytest.MonkeyPatch,
|
||
|
|
) -> None:
|
||
|
|
"""批准候选必须原子生成当前画像快照和两个投影事件。"""
|
||
|
|
customer_id = uuid4().int % 10**15 + 10**15
|
||
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
||
|
|
async with SessionFactory() as session, session.begin():
|
||
|
|
candidate = MemoryUnit(
|
||
|
|
memory_uuid=str(uuid4()), customer_id=customer_id,
|
||
|
|
memory_key="preference:risk_level", content="稳健型",
|
||
|
|
memory_type="preference", source_type="AI对话提取",
|
||
|
|
source_confidence=0.9, confidence=0.9, evidence_count=1,
|
||
|
|
conflict_count=0, recall_count=0, status="verified", valid_from=now,
|
||
|
|
version=1, created_at=now, updated_at=now,
|
||
|
|
)
|
||
|
|
session.add(candidate)
|
||
|
|
await session.flush()
|
||
|
|
candidate_id = int(candidate.id)
|
||
|
|
monkeypatch.setattr(
|
||
|
|
"app.service.customer_profile_candidate_service.get_memory_cache_adapter",
|
||
|
|
lambda: None,
|
||
|
|
)
|
||
|
|
try:
|
||
|
|
async with SessionFactory() as session, session.begin():
|
||
|
|
target = await session.scalar(
|
||
|
|
select(MemoryUnit).where(MemoryUnit.id == candidate_id).with_for_update()
|
||
|
|
)
|
||
|
|
assert target is not None
|
||
|
|
await CustomerProfileCandidateService()._promote(session, target, reviewer_id=9003)
|
||
|
|
|
||
|
|
async with SessionFactory() as session:
|
||
|
|
snapshot = await session.scalar(
|
||
|
|
select(ProfileSnapshot).where(
|
||
|
|
ProfileSnapshot.customer_id == customer_id,
|
||
|
|
ProfileSnapshot.current_customer_id == customer_id,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
assert snapshot is not None
|
||
|
|
assert snapshot.version == 1
|
||
|
|
assert snapshot.snapshot["customer_service_preferences"][
|
||
|
|
"preference:risk_level"
|
||
|
|
]["value"] == "稳健型"
|
||
|
|
events = list(await session.scalars(
|
||
|
|
select(MemorySyncOutbox).where(
|
||
|
|
MemorySyncOutbox.aggregate_uuid == snapshot.profile_uuid
|
||
|
|
)
|
||
|
|
))
|
||
|
|
assert {event.target_store for event in events} == {"milvus", "neo4j"}
|
||
|
|
finally:
|
||
|
|
async with SessionFactory() as session, session.begin():
|
||
|
|
# Outbox uses the snapshot UUID as an aggregate reference; remove it first.
|
||
|
|
await session.execute(delete(MemorySyncOutbox).where(
|
||
|
|
MemorySyncOutbox.aggregate_uuid.in_(
|
||
|
|
select(ProfileSnapshot.profile_uuid).where(
|
||
|
|
ProfileSnapshot.customer_id == customer_id
|
||
|
|
)
|
||
|
|
)
|
||
|
|
))
|
||
|
|
# Delete snapshots by customer directly to avoid MySQL error 1093
|
||
|
|
# (target table referenced by its own subquery).
|
||
|
|
await session.execute(delete(ProfileSnapshot).where(
|
||
|
|
ProfileSnapshot.customer_id == customer_id
|
||
|
|
))
|
||
|
|
await session.execute(delete(MemoryConflict).where(
|
||
|
|
MemoryConflict.left_memory_id.in_(
|
||
|
|
select(MemoryUnit.id).where(MemoryUnit.customer_id == customer_id)
|
||
|
|
)
|
||
|
|
| MemoryConflict.right_memory_id.in_(
|
||
|
|
select(MemoryUnit.id).where(MemoryUnit.customer_id == customer_id)
|
||
|
|
)
|
||
|
|
))
|
||
|
|
await session.execute(delete(MemoryUnit).where(MemoryUnit.customer_id == customer_id))
|