JhHardwareWRS_BackPoint/tests/test_erp_login.py

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-ticket-abc.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-ticket-abc.png",
        "page": "pages/erpLoginConfirm/erpLoginConfirm",
        "scene": "ticket=ticket-abc&session_id=123",
    }
    assert calls == [{
        "key": "erp-login-ticket-abc",
        "page": "pages/erpLoginConfirm/erpLoginConfirm",
        "scene": "ticket=ticket-abc&session_id=123",
        "public_base_url": "https://miniapp.example.com",
        "directory": "erp-login-qrcodes",
        "label": "ERP扫码登录",
    }]


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_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_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#",
    }]