.env.example

@@ -15,6 +15,3 @@ WECHAT_APP_SECRET=
 PUBLIC_BASE_URL=http://127.0.0.1:8000
 UPLOAD_DIR=uploads
 
-ERP_API_BASE_URL=http://127.0.0.1:8000
-ERP_QR_LOGIN_SHARED_SECRET=replace-with-same-secret-on-erp-backend
-ERP_LOGIN_QRCODE_DIR=erp-login-qrcodes

app/config.py

@@ -27,9 +27,6 @@ class Settings(BaseSettings):
     wechat_qr_env_version: str = "release"
     public_base_url: str = "http://127.0.0.1:8000"
     upload_dir: str = "uploads"
-    erp_api_base_url: str = "http://127.0.0.1:8000"
-    erp_qr_login_shared_secret: str = ""
-    erp_login_qrcode_dir: str = "erp-login-qrcodes"
 
     @property
     def database_url(self) -> str:

app/main.py

@@ -16,7 +16,6 @@ from app.routers import (
     dashboard,
     devices,
     equipment,
-    erp_login,
     monitor,
     notices,
     people,
@@ -93,7 +92,6 @@ def create_app() -> FastAPI:
     app.include_router(dashboard.router)
     app.include_router(devices.router)
     app.include_router(equipment.router)
-    app.include_router(erp_login.router)
     app.include_router(monitor.router)
     app.include_router(notices.router)
     app.include_router(reconciliation.router)

app/routers/erp_login.py

@@ -1,199 +0,0 @@
-import json
-from typing import Any
-from urllib.parse import urlsplit
-from uuid import uuid4
-
-import httpx
-from fastapi import APIRouter, Header, HTTPException, Query, Request
-
-from app.config import settings
-from app.schemas import (
-    ErpLoginActionResponse,
-    ErpLoginConfirmRequest,
-    ErpLoginPreviewResponse,
-    ErpLoginQrcodeRequest,
-    ErpLoginQrcodeResponse,
-)
-from app.services.erp_login_security import (
-    SERVICE_NONCE_HEADER,
-    SERVICE_SIGNATURE_HEADER,
-    SERVICE_TIMESTAMP_HEADER,
-    signed_headers,
-    verify_service_signature,
-)
-from app.services.wechat import WechatConfigError, create_miniapp_qrcode, get_phone_number
-from app.timezone import now
-
-
-router = APIRouter(prefix="/api/erp-login", tags=["erp-login"])
-
-ERP_LOGIN_PAGE = "pages/erpLoginConfirm/erpLoginConfirm"
-WECHAT_SCENE_MAX_LENGTH = 32
-
-
-def _json_body(payload: dict[str, Any]) -> bytes:
-    return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
-
-
-def _erp_url(path: str) -> str:
-    return f"{settings.erp_api_base_url.rstrip('/')}{path}"
-
-
-def _require_session_id(session_id: int | None) -> int:
-    if session_id is None:
-        raise HTTPException(status_code=400, detail="缺少ERP扫码登录会话ID")
-    return session_id
-
-
-def _scene(ticket: str, session_id: int) -> str:
-    scene = f"t={ticket}&s={session_id}"
-    if len(scene) > WECHAT_SCENE_MAX_LENGTH:
-        raise HTTPException(status_code=422, detail="ERP扫码登录二维码参数过长")
-    return scene
-
-
-def _action_response(data: dict[str, Any]) -> ErpLoginActionResponse:
-    return ErpLoginActionResponse(
-        status=str(data.get("status") or ""),
-        failure_reason=data.get("failure_reason"),
-    )
-
-
-async def _read_erp_json(response: httpx.Response) -> dict[str, Any]:
-    try:
-        response.raise_for_status()
-    except httpx.HTTPStatusError as exc:
-        raise HTTPException(status_code=502, detail="ERP扫码登录服务暂时不可用") from exc
-    try:
-        data = response.json()
-    except ValueError as exc:
-        raise HTTPException(status_code=502, detail="ERP扫码登录响应无效") from exc
-    if not isinstance(data, dict):
-        raise HTTPException(status_code=502, detail="ERP扫码登录响应无效")
-    return data
-
-
-async def _get_erp_json(path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
-    url = _erp_url(path)
-    body = b""
-    headers = signed_headers(
-        "GET",
-        urlsplit(url).path,
-        body,
-        settings.erp_qr_login_shared_secret,
-    )
-    try:
-        async with httpx.AsyncClient(timeout=15) as client:
-            response = await client.get(url, params=params, headers=headers)
-    except httpx.RequestError as exc:
-        raise HTTPException(status_code=502, detail="ERP扫码登录服务暂时不可用") from exc
-    return await _read_erp_json(response)
-
-
-async def _post_erp_json(path: str, payload: dict[str, Any]) -> dict[str, Any]:
-    url = _erp_url(path)
-    body = _json_body(payload)
-    headers = signed_headers(
-        "POST",
-        urlsplit(url).path,
-        body,
-        settings.erp_qr_login_shared_secret,
-    )
-    headers["content-type"] = "application/json"
-    try:
-        async with httpx.AsyncClient(timeout=15) as client:
-            response = await client.post(url, content=body, headers=headers)
-    except httpx.RequestError as exc:
-        raise HTTPException(status_code=502, detail="ERP扫码登录服务暂时不可用") from exc
-    return await _read_erp_json(response)
-
-
-@router.post("/qrcode", response_model=ErpLoginQrcodeResponse)
-async def create_erp_login_qrcode(
-    payload: ErpLoginQrcodeRequest,
-    request: Request,
-    timestamp: str | None = Header(default=None, alias=SERVICE_TIMESTAMP_HEADER),
-    nonce: str | None = Header(default=None, alias=SERVICE_NONCE_HEADER),
-    signature: str | None = Header(default=None, alias=SERVICE_SIGNATURE_HEADER),
-) -> ErpLoginQrcodeResponse:
-    body = await request.body()
-    verify_service_signature(
-        "POST",
-        request.url.path,
-        timestamp,
-        nonce,
-        body,
-        signature,
-        settings.erp_qr_login_shared_secret,
-    )
-    scene = _scene(payload.ticket, payload.session_id)
-    try:
-        qr_url = await create_miniapp_qrcode(
-            key=f"erp-login-{payload.session_id}",
-            page=ERP_LOGIN_PAGE,
-            scene=scene,
-            public_base_url=settings.public_base_url,
-            directory=settings.erp_login_qrcode_dir,
-            label="ERP扫码登录",
-        )
-    except WechatConfigError as exc:
-        raise HTTPException(status_code=400, detail=str(exc)) from exc
-    except RuntimeError as exc:
-        raise HTTPException(status_code=502, detail=str(exc)) from exc
-    return ErpLoginQrcodeResponse(qr_url=qr_url, page=ERP_LOGIN_PAGE, scene=scene)
-
-
-@router.get("/sessions/{ticket}", response_model=ErpLoginPreviewResponse)
-async def preview_erp_login_session(
-    ticket: str,
-    session_id: int | None = Query(default=None),
-) -> ErpLoginPreviewResponse:
-    session_id = _require_session_id(session_id)
-    data = await _get_erp_json(
-        f"/api/auth/qr-login/sessions/{session_id}/preview",
-        params={"ticket": ticket},
-    )
-    await _post_erp_json(
-        f"/api/auth/qr-login/sessions/{session_id}/scanned",
-        {"ticket": ticket},
-    )
-    return ErpLoginPreviewResponse(**data)
-
-
-@router.post("/sessions/{ticket}/confirm", response_model=ErpLoginActionResponse)
-async def confirm_erp_login_session(
-    ticket: str,
-    payload: ErpLoginConfirmRequest,
-    session_id: int | None = Query(default=None),
-) -> ErpLoginActionResponse:
-    session_id = _require_session_id(session_id)
-    try:
-        phone = await get_phone_number(payload.phone_code)
-    except WechatConfigError as exc:
-        raise HTTPException(status_code=400, detail=str(exc)) from exc
-    except RuntimeError as exc:
-        raise HTTPException(status_code=502, detail=str(exc)) from exc
-
-    data = await _post_erp_json(
-        f"/api/auth/qr-login/sessions/{session_id}/confirm",
-        {
-            "ticket": ticket,
-            "phone": phone,
-            "confirmed_at": now().isoformat(),
-            "nonce": uuid4().hex,
-        },
-    )
-    return _action_response(data)
-
-
-@router.post("/sessions/{ticket}/cancel", response_model=ErpLoginActionResponse)
-async def cancel_erp_login_session(
-    ticket: str,
-    session_id: int | None = Query(default=None),
-) -> ErpLoginActionResponse:
-    session_id = _require_session_id(session_id)
-    data = await _post_erp_json(
-        f"/api/auth/qr-login/sessions/{session_id}/cancel",
-        {"ticket": ticket},
-    )
-    return _action_response(data)

app/schemas.py

@@ -778,35 +778,6 @@ class ReconciliationEntryUpdate(BaseModel):
     return_qty: float | None = Field(default=None, ge=0)
 
 
-class ErpLoginQrcodeRequest(BaseModel):
-    ticket: str = Field(min_length=1, max_length=16)
-    session_id: int = Field(ge=1)
-
-
-class ErpLoginQrcodeResponse(BaseModel):
-    qr_url: str | None = None
-    page: str
-    scene: str
-
-
-class ErpLoginPreviewResponse(BaseModel):
-    session_id: int
-    system_name: str
-    started_at: datetime
-    expires_at: datetime
-    request_ip_hint: str | None = None
-    user_agent_hint: str | None = None
-
-
-class ErpLoginConfirmRequest(BaseModel):
-    phone_code: str = Field(min_length=1, max_length=256)
-
-
-class ErpLoginActionResponse(BaseModel):
-    status: str
-    failure_reason: str | None = None
-
-
 PersonnelOut.model_rebuild()
 TokenResponse.model_rebuild()
 LoginSelectionResponse.model_rebuild()

app/services/erp_login_security.py

@@ -1,99 +0,0 @@
-import hashlib
-import hmac
-import time
-from uuid import uuid4
-
-from fastapi import HTTPException
-
-
-SERVICE_TIMESTAMP_HEADER = "X-ERP-QR-Timestamp"
-SERVICE_NONCE_HEADER = "X-ERP-QR-Nonce"
-SERVICE_SIGNATURE_HEADER = "X-ERP-QR-Signature"
-
-_seen_nonces: dict[str, int] = {}
-
-
-def build_service_signature(
-    method: str,
-    path: str,
-    timestamp: str,
-    nonce: str,
-    body: bytes,
-    secret: str,
-) -> str:
-    message = "\n".join([
-        method.upper(),
-        path,
-        timestamp,
-        nonce,
-        body.decode("utf-8"),
-    ])
-    return hmac.new(secret.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).hexdigest()
-
-
-def signed_headers(method: str, path: str, body: bytes, secret: str) -> dict[str, str]:
-    timestamp = str(int(time.time()))
-    nonce = uuid4().hex
-    signature = build_service_signature(method, path, timestamp, nonce, body, secret)
-    return {
-        SERVICE_TIMESTAMP_HEADER: timestamp,
-        SERVICE_NONCE_HEADER: nonce,
-        SERVICE_SIGNATURE_HEADER: signature,
-    }
-
-
-def _cleanup_seen_nonces(now_ts: int, ttl_seconds: int) -> None:
-    expired = [
-        nonce
-        for nonce, seen_at in _seen_nonces.items()
-        if now_ts - seen_at > ttl_seconds
-    ]
-    for nonce in expired:
-        _seen_nonces.pop(nonce, None)
-
-
-def verify_service_signature(
-    method: str,
-    path: str,
-    timestamp: str | None,
-    nonce: str | None,
-    body: bytes,
-    signature: str | None,
-    secret: str,
-    ttl_seconds: int = 300,
-) -> None:
-    if not secret:
-        raise HTTPException(status_code=500, detail="ERP扫码登录服务密钥未配置")
-
-    try:
-        timestamp_value = int(str(timestamp))
-    except (TypeError, ValueError) as exc:
-        raise HTTPException(status_code=401, detail="ERP服务签名无效") from exc
-
-    now_ts = int(time.time())
-    if abs(now_ts - timestamp_value) > ttl_seconds:
-        raise HTTPException(status_code=401, detail="ERP服务签名已过期")
-
-    _cleanup_seen_nonces(now_ts, ttl_seconds)
-    nonce_value = str(nonce or "")
-    if not nonce_value or nonce_value in _seen_nonces:
-        raise HTTPException(status_code=401, detail="ERP服务签名无效")
-
-    try:
-        expected = build_service_signature(
-            method,
-            path,
-            str(timestamp),
-            nonce_value,
-            body,
-            secret,
-        )
-    except UnicodeDecodeError as exc:
-        raise HTTPException(status_code=401, detail="ERP服务签名无效") from exc
-
-    if not hmac.compare_digest(expected, str(signature or "")):
-        raise HTTPException(status_code=401, detail="ERP服务签名无效")
-
-    # This in-process nonce store only protects one running process. Production
-    # multi-worker deployments should move replay tracking to shared storage.
-    _seen_nonces[nonce_value] = now_ts

app/services/wechat.py

@@ -190,24 +190,6 @@ async def create_device_qrcode(
     scene: str,
     public_base_url: str,
     label: str | None = None,
-) -> str | None:
-    return await create_miniapp_qrcode(
-        key=device_no,
-        page=page,
-        scene=scene,
-        public_base_url=public_base_url,
-        directory="qrcodes",
-        label=label,
-    )
-
-
-async def create_miniapp_qrcode(
-    key: str,
-    page: str,
-    scene: str,
-    public_base_url: str,
-    directory: str,
-    label: str | None = None,
 ) -> str | None:
     token = await get_access_token()
     url = f"https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token={token}"
@@ -223,15 +205,12 @@ async def create_miniapp_qrcode(
     content_type = response.headers.get("content-type", "")
     if "application/json" in content_type:
         data = response.json()
-        raise RuntimeError(data.get("errmsg") or "生成小程序码失败")
+        raise RuntimeError(data.get("errmsg") or "生成设备小程序码失败")
 
-    qr_dir = settings.upload_path / directory
+    qr_dir = settings.upload_path / "qrcodes"
     qr_dir.mkdir(parents=True, exist_ok=True)
     suffix = _qrcode_file_suffix(content_type)
-    file_name = _qrcode_file_name(key, suffix)
+    file_name = _qrcode_file_name(device_no, suffix)
     file_path = Path(qr_dir) / file_name
-    label_text = label if label is not None else key
+    file_path.write_bytes(_draw_qrcode_label(response.content, suffix, label or device_no))
-    file_path.write_bytes(_draw_qrcode_label(response.content, suffix, label_text))
+    return f"{public_base_url.rstrip('/')}/uploads/qrcodes/{quote(file_name, safe='')}"
-    encoded_directory = quote(directory.strip("/"), safe="")
-    encoded_file_name = quote(file_name, safe="")
-    return f"{public_base_url.rstrip('/')}/uploads/{encoded_directory}/{encoded_file_name}"

tests/test_erp_login.py

@@ -1,392 +0,0 @@
-import asyncio
-import hmac
-import json
-import time
-from hashlib import sha256
-from uuid import uuid4
-
-import httpx
-from fastapi.testclient import TestClient
-
-from app.config import settings
-from app.main import create_app
-from app.services import wechat
-
-
-SERVICE_SECRET = "test-erp-secret"
-TIMESTAMP_HEADER = "X-ERP-QR-Timestamp"
-NONCE_HEADER = "X-ERP-QR-Nonce"
-SIGNATURE_HEADER = "X-ERP-QR-Signature"
-
-
-def _json_bytes(payload: dict) -> bytes:
-    return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
-
-
-def _test_signature(
-    method: str,
-    path: str,
-    timestamp: str,
-    nonce: str,
-    body: bytes,
-    secret: str = SERVICE_SECRET,
-) -> str:
-    message = "\n".join([
-        method.upper(),
-        path,
-        timestamp,
-        nonce,
-        body.decode("utf-8"),
-    ])
-    return hmac.new(secret.encode("utf-8"), message.encode("utf-8"), sha256).hexdigest()
-
-
-def _signed_headers(
-    method: str,
-    path: str,
-    body: bytes,
-    secret: str = SERVICE_SECRET,
-) -> dict[str, str]:
-    timestamp = str(int(time.time()))
-    nonce = uuid4().hex
-    return {
-        TIMESTAMP_HEADER: timestamp,
-        NONCE_HEADER: nonce,
-        SIGNATURE_HEADER: _test_signature(method, path, timestamp, nonce, body, secret),
-        "content-type": "application/json",
-    }
-
-
-def _client(monkeypatch, erp_base_url: str = "http://erp.local") -> TestClient:
-    monkeypatch.setattr(settings, "erp_qr_login_shared_secret", SERVICE_SECRET)
-    monkeypatch.setattr(settings, "erp_api_base_url", erp_base_url)
-    monkeypatch.setattr(settings, "erp_login_qrcode_dir", "erp-login-qrcodes")
-    monkeypatch.setattr(settings, "public_base_url", "https://miniapp.example.com")
-    return TestClient(create_app())
-
-
-def test_erp_login_router_registered(monkeypatch):
-    client = _client(monkeypatch)
-
-    response = client.get("/api/erp-login/sessions/ticket-abc")
-
-    assert response.status_code == 400
-    assert response.json()["detail"] == "缺少ERP扫码登录会话ID"
-
-
-def test_qrcode_rejects_missing_or_bad_service_signature(monkeypatch):
-    client = _client(monkeypatch)
-    body = _json_bytes({"ticket": "ticket-abc", "session_id": 123})
-
-    missing = client.post(
-        "/api/erp-login/qrcode",
-        content=body,
-        headers={"content-type": "application/json"},
-    )
-    bad_headers = _signed_headers("POST", "/api/erp-login/qrcode", body)
-    bad_headers[SIGNATURE_HEADER] = "bad-signature"
-    bad = client.post("/api/erp-login/qrcode", content=body, headers=bad_headers)
-
-    assert missing.status_code == 401
-    assert bad.status_code == 401
-
-
-def test_qrcode_signed_request_uses_expected_miniapp_qrcode_contract(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-123.png"
-
-    monkeypatch.setattr(erp_login, "create_miniapp_qrcode", fake_create_miniapp_qrcode)
-    client = _client(monkeypatch)
-    body = _json_bytes({"ticket": "ticket-abc", "session_id": 123})
-
-    response = client.post(
-        "/api/erp-login/qrcode",
-        content=body,
-        headers=_signed_headers("POST", "/api/erp-login/qrcode", body),
-    )
-
-    assert response.status_code == 200
-    assert response.json() == {
-        "qr_url": "https://miniapp.example.com/uploads/erp-login-qrcodes/erp-login-123.png",
-        "page": "pages/erpLoginConfirm/erpLoginConfirm",
-        "scene": "t=ticket-abc&s=123",
-    }
-    assert "ticket-abc" not in calls[0]["key"]
-    assert calls == [{
-        "key": "erp-login-123",
-        "page": "pages/erpLoginConfirm/erpLoginConfirm",
-        "scene": "t=ticket-abc&s=123",
-        "public_base_url": "https://miniapp.example.com",
-        "directory": "erp-login-qrcodes",
-        "label": "ERP扫码登录",
-    }]
-    assert len(response.json()["scene"]) <= 32
-
-
-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_qrcode_rejects_scene_values_over_wechat_limit(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-long.png"
-
-    monkeypatch.setattr(erp_login, "create_miniapp_qrcode", fake_create_miniapp_qrcode)
-    client = _client(monkeypatch)
-    body = _json_bytes({"ticket": "x" * 17, "session_id": 123456789})
-
-    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)
-
-    response = client.get("/api/erp-login/sessions/ticket-abc")
-
-    assert response.status_code == 400
-    assert response.json()["detail"] == "缺少ERP扫码登录会话ID"
-
-
-def test_preview_marks_erp_session_scanned(monkeypatch):
-    from app.routers import erp_login
-
-    original_async_client = httpx.AsyncClient
-    calls = []
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        calls.append((request.method, request.url.path, request.content))
-        if request.method == "GET":
-            assert request.url.path == "/api/auth/qr-login/sessions/123/preview"
-            assert request.url.params["ticket"] == "ticket-abc"
-            return httpx.Response(
-                200,
-                json={
-                    "session_id": 123,
-                    "system_name": "嘉恒智能五金 ERP",
-                    "started_at": "2026-07-05T08:00:00",
-                    "expires_at": "2026-07-05T08:02:00",
-                    "request_ip_hint": "127.0.0.1",
-                    "user_agent_hint": "Chrome",
-                },
-            )
-        if request.method == "POST":
-            assert request.url.path == "/api/auth/qr-login/sessions/123/scanned"
-            assert json.loads(request.content.decode("utf-8")) == {"ticket": "ticket-abc"}
-            return httpx.Response(200, json={"status": "SCANNED"})
-        raise AssertionError(f"Unexpected request: {request.method} {request.url.path}")
-
-    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 == 200
-    assert [call[0] for call in calls] == ["GET", "POST"]
-
-
-def test_confirm_gets_phone_signs_erp_request_with_base_path_and_hides_auth_session(monkeypatch):
-    from app.routers import erp_login
-
-    async def fake_get_phone_number(phone_code: str) -> str:
-        assert phone_code == "wx-phone-code"
-        return "13800000000"
-
-    captured: dict[str, object] = {}
-    original_async_client = httpx.AsyncClient
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        captured["url"] = str(request.url)
-        captured["body"] = request.content
-        captured["headers"] = request.headers
-        assert request.url.path == "/erp-prefix/api/auth/qr-login/sessions/123/confirm"
-        raw_header_names = {name.decode("ascii") for name, _ in request.headers.raw}
-        assert TIMESTAMP_HEADER in raw_header_names
-        assert NONCE_HEADER in raw_header_names
-        assert SIGNATURE_HEADER in raw_header_names
-        assert "x-service-timestamp" not in request.headers
-        timestamp = request.headers[TIMESTAMP_HEADER]
-        nonce = request.headers[NONCE_HEADER]
-        expected = _test_signature(
-            "POST",
-            "/erp-prefix/api/auth/qr-login/sessions/123/confirm",
-            timestamp,
-            nonce,
-            request.content,
-        )
-        assert hmac.compare_digest(request.headers[SIGNATURE_HEADER], expected)
-        return httpx.Response(
-            200,
-            json={
-                "status": "confirmed",
-                "failure_reason": None,
-                "access_token": "must-not-leak",
-                "user": {"phone": "13800000000"},
-            },
-        )
-
-    transport = httpx.MockTransport(handler)
-
-    def fake_async_client(**kwargs):
-        _ = kwargs
-        return original_async_client(transport=transport)
-
-    monkeypatch.setattr(erp_login, "get_phone_number", fake_get_phone_number)
-    monkeypatch.setattr(erp_login.httpx, "AsyncClient", fake_async_client)
-    client = _client(monkeypatch, erp_base_url="http://erp.local/erp-prefix")
-
-    response = client.post(
-        "/api/erp-login/sessions/ticket-abc/confirm?session_id=123",
-        json={"phone_code": "wx-phone-code"},
-    )
-
-    assert response.status_code == 200
-    assert response.json() == {"status": "confirmed", "failure_reason": None}
-    assert "access_token" not in response.json()
-    erp_body = json.loads(captured["body"].decode("utf-8"))
-    assert erp_body["ticket"] == "ticket-abc"
-    assert erp_body["phone"] == "13800000000"
-    assert erp_body["confirmed_at"]
-    assert erp_body["nonce"]
-
-
-def test_erp_http_status_error_uses_generic_gateway_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(403, json={"detail": "do-not-leak"})
-
-    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_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 = []
-
-    async def fake_create_miniapp_qrcode(**kwargs):
-        calls.append(kwargs)
-        return "https://miniapp.example.com/uploads/qrcodes/device.png"
-
-    monkeypatch.setattr(wechat, "create_miniapp_qrcode", fake_create_miniapp_qrcode, raising=False)
-
-    result = asyncio.run(
-        wechat.create_device_qrcode(
-            "28#",
-            page="pages/clock/clock",
-            scene="mold=abc",
-            public_base_url="https://miniapp.example.com",
-            label="模具 28#",
-        )
-    )
-
-    assert result == "https://miniapp.example.com/uploads/qrcodes/device.png"
-    assert calls == [{
-        "key": "28#",
-        "page": "pages/clock/clock",
-        "scene": "mold=abc",
-        "public_base_url": "https://miniapp.example.com",
-        "directory": "qrcodes",
-        "label": "模具 28#",
-    }]