X-Financial/server/tests/test_expense_application_memory.py

from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from decimal import Decimal

import pytest
from auth_helpers import install_legacy_header_auth_override
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker

from app.api.deps import CurrentUserContext, get_db
from app.main import create_app
from app.models.ai_learning import AIDecision, AIDecisionFeedback, WorkflowOutcome
from app.models.ai_memory import MemoryEntry, MemoryEvidenceLink
from app.models.expense_case import BusinessEvent, ExpenseCase
from app.models.financial_record import ExpenseClaim
from app.schemas.expense_application_preview import ExpenseApplicationPreviewDecisionCreate
from app.schemas.ontology import OntologyParseResult, OntologyPermission
from app.schemas.orchestrator import OrchestratorRequest
from app.schemas.reimbursement import ExpenseApplicationPreviewActionPayload
from app.services.expense_application_memory import (
    MEMORY_CANDIDATE_TTL,
    ExpenseApplicationMemoryService,
)
from app.services.expense_application_preview_workflow import (
    ExpenseApplicationPreviewWorkflow,
)
from app.services.orchestrator_expense_application_workflow import (
    OrchestratorExpenseApplicationWorkflow,
)
from app.test_helpers.db import build_in_memory_session_factory


def _user(
    *,
    tenant_id: str = "tenant-memory",
    employee_id: str = "employee-memory-owner",
) -> CurrentUserContext:
    return CurrentUserContext(
        username=f"{employee_id}@example.com",
        name="记忆测试员工",
        role_codes=["user"],
        is_admin=False,
        tenant_id=tenant_id,
        employee_id=employee_id,
        employee_no="E-MEMORY-001",
        department_name="交付部",
        position="实施顾问",
        grade="P4",
        auth_session_id=f"session-{employee_id}",
    )


def _seed_learning_evidence(
    db: Session,
    *,
    current_user: CurrentUserContext,
    index: int,
    transport_mode: str = "火车",
    evidence_at: datetime | None = None,
    approved: bool = False,
    expect_memory_entry: bool = True,
) -> tuple[
    MemoryEntry | None,
    ExpenseCase,
    AIDecision,
    AIDecisionFeedback,
    WorkflowOutcome,
]:
    tenant_id = current_user.tenant_id
    suffix = f"{tenant_id}-{current_user.employee_id}-{index}-{uuid.uuid4().hex[:6]}"
    occurred_at = evidence_at or datetime.now(UTC)
    claim = ExpenseClaim(
        id=str(uuid.uuid4()),
        claim_no=f"CLM-{suffix}"[:50],
        employee_name=current_user.name,
        department_name=current_user.department_name,
        expense_type="travel",
        reason="客户现场实施",
        location="上海",
        amount=Decimal("1800.00"),
        occurred_at=occurred_at,
        submitted_at=occurred_at,
        status="submitted",
        approval_stage="审批中",
        risk_flags_json=[],
    )
    expense_case = ExpenseCase(
        id=str(uuid.uuid4()),
        tenant_id=tenant_id,
        case_no=f"CASE-{suffix}"[:80],
        scene_code="travel",
        title="差旅申请",
        owner_employee_id=current_user.employee_id,
        current_stage="application",
        status="active",
    )
    submitted_event = BusinessEvent(
        id=str(uuid.uuid4()),
        tenant_id=tenant_id,
        expense_case_id=expense_case.id,
        aggregate_type="expense_claim",
        aggregate_id=claim.id,
        event_type="application_submitted",
        event_version=1,
        idempotency_key=f"submitted:{suffix}"[:120],
        correlation_id=f"corr:{suffix}"[:64],
        actor_id=current_user.employee_id,
        actor_type="user",
        payload_json={},
        delivery_status="pending",
        occurred_at=occurred_at,
    )
    db.add_all([claim, expense_case, submitted_event])
    db.flush()

    decision = AIDecision(
        id=str(uuid.uuid4()),
        tenant_id=tenant_id,
        expense_case_id=expense_case.id,
        business_event_id=submitted_event.id,
        expense_claim_id=claim.id,
        correlation_id=f"corr:{suffix}"[:64],
        subject_type="expense_claim",
        subject_id=claim.id,
        decision_type="expense_application_submit",
        decision_source="server_preview",
        status="edited",
        automation_mode="human_confirmed",
        confidence=Decimal("1.0000"),
        suggestion_json={},
        evidence_json={},
        version_json={},
        schema_version=1,
        training_eligible=True,
        idempotency_key=f"decision:{suffix}"[:120],
        content_fingerprint=f"sha256:{uuid.uuid4().hex}",
    )
    feedback = AIDecisionFeedback(
        id=str(uuid.uuid4()),
        tenant_id=tenant_id,
        decision_id=decision.id,
        expense_claim_id=claim.id,
        correlation_id=f"corr:{suffix}"[:64],
        feedback_type="edited",
        action_type="submit",
        actor_id=current_user.employee_id,
        actor_type="user",
        evidence_source="server_preview_action",
        verification_status="server_verified",
        training_eligible=True,
        final_value_json={"transport_mode": transport_mode},
        changed_fields_json=[{"field_key": "transport_mode"}],
        idempotency_key=f"feedback:{suffix}"[:120],
        content_fingerprint=f"sha256:{uuid.uuid4().hex}",
    )
    outcome = WorkflowOutcome(
        id=str(uuid.uuid4()),
        tenant_id=tenant_id,
        expense_case_id=expense_case.id,
        decision_id=decision.id,
        business_event_id=submitted_event.id,
        expense_claim_id=claim.id,
        correlation_id=f"corr:{suffix}"[:64],
        outcome_type="application_submitted",
        outcome_status="verified",
        actor_id=current_user.employee_id,
        actor_type="user",
        result_json={},
        idempotency_key=f"outcome:{suffix}"[:120],
        content_fingerprint=f"sha256:{uuid.uuid4().hex}",
        effective_at=occurred_at,
    )
    db.add_all([decision, feedback, outcome])
    if approved:
        db.add(
            BusinessEvent(
                id=str(uuid.uuid4()),
                tenant_id=tenant_id,
                expense_case_id=expense_case.id,
                aggregate_type="expense_claim",
                aggregate_id=claim.id,
                event_type="application_approved",
                event_version=1,
                idempotency_key=f"approved:{suffix}"[:120],
                correlation_id=f"approval:{suffix}"[:64],
                actor_id="approver",
                actor_type="user",
                payload_json={},
                delivery_status="pending",
                occurred_at=occurred_at + timedelta(hours=1),
            )
        )
    db.flush()

    receipt = ExpenseApplicationMemoryService(db).record_transport_edit_evidence(
        current_user=current_user,
        decision=decision,
        feedback=feedback,
        outcome=outcome,
        claim=claim,
        transport_mode=transport_mode,
    )
    if not expect_memory_entry:
        assert receipt is None
        return None, expense_case, decision, feedback, outcome
    assert receipt is not None
    evidence = db.get(MemoryEvidenceLink, receipt.evidence_id)
    assert evidence is not None
    evidence.created_at = occurred_at
    db.flush()
    entry = db.get(MemoryEntry, receipt.memory_id)
    assert entry is not None
    return entry, expense_case, decision, feedback, outcome


def _seed_active_memory(
    db: Session,
    *,
    current_user: CurrentUserContext,
    transport_mode: str = "火车",
) -> tuple[MemoryEntry, list[ExpenseCase]]:
    now = datetime.now(UTC)
    cases: list[ExpenseCase] = []
    entry: MemoryEntry | None = None
    for index, (days_ago, approved) in enumerate(((10, True), (5, True), (0, False)), 1):
        entry, expense_case, *_ = _seed_learning_evidence(
            db,
            current_user=current_user,
            index=index,
            transport_mode=transport_mode,
            evidence_at=now - timedelta(days=days_ago),
            approved=approved,
        )
        cases.append(expense_case)
    facts: dict[str, object] = {}
    applications = ExpenseApplicationMemoryService(db).apply_active_transport_memory(
        facts,
        current_user,
    )
    assert entry is not None
    assert applications and facts["transport_mode"] == transport_mode
    db.flush()
    db.refresh(entry)
    assert entry.status == "active"
    return entry, cases


def _add_returned_event(
    db: Session,
    *,
    current_user: CurrentUserContext,
    expense_case: ExpenseCase,
) -> None:
    approved_at = db.scalar(
        select(BusinessEvent.occurred_at).where(
            BusinessEvent.expense_case_id == expense_case.id,
            BusinessEvent.event_type == "application_approved",
        )
    )
    assert approved_at is not None
    db.add(
        BusinessEvent(
            id=str(uuid.uuid4()),
            tenant_id=current_user.tenant_id,
            expense_case_id=expense_case.id,
            aggregate_type="expense_claim",
            aggregate_id="returned-claim",
            event_type="application_returned",
            event_version=1,
            idempotency_key=f"returned:{uuid.uuid4().hex}",
            correlation_id=f"return:{uuid.uuid4().hex}"[:64],
            actor_id="approver",
            actor_type="user",
            payload_json={},
            delivery_status="pending",
            # 同一时间戳也必须由退回事件保守覆盖批准，不能依赖随机 UUID 排序。
            occurred_at=approved_at,
        )
    )
    db.flush()


def test_memory_activates_after_three_cases_two_approvals_and_seven_days() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        entry, _ = _seed_active_memory(db, current_user=current_user)

        assert entry.evidence_count == 3
        assert entry.approved_evidence_count == 2
        assert entry.active_expires_at is not None
        assert entry.activated_at is not None

        explicit_facts = {"transport_mode": "汽车"}
        assert (
            ExpenseApplicationMemoryService(db).apply_active_transport_memory(
                explicit_facts,
                current_user,
            )
            == []
        )
        assert explicit_facts["transport_mode"] == "汽车"


def test_latest_returned_event_reverses_approval_and_suppresses_active_memory() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        entry, cases = _seed_active_memory(db, current_user=current_user)
        _add_returned_event(db, current_user=current_user, expense_case=cases[1])

        facts: dict[str, object] = {}
        assert (
            ExpenseApplicationMemoryService(db).apply_active_transport_memory(
                facts,
                current_user,
            )
            == []
        )
        db.refresh(entry)
        assert entry.approved_evidence_count == 1
        assert entry.status == "suppressed"
        assert entry.suppressed_at is not None
        assert facts == {}


def test_invalidated_feedback_and_reversed_outcome_remove_active_evidence() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        entry, _ = _seed_active_memory(db, current_user=current_user)
        evidence = list(
            db.scalars(
                select(MemoryEvidenceLink)
                .where(MemoryEvidenceLink.memory_entry_id == entry.id)
                .order_by(MemoryEvidenceLink.created_at.asc())
            ).all()
        )
        assert len(evidence) == 3
        invalidated_feedback = db.get(AIDecisionFeedback, evidence[0].feedback_id)
        reversed_outcome = db.get(WorkflowOutcome, evidence[1].outcome_id)
        assert invalidated_feedback is not None and reversed_outcome is not None
        invalidated_feedback.verification_status = "invalidated"
        invalidated_feedback.training_eligible = False
        reversed_outcome.outcome_status = "reversed"
        db.flush()

        assert ExpenseApplicationMemoryService(db).apply_active_transport_memory(
            {}, current_user
        ) == []
        db.refresh(entry)
        assert entry.status == "suppressed"
        assert entry.evidence_count == 1
        assert entry.approved_evidence_count == 0


def test_opposite_evidence_immediately_suppresses_old_active_memory() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        old_entry, _ = _seed_active_memory(db, current_user=current_user)

        new_entry, *_ = _seed_learning_evidence(
            db,
            current_user=current_user,
            index=4,
            transport_mode="飞机",
        )
        db.refresh(old_entry)
        assert old_entry.status == "suppressed"
        assert old_entry.suppressed_at is not None
        assert new_entry.status == "candidate"
        assert ExpenseApplicationMemoryService(db).apply_active_transport_memory(
            {}, current_user
        ) == []


def test_non_whitelisted_correction_suppresses_old_active_without_storing_value() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        old_entry, _ = _seed_active_memory(db, current_user=current_user)
        entry_count = len(list(db.scalars(select(MemoryEntry)).all()))

        new_entry, *_ = _seed_learning_evidence(
            db,
            current_user=current_user,
            index=4,
            transport_mode="汽车",
            expect_memory_entry=False,
        )
        assert new_entry is None
        db.refresh(old_entry)
        assert old_entry.status == "suppressed"
        assert old_entry.suppressed_at is not None
        assert len(list(db.scalars(select(MemoryEntry)).all())) == entry_count
        assert ExpenseApplicationMemoryService(db).apply_active_transport_memory(
            {}, current_user
        ) == []


def test_expiry_sets_expired_at_and_replay_does_not_extend_ttl() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        entry, _, decision, feedback, outcome = _seed_learning_evidence(
            db,
            current_user=current_user,
            index=1,
        )
        original_expiry = datetime.now(UTC) + timedelta(days=30)
        entry.candidate_expires_at = original_expiry
        db.flush()
        claim = db.get(ExpenseClaim, decision.expense_claim_id)
        assert claim is not None

        replay = ExpenseApplicationMemoryService(db).record_transport_edit_evidence(
            current_user=current_user,
            decision=decision,
            feedback=feedback,
            outcome=outcome,
            claim=claim,
            transport_mode="火车",
        )
        db.refresh(entry)
        assert replay is not None
        assert entry.evidence_count == 1
        assert entry.candidate_expires_at == original_expiry.replace(tzinfo=None)

        old_created_at = datetime.now(UTC) - MEMORY_CANDIDATE_TTL - timedelta(days=2)
        entry.created_at = old_created_at
        entry.candidate_expires_at = datetime.now(UTC) - timedelta(days=1)
        db.flush()
        memories = ExpenseApplicationMemoryService(db).list_current_user_memories(
            current_user
        )
        db.refresh(entry)
        assert memories.items[0].status == "expired"
        assert entry.status == "expired"
        assert entry.expired_at is not None


def test_expired_entry_gets_new_generation_for_new_evidence() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        old_entry, *_ = _seed_learning_evidence(
            db,
            current_user=current_user,
            index=1,
        )
        old_entry.created_at = datetime.now(UTC) - MEMORY_CANDIDATE_TTL - timedelta(days=2)
        old_entry.candidate_expires_at = datetime.now(UTC) - timedelta(days=1)
        db.flush()

        new_entry, *_ = _seed_learning_evidence(
            db,
            current_user=current_user,
            index=2,
        )
        db.refresh(old_entry)
        assert new_entry.id != old_entry.id
        assert new_entry.generation == old_entry.generation + 1
        assert new_entry.status == "candidate"
        assert old_entry.status == "expired"
        assert old_entry.expired_at is not None


def test_replay_after_revoke_or_suppression_never_recreates_memory() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        revoked_entry, _, decision, feedback, outcome = _seed_learning_evidence(
            db,
            current_user=current_user,
            index=1,
        )
        claim = db.get(ExpenseClaim, decision.expense_claim_id)
        assert claim is not None
        assert ExpenseApplicationMemoryService(db).revoke_current_user_memory(
            revoked_entry.id,
            current_user,
        ) is not None
        entry_count = len(list(db.scalars(select(MemoryEntry)).all()))

        revoked_replay = ExpenseApplicationMemoryService(db).record_transport_edit_evidence(
            current_user=current_user,
            decision=decision,
            feedback=feedback,
            outcome=outcome,
            claim=claim,
            transport_mode="火车",
        )
        assert revoked_replay is not None
        assert revoked_replay.status == "revoked"
        assert len(list(db.scalars(select(MemoryEntry)).all())) == entry_count

    with build_in_memory_session_factory()() as db:
        current_user = _user()
        suppressed_entry, _ = _seed_active_memory(db, current_user=current_user)
        evidence = db.scalar(
            select(MemoryEvidenceLink)
            .where(MemoryEvidenceLink.memory_entry_id == suppressed_entry.id)
            .order_by(MemoryEvidenceLink.created_at.asc())
        )
        assert evidence is not None
        decision = db.get(AIDecision, evidence.decision_id)
        feedback = db.get(AIDecisionFeedback, evidence.feedback_id)
        outcome = db.get(WorkflowOutcome, evidence.outcome_id)
        assert decision is not None and feedback is not None and outcome is not None
        claim = db.get(ExpenseClaim, decision.expense_claim_id)
        assert claim is not None
        _seed_learning_evidence(
            db,
            current_user=current_user,
            index=4,
            transport_mode="飞机",
        )
        db.refresh(suppressed_entry)
        assert suppressed_entry.status == "suppressed"
        entry_count = len(list(db.scalars(select(MemoryEntry)).all()))

        suppressed_replay = ExpenseApplicationMemoryService(
            db
        ).record_transport_edit_evidence(
            current_user=current_user,
            decision=decision,
            feedback=feedback,
            outcome=outcome,
            claim=claim,
            transport_mode="火车",
        )
        assert suppressed_replay is not None
        assert suppressed_replay.status == "suppressed"
        assert len(list(db.scalars(select(MemoryEntry)).all())) == entry_count


def test_memory_isolated_by_tenant_and_owner_and_revoke_clears_value() -> None:
    with build_in_memory_session_factory()() as db:
        owner = _user()
        entry, _ = _seed_active_memory(db, current_user=owner)
        service = ExpenseApplicationMemoryService(db)

        assert service.list_current_user_memories(_user(employee_id="employee-other")).items == []
        assert service.list_current_user_memories(
            _user(tenant_id="tenant-other")
        ).items == []
        assert service.revoke_current_user_memory(
            entry.id,
            _user(employee_id="employee-other"),
        ) is None

        revoked = service.revoke_current_user_memory(entry.id, owner)
        assert revoked is not None
        db.refresh(entry)
        assert entry.status == "revoked"
        assert entry.value_json == {}
        assert entry.value_fingerprint == ""
        assert service.apply_active_transport_memory({}, owner) == []


def test_memory_rejects_same_tenant_evidence_from_another_owner() -> None:
    with build_in_memory_session_factory()() as db:
        owner = _user()
        _, _, decision, feedback, outcome = _seed_learning_evidence(
            db,
            current_user=owner,
            index=1,
        )
        claim = db.get(ExpenseClaim, decision.expense_claim_id)
        assert claim is not None

        other_owner = _user(employee_id="employee-other")
        with pytest.raises(PermissionError, match="当前登录人"):
            ExpenseApplicationMemoryService(db).record_transport_edit_evidence(
                current_user=other_owner,
                decision=decision,
                feedback=feedback,
                outcome=outcome,
                claim=claim,
                transport_mode="火车",
            )
        assert (
            ExpenseApplicationMemoryService(db)
            .list_current_user_memories(other_owner)
            .items
            == []
        )


@pytest.mark.parametrize(
    ("field_name", "invalid_value"),
    [
        ("event_type", "claim_draft_updated"),
        ("aggregate_type", "expense_case"),
        ("aggregate_id", "another-claim"),
    ],
)
def test_memory_rejects_non_submit_or_wrong_claim_business_event(
    field_name: str,
    invalid_value: str,
) -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        _, _, decision, feedback, outcome = _seed_learning_evidence(
            db,
            current_user=current_user,
            index=1,
        )
        business_event = db.get(BusinessEvent, decision.business_event_id)
        claim = db.get(ExpenseClaim, decision.expense_claim_id)
        assert business_event is not None
        assert claim is not None
        setattr(business_event, field_name, invalid_value)
        db.flush()

        with pytest.raises(ValueError, match="提交业务事件语义不一致"):
            ExpenseApplicationMemoryService(db).record_transport_edit_evidence(
                current_user=current_user,
                decision=decision,
                feedback=feedback,
                outcome=outcome,
                claim=claim,
                transport_mode="火车",
            )


def test_memory_read_failure_degrades_without_mutating_facts(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    with build_in_memory_session_factory()() as db:
        service = ExpenseApplicationMemoryService(db)
        facts: dict[str, object] = {}

        def fail_resolve(_current_user: CurrentUserContext) -> MemoryEntry | None:
            raise RuntimeError("memory unavailable")

        monkeypatch.setattr(service, "_resolve_active_entry", fail_resolve)
        assert service.apply_active_transport_memory(facts, _user()) == []
        assert facts == {}


def test_preview_and_orchestrator_apply_active_memory() -> None:
    with build_in_memory_session_factory()() as db:
        current_user = _user()
        _seed_active_memory(db, current_user=current_user)

        preview = ExpenseApplicationPreviewWorkflow(db).issue(
            ExpenseApplicationPreviewDecisionCreate(
                message=(
                    "申请时间：2026-07-20 至 2026-07-22\n"
                    "地点：上海\n事由：客户现场实施\n天数：3天"
                ),
                conversation_id="conversation-memory-preview",
                request_id="request-memory-preview",
            ),
            current_user,
        )
        assert preview.application_preview["fields"]["transportMode"] == "火车"
        assert preview.application_preview["fields"]["amount"]
        assert preview.application_preview["fields"]["transportEstimatedAmount"]
        assert "交通" in preview.application_preview["fields"]["policyEstimate"]
        assert preview.application_preview["memoryApplications"][0]["status"] == "applied"

        outcome = OrchestratorExpenseApplicationWorkflow(db).execute(
            payload=OrchestratorRequest(
                source="user_message",
                user_id=current_user.username,
                message=(
                    "申请时间：2026-08-01 至 2026-08-03\n"
                    "地点：北京\n事由：客户现场验收\n天数：3天\n申请金额：2000元"
                ),
            ),
            current_user=current_user,
            run_id="run-memory-orchestrator",
            conversation_id="conversation-memory-orchestrator",
            ontology=OntologyParseResult(
                scenario="expense",
                intent="operate",
                permission=OntologyPermission(
                    level="approval_required",
                    allowed=True,
                    reason="test",
                ),
                confidence=1.0,
                run_id="run-memory-orchestrator",
            ),
            context_json={},
            selected_capability_codes=[],
        )
        assert outcome is not None
        assert outcome.result["application_preview"]["fields"]["transportMode"] == "火车"
        assert outcome.result["application_preview"]["memoryApplications"][0][
            "status"
        ] == "applied"
        assert "补充出行方式" not in str(outcome.result)

        edited_fields = dict(preview.application_preview["fields"])
        edited_fields["transportMode"] = "飞机"
        action = ExpenseApplicationPreviewWorkflow(db).execute(
            ExpenseApplicationPreviewActionPayload(
                source="user_message",
                user_id=current_user.username,
                conversation_id="conversation-memory-preview",
                action_type="submit",
                decision_id=preview.decision_id,
                request_id="request-memory-submit",
                message=(
                    "申请时间：2026-07-20 至 2026-07-22\n"
                    "地点：上海\n事由：客户现场实施\n天数：3天\n"
                    "出行方式：飞机\n申请金额：1800元\n确认提交"
                ),
                context_json={
                    "application_preview": {
                        "modelReviewStatus": "server_registered",
                        "fields": edited_fields,
                    }
                },
            ),
            current_user,
        )
        assert action.result.learning_receipts
        assert action.result.learning_receipts[0]["value"] == "飞机"
        assert action.result.learning_receipts[0]["status"] == "candidate"


def test_memory_api_enforces_owner_and_allows_owner_revoke() -> None:
    session_factory: sessionmaker[Session] = build_in_memory_session_factory()
    with session_factory() as db:
        owner = _user()
        entry, _ = _seed_active_memory(db, current_user=owner)
        memory_id = entry.id
        db.commit()

    app = create_app()
    install_legacy_header_auth_override(app)

    def override_db():
        with session_factory() as db:
            yield db

    app.dependency_overrides[get_db] = override_db
    client = TestClient(app)
    owner_headers = {
        "X-Auth-Username": owner.username,
        "X-Auth-Name": "Memory Owner",
        "X-Auth-Employee-Id": owner.employee_id,
        "X-Auth-Tenant-Id": owner.tenant_id,
        "X-Auth-Role-Codes": "user",
    }
    other_headers = {**owner_headers, "X-Auth-Employee-Id": "employee-other"}

    assert client.get(
        "/api/v1/expense-application-memories/me",
        headers=other_headers,
    ).json() == {"items": []}
    assert client.delete(
        f"/api/v1/expense-application-memories/{memory_id}",
        headers=other_headers,
    ).status_code == 404

    response = client.delete(
        f"/api/v1/expense-application-memories/{memory_id}",
        headers=owner_headers,
    )
    assert response.status_code == 200, response.text
    with session_factory() as db:
        revoked_entry = db.scalar(select(MemoryEntry).where(MemoryEntry.id == memory_id))
        assert revoked_entry is not None
        assert revoked_entry.status == "revoked"
        assert revoked_entry.value_json == {}