feat(strategy): P0 战略基座 — strategy_charter + GET/PUT /api/cma/charter + P组入口
ci/woodpecker/push/main Pipeline was successful

- 表 strategy_charter(一账套一行,entity_id 唯一):mission/vision/vision_year/vision_measurable/values_text/value_items(JSON) 全 NOT NULL
- 迁移脚本 backend/scripts/migrate_strategy_charter_20261021.py:--check/--apply/--down(回滚先 mysqldump,有数据需 --force),零 INSERT
- API:GET 无行 → filled=false 空态(禁示例数据);PUT 字段白名单 + 账套隔离三道 + vision_year 4位年份 + value_items [{name,meaning}] 结构校验
- 前端:P 组「战略章程」入口 + MyDialog 编辑(ceo/finance 可写,其余只读);战略地图页顶部只读章程条
- 用例 43 条 + auth-surface 58 条 = 101 passed;vite build rc=0
This commit is contained in:
wecom-fullstack
2026-10-21 18:54:09 +08:00
parent 7022a05928
commit 0a6525edab
11 changed files with 1082 additions and 6 deletions
@@ -0,0 +1,249 @@
# -*- coding: utf-8 -*-
"""战略基座(strategy_charter + /api/cma/charter)验收用例 — 2026-10-21
任务:`project-20261021-cma-strategy-charter-p0-fullstack`(P0 战略基座)
覆盖(每条都是硬断言,不做“大概通过”):
A. 禁止示例数据:库内零行时 GET 200 + filled=false + data=null,且表内 rowcount=0;
B. 正常写入:PUT 落一行,六字段逐字回读(含 value_items 数组);GET 再读 filled=true;
C. 账套隔离:e1/e2 各写各的,互相 GET 不到对方;token=e1 显式 body.entity_id=2 → 403;
GET ?entity_id=<对方> → 403(平台统一语义);
D. 不可空:缺 vision_year / vision_measurable / mission / vision / values_text / value_items → 400;
显式 null → 400;空串/空白 → 400;
E. 值域:vision_year 非 4 位年份 → 400(int 2030 归一为 "2030");
value_items 非数组 / 条目缺 name|meaning / 含非白名单键 → 400;
F. 字段白名单:非白名单字段(含 id/updated_by/entity_id 手写审计字段)→ 400;
G. 鉴权/角色:无 token → 401;business 角色 PUT → 403(GET 仍 200)。
本套件**不连生产库**(conftest 已换内存 SQLite),也不插入任何“示例章程”。
"""
import hashlib
import pytest
from app.models import Entity, StrategyCharter, User, UserEntity
from tests.conftest import (
TEST_USER_PASSWORD,
auth_header,
create_test_user,
)
CHARTER = "/api/cma/charter"
VALID = {
"mission": "让每一家中国企业的经营决策都有据可依",
"vision": "成为西北最被信任的管理会计数字化伙伴",
"vision_year": "2030",
"vision_measurable": "2030 年服务 300 家付费企业、续约率 ≥85%(年报复核)",
"values_text": "客户成功|数据说话|长期主义",
"value_items": [
{"name": "客户成功", "meaning": "客户指标不改善的方案,内部再漂亮也不上"},
{"name": "数据说话", "meaning": "决策冲突时以可复核口径为准,不以职级定输赢"},
],
}
def _ensure_entity(db, eid: int, name: str) -> Entity:
ent = db.query(Entity).filter(Entity.id == eid).first()
if not ent:
ent = Entity(id=eid, name=name, short_name=name, status="active")
db.add(ent)
db.commit()
db.refresh(ent)
return ent
def _user_with_entities(db, username: str, role: str = "ceo", entity_ids=(1,)) -> User:
user = db.query(User).filter(User.username == username).first()
if not user:
user = create_test_user(
db, username=username, role=role,
password_hash=hashlib.sha256(TEST_USER_PASSWORD.encode()).hexdigest())
have = {ue.entity_id for ue in db.query(UserEntity).filter(UserEntity.user_id == user.id).all()}
for eid in entity_ids:
if eid not in have:
db.add(UserEntity(user_id=user.id, entity_id=eid))
db.commit()
return user
def _login(client, username: str, entity_id: int) -> str:
resp = client.post("/api/cma/auth/login", json={
"username": username, "password": TEST_USER_PASSWORD, "entity_id": entity_id,
})
assert resp.status_code == 200, resp.text
return resp.json()["token"]
@pytest.fixture
def ctx(client, db):
"""两账套 + 两个角色用户(ceo@e1+e2 / business@e1)"""
_ensure_entity(db, 1, "测试酣客")
_ensure_entity(db, 2, "测试博海")
_user_with_entities(db, "ceo1", "ceo", (1, 2))
_user_with_entities(db, "biz1", "business", (1,))
return {
"e1": auth_header(_login(client, "ceo1", 1)),
"e2": auth_header(_login(client, "ceo1", 2)),
"biz": auth_header(_login(client, "biz1", 1)),
}
# ── A. 禁止示例数据 + 空态契约 ────────────────────────────────────────────
def test_get_empty_state_no_seed_rows(client, db, ctx):
resp = client.get(CHARTER, headers=ctx["e1"])
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["filled"] is False
assert body["data"] is None
assert body["entity_id"] == 1
# 派单红线:禁止示例/默认文案 —— 表内零行
assert db.query(StrategyCharter).count() == 0
# ── B. 正常写入 + 回读 ───────────────────────────────────────────────────
def test_put_then_get_roundtrip(client, db, ctx):
resp = client.put(CHARTER, headers=ctx["e1"], json=VALID)
assert resp.status_code == 200, resp.text
data = resp.json()["data"]
for k, v in VALID.items():
assert data[k] == v, k
assert data["entity_id"] == 1
assert data["updated_by_name"] == "测试管理员"
row = db.query(StrategyCharter).filter(StrategyCharter.entity_id == 1).one()
assert row.vision_year == "2030"
assert row.vision_measurable == VALID["vision_measurable"]
assert row.value_items == VALID["value_items"]
got = client.get(CHARTER, headers=ctx["e1"]).json()
assert got["filled"] is True
assert got["data"]["mission"] == VALID["mission"]
assert got["data"]["vision_year"] == "2030"
def test_put_is_upsert_single_row_per_entity(client, db, ctx):
assert client.put(CHARTER, headers=ctx["e1"], json=VALID).status_code == 200
changed = dict(VALID, vision="改过的愿景", vision_year="2035")
assert client.put(CHARTER, headers=ctx["e1"], json=changed).status_code == 200
rows = db.query(StrategyCharter).filter(StrategyCharter.entity_id == 1).all()
assert len(rows) == 1, "每账套必须单行(唯一键 uq_strategy_charter_entity)"
assert rows[0].vision == "改过的愿景"
assert rows[0].vision_year == "2035"
def test_int_vision_year_normalized_to_string(client, db, ctx):
resp = client.put(CHARTER, headers=ctx["e1"], json=dict(VALID, vision_year=2030))
assert resp.status_code == 200, resp.text
assert resp.json()["data"]["vision_year"] == "2030"
# ── C. 账套隔离 ─────────────────────────────────────────────────────────
def test_entity_isolation_both_directions(client, db, ctx):
e2_payload = dict(VALID, mission="博海使命", vision="博海愿景")
assert client.put(CHARTER, headers=ctx["e1"], json=VALID).status_code == 200
assert client.put(CHARTER, headers=ctx["e2"], json=e2_payload).status_code == 200
d1 = client.get(CHARTER, headers=ctx["e1"]).json()["data"]
d2 = client.get(CHARTER, headers=ctx["e2"]).json()["data"]
assert d1["entity_id"] == 1 and d1["mission"] == VALID["mission"]
assert d2["entity_id"] == 2 and d2["mission"] == "博海使命"
assert d1["mission"] != d2["mission"]
assert db.query(StrategyCharter).count() == 2
def test_put_with_foreign_entity_id_in_body_forbidden(client, db, ctx):
resp = client.put(CHARTER, headers=ctx["e1"], json=dict(VALID, entity_id=2))
assert resp.status_code == 403, resp.text
assert "无权写入" in resp.json()["detail"]
assert db.query(StrategyCharter).count() == 0, "403 必须零写入"
def test_get_with_explicit_other_entity_query_forbidden(client, ctx):
resp = client.get(f"{CHARTER}?entity_id=2", headers=ctx["e1"])
assert resp.status_code == 403, resp.text
def test_entity_without_charter_sees_empty_state(client, db, ctx):
"""e1 有章程、e2 没有 ⇒ e2 读到 filled=false(不串 e1 数据)"""
assert client.put(CHARTER, headers=ctx["e1"], json=VALID).status_code == 200
got = client.get(CHARTER, headers=ctx["e2"]).json()
assert got["filled"] is False and got["data"] is None and got["entity_id"] == 2
# ── D. 不可空约束 ───────────────────────────────────────────────────────
@pytest.mark.parametrize("field", ["vision_year", "vision_measurable", "mission", "vision",
"values_text", "value_items"])
def test_required_fields_cannot_be_missing(client, ctx, field):
payload = {k: v for k, v in VALID.items() if k != field}
resp = client.put(CHARTER, headers=ctx["e1"], json=payload)
assert resp.status_code == 400, resp.text
assert field in resp.json()["detail"]
@pytest.mark.parametrize("field", ["vision_year", "vision_measurable", "mission", "vision",
"values_text", "value_items"])
def test_required_fields_cannot_be_null(client, ctx, field):
resp = client.put(CHARTER, headers=ctx["e1"], json=dict(VALID, **{field: None}))
assert resp.status_code == 400, resp.text
@pytest.mark.parametrize("field", ["mission", "vision", "vision_measurable", "values_text"])
def test_blank_text_rejected(client, ctx, field):
resp = client.put(CHARTER, headers=ctx["e1"], json=dict(VALID, **{field: " "}))
assert resp.status_code == 400, resp.text
assert field in resp.json()["detail"]
def test_empty_value_items_rejected(client, ctx):
resp = client.put(CHARTER, headers=ctx["e1"], json=dict(VALID, value_items=[]))
assert resp.status_code == 400, resp.text
assert "value_items" in resp.json()["detail"]
# ── E. 值域 ─────────────────────────────────────────────────────────────
@pytest.mark.parametrize("year", ["", "203", "20300", "abcd", 2030.5, True, " "])
def test_vision_year_must_be_four_digits(client, ctx, year):
resp = client.put(CHARTER, headers=ctx["e1"], json=dict(VALID, vision_year=year))
assert resp.status_code == 400, (year, resp.text)
@pytest.mark.parametrize("items", [
"客户成功", # 非数组
[{"name": "客户成功"}], # 缺 meaning
[{"name": "", "meaning": "x"}], # name 空
[{"name": "客户成功", "meaning": "x", "extra": 1}], # 非白名单键
[1, 2], # 条目非对象
])
def test_value_items_shape(client, ctx, items):
resp = client.put(CHARTER, headers=ctx["e1"], json=dict(VALID, value_items=items))
assert resp.status_code == 400, (items, resp.text)
# ── F. 字段白名单 ───────────────────────────────────────────────────────
@pytest.mark.parametrize("bad", [{"entity_id_x": 2}, {"id": 99}, {"updated_by": 7},
{"verification_status": "verified"}])
def test_unknown_fields_rejected(client, ctx, bad):
resp = client.put(CHARTER, headers=ctx["e1"], json=dict(VALID, **bad))
assert resp.status_code == 400, resp.text
assert "非白名单字段" in resp.json()["detail"]
# ── G. 鉴权 / 角色 ─────────────────────────────────────────────────────
def test_no_token_401(client, ctx):
assert client.get(CHARTER).status_code == 401
assert client.put(CHARTER, json=VALID).status_code == 401
def test_business_role_can_read_but_not_write(client, db, ctx):
assert client.get(CHARTER, headers=ctx["biz"]).status_code == 200
resp = client.put(CHARTER, headers=ctx["biz"], json=VALID)
assert resp.status_code == 403, resp.text
assert db.query(StrategyCharter).count() == 0