197 lines
8.1 KiB
Python
197 lines
8.1 KiB
Python
"""auto-verify 验证引擎修复测试 (2026-08-30 P1)
|
||
|
||
覆盖 verify.py 4 项缺陷修复:
|
||
1. OKR progress 防重复累加（同一 plan 重复 verify 不再 +15%）
|
||
2. KPIValue 回填 entity_id 多租户隔离
|
||
3. 取 KPI 最新值按 period <= 当前月 过滤（跨月验证不取未来/历史期间）
|
||
4. status "done" → "completed"（枚举外值修正）
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import ActionPlan, KPIValue, Objective
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
||
|
||
BASE = "/api/cma/verify"
|
||
|
||
|
||
def _seed_objective(db: Session, progress: int = 10) -> Objective:
|
||
obj = Objective(
|
||
entity_id=1,
|
||
title="测试目标",
|
||
quarter="2026Q3",
|
||
owner="任富海",
|
||
progress=progress,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def _seed_plan(db: Session, kpi_id: int, objective_id: int, rule: dict, status: str = "pending") -> ActionPlan:
|
||
plan = ActionPlan(
|
||
kpi_id=kpi_id,
|
||
objective_id=objective_id,
|
||
title="测试行动计划",
|
||
status=status,
|
||
priority="high",
|
||
auto_verify_rule=rule,
|
||
)
|
||
db.add(plan)
|
||
db.commit()
|
||
db.refresh(plan)
|
||
return plan
|
||
|
||
|
||
def _make_rule(kpi_code: str) -> dict:
|
||
return {
|
||
"kpi_code": kpi_code,
|
||
"condition": "LESS_THAN", # actual < target → passed
|
||
"target_value": 80,
|
||
"notify": False, # 测试不发企微
|
||
}
|
||
|
||
|
||
class TestOkrIdempotent:
|
||
def test_repeat_verify_does_not_accumulate(self, client: TestClient, db: Session):
|
||
"""缺陷1修复：同一 plan 重复 verify，OKR progress 只累加一次"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="IDEM_001", kpi_name="幂等KPI")
|
||
obj = _seed_objective(db, progress=10)
|
||
plan = _seed_plan(db, kpi.id, obj.id, _make_rule("IDEM_001"))
|
||
|
||
# 第一次验证通过 → +15%
|
||
r1 = client.post(f"{BASE}/{plan.id}", headers=auth_header(token), json={"actual_value": 50})
|
||
assert r1.status_code == 200
|
||
assert r1.json()["passed"] is True
|
||
assert r1.json()["okr_progress"]["updated"] is True
|
||
assert r1.json()["okr_progress"]["after"] == 25 # 10 + 15
|
||
|
||
# 第二次验证通过 → 不再累加（保持 25）
|
||
r2 = client.post(f"{BASE}/{plan.id}", headers=auth_header(token), json={"actual_value": 40})
|
||
assert r2.status_code == 200
|
||
assert r2.json()["okr_progress"]["updated"] is False
|
||
assert r2.json()["okr_progress"]["reason"] == "already_verified"
|
||
assert r2.json()["okr_progress"]["after"] == 25 # 保持原值
|
||
|
||
db.expire_all()
|
||
assert db.query(Objective).filter(Objective.id == obj.id).first().progress == 25
|
||
|
||
def test_force_recalc_accumulates(self, client: TestClient, db: Session):
|
||
"""缺陷1修复：force_recalc=True 保留强制重新累加入口"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="IDEM_002", kpi_name="幂等KPI2")
|
||
obj = _seed_objective(db, progress=10)
|
||
plan = _seed_plan(db, kpi.id, obj.id, _make_rule("IDEM_002"))
|
||
|
||
client.post(f"{BASE}/{plan.id}", headers=auth_header(token), json={"actual_value": 50})
|
||
r2 = client.post(f"{BASE}/{plan.id}", headers=auth_header(token),
|
||
json={"actual_value": 40, "force_recalc": True})
|
||
assert r2.json()["okr_progress"]["updated"] is True
|
||
assert r2.json()["okr_progress"]["after"] == 40 # 25 + 15
|
||
|
||
|
||
class TestEntityBackfill:
|
||
def test_kpi_value_gets_entity_id(self, client: TestClient, db: Session):
|
||
"""缺陷2修复：回填的 KPIValue 带 entity_id（与 KPI 定义一致，非默认1）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
# entity_id=2 的 KPI（模拟第二个账套）
|
||
kpi = create_test_kpi(db, kpi_code="ENT_001", kpi_name="多租户KPI", entity_id=2)
|
||
obj = _seed_objective(db)
|
||
plan = _seed_plan(db, kpi.id, obj.id, _make_rule("ENT_001"))
|
||
|
||
r = client.post(f"{BASE}/{plan.id}", headers=auth_header(token), json={"actual_value": 60})
|
||
assert r.status_code == 200
|
||
|
||
val = db.query(KPIValue).filter(KPIValue.source_batch == f"verify-plan-{plan.id}").first()
|
||
assert val is not None
|
||
assert val.entity_id == 2 # 从 KPI 定义继承，而非默认 1
|
||
assert val.kpi_id == kpi.id
|
||
|
||
|
||
class TestPeriodFilter:
|
||
def test_latest_value_respects_period_limit(self, client: TestClient, db: Session):
|
||
"""缺陷3修复：缺省 actual 时只取 period <= 当前月的值，不取未来期间"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="PER_001", kpi_name="期间KPI")
|
||
|
||
# 未来月(2026-09)有值 200 → GREATER_THAN 100 会通过；但当前月前(2026-07)值为 50 → 应取到 50
|
||
from datetime import datetime
|
||
db.add_all([
|
||
KPIValue(kpi_id=kpi.id, entity_id=1, period="2026-07", actual_value=50,
|
||
source_type="manual", data_status="verified"),
|
||
KPIValue(kpi_id=kpi.id, entity_id=1, period="2099-12", actual_value=200,
|
||
source_type="manual", data_status="verified"),
|
||
])
|
||
db.commit()
|
||
|
||
obj = _seed_objective(db)
|
||
plan = _seed_plan(db, kpi.id, obj.id, {
|
||
"kpi_code": "PER_001",
|
||
"condition": "GREATER_THAN", # actual > target
|
||
"target_value": 100,
|
||
"notify": False,
|
||
})
|
||
|
||
# 缺省 actual → 应取 2026-07 的 50 → 不通过（若错误取到 2099-12 的 200 则会通过）
|
||
r = client.post(f"{BASE}/{plan.id}", headers=auth_header(token), json={})
|
||
assert r.status_code == 200
|
||
assert r.json()["passed"] is False
|
||
assert r.json()["kpi_current_after"] == 50
|
||
|
||
def test_explicit_period_override(self, client: TestClient, db: Session):
|
||
"""缺陷3修复：调用方显式传 period 覆盖默认当前月"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="PER_002", kpi_name="期间KPI2")
|
||
db.add_all([
|
||
KPIValue(kpi_id=kpi.id, entity_id=1, period="2026-06", actual_value=30,
|
||
source_type="manual", data_status="verified"),
|
||
KPIValue(kpi_id=kpi.id, entity_id=1, period="2026-07", actual_value=90,
|
||
source_type="manual", data_status="verified"),
|
||
])
|
||
db.commit()
|
||
|
||
obj = _seed_objective(db)
|
||
plan = _seed_plan(db, kpi.id, obj.id, {
|
||
"kpi_code": "PER_002",
|
||
"condition": "GREATER_THAN",
|
||
"target_value": 50,
|
||
"notify": False,
|
||
})
|
||
|
||
# 显式 period=2026-06 → 取 30 → 不通过
|
||
r1 = client.post(f"{BASE}/{plan.id}", headers=auth_header(token), json={"period": "2026-06"})
|
||
assert r1.json()["kpi_current_after"] == 30
|
||
assert r1.json()["passed"] is False
|
||
# 显式 period=2026-07 → 取 90 → 通过
|
||
r2 = client.post(f"{BASE}/{plan.id}", headers=auth_header(token), json={"period": "2026-07"})
|
||
assert r2.json()["kpi_current_after"] == 90
|
||
assert r2.json()["passed"] is True
|
||
|
||
|
||
class TestStatusEnum:
|
||
def test_passed_plan_status_is_completed(self, client: TestClient, db: Session):
|
||
"""缺陷4修复：验证通过后 plan.status 写入枚举内值 completed，而非 done"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="ST_001", kpi_name="状态KPI")
|
||
obj = _seed_objective(db)
|
||
plan = _seed_plan(db, kpi.id, obj.id, _make_rule("ST_001"))
|
||
|
||
r = client.post(f"{BASE}/{plan.id}", headers=auth_header(token), json={"actual_value": 50})
|
||
assert r.status_code == 200
|
||
assert r.json()["passed"] is True
|
||
|
||
db.expire_all()
|
||
refreshed = db.query(ActionPlan).filter(ActionPlan.id == plan.id).first()
|
||
assert refreshed.status == "completed"
|
||
assert refreshed.status != "done"
|
||
assert refreshed.progress == 100
|