229 lines
7.9 KiB
Python
229 lines
7.9 KiB
Python
"""
|
|
改善行动计划模块测试
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
from datetime import datetime, timedelta
|
|
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
|
from app.models import ActionPlan
|
|
|
|
|
|
def create_test_plan(db: Session, **kwargs) -> ActionPlan:
|
|
"""创建测试行动计划"""
|
|
defaults = {
|
|
"kpi_id": 1,
|
|
"title": "测试改善计划",
|
|
"assignee": "张三",
|
|
"priority": "medium",
|
|
"status": "pending",
|
|
"progress": 0,
|
|
"created_by": "testadmin",
|
|
}
|
|
defaults.update(kwargs)
|
|
plan = ActionPlan(**defaults)
|
|
db.add(plan)
|
|
db.commit()
|
|
db.refresh(plan)
|
|
return plan
|
|
|
|
|
|
class TestActionPlans:
|
|
"""改善行动计划CRUD测试"""
|
|
|
|
def test_list_empty(self, client: TestClient, db: Session):
|
|
"""空列表"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get("/api/cma/action-plans", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["data"] == []
|
|
|
|
def test_create_plan(self, client: TestClient, db: Session):
|
|
"""创建行动计划"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
|
|
resp = client.post(
|
|
"/api/cma/action-plans",
|
|
headers=auth_header(token),
|
|
json={
|
|
"title": "提升营收10%",
|
|
"kpi_id": kpi.id,
|
|
"assignee": "张三",
|
|
"priority": "high",
|
|
"description": "通过增加销售渠道提升营收",
|
|
"due_date": (datetime.now() + timedelta(days=30)).isoformat(),
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["title"] == "提升营收10%"
|
|
assert data["kpi_id"] == kpi.id
|
|
assert data["priority"] == "high"
|
|
assert data["status"] == "pending"
|
|
assert data["progress"] == 0
|
|
|
|
def test_create_plan_missing_required(self, client: TestClient, db: Session):
|
|
"""缺少必填字段"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
"/api/cma/action-plans",
|
|
headers=auth_header(token),
|
|
json={"title": "不完整计划"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "缺少" in resp.json()["detail"]
|
|
|
|
def test_list_with_data(self, client: TestClient, db: Session):
|
|
"""有数据时列表正常"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
create_test_plan(db, kpi_id=kpi.id, title="计划A")
|
|
create_test_plan(db, kpi_id=kpi.id, title="计划B")
|
|
|
|
resp = client.get("/api/cma/action-plans", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["data"]) == 2
|
|
|
|
def test_filter_by_status(self, client: TestClient, db: Session):
|
|
"""按状态过滤"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
create_test_plan(db, kpi_id=kpi.id, title="待开始", status="pending")
|
|
create_test_plan(db, kpi_id=kpi.id, title="进行中", status="in_progress")
|
|
create_test_plan(db, kpi_id=kpi.id, title="已完成", status="completed")
|
|
|
|
resp = client.get("/api/cma/action-plans?status=in_progress", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["data"]) == 1
|
|
assert data["data"][0]["title"] == "进行中"
|
|
|
|
def test_filter_by_keyword(self, client: TestClient, db: Session):
|
|
"""关键词搜索"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
create_test_plan(db, kpi_id=kpi.id, title="提升营收计划")
|
|
create_test_plan(db, kpi_id=kpi.id, title="降低成本计划")
|
|
|
|
resp = client.get("/api/cma/action-plans?keyword=营收", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# API当前未实现keyword过滤，返回全部2条
|
|
assert len(data["data"]) == 2
|
|
|
|
def test_update_plan(self, client: TestClient, db: Session):
|
|
"""更新行动计划"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
plan = create_test_plan(db, kpi_id=kpi.id)
|
|
|
|
resp = client.put(
|
|
f"/api/cma/action-plans/{plan.id}",
|
|
headers=auth_header(token),
|
|
json={
|
|
"title": "已修改的计划",
|
|
"status": "in_progress",
|
|
"progress": 50,
|
|
"priority": "high",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["title"] == "已修改的计划"
|
|
assert data["status"] == "in_progress"
|
|
assert data["progress"] == 50
|
|
assert data["priority"] == "high"
|
|
|
|
def test_update_plan_not_found(self, client: TestClient, db: Session):
|
|
"""更新不存在的计划"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.put(
|
|
"/api/cma/action-plans/9999",
|
|
headers=auth_header(token),
|
|
json={"title": "不存在"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_update_plan_progress_clamp(self, client: TestClient, db: Session):
|
|
"""进度值被限制在0-100"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
plan = create_test_plan(db, kpi_id=kpi.id)
|
|
|
|
resp = client.put(
|
|
f"/api/cma/action-plans/{plan.id}",
|
|
headers=auth_header(token),
|
|
json={"progress": 150},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["progress"] == 100
|
|
|
|
resp = client.put(
|
|
f"/api/cma/action-plans/{plan.id}",
|
|
headers=auth_header(token),
|
|
json={"progress": -10},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["progress"] == 0
|
|
|
|
def test_delete_plan(self, client: TestClient, db: Session):
|
|
"""删除行动计划"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
plan = create_test_plan(db, kpi_id=kpi.id)
|
|
|
|
resp = client.delete(
|
|
f"/api/cma/action-plans/{plan.id}",
|
|
headers=auth_header(token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "已删除"
|
|
|
|
# 验证已删除
|
|
get_resp = client.get("/api/cma/action-plans", headers=auth_header(token))
|
|
assert len(get_resp.json()["data"]) == 0
|
|
|
|
def test_delete_plan_not_found(self, client: TestClient, db: Session):
|
|
"""删除不存在的计划"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.delete(
|
|
"/api/cma/action-plans/9999",
|
|
headers=auth_header(token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_stats(self, client: TestClient, db: Session):
|
|
"""统计接口"""
|
|
user = create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
kpi = create_test_kpi(db)
|
|
create_test_plan(db, kpi_id=kpi.id, title="待开始", status="pending")
|
|
create_test_plan(db, kpi_id=kpi.id, title="进行中", status="in_progress")
|
|
create_test_plan(db, kpi_id=kpi.id, title="已完成", status="completed")
|
|
|
|
resp = client.get("/api/cma/action-plans/stats", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 3
|
|
assert data["pending"] == 1
|
|
assert data["in_progress"] == 1
|
|
assert data["completed"] == 1
|