强制扫码登录二维码携带会话ID

app/routers/erp_login.py

@@ -44,11 +44,8 @@ def _require_session_id(session_id: int | None) -> int:
     return session_id
 
 
-def _scene(ticket: str, session_id: int | None) -> str:
-    scene = f"ticket={ticket}"
-    if session_id is not None:
-        scene = f"{scene}&session_id={session_id}"
-    return scene
+def _scene(ticket: str, session_id: int) -> str:
+    return f"ticket={ticket}&session_id={session_id}"
 
 
 def _action_response(data: dict[str, Any]) -> ErpLoginActionResponse:

app/schemas.py

@@ -780,7 +780,7 @@ class ReconciliationEntryUpdate(BaseModel):
 
 class ErpLoginQrcodeRequest(BaseModel):
     ticket: str = Field(min_length=1, max_length=256)
-    session_id: int | None = None
+    session_id: int = Field(ge=1)
 
 
 class ErpLoginQrcodeResponse(BaseModel):

tests/test_erp_login.py

@@ -126,6 +126,29 @@ def test_qrcode_signed_request_uses_expected_miniapp_qrcode_contract(monkeypatch
     }]
 
 
+def test_qrcode_requires_session_id(monkeypatch):
+    from app.routers import erp_login
+
+    calls = []
+
+    async def fake_create_miniapp_qrcode(**kwargs):
+        calls.append(kwargs)
+        return "https://miniapp.example.com/uploads/erp-login-qrcodes/erp-login-ticket-abc.png"
+
+    monkeypatch.setattr(erp_login, "create_miniapp_qrcode", fake_create_miniapp_qrcode)
+    client = _client(monkeypatch)
+    body = _json_bytes({"ticket": "ticket-abc"})
+
+    response = client.post(
+        "/api/erp-login/qrcode",
+        content=body,
+        headers=_signed_headers("POST", "/api/erp-login/qrcode", body),
+    )
+
+    assert response.status_code == 422
+    assert calls == []
+
+
 def test_preview_requires_session_id(monkeypatch):
     client = _client(monkeypatch)
 
@@ -224,6 +247,53 @@ def test_erp_http_status_error_uses_generic_gateway_error(monkeypatch):
     assert response.json()["detail"] == "ERP扫码登录服务暂时不可用"
 
 
+def test_erp_request_error_uses_generic_gateway_error(monkeypatch):
+    from app.routers import erp_login
+
+    original_async_client = httpx.AsyncClient
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("boom", request=request)
+
+    transport = httpx.MockTransport(handler)
+
+    def fake_async_client(**kwargs):
+        _ = kwargs
+        return original_async_client(transport=transport)
+
+    monkeypatch.setattr(erp_login.httpx, "AsyncClient", fake_async_client)
+    client = _client(monkeypatch)
+
+    response = client.get("/api/erp-login/sessions/ticket-abc?session_id=123")
+
+    assert response.status_code == 502
+    assert response.json()["detail"] == "ERP扫码登录服务暂时不可用"
+
+
+def test_erp_invalid_json_uses_invalid_response_error(monkeypatch):
+    from app.routers import erp_login
+
+    original_async_client = httpx.AsyncClient
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/api/auth/qr-login/sessions/123/preview"
+        return httpx.Response(200, content=b"not-json")
+
+    transport = httpx.MockTransport(handler)
+
+    def fake_async_client(**kwargs):
+        _ = kwargs
+        return original_async_client(transport=transport)
+
+    monkeypatch.setattr(erp_login.httpx, "AsyncClient", fake_async_client)
+    client = _client(monkeypatch)
+
+    response = client.get("/api/erp-login/sessions/ticket-abc?session_id=123")
+
+    assert response.status_code == 502
+    assert response.json()["detail"] == "ERP扫码登录响应无效"
+
+
 def test_create_device_qrcode_delegates_to_miniapp_qrcode_directory(monkeypatch):
     calls = []