merge: test框架

2026-05-28 17:42:15 +08:00
parent f79db8ccba 0d90a8e8f3
commit 2c92168e0e
12 changed files with 501 additions and 0 deletions
@@ -99,6 +99,10 @@ def get_kpi(kpi_id: int, db: Session = Depends(get_db)):

@router.post("")
 def create_kpi(data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
+    # 检查编码唯一性
+    existing = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == data.get("kpi_code", "")).first()
+    if existing:
+        raise HTTPException(400, f"KPI编码 {data['kpi_code']} 已存在")
    kpi = KPIDefinition(**data)
    db.add(kpi)
    db.commit()
@@ -0,0 +1,13 @@
+"""
+管理会计OS 测试配置
+
+使用 SQLite 内存数据库进行测试，避免依赖外部 MySQL。
+"""
+import os
+
+# 在导入任何app模块之前设置数据库环境变量
+os.environ["CMA_DB_USER"] = "test"
+os.environ["CMA_DB_PASS"] = "test"
+os.environ["CMA_DB_HOST"] = "localhost"
+os.environ["CMA_DB_PORT"] = "3306"
+os.environ["CMA_DB_NAME"] = "test"
@@ -0,0 +1,147 @@
+"""
+管理会计OS 测试配置
+
+使用 SQLite 内存数据库进行测试，避免依赖外部 MySQL。
+测试前自动建表，测试后自动清理。
+
+重要：此文件在pytest收集测试时最先加载，确保环境变量在app模块导入前注入。
+"""
+import os
+
+# 必须在任何app模块导入之前设置环境变量（通过 pytest.ini 的 python_files 保证加载顺序）
+os.environ.setdefault("CMA_DB_USER", "test")
+os.environ.setdefault("CMA_DB_PASS", "test")
+os.environ.setdefault("CMA_DB_HOST", "localhost")
+os.environ.setdefault("CMA_DB_PORT", "3306")
+os.environ.setdefault("CMA_DB_NAME", "test")
+
+import pytest
+from typing import Generator
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.pool import StaticPool
+
+# 在 import app 模块之前就先覆盖掉 database.py 的 engine
+# 方式：直接 monkey-patch database 模块
+from app import database as db_module
+from app.database import Base
+
+# SQLite 内存引擎
+TEST_ENGINE = create_engine(
+    "sqlite:///:memory:",
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool,
+)
+TEST_SESSION_LOCAL = sessionmaker(autocommit=False, autoflush=False, bind=TEST_ENGINE)
+
+# 替换 database 模块的全局引擎
+db_module._engine = TEST_ENGINE
+db_module._SessionLocal = TEST_SESSION_LOCAL
+
+# 然后导入models（它们依赖Base）
+from app.models import (
+    User, StrategicMap, KPIDefinition, KPIValue, KPIAlert,
+    ActionPlan, OrgNode, StrategicMapVersion,
+)
+import hashlib
+
+
+@pytest.fixture(autouse=True)
+def setup_db():
+    """每个测试函数自动初始化和清理数据库"""
+    Base.metadata.create_all(bind=TEST_ENGINE)
+    yield
+    Base.metadata.drop_all(bind=TEST_ENGINE)
+
+
+@pytest.fixture
+def db() -> Generator[Session, None, None]:
+    """提供数据库 session"""
+    session = TEST_SESSION_LOCAL()
+    try:
+        yield session
+    finally:
+        session.close()
+
+
+@pytest.fixture
+def client(db) -> Generator[TestClient, None, None]:
+    """提供测试 HTTP 客户端"""
+    from app.main import app
+
+    # 重写依赖，使用测试数据库
+    app.dependency_overrides[db_module.get_db] = lambda: db
+
+    with TestClient(app) as c:
+        yield c
+
+    app.dependency_overrides.clear()
+
+
+# ── 测试数据工厂 ──
+
+def create_test_user(db: Session, **kwargs) -> User:
+    """创建测试用户"""
+    defaults = {
+        "username": "testadmin",
+        "password_hash": hashlib.sha256("admin123".encode()).hexdigest(),
+        "name": "测试管理员",
+        "role": "ceo",
+    }
+    defaults.update(kwargs)
+    user = User(**defaults)
+    db.add(user)
+    db.commit()
+    db.refresh(user)
+    return user
+
+
+def get_token_for_user(client: TestClient, username: str = "testadmin", password: str = "admin123") -> str:
+    """获取测试用户的token"""
+    resp = client.post("/api/cma/auth/login", json={
+        "username": username,
+        "password": password,
+    })
+    return resp.json()["token"]
+
+
+def auth_header(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def create_test_kpi(db: Session, **kwargs) -> KPIDefinition:
+    """创建测试KPI"""
+    defaults = {
+        "kpi_code": "TEST_001",
+        "kpi_name": "测试KPI",
+        "dimension": "finance",
+        "target_value": 100.0,
+        "unit": "%",
+        "status": "active",
+    }
+    defaults.update(kwargs)
+    kpi = KPIDefinition(**defaults)
+    db.add(kpi)
+    db.commit()
+    db.refresh(kpi)
+    return kpi
+
+
+def create_test_map(db: Session, **kwargs) -> StrategicMap:
+    """创建测试战略地图"""
+    defaults = {
+        "title": "测试地图",
+        "status": "draft",
+        "dimensions": [
+            {"key": "finance", "name": "财务维度", "icon": "💰", "color": "#409eff", "objectives": []},
+            {"key": "customer", "name": "客户维度", "icon": "🤝", "color": "#67c23a", "objectives": []},
+        ],
+        "canvas_data": {"connections": []},
+    }
+    defaults.update(kwargs)
+    m = StrategicMap(**defaults)
+    db.add(m)
+    db.commit()
+    db.refresh(m)
+    return m
@@ -0,0 +1,94 @@
+"""
+认证模块测试
+"""
+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
+
+
+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": "admin123",
+        })
+
+        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": "admin123",
+        })
+
+        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 == 403  # HTTPBearer auto_error
+
+    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",
+        })
+        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": "admin123",
+            "name": "重复用户",
+        })
+
+        assert resp.status_code == 400
@@ -0,0 +1,118 @@
+"""
+KPI字典模块测试
+"""
+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, create_test_kpi
+
+
+class TestKPIs:
+    """KPI字典CRUD测试"""
+
+    def test_list_kpis_empty(self, client: TestClient, db: Session):
+        """空列表"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        resp = client.get("/api/cma/kpis", headers=auth_header(token))
+        assert resp.status_code == 200
+        assert resp.json()["data"] == []
+
+    def test_create_kpi(self, client: TestClient, db: Session):
+        """创建KPI"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        resp = client.post(
+            "/api/cma/kpis",
+            headers=auth_header(token),
+            json={
+                "kpi_code": "F_REVENUE_002",
+                "kpi_name": "测试收入指标",
+                "dimension": "finance",
+                "target_value": 1000000,
+                "unit": "元",
+            },
+        )
+        assert resp.status_code == 200
+        assert resp.json()["kpi_code"] == "F_REVENUE_002"
+
+    def test_create_kpi_duplicate_code(self, client: TestClient, db: Session):
+        """重复KPI编码被拒绝"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        # 先创建一个
+        client.post(
+            "/api/cma/kpis",
+            headers=auth_header(token),
+            json={
+                "kpi_code": "F_REVENUE_003",
+                "kpi_name": "收入指标",
+                "dimension": "finance",
+            },
+        )
+
+        # 重复创建
+        resp = client.post(
+            "/api/cma/kpis",
+            headers=auth_header(token),
+            json={
+                "kpi_code": "F_REVENUE_003",
+                "kpi_name": "重复编码",
+                "dimension": "finance",
+            },
+        )
+        assert resp.status_code == 400
+
+    def test_update_kpi(self, client: TestClient, db: Session):
+        """编辑KPI"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="F_TEST_001")
+
+        resp = client.put(
+            f"/api/cma/kpis/{kpi.id}",
+            headers=auth_header(token),
+            json={"kpi_name": "已编辑指标", "target_value": 200},
+        )
+        assert resp.status_code == 200
+        assert resp.json()["kpi_name"] == "已编辑指标"
+        assert resp.json()["target_value"] == 200
+
+    def test_delete_kpi(self, client: TestClient, db: Session):
+        """删除KPI（软删除）"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="F_DEL_001")
+
+        resp = client.delete(
+            f"/api/cma/kpis/{kpi.id}",
+            headers=auth_header(token),
+        )
+        assert resp.status_code == 200
+
+        # 验证已被软删除（status变为非active）
+        get_resp = client.get(
+            f"/api/cma/kpis/{kpi.id}",
+            headers=auth_header(token),
+        )
+        assert get_resp.status_code == 200
+        assert get_resp.json()["status"] != "active"
+
+    def test_get_kpi_by_code(self, client: TestClient, db: Session):
+        """按编码查询KPI（通过列表过滤）"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        create_test_kpi(db, kpi_code="F_CODE_001", kpi_name="编码查询测试")
+
+        # 通过列表+参数过滤
+        resp = client.get(
+            "/api/cma/kpis?code=F_CODE_001",
+            headers=auth_header(token),
+        )
+        assert resp.status_code == 200
+        data = resp.json()["data"]
+        assert len(data) >= 1
+        assert data[0]["kpi_name"] == "编码查询测试"
@@ -0,0 +1,125 @@
+"""
+战略地图模块测试
+"""
+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
+
+
+class TestMaps:
+    """战略地图CRUD测试"""
+
+    def test_list_maps_empty(self, client: TestClient, db: Session):
+        """空列表"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        resp = client.get("/api/cma/maps", headers=auth_header(token))
+        assert resp.status_code == 200
+        assert resp.json()["data"] == []
+
+    def test_create_with_template(self, client: TestClient, db: Session):
+        """创建带模板的地图"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        resp = client.post(
+            "/api/cma/maps/create-with-template",
+            headers=auth_header(token),
+            json={"title": "测试模板地图"},
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["title"] == "测试模板地图"
+        assert data["status"] == "draft"
+        assert len(data["dimensions"]) == 4
+        assert len(data["dimensions"][0]["objectives"]) > 0
+
+    def test_update_map(self, client: TestClient, db: Session):
+        """编辑地图"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        create_resp = client.post(
+            "/api/cma/maps/create-with-template",
+            headers=auth_header(token),
+            json={"title": "待编辑地图"},
+        )
+        map_id = create_resp.json()["id"]
+
+        update_resp = client.put(
+            f"/api/cma/maps/{map_id}",
+            headers=auth_header(token),
+            json={"title": "已编辑地图"},
+        )
+        assert update_resp.status_code == 200
+        assert update_resp.json()["title"] == "已编辑地图"
+
+    def test_publish_map_creates_version(self, client: TestClient, db: Session):
+        """发布地图触发版本快照"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        create_resp = client.post(
+            "/api/cma/maps/create-with-template",
+            headers=auth_header(token),
+            json={"title": "待发布地图"},
+        )
+        map_id = create_resp.json()["id"]
+
+        # 发布
+        client.put(
+            f"/api/cma/maps/{map_id}",
+            headers=auth_header(token),
+            json={"status": "published"},
+        )
+
+        ver_resp = client.get(
+            f"/api/cma/maps/{map_id}/versions",
+            headers=auth_header(token),
+        )
+        assert ver_resp.status_code == 200
+        versions = ver_resp.json()["data"]
+        assert len(versions) >= 1
+        assert versions[0]["version"] == "v1.0"
+
+    def test_add_connection(self, client: TestClient, db: Session):
+        """添加因果连线"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        create_resp = client.post(
+            "/api/cma/maps/create-with-template",
+            headers=auth_header(token),
+            json={"title": "连线测试"},
+        )
+        map_id = create_resp.json()["id"]
+
+        resp = client.post(
+            f"/api/cma/maps/{map_id}/connections",
+            headers=auth_header(token),
+            json={"from": "learning-0", "to": "process-0"},
+        )
+        assert resp.status_code == 200
+        assert len(resp.json()["connections"]) == 1
+
+    def test_same_dim_connection_fails(self, client: TestClient, db: Session):
+        """同维度连线被拒绝"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+
+        create_resp = client.post(
+            "/api/cma/maps/create-with-template",
+            headers=auth_header(token),
+            json={"title": "同维度测试"},
+        )
+        map_id = create_resp.json()["id"]
+
+        resp = client.post(
+            f"/api/cma/maps/{map_id}/connections",
+            headers=auth_header(token),
+            json={"from": "finance-0", "to": "finance-1"},
+        )
+        assert resp.status_code == 400
+        assert "不能" in resp.json()["detail"]