harness_zbt/backend/tests/test_mysql_langgraph_checkpoint.py

from datetime import UTC, datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from zbt.domains.surrender.models import SurrenderRequest
from zbt.domains.surrender.workflow import SurrenderWorkflow
from zbt.infrastructure.mysql.core_models import (
    LangGraphBlobRecord,
    LangGraphCheckpointRecord,
    LangGraphWriteRecord,
)
from zbt.infrastructure.mysql.langgraph_checkpoint import (
    MySqlLangGraphCheckpointSaver,
)


def test_workflow_resumes_from_database_after_runtime_is_recreated(tmp_path) -> None:
    engine = create_engine(f"sqlite:///{tmp_path / 'checkpoint.db'}")
    for table in (
        LangGraphCheckpointRecord.__table__,
        LangGraphBlobRecord.__table__,
        LangGraphWriteRecord.__table__,
    ):
        table.create(engine)
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    request = SurrenderRequest(
        id="01KZ0000000000000000000091",
        request_no="ZBT-TB-20260726-0091",
        user_id="01KZ0000000000000000000092",
        policy_id="01KZ0000000000000000000093",
        idempotency_key="checkpoint-test-001",
        reason="保障计划调整",
        status="WAITING_APPROVAL",
        refund_amount_cents=30_600,
        currency="CNY",
        rule_version="surrender-v1",
        calculation={},
        workflow_thread_id="surrender:01KZ0000000000000000000091",
        created_at=datetime(2026, 7, 26, tzinfo=UTC),
    )

    first_runtime = SurrenderWorkflow(MySqlLangGraphCheckpointSaver(factory))
    paused = first_runtime.start(request)
    del first_runtime

    restarted_runtime = SurrenderWorkflow(MySqlLangGraphCheckpointSaver(factory))
    resumed = restarted_runtime.resume_review(
        request,
        approved=True,
        note="审批通过",
        reviewed_by="01KZ0000000000000000000094",
    )

    assert paused["paused"] is True
    assert paused["step"] == "OPERATIONS_APPROVAL"
    assert paused["risk_level"] == "HIGH"
    assert paused["handoff"]["request_no"] == request.request_no
    assert resumed["workflow_status"] == "REFUND_PENDING"