273 lines
10 KiB
Python
273 lines
10 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())
|
|
|
|
assert client.get("/api/auth/tenants").json() == {
|
|
"tenants": [
|
|
{"code": "platform", "name": "平台"},
|
|
{"code": "tenant-code", "name": "测试租户"},
|
|
],
|
|
}
|
|
assert client.get("/api/auth/login/precheck", params={
|
|
"username": "operator", "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",
|
|
"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_supports_platform_tenant_missing_from_public_list(jms_provider):
|
|
provider, _ = jms_provider
|
|
assert await provider._resolve_tenant("platform") == {
|
|
"code": "platform",
|
|
"name": "平台",
|
|
}
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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
|