feat: P0/P1/P2全部功能 — 四层泳道/视角切换/KPI看板/预警/差异反打/预算/知识面板/回顾会/情景预测/Excel导入/角色权限

backend/tests/test_users.py

@@ -0,0 +1,141 @@
+"""用户管理模块测试 — 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",
+        })
+        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