Hermes CI Fix
0d90a8e8f3
- 测试框架：SQLite内存数据库，每个测试自动建表/清理 - auth测试：登录/注册/token验证（6个用例） - kpis测试：增删改查+重复编码检查（6个用例） - maps测试：创建/编辑/发布/连线/同维度限制（6个用例） - 后端修复：KPI创建时检查编码唯一性（原为数据库抛500）
148 lines
4.0 KiB
Python
148 lines
4.0 KiB
Python
"""
|
|
管理会计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
|