96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
|
"""S2 接缝适配器单测：宿主 AuthContext → 模块 AuthContext。
|
||
|
|
|
||
|
|
适配器当前为预制件（未接线），本测试保证它随时可用、且语义是 fail-closed。
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from types import SimpleNamespace
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.api.auth_adapter import (
|
||
|
|
HostAuthAdapterError,
|
||
|
|
from_host_auth,
|
||
|
|
perm_matches,
|
||
|
|
resolve_agent_type,
|
||
|
|
)
|
||
|
|
from app.utils.trace import current_trace, reset_trace, set_trace
|
||
|
|
|
||
|
|
|
||
|
|
def _host(**kw):
|
||
|
|
"""构造一个宿主风格的 AuthContext（字段与 main 的 schemas.AuthContext 对齐）。"""
|
||
|
|
base = dict(
|
||
|
|
sub="STAFF-90001",
|
||
|
|
token_type="staff",
|
||
|
|
roles=["risk_officer"],
|
||
|
|
permissions=["risk:alert:write"],
|
||
|
|
tenant_id="default",
|
||
|
|
trace_id="trace-host-001",
|
||
|
|
agent_type="risk",
|
||
|
|
jti="jti-001",
|
||
|
|
customer_id=None,
|
||
|
|
)
|
||
|
|
base.update(kw)
|
||
|
|
return SimpleNamespace(**base)
|
||
|
|
|
||
|
|
|
||
|
|
def test_sub_maps_to_actor_id():
|
||
|
|
ctx = from_host_auth(_host(), bind_trace=False)
|
||
|
|
assert ctx.actor_id == "STAFF-90001"
|
||
|
|
|
||
|
|
|
||
|
|
def test_roles_and_permissions_passthrough():
|
||
|
|
ctx = from_host_auth(_host(roles=["risk_officer", "risk_demo"]), bind_trace=False)
|
||
|
|
assert ctx.has_role("risk_officer", "compliance") is True
|
||
|
|
assert ctx.has_permission("risk:alert:write") is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_customer_token_fallbacks_customer_id_to_sub():
|
||
|
|
ctx = from_host_auth(_host(sub="CUST-1001", token_type="customer"), bind_trace=False)
|
||
|
|
assert ctx.customer_id == "CUST-1001"
|
||
|
|
assert ctx.is_customer() is False # 角色集合仍为空，归属判定不因 token_type 放宽
|
||
|
|
|
||
|
|
|
||
|
|
def test_missing_sub_is_fail_closed():
|
||
|
|
bad = SimpleNamespace(sub=None, actor_id=None, roles=[], token_type="staff")
|
||
|
|
with pytest.raises(HostAuthAdapterError):
|
||
|
|
from_host_auth(bad, bind_trace=False)
|
||
|
|
|
||
|
|
|
||
|
|
def test_empty_sub_string_is_fail_closed():
|
||
|
|
bad = SimpleNamespace(sub="", roles=[], token_type="staff")
|
||
|
|
with pytest.raises(HostAuthAdapterError):
|
||
|
|
from_host_auth(bad, bind_trace=False)
|
||
|
|
|
||
|
|
|
||
|
|
def test_trace_binding_writes_contextvar():
|
||
|
|
token = set_trace("before")
|
||
|
|
try:
|
||
|
|
from_host_auth(_host(trace_id="trace-host-999"), bind_trace=True)
|
||
|
|
assert current_trace() == "trace-host-999"
|
||
|
|
finally:
|
||
|
|
reset_trace(token)
|
||
|
|
|
||
|
|
|
||
|
|
def test_trace_not_bound_when_disabled():
|
||
|
|
token = set_trace("keep-me")
|
||
|
|
try:
|
||
|
|
from_host_auth(_host(trace_id="trace-host-999"), bind_trace=False)
|
||
|
|
assert current_trace() == "keep-me"
|
||
|
|
finally:
|
||
|
|
reset_trace(token)
|
||
|
|
|
||
|
|
|
||
|
|
def test_perm_matches_exact_and_wildcard():
|
||
|
|
perms = ["risk:alert:read", "audit:*"]
|
||
|
|
assert perm_matches("risk:alert:read", perms) is True
|
||
|
|
assert perm_matches("audit:read:all", perms) is True # 命中 audit:* 通配
|
||
|
|
assert perm_matches("trade:execute", perms) is False
|
||
|
|
assert perm_matches("", perms) is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_resolve_agent_type():
|
||
|
|
assert resolve_agent_type(_host(agent_type="risk")) == "risk"
|
||
|
|
assert resolve_agent_type(SimpleNamespace(agent_type=None)) == "risk" # 默认回退
|