From 8ae25245cfb84a0789ff846864eec2168398090d Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 14 Jun 2026 09:56:14 +0800 Subject: [PATCH] test: add alipay sim user e2e flow --- admin/routes/web_public.py | 168 ++++++++++++++---- admin/tests/test_web_public_alipay_sim_e2e.py | 132 ++++++++++++++ 2 files changed, 262 insertions(+), 38 deletions(-) create mode 100644 admin/tests/test_web_public_alipay_sim_e2e.py diff --git a/admin/routes/web_public.py b/admin/routes/web_public.py index 7aa1123..74df478 100644 --- 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""" - Mock 支付 + {escape(title)}
-

Mock 支付沙箱

+

{escape(title)}

订单支付单号：{escape(payment_id)}

支付金额：¥{amount_cents / 100:.2f}

-
- + +
@@ -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: diff --git a/admin/tests/test_web_public_alipay_sim_e2e.py b/admin/tests/test_web_public_alipay_sim_e2e.py new file mode 100644 index 0000000..e616bf1 --- /dev/null +++ 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( + "

志愿方案报告

已生成。

", 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"