test: pytest测试框架 + API测试
- 测试框架:SQLite内存数据库,每个测试自动建表/清理 - auth测试:登录/注册/token验证(6个用例) - kpis测试:增删改查+重复编码检查(6个用例) - maps测试:创建/编辑/发布/连线/同维度限制(6个用例) - 后端修复:KPI创建时检查编码唯一性(原为数据库抛500)
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user