from datetime import UTC, date, datetime import pytest from fastapi.testclient import TestClient from zbt.commands.seed import build_seed_manifest from zbt.core.config import Settings from zbt.core.errors import AppError from zbt.core.passwords import PasswordService from zbt.domains.catalog.repository import InMemoryCatalogRepository from zbt.domains.enrollment.models import OutboxEvent, Policy from zbt.domains.enrollment.repository import InMemoryEnrollmentRepository from zbt.domains.identity.models import AdminUser from zbt.domains.identity.repository import InMemoryIdentityRepository from zbt.domains.surrender.migration import ManualRefundMigrationService from zbt.domains.surrender.models import RefundAttempt, SurrenderRequest from zbt.domains.surrender.recovery import RefundRecoveryService from zbt.domains.surrender.repository import InMemorySurrenderRepository from zbt.main import create_app NOW = datetime(2026, 7, 30, 10, 0, tzinfo=UTC) REQUEST_ID = "01KZ0000000000000000000101" POLICY_ID = "01KZ0000000000000000000102" EVENT_ID = "01KZ0000000000000000000103" def admin() -> AdminUser: return AdminUser( id="01KZ0000000000000000000104", username="operator01", password_hash="hash", display_name="运营人员", status="ACTIVE", roles=("OPERATOR",), permissions=("surrender:review",), data_scope="MASKED_ALL", ) def build_recovery( *, attempt_status: str = "SUBMITTED", ) -> tuple[ RefundRecoveryService, InMemoryEnrollmentRepository, InMemorySurrenderRepository, ]: enrollments = InMemoryEnrollmentRepository() surrenders = InMemorySurrenderRepository() enrollments.save_policy( Policy( id=POLICY_ID, policy_no="POL-20260730-TEST01", order_id="01KZ0000000000000000000105", user_id="01KZ0000000000000000000106", product_id="01KZ0000000000000000000107", product_version_id="01KZ0000000000000000000108", plan_id="01KZ0000000000000000000109", premium_cents=36_500, currency="CNY", coverage_start=date(2026, 7, 1), coverage_end=date(2027, 6, 30), status="SURRENDERING", issued_at=NOW, ) ) surrenders.save_request( SurrenderRequest( id=REQUEST_ID, request_no="ZBT-TB-20260730-1001", user_id="01KZ0000000000000000000106", policy_id=POLICY_ID, idempotency_key="recovery-test", reason="保障计划调整", status="REFUND_PENDING", refund_amount_cents=30_000, currency="CNY", rule_version="surrender-v1", calculation={}, workflow_thread_id="recovery-thread", created_at=NOW, ) ) surrenders.save_refund_attempt( RefundAttempt( id="01KZ0000000000000000000110", surrender_request_id=REQUEST_ID, attempt_no=1, refund_no="ZBT-RF-20260730-1001", provider="INTERNAL_REFUND", idempotency_key="recovery-refund", amount_cents=30_000, currency="CNY", status=attempt_status, provider_refund_no=None, error_code=None, error_message=None, created_at=NOW, updated_at=NOW, ) ) enrollments.save_outbox_event( OutboxEvent( id=EVENT_ID, aggregate_type="SURRENDER_REQUEST", aggregate_id=REQUEST_ID, event_type="SURRENDER_REFUND_REQUESTED", payload={"request_id": REQUEST_ID}, status="DEAD", created_at=NOW, attempt_count=5, lease_owner="worker-old", leased_until=NOW, last_error="数据库临时不可用", ) ) return ( RefundRecoveryService(enrollments, surrenders, lambda: NOW), enrollments, surrenders, ) def test_dead_refund_task_can_be_requeued_with_a_fresh_retry_budget() -> None: service, enrollments, surrenders = build_recovery() result = service.requeue(admin(), EVENT_ID, note="数据库已恢复，重新执行") recovered_event = enrollments.get_outbox_event(EVENT_ID) request = surrenders.get_request(REQUEST_ID) policy = enrollments.get_policy(POLICY_ID) assert result["event_status"] == "PENDING" assert recovered_event is not None assert recovered_event.attempt_count == 0 assert recovered_event.last_error is None assert recovered_event.payload["previous_attempt_count"] == 5 assert request is not None and request.status == "REFUND_PENDING" assert policy is not None and policy.status == "SURRENDERING" assert surrenders.list_timeline_events(REQUEST_ID)[-1].event_type == ( "REFUND_TASK_REQUEUED" ) def test_dead_refund_task_can_be_cancelled_and_restore_policy() -> None: service, enrollments, surrenders = build_recovery() result = service.cancel(admin(), EVENT_ID, note="客户已改为继续保障") event = enrollments.get_outbox_event(EVENT_ID) request = surrenders.get_request(REQUEST_ID) policy = enrollments.get_policy(POLICY_ID) attempt = surrenders.get_latest_refund_attempt(REQUEST_ID) assert result["event_status"] == "CANCELLED" assert event is not None and event.status == "CANCELLED" assert request is not None and request.status == "COMPENSATED" assert policy is not None and policy.status == "ACTIVE" assert attempt is not None and attempt.status == "CANCELLED" assert surrenders.list_timeline_events(REQUEST_ID)[-1].event_type == ( "REFUND_TASK_CANCELLED" ) @pytest.mark.parametrize("operation", ["requeue", "cancel"]) def test_succeeded_refund_cannot_be_recovered(operation: str) -> None: service, _, _ = build_recovery(attempt_status="SUCCEEDED") with pytest.raises(AppError) as captured: getattr(service, operation)(admin(), EVENT_ID, note="错误操作") assert captured.value.code == "REFUND_TASK_ALREADY_SETTLED" def test_admin_can_requeue_dead_refund_task_via_operations_api() -> None: _, enrollments, surrenders = build_recovery() identities = InMemoryIdentityRepository() for user in build_seed_manifest( PasswordService().hash("zaq1XSW@") ).admin_users: identities.save_admin_user(user) app = create_app( settings=Settings( app_env="test", jwt_access_secret="a" * 32, jwt_refresh_secret="b" * 32, field_encryption_key="c" * 32, ), identity_repository=identities, catalog_repository=InMemoryCatalogRepository(), enrollment_repository=enrollments, surrender_repository=surrenders, clock=lambda: NOW, ) with TestClient(app) as client: login = client.post( "/api/v1/admin/auth/login", json={"username": "operator01", "password": "zaq1XSW@"}, ) token = login.json()["data"]["tokens"]["access_token"] response = client.post( f"/api/v1/admin/refund-operations/{EVENT_ID}/requeue", headers={"Authorization": f"Bearer {token}"}, json={"note": "基础设施已经恢复"}, ) assert response.status_code == 200 assert response.json()["data"]["event_status"] == "PENDING" assert response.json()["data"]["policy_status"] == "SURRENDERING" def test_manual_intervention_request_is_migrated_to_dead_task_once() -> None: enrollments = InMemoryEnrollmentRepository() surrenders = InMemorySurrenderRepository() surrenders.save_request( SurrenderRequest( id=REQUEST_ID, request_no="ZBT-TB-20260730-1002", user_id="01KZ0000000000000000000106", policy_id=POLICY_ID, idempotency_key="legacy-manual-intervention", reason="历史退款连续失败", status="MANUAL_INTERVENTION", refund_amount_cents=30_000, currency="CNY", rule_version="surrender-v1", calculation={}, workflow_thread_id="legacy-thread", created_at=NOW, ) ) service = ManualRefundMigrationService(enrollments, surrenders, lambda: NOW) first = service.migrate() second = service.migrate() events = [ event for event in enrollments.list_outbox_events( event_type="SURRENDER_REFUND_REQUESTED" ) if event.aggregate_id == REQUEST_ID ] timeline = surrenders.list_timeline_events(REQUEST_ID) request = surrenders.get_request(REQUEST_ID) assert first == {"scanned": 1, "migrated": 1} assert second == {"scanned": 1, "migrated": 0} assert len(events) == 1 assert events[0].status == "DEAD" assert events[0].attempt_count == 5 assert events[0].payload["migration_key"] == "manual-intervention-v1" assert request is not None and request.status == "MANUAL_INTERVENTION" assert [event.event_type for event in timeline] == ["REFUND_TASK_MIGRATED"]