test: fix mysql profile snapshot cleanup

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))