Files
X-Financial/server/tests/test_expense_application_memory.py
caoxiaozhu 787bc3a481 feat(platform): close AI expense value loop
Add tenant-safe value, telemetry, connector, commercial, and production-readiness foundations.
2026-07-17 14:14:08 +08:00

785 lines
28 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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.employee import Employee
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()
manager = Employee(
id="employee-memory-manager",
tenant_id=current_user.tenant_id,
employee_no="E-MEMORY-MANAGER",
name="记忆测试经理",
email="employee-memory-manager@example.com",
)
employee = Employee(
id=current_user.employee_id,
tenant_id=current_user.tenant_id,
employee_no=current_user.employee_no,
name=current_user.name,
email=current_user.username,
manager=manager,
)
db.add_all([manager, employee])
db.flush()
_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 == {}