Hermes CI Fix
a13a080381
- 新增8个测试文件(bot_bridge/kpi_causality/cash/predict/reports/tax_compliance/expenses/probe_cost) - 增强 budget/auth/users + conftest账套模式适配 - 测试驱动修复: bot_bridge导入batch_id→source_batch; cash_forecast extra空dict - 全量: 451 passed, 1 xfailed; 报告 docs/cma-test-coverage-report.md
142 lines
4.7 KiB
Python
142 lines
4.7 KiB
Python
"""用户管理模块测试 — CRUD"""
|
|
import hashlib
|
|
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
|
|
from app.models import User
|
|
|
|
|
|
class TestUsers:
|
|
"""用户管理CRUD测试"""
|
|
|
|
BASE = "/api/cma/users"
|
|
|
|
def test_list_users_empty(self, client: TestClient, db: Session):
|
|
"""空列表（当前只有一个默认创建的管理员）"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get(self.BASE, headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert len(data) >= 1
|
|
|
|
def test_create_user(self, client: TestClient, db: Session):
|
|
"""创建新用户"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
self.BASE,
|
|
headers=auth_header(token),
|
|
json={
|
|
"username": "newuser001",
|
|
"password": "test12345",
|
|
"name": "新用户",
|
|
"role": "finance",
|
|
"phone": "13800138000",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["username"] == "newuser001"
|
|
assert data["name"] == "新用户"
|
|
assert data["role"] == "finance"
|
|
assert data["phone"] == "13800138000"
|
|
assert "password" not in data # 不返回密码
|
|
|
|
def test_create_duplicate_username(self, client: TestClient, db: Session):
|
|
"""重复用户名被拒绝"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
# 同名用户
|
|
client.post(
|
|
self.BASE,
|
|
headers=auth_header(token),
|
|
json={"username": "dupuser", "password": "pass123", "name": "重复"},
|
|
)
|
|
resp = client.post(
|
|
self.BASE,
|
|
headers=auth_header(token),
|
|
json={"username": "dupuser", "password": "pass456", "name": "重复"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "已存在" in resp.json()["detail"]
|
|
|
|
def test_update_user(self, client: TestClient, db: Session):
|
|
"""更新用户信息"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
# 创建待更新用户
|
|
create_resp = client.post(
|
|
self.BASE,
|
|
headers=auth_header(token),
|
|
json={"username": "updatable", "password": "oldpass", "name": "旧名称"},
|
|
)
|
|
user_id = create_resp.json()["id"]
|
|
|
|
resp = client.put(
|
|
f"{self.BASE}/{user_id}",
|
|
headers=auth_header(token),
|
|
json={"name": "新名称", "role": "business"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "新名称"
|
|
assert resp.json()["role"] == "business"
|
|
|
|
def test_update_user_password(self, client: TestClient, db: Session):
|
|
"""更新用户密码"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
create_resp = client.post(
|
|
self.BASE,
|
|
headers=auth_header(token),
|
|
json={"username": "passuser", "password": "oldpass", "name": "密码测试"},
|
|
)
|
|
user_id = create_resp.json()["id"]
|
|
|
|
client.put(
|
|
f"{self.BASE}/{user_id}",
|
|
headers=auth_header(token),
|
|
json={"password": "newsecret"},
|
|
)
|
|
|
|
# 验证可以用新密码登录
|
|
login_resp = client.post("/api/cma/auth/login", json={
|
|
"username": "passuser", "password": "newsecret", "entity_id": 1,
|
|
})
|
|
assert login_resp.status_code == 200
|
|
|
|
def test_delete_user(self, client: TestClient, db: Session):
|
|
"""删除用户"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
create_resp = client.post(
|
|
self.BASE,
|
|
headers=auth_header(token),
|
|
json={"username": "todelete", "password": "pass", "name": "待删除"},
|
|
)
|
|
user_id = create_resp.json()["id"]
|
|
|
|
resp = client.delete(f"{self.BASE}/{user_id}", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "已删除"
|
|
|
|
# 验证已删除
|
|
list_resp = client.get(self.BASE, headers=auth_header(token))
|
|
ids = [u["id"] for u in list_resp.json()["data"]]
|
|
assert user_id not in ids
|
|
|
|
def test_delete_user_not_found(self, client: TestClient, db: Session):
|
|
"""删除不存在的用户"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.delete(f"{self.BASE}/99999", headers=auth_header(token))
|
|
assert resp.status_code == 404
|