diff --git a/tests/integration/test_customer_profile_candidate_mysql.py b/tests/integration/test_customer_profile_candidate_mysql.py new file mode 100644 index 0000000..6581241 --- /dev/null +++ b/tests/integration/test_customer_profile_candidate_mysql.py @@ -0,0 +1,88 @@ +"""客户画像候选批准到画像快照的真实 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))