require_auth now uses HTTPBearer(auto_error=False) and raises 401 explicitly when the Authorization header is missing or non-Bearer, matching require_auth_401. Unauthenticated requests are 401 for every user endpoint; authenticated-but-forbidden (cross-entity / role) stays 403. - app/auth_middleware.py: single optional bearer, 401 on missing token (require_auth_401 kept as a compatibility alias) - tests: update 4 pre-existing cases that asserted 403 without a token (test_auth, test_reports, test_kpi_causality, test_expenses, test_api_auth_surface_20260914) - intentional behavior change - add tests/test_auth_401_unification_20260915.py: sample + full sweep (all AUTHED GET routes -> 401), invalid token -> 401, cross-entity and role-denied still 403, bot channel (X-BOT-KEY) and public paths unchanged
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
"""
|
||
认证模块测试
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
|
||
|
||
|
||
class TestAuth:
|
||
"""用户认证测试"""
|
||
|
||
def test_login_success(self, client: TestClient, db: Session):
|
||
"""登录成功"""
|
||
create_test_user(db)
|
||
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": "testadmin",
|
||
"password": TEST_USER_PASSWORD,
|
||
"entity_id": 1,
|
||
})
|
||
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "token" in data
|
||
assert data["user"]["username"] == "testadmin"
|
||
assert data["user"]["role"] == "ceo"
|
||
|
||
def test_login_wrong_password(self, client: TestClient, db: Session):
|
||
"""密码错误"""
|
||
create_test_user(db)
|
||
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": "testadmin",
|
||
"password": "wrongpass",
|
||
})
|
||
|
||
assert resp.status_code == 401
|
||
|
||
def test_login_nonexistent_user(self, client: TestClient):
|
||
"""用户不存在"""
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": "nobody",
|
||
"password": TEST_USER_PASSWORD,
|
||
})
|
||
|
||
assert resp.status_code == 401
|
||
|
||
def test_me_with_valid_token(self, client: TestClient, db: Session):
|
||
"""有效token获取用户信息"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.get("/api/cma/auth/me", headers={
|
||
"Authorization": f"Bearer {token}"
|
||
})
|
||
|
||
assert resp.status_code == 200
|
||
assert resp.json()["username"] == "testadmin"
|
||
|
||
def test_me_without_token(self, client: TestClient):
|
||
"""无token访问需要认证的接口"""
|
||
resp = client.get("/api/cma/auth/me")
|
||
assert resp.status_code == 401 # 语义统一(2026-09-15):未认证=401(原 HTTPBearer auto_error 给 403)
|
||
|
||
def test_register(self, client: TestClient, db: Session):
|
||
"""注册新用户"""
|
||
resp = client.post("/api/cma/auth/register", json={
|
||
"username": "newuser",
|
||
"password": "newpass123",
|
||
"name": "新用户",
|
||
"role": "business",
|
||
})
|
||
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "注册成功"
|
||
|
||
# 验证可以登录
|
||
login_resp = client.post("/api/cma/auth/login", json={
|
||
"username": "newuser",
|
||
"password": "newpass123",
|
||
"entity_id": 1,
|
||
})
|
||
assert login_resp.status_code == 200
|
||
|
||
def test_register_duplicate(self, client: TestClient, db: Session):
|
||
"""重复用户名注册"""
|
||
create_test_user(db)
|
||
|
||
resp = client.post("/api/cma/auth/register", json={
|
||
"username": "testadmin",
|
||
"password": TEST_USER_PASSWORD,
|
||
"name": "重复用户",
|
||
})
|
||
|
||
assert resp.status_code == 400
|