test: add alipay sim user e2e flow

2026-06-14 09:56:14 +08:00
parent 85605716b7
commit 8ae25245cf
2 changed files with 262 additions and 38 deletions
--- a/admin/routes/web_public.py
+++ b/admin/routes/web_public.py
@@ -149,17 +149,8 @@ def mock_payment_page(
    token: str,
    settings: Settings = Depends(get_settings_dep),
 ) -> HTMLResponse:
-    service = PaymentService.for_db(
-        settings.orders_db_path,
-        base_url=settings.payment_base_url,
-        webhook_secret=settings.payment_webhook_secret,
-        provider_name=settings.payment_provider,
-    )
-    payment = service.get_payment(payment_id)
-    if payment is None:
-        raise HTTPException(status_code=404, detail="payment not found")
-    return HTMLResponse(
-        _render_mock_payment_page(payment_id, payment.amount_cents, token)
+    return _render_simulated_payment_page(
+        payment_id, token, settings, provider_slug="mock"
    )


@@ -169,31 +160,31 @@ def complete_mock_payment(
    token: str,
    settings: Settings = Depends(get_settings_dep),
 ) -> RedirectResponse:
-    service = PaymentService.for_db(
-        settings.orders_db_path,
-        base_url=settings.payment_base_url,
-        webhook_secret=settings.payment_webhook_secret,
-        provider_name=settings.payment_provider,
+    return _complete_simulated_payment(
+        payment_id, token, settings, provider_slug="mock"
    )
-    payment = service.get_payment(payment_id)
-    if payment is None:
-        raise HTTPException(status_code=404, detail="payment not found")
-    try:
-        portal_payload = verify_portal_token(token, settings.jwt_secret)
-    except PortalTokenError as exc:
-        raise HTTPException(status_code=401, detail=str(exc)) from exc
-    if (
-        str(portal_payload["order_id"]) != payment.order_id
-        or payment.checkout_token != token
-    ):
-        raise HTTPException(status_code=403, detail="payment token mismatch")
-    payload, headers = service.provider.build_webhook_request(
-        payment_id=payment.id,
-        amount_cents=payment.amount_cents,
-        provider_trade_no=f"MOCK-{payment.id}",
+
+
+@router.get("/pay/alipay-sim/{payment_id}", include_in_schema=False)
+def alipay_sim_payment_page(
+    payment_id: str,
+    token: str,
+    settings: Settings = Depends(get_settings_dep),
+) -> HTMLResponse:
+    return _render_simulated_payment_page(
+        payment_id, token, settings, provider_slug="alipay-sim"
+    )
+
+
+@router.post("/pay/alipay-sim/{payment_id}/complete", include_in_schema=False)
+def complete_alipay_sim_payment(
+    payment_id: str,
+    token: str,
+    settings: Settings = Depends(get_settings_dep),
+) -> RedirectResponse:
+    return _complete_simulated_payment(
+        payment_id, token, settings, provider_slug="alipay-sim"
    )
-    service.handle_webhook(payload, headers["X-Mock-Signature"])
-    return RedirectResponse(url=f"/portal/{token}/status", status_code=303)


@router.get("/portal/{token}/info", include_in_schema=False)
@@ -533,16 +524,48 @@ def _render_checkout_page(service_version: str) -> str:


 def _render_mock_payment_page(payment_id: str, amount_cents: int, token: str) -> str:
+    return _render_simulated_payment_html(
+        title="Mock 支付沙箱",
+        provider_slug="mock",
+        payment_id=payment_id,
+        amount_cents=amount_cents,
+        token=token,
+        submit_label="模拟支付成功并返回订单页",
+    )
+
+
+def _render_alipay_sim_payment_page(
+    payment_id: str, amount_cents: int, token: str
+) -> str:
+    return _render_simulated_payment_html(
+        title="支付宝模拟收银台",
+        provider_slug="alipay-sim",
+        payment_id=payment_id,
+        amount_cents=amount_cents,
+        token=token,
+        submit_label="模拟支付宝支付成功并返回订单页",
+    )
+
+
+def _render_simulated_payment_html(
+    *,
+    title: str,
+    provider_slug: str,
+    payment_id: str,
+    amount_cents: int,
+    token: str,
+    submit_label: str,
+) -> str:
    return f"""<!doctype html>
 <html lang=\"zh-CN\">
-  <head><meta charset=\"utf-8\" /><meta name=\"viewport\" content=\"width=device-width, initial-scale=1\" /><title>Mock 支付</title></head>
+  <head><meta charset=\"utf-8\" /><meta name=\"viewport\" content=\"width=device-width, initial-scale=1\" /><title>{escape(title)}</title></head>
  <body style=\"font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;background:#f5f7fb;padding:24px;\">
    <main style=\"max-width:640px;margin:0 auto;background:#fff;border:1px solid #dbe3f0;border-radius:18px;padding:24px;\">
-      <h1>Mock 支付沙箱</h1>
+      <h1>{escape(title)}</h1>
      <p>订单支付单号：{escape(payment_id)}</p>
      <p>支付金额：¥{amount_cents / 100:.2f}</p>
-      <form method=\"post\" action=\"/pay/mock/{escape(payment_id)}/complete?token={escape(token)}\">
-        <button type=\"submit\" style=\"border:none;border-radius:12px;background:#1f6feb;color:#fff;font-weight:700;padding:12px 18px;cursor:pointer;\">模拟支付成功并返回订单页</button>
+      <form method=\"post\" action=\"/pay/{escape(provider_slug)}/{escape(payment_id)}/complete?token={escape(token)}\">
+        <button type=\"submit\" style=\"border:none;border-radius:12px;background:#1f6feb;color:#fff;font-weight:700;padding:12px 18px;cursor:pointer;\">{escape(submit_label)}</button>
      </form>
    </main>
  </body>
@@ -550,6 +573,75 @@ def _render_mock_payment_page(payment_id: str, amount_cents: int, token: str) ->
 """


+def _render_simulated_payment_page(
+    payment_id: str,
+    token: str,
+    settings: Settings,
+    *,
+    provider_slug: str,
+) -> HTMLResponse:
+    service = PaymentService.for_db(
+        settings.orders_db_path,
+        base_url=settings.payment_base_url,
+        webhook_secret=settings.payment_webhook_secret,
+        provider_name=settings.payment_provider,
+    )
+    payment = service.get_payment(payment_id)
+    if payment is None:
+        raise HTTPException(status_code=404, detail="payment not found")
+    if provider_slug == "mock":
+        body = _render_mock_payment_page(payment_id, payment.amount_cents, token)
+    elif provider_slug == "alipay-sim":
+        body = _render_alipay_sim_payment_page(payment_id, payment.amount_cents, token)
+    else:
+        raise HTTPException(
+            status_code=404, detail="unsupported simulated payment page"
+        )
+    return HTMLResponse(body)
+
+
+def _complete_simulated_payment(
+    payment_id: str,
+    token: str,
+    settings: Settings,
+    *,
+    provider_slug: str,
+) -> RedirectResponse:
+    service = PaymentService.for_db(
+        settings.orders_db_path,
+        base_url=settings.payment_base_url,
+        webhook_secret=settings.payment_webhook_secret,
+        provider_name=settings.payment_provider,
+    )
+    payment = service.get_payment(payment_id)
+    if payment is None:
+        raise HTTPException(status_code=404, detail="payment not found")
+    try:
+        portal_payload = verify_portal_token(token, settings.jwt_secret)
+    except PortalTokenError as exc:
+        raise HTTPException(status_code=401, detail=str(exc)) from exc
+    if (
+        str(portal_payload["order_id"]) != payment.order_id
+        or payment.checkout_token != token
+    ):
+        raise HTTPException(status_code=403, detail="payment token mismatch")
+    payload, headers = service.provider.build_webhook_request(
+        payment_id=payment.id,
+        amount_cents=payment.amount_cents,
+        provider_trade_no=f"SIM-{payment.id}",
+    )
+    if provider_slug == "mock":
+        signature = headers["X-Mock-Signature"]
+    elif provider_slug == "alipay-sim":
+        signature = headers["X-Alipay-Sim-Signature"]
+    else:
+        raise HTTPException(
+            status_code=404, detail="unsupported simulated payment completion"
+        )
+    service.handle_webhook(payload, signature)
+    return RedirectResponse(url=f"/portal/{token}/status", status_code=303)
+
+
 def _render_info_page(
    order: Order, token: str, payload: dict[str, Any], stage: str
 ) -> str:
--- a/admin/tests/test_web_public_alipay_sim_e2e.py
+++ b/admin/tests/test_web_public_alipay_sim_e2e.py
@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+
+def test_alipay_sim_public_user_e2e_flow(tmp_path, monkeypatch):
+    admin_db = tmp_path / "admin.db"
+    orders_db = tmp_path / "orders.db"
+    share_db = tmp_path / "share.db"
+    share_reports = tmp_path / "share_reports"
+    share_reports.mkdir()
+
+    monkeypatch.setenv("GAOKAO_ENV", "dev")
+    monkeypatch.setenv("GAOKAO_DB_PATH", str(admin_db))
+    monkeypatch.setenv("GAOKAO_ORDERS_DB_PATH", str(orders_db))
+    monkeypatch.setenv("GAOKAO_SHARE_DB_PATH", str(share_db))
+    monkeypatch.setenv("GAOKAO_SHARE_REPORT_DIR", str(share_reports))
+    monkeypatch.setenv("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-web-self-service")
+    monkeypatch.setenv("GAOKAO_JWT_SECRET", "x" * 64)
+    monkeypatch.setenv("GAOKAO_JWT_EXP_MIN", "5")
+    monkeypatch.setenv("GAOKAO_ADMIN_USER", "admin")
+    monkeypatch.setenv("GAOKAO_ADMIN_PASS", "test-pass-123")
+    monkeypatch.setenv("GAOKAO_PAYMENT_PROVIDER", "alipay_sim")
+    monkeypatch.setenv("GAOKAO_PAYMENT_BASE_URL", "http://testserver")
+    monkeypatch.setenv("GAOKAO_PAYMENT_WEBHOOK_SECRET", "alipay-sim-secret")
+
+    from admin.app import create_app
+    from admin.config import load_settings
+    from data.orders.dao import OrdersDAO
+    from data.orders.intake_store import IntakeStore
+
+    settings = load_settings()
+    app = create_app(settings)
+
+    with TestClient(app) as client:
+        create_resp = client.post(
+            "/api/public/orders",
+            json={
+                "service_version": "standard",
+                "amount_cents": 9900,
+                "customer_name": "张家长",
+                "customer_phone": "13800138000",
+                "candidate_name": "张三",
+                "candidate_province": "湖南",
+            },
+        )
+        assert create_resp.status_code == 201, create_resp.text
+        created = create_resp.json()
+        assert created["checkout_url"].startswith("/pay/alipay-sim/")
+
+        payment_id = created["checkout_url"].split("/pay/alipay-sim/")[1].split("?")[0]
+        portal_status_url = created["portal_status_url"]
+        portal_info_url = created["portal_info_url"]
+        token = portal_status_url.split("/portal/")[1].split("/status")[0]
+
+        checkout_page = client.get(created["checkout_url"])
+        assert checkout_page.status_code == 200, checkout_page.text
+        assert "支付宝模拟收银台" in checkout_page.text
+
+        before_pay_status = client.get(portal_status_url)
+        assert before_pay_status.status_code == 200, before_pay_status.text
+        assert "待支付" in before_pay_status.text
+
+        pay_resp = client.post(
+            f"/pay/alipay-sim/{payment_id}/complete?token={token}",
+            follow_redirects=False,
+        )
+        assert pay_resp.status_code == 303, pay_resp.text
+        assert pay_resp.headers["location"].endswith(f"/portal/{token}/status")
+
+        after_pay_status = client.get(portal_status_url)
+        assert after_pay_status.status_code == 200, after_pay_status.text
+        assert "待填写资料" in after_pay_status.text
+
+        info_page = client.get(portal_info_url)
+        assert info_page.status_code == 200, info_page.text
+        assert "考生资料填写" in info_page.text
+
+        submit_info = client.post(
+            portal_info_url,
+            json={
+                "mode": "submit",
+                "candidate_score": 578,
+                "candidate_rank": 12034,
+                "candidate_subjects": ["物理", "化学", "生物"],
+                "candidate_interests": "计算机",
+                "guardian_notes": "更看重省内城市",
+            },
+        )
+        assert submit_info.status_code == 200, submit_info.text
+        assert submit_info.json()["stage"] == "info_submitted"
+
+        with OrdersDAO.connect(settings.orders_db_path) as dao:
+            order = dao.get(created["order_id"])
+            dao.transition_status(
+                order.id, "serving", actor="test", reason="processing"
+            )
+            report_path = tmp_path / "report.html"
+            pdf_path = tmp_path / "report.pdf"
+            report_path.write_text(
+                "<h1>志愿方案报告</h1><p>已生成。</p>", encoding="utf-8"
+            )
+            pdf_path.write_bytes(b"%PDF-1.4\nalipay-sim-report\n")
+            dao.update(
+                order.id,
+                {"audit_report": str(report_path), "pdf_path": str(pdf_path)},
+                actor="test",
+                reason="attach_report",
+            )
+            dao.transition_status(
+                order.id, "delivered", actor="test", reason="report_ready"
+            )
+
+        report_status = client.get(portal_status_url)
+        assert report_status.status_code == 200, report_status.text
+        assert "报告已就绪" in report_status.text
+
+        report_view = client.get(f"/portal/{token}/report")
+        assert report_view.status_code == 200, report_view.text
+        assert "志愿方案报告" in report_view.text
+
+        report_pdf = client.get(f"/portal/{token}/report.pdf")
+        assert report_pdf.status_code == 200, report_pdf.text
+        assert report_pdf.headers["content-type"].startswith("application/pdf")
+
+        intake = IntakeStore.for_db(settings.orders_db_path)
+        try:
+            record = intake.get(created["order_id"])
+        finally:
+            intake.close()
+        assert record is not None
+        assert record.status == "submitted"