服务链路自动化测试实战教程

# loan_status.py 状态常量与合法迁移表
UPLOADED = "UPLOADED"        # 合同已上传
OCR_PARSING = "OCR_PARSING"  # OCR 识别中
EXTRACTED = "EXTRACTED"      # 字段抽取完成
REVIEWING = "REVIEWING"      # 审核中
APPROVED = "APPROVED"        # 审核通过
STORED = "STORED"            # 已入库
DISPLAYED = "DISPLAYED"      # 已对客展示
REJECTED = "REJECTED"        # 审核拒绝

VALID_TRANSITIONS = {
    "UPLOADED": {"OCR_PARSING"},
    "OCR_PARSING": {"EXTRACTED", "OCR_FAILED"},
    "EXTRACTED": {"REVIEWING", "REJECTED"},
    "REVIEWING": {"APPROVED", "REJECTED"},
    "APPROVED": {"STORED"},
    "STORED": {"DISPLAYED"},
}

# 所有环节共用一个 loanId，traceId 贯穿日志
import uuid

def create_loan_id(contract_no):
    return "LOAN-" + contract_no + "-" + uuid.uuid4().hex[:8]

def build_headers(loan_id, trace_id=None):
    return {
        "X-Loan-Id": loan_id,
        "X-Trace-Id": trace_id or uuid.uuid4().hex,
    }

class ChainClient:
    def __init__(self, base_url, loan_id):
        self.base_url = base_url
        self.headers = build_headers(loan_id)

    def upload_contract(self, file_name):
        return self._post("/v1/contracts/upload", {"file": file_name})

    def ocr_result(self):
        return self._get("/v1/contracts/ocr")

    def approve(self):
        return self._post("/v1/loans/review", {"decision": "APPROVE"})

    def _post(self, path, body):
        return requests.post(self.base_url + path, json=body, headers=self.headers)

import pytest

@pytest.fixture
def loan_flow(chain_client_factory):
    client = chain_client_factory(create_loan_id("2025-0001"))
    yield client
    client.archive_loan()  # 精确清理，避免污染下次运行

def test_上传后OCR能识别同一份合同(loan_flow):
    upload = loan_flow.upload_contract("sample_contract.pdf")
    assert upload.json()["loanId"] == loan_flow.headers["X-Loan-Id"]

    parsed = loan_flow.wait_status("EXTRACTED", timeout=30)
    assert parsed["borrowerName"]          # 借款人字段非空
    assert parsed["loanAmount"] > 0        # 金额为正数

def test_状态只能按合法方向迁移(loan_flow):
    assert loan_flow.status() == "UPLOADED"

    loan_flow.start_ocr()
    assert loan_flow.status() == "OCR_PARSING"

    loan_flow.finish_extract()
    assert loan_flow.status() == "EXTRACTED"

    # 非法迁移：EXTRACTED 不允许直接跳到 STORED
    resp = loan_flow.force_store()
    assert resp.json()["businessCode"] == "ILLEGAL_TRANSITION"
    assert loan_flow.status() == "EXTRACTED"

def test_重复上传同一合同只创建一个记录(loan_flow):
    first = loan_flow.upload_contract("sample_contract.pdf")
    second = loan_flow.upload_contract("sample_contract.pdf")
    assert first.status_code == 200
    assert second.status_code == 200
    assert second.json()["businessCode"] == "DUPLICATE_IGNORED"
    assert loan_flow.contract_count() == 1

def test_OCR回调重复到达只更新一次(loan_flow):
    loan_flow.start_ocr()
    payload = {"loanId": loan_flow.headers["X-Loan-Id"],
               "fields": {"borrowerName": "张三", "loanAmount": 100000}}
    for _ in range(3):
        loan_flow.callback_ocr(payload)
    result = loan_flow.ocr_result()
    assert result["version"] == 1          # 版本只递增一次

def test_OCR超时后按退避策略恢复(loan_flow):
    loan_flow.make_ocr_slow(delay=60)     # 注入 60 秒延迟
    loan_flow.start_ocr()

    assert loan_flow.wait_status("OCR_RETRYING", timeout=5)

    loan_flow.recover_ocr()               # 恢复服务
    assert loan_flow.wait_status("EXTRACTED", timeout=30)

def test_OCR降级为人工补录后仍能走到审核(loan_flow):
    loan_flow.degrade_ocr()               # 关闭自动抽取
    loan_flow.start_ocr()
    assert loan_flow.wait_status("MANUAL_FILL", timeout=10)

    loan_flow.fill_fields({"borrowerName": "张三", "loanAmount": 100000})
    assert loan_flow.wait_status("REVIEWING", timeout=10)

def test_入库失败后补偿回滚至可重试状态(loan_flow):
    loan_flow.approve()
    loan_flow.make_store_fail(fail_once=True)

    resp = loan_flow.store()
    assert resp.json()["status"] == "STORE_FAILED"

    # 补偿任务回滚抽取结果，恢复 EXTRACTED 供再次审核
    loan_flow.run_compensation()
    assert loan_flow.wait_status("EXTRACTED", timeout=10)

    loan_flow.store()
    assert loan_flow.wait_status("STORED", timeout=10)
    assert loan_flow.status() == "STORED"

NODES = ["upload", "ocr", "extract", "review", "store", "display"]

class ChainFailed(Exception):
    def __init__(self, node, report):
        self.node = node
        self.report = report

def run_chain(loan_flow, plan):
    report = {}
    for node in NODES:
        try:
            result = plan[node](loan_flow)
            report[node] = {"status": "PASS", "detail": result}
        except AssertionError as exc:
            report[node] = {"status": "FAIL", "detail": str(exc)}
            raise ChainFailed(node, report)
    return report

def test_完整贷款链路_六节点全部通过(loan_flow):
    plan = {
        "upload": lambda f: f.assert_upload_ok(),
        "ocr": lambda f: f.assert_status("EXTRACTED", 30),
        "extract": lambda f: f.assert_fields_ok(),
        "review": lambda f: f.assert_approved(),
        "store": lambda f: f.assert_status("STORED", 10),
        "display": lambda f: f.assert_visible(),
    }
    report = run_chain(loan_flow, plan)
    assert all(item["status"] == "PASS" for item in report.values())

def run_suite(runner, cases):
    results = []
    for name, case in cases:
        try:
            case()
            results.append({"name": name, "result": "PASS"})
        except ChainFailed as failure:
            results.append({
                "name": name,
                "result": "FAIL",
                "node": failure.node,       # 失败直接定位到节点
                "evidence": failure.report, # 携带已通过节点的证据
            })
    return results

# 输出示例（JSON）
# [{"name": "贷款链路-六节点", "result": "FAIL",
#   "node": "store", "evidence": {...}}]