393 lines
13 KiB
Python
393 lines
13 KiB
Python
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#",
|
|
}]
|