355 lines
13 KiB
Python
355 lines
13 KiB
Python
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.auth.providers import AuthError, JmsAuthProvider
|
||
from server.integrations.jms_auth_client import JmsAuthClient, JmsAuthClientError
|
||
|
||
|
||
class FakeJmsAuthClient:
|
||
def __init__(self) -> None:
|
||
self.login_payload: dict[str, Any] | None = None
|
||
self.logged_out = False
|
||
self.action = ""
|
||
self.user_state = "0"
|
||
self.tenant_state = "0"
|
||
|
||
async def precheck(self, *, tenant_code: str, username: str = "") -> dict[str, Any]:
|
||
assert tenant_code == "tenant-code"
|
||
return {"requireCaptcha": username == "captcha-user"}
|
||
|
||
async def tenants(self) -> list[dict[str, Any]]:
|
||
return [
|
||
{
|
||
"code": "tenant-code", "name": "测试租户", "state": "0", "deleted": 0,
|
||
"contact": "不得透传",
|
||
},
|
||
{"code": "disabled", "name": "停用租户", "state": "1", "deleted": 0},
|
||
]
|
||
|
||
async def captcha(self) -> dict[str, str]:
|
||
return {"captchaId": "captcha-id", "image": "data:image/png;base64,AA=="}
|
||
|
||
async def login(self, **payload: Any) -> dict[str, Any]:
|
||
self.login_payload = payload
|
||
return {
|
||
"tokenName": "X-JMS-Token",
|
||
"tokenValue": "remote-token",
|
||
"action": self.action,
|
||
}
|
||
|
||
async def profile(self, *, token_name: str, token_value: str) -> dict[str, Any]:
|
||
assert (token_name, token_value) == ("X-JMS-Token", "remote-token")
|
||
return {
|
||
"id": "9007199254740001",
|
||
"username": "operator",
|
||
"fullname": "计划员",
|
||
"state": self.user_state,
|
||
"roles": [{"code": "planner"}, {"code": "scheduler"}],
|
||
}
|
||
|
||
async def tenant(self, *, token_name: str, token_value: str) -> dict[str, Any]:
|
||
assert (token_name, token_value) == ("X-JMS-Token", "remote-token")
|
||
return {"uuid": "tenant-uuid-001", "code": "tenant-code", "state": self.tenant_state}
|
||
|
||
async def logout(self, *, token_name: str, token_value: str) -> None:
|
||
assert (token_name, token_value) == ("X-JMS-Token", "remote-token")
|
||
self.logged_out = True
|
||
|
||
|
||
@pytest.fixture()
|
||
def jms_provider(monkeypatch) -> tuple[JmsAuthProvider, FakeJmsAuthClient]:
|
||
monkeypatch.delenv("JMS_AUTH_TENANT_CODE", raising=False)
|
||
monkeypatch.delenv("JMS_AUTH_TENANT_NAME", raising=False)
|
||
monkeypatch.setenv("JMS_AUTH_SESSION_SECRET", "test-session-secret-with-at-least-32-chars")
|
||
fake = FakeJmsAuthClient()
|
||
return JmsAuthProvider(client=fake), fake # type: ignore[arg-type]
|
||
|
||
|
||
def test_gateway_uses_real_login_contract_and_revalidates_session(jms_provider, monkeypatch):
|
||
provider, fake = jms_provider
|
||
import server.auth.middleware as auth_middleware
|
||
import server.gateway.app as gateway_app
|
||
|
||
monkeypatch.setattr(auth_middleware, "get_auth_provider", lambda: provider)
|
||
monkeypatch.setattr(gateway_app, "get_auth_provider", lambda: provider)
|
||
client = TestClient(gateway_app.create_app())
|
||
|
||
tenants_resp = client.get("/api/auth/tenants")
|
||
assert tenants_resp.status_code == 410
|
||
assert "企业名称" in tenants_resp.json()["error"]["message"]
|
||
|
||
assert client.get("/api/auth/login/precheck", params={
|
||
"username": "operator", "tenantName": "测试租户", "tenantCode": "tenant-code",
|
||
}).json() == {
|
||
"requireCaptcha": False,
|
||
}
|
||
assert client.get("/api/auth/captcha").json() == {
|
||
"captchaId": "captcha-id",
|
||
"image": "data:image/png;base64,AA==",
|
||
}
|
||
response = client.post("/api/auth/login", json={
|
||
"method": "password",
|
||
"tenantName": "测试租户",
|
||
"tenantCode": "tenant-code",
|
||
"username": "operator",
|
||
"password": "secret",
|
||
"captchaId": "captcha-id",
|
||
"captchaCode": "A7K9",
|
||
})
|
||
assert response.status_code == 200
|
||
assert response.json()["user"] == {
|
||
"user_id": "9007199254740001",
|
||
"username": "operator",
|
||
"fullname": "计划员",
|
||
"tenant_uuid": "tenant-uuid-001",
|
||
"roles": ["planner", "scheduler"],
|
||
"expires_at": None,
|
||
"auth_kind": "user",
|
||
"license_type": None,
|
||
"license_activated_at": None,
|
||
"license_expires_at": None,
|
||
"activation_id": None,
|
||
}
|
||
assert fake.login_payload == {
|
||
"tenant_code": "tenant-code",
|
||
"tenant_name": "测试租户",
|
||
"username": "operator",
|
||
"password": "secret",
|
||
"captcha_id": "captcha-id",
|
||
"captcha_code": "A7K9",
|
||
}
|
||
assert client.get("/api/auth/me").status_code == 200
|
||
assert client.post("/api/auth/refresh").status_code == 200
|
||
assert client.post("/api/auth/logout").status_code == 200
|
||
assert fake.logged_out is True
|
||
assert client.get("/api/auth/me").status_code == 401
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_provider_rejects_disabled_user_and_required_security_action(jms_provider):
|
||
provider, fake = jms_provider
|
||
fake.user_state = "1"
|
||
with pytest.raises(AuthError, match="用户已被停用") as disabled:
|
||
await provider.login({
|
||
"method": "password", "tenantCode": "tenant-code",
|
||
"username": "operator", "password": "secret",
|
||
})
|
||
assert disabled.value.status_code == 403
|
||
|
||
fake.user_state = "0"
|
||
fake.action = "changePassword"
|
||
with pytest.raises(AuthError, match="changePassword") as action:
|
||
await provider.login({
|
||
"method": "password", "tenantCode": "tenant-code",
|
||
"username": "operator", "password": "secret",
|
||
})
|
||
assert action.value.status_code == 409
|
||
assert fake.logged_out is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_jms_http_client_sends_documented_query_body_and_token_header():
|
||
requests: list[httpx.Request] = []
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
requests.append(request)
|
||
path = request.url.path
|
||
if path == "/admin/index/tenant":
|
||
return httpx.Response(200, json={"code": 0, "data": [{
|
||
"code": "tenant-code", "name": "测试租户", "state": 0,
|
||
}]})
|
||
if path == "/admin/index/login/precheck":
|
||
return httpx.Response(200, json={"code": 0, "data": {"requireCaptcha": False}})
|
||
if path == "/admin/index/captcha":
|
||
return httpx.Response(200, json={"code": 0, "data": {
|
||
"captchaId": "cid", "image": "data:image/png;base64,AA==",
|
||
}})
|
||
if path == "/admin/index/login":
|
||
return httpx.Response(200, json={"code": 0, "data": {
|
||
"tokenName": "X-JMS-Token", "tokenValue": "remote-token",
|
||
}})
|
||
if path == "/admin/auth/profile":
|
||
return httpx.Response(200, json={"code": 0, "data": {"id": "1"}})
|
||
if path == "/admin/auth/tenant":
|
||
return httpx.Response(200, json={"code": 0, "data": {"uuid": "tenant"}})
|
||
if path == "/admin/auth/logout":
|
||
return httpx.Response(200, json={"code": 0, "data": None})
|
||
raise AssertionError(path)
|
||
|
||
client = JmsAuthClient(
|
||
base_url="https://jms.example.test",
|
||
transport=httpx.MockTransport(handler),
|
||
)
|
||
await client.tenants()
|
||
await client.precheck(tenant_code="tenant-code", username="operator")
|
||
await client.captcha()
|
||
await client.login(
|
||
tenant_code="tenant-code",
|
||
tenant_name="测试租户",
|
||
username="operator",
|
||
password="secret",
|
||
captcha_id="cid",
|
||
captcha_code="A7K9",
|
||
)
|
||
await client.profile(token_name="X-JMS-Token", token_value="remote-token")
|
||
await client.tenant(token_name="X-JMS-Token", token_value="remote-token")
|
||
await client.logout(token_name="X-JMS-Token", token_value="remote-token")
|
||
|
||
login_request = requests[3]
|
||
assert dict(login_request.url.params) == {}
|
||
assert login_request.headers["content-type"].startswith("application/x-www-form-urlencoded")
|
||
assert dict(httpx.QueryParams(login_request.content.decode("utf-8"))) == {
|
||
"tenantCode": "tenant-code",
|
||
"tenantName": "测试租户",
|
||
"username": "operator",
|
||
"password": "secret",
|
||
"device": "PC",
|
||
"captchaId": "cid",
|
||
"captchaCode": "A7K9",
|
||
}
|
||
assert all(request.headers["X-JMS-Token"] == "remote-token" for request in requests[4:])
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_provider_resolves_platform_by_display_name(jms_provider):
|
||
provider, fake = jms_provider
|
||
assert provider._select_tenant(tenant_name="平台") == {
|
||
"code": "platform",
|
||
"name": "平台",
|
||
}
|
||
await provider.login({
|
||
"method": "password",
|
||
"tenantName": "平台",
|
||
"username": "operator",
|
||
"password": "secret",
|
||
})
|
||
# Fake remote tenant still returns tenant-code; login by name alone uses alias code
|
||
assert fake.login_payload["tenant_code"] == "platform"
|
||
assert fake.login_payload["tenant_name"] == "平台"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_provider_login_by_tenant_name_without_public_list(jms_provider):
|
||
provider, fake = jms_provider
|
||
await provider.login({
|
||
"method": "password",
|
||
"tenantName": "测试租户",
|
||
"tenantCode": "tenant-code",
|
||
"username": "operator",
|
||
"password": "secret",
|
||
})
|
||
assert fake.login_payload == {
|
||
"tenant_code": "tenant-code",
|
||
"tenant_name": "测试租户",
|
||
"username": "operator",
|
||
"password": "secret",
|
||
"captcha_id": "",
|
||
"captcha_code": "",
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_jms_http_client_rejects_remote_errors():
|
||
transport = httpx.MockTransport(lambda request: httpx.Response(
|
||
401, json={"code": 401, "message": "未提供登录凭证"},
|
||
))
|
||
client = JmsAuthClient(base_url="https://jms.example.test", transport=transport)
|
||
with pytest.raises(JmsAuthClientError, match="未提供登录凭证") as exc:
|
||
await client.profile(token_name="X-JMS-Token", token_value="bad")
|
||
assert exc.value.status_code == 401
|
||
|
||
|
||
def test_jms_http_client_requires_explicit_base_url(monkeypatch):
|
||
monkeypatch.delenv("JMS_AUTH_BASE_URL", raising=False)
|
||
with pytest.raises(JmsAuthClientError, match="JMS_AUTH_BASE_URL") as exc:
|
||
JmsAuthClient()
|
||
assert exc.value.code == "AUTH_CONFIG_INVALID"
|
||
assert exc.value.status_code == 503
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize(
|
||
("envelope", "expected_code", "expected_message", "expected_status"),
|
||
[
|
||
(
|
||
{"code": 500, "message": "系统异常，请稍后重试", "error": [
|
||
'400 BAD_REQUEST "Validation failure"',
|
||
]},
|
||
"AUTH_INVALID_CREDENTIALS", "账号或密码错误", 401,
|
||
),
|
||
(
|
||
{"code": 400, "message": "图形验证码错误"},
|
||
"AUTH_CAPTCHA_INVALID", "验证码错误，请重新输入", 400,
|
||
),
|
||
(
|
||
{"code": 500, "message": "系统异常，请稍后重试"},
|
||
"AUTH_UPSTREAM_ERROR", "认证服务暂时不可用，请稍后重试", 502,
|
||
),
|
||
],
|
||
)
|
||
async def test_jms_http_client_translates_login_failures(
|
||
envelope, expected_code, expected_message, expected_status,
|
||
):
|
||
transport = httpx.MockTransport(lambda request: httpx.Response(500, json=envelope))
|
||
client = JmsAuthClient(base_url="https://jms.example.test", transport=transport)
|
||
with pytest.raises(JmsAuthClientError, match=expected_message) as exc:
|
||
await client.login(
|
||
tenant_code="tenant-code",
|
||
tenant_name="测试租户",
|
||
username="operator",
|
||
password="wrong-password",
|
||
)
|
||
assert exc.value.code == expected_code
|
||
assert exc.value.status_code == expected_status
|
||
|
||
|
||
async def _no_role_profile(**kwargs: Any) -> dict[str, Any]:
|
||
return {
|
||
"id": "9007199254740002",
|
||
"username": "operator2",
|
||
"fullname": "计划员2",
|
||
"state": "0",
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_profile_without_roles_defaults_to_planner(jms_provider, monkeypatch):
|
||
"""JMS profile 无 roles 字段（真实平台实测）→ 默认 planner，保证 P2 门禁可用。"""
|
||
monkeypatch.delenv("JMS_AUTH_DEFAULT_ROLE", raising=False)
|
||
provider, fake = jms_provider
|
||
monkeypatch.setattr(fake, "profile", _no_role_profile)
|
||
identity, _token = await provider.login({
|
||
"method": "password", "tenantCode": "tenant-code",
|
||
"username": "operator", "password": "x",
|
||
})
|
||
assert "planner" in identity.roles
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_profile_without_roles_env_override(jms_provider, monkeypatch):
|
||
monkeypatch.setenv("JMS_AUTH_DEFAULT_ROLE", "scheduler")
|
||
provider, fake = jms_provider
|
||
monkeypatch.setattr(fake, "profile", _no_role_profile)
|
||
identity, _token = await provider.login({
|
||
"method": "password", "tenantCode": "tenant-code",
|
||
"username": "operator", "password": "x",
|
||
})
|
||
assert identity.roles == ("scheduler",)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_profile_with_roles_kept(jms_provider, monkeypatch):
|
||
"""JMS 明确返回角色时不得被默认值覆盖。"""
|
||
monkeypatch.delenv("JMS_AUTH_DEFAULT_ROLE", raising=False)
|
||
provider, fake = jms_provider
|
||
identity, _token = await provider.login({
|
||
"method": "password", "tenantCode": "tenant-code",
|
||
"username": "operator", "password": "x",
|
||
})
|
||
assert set(identity.roles) == {"planner", "scheduler"}
|