Pine
b9b5e2e2ff
- Task 加 task_code/tags/display_priority/claimed_by/claimed_at/doing_at；新增 TaskClaim 流水表。
- TaskService 加 claimed/doing/completed 状态机(claim/start_doing/complete)，grab 并入 claim。
- TaskRepository 加 list_published/get_by_code/update/claim/set_doing + TaskClaimRepository；挂 Database。
- 端口：
· opc /tasks/grab-by-code、/tasks/{id}/doing、/tasks/{id}/complete
· operator POST /tasks(auto task_code) + PATCH /tasks/:id
· park /park/api/tasks(大屏展示,含 scan_payload 二维码载荷)
· training /api/tasks/claim-by-code、/api/tasks/my(小程序 C 端账号→OPC 身份 find-or-create 领单)
- 迁移 0008(tasks 加列 + task_claims)已应用+stamp；seed 补 task_code/tags/grab demo。
- tests/test_task_claim.py(自包含内存库, 状态机+流水+list_published), 直接 async 校验通过。
Co-Authored-By: Claude <noreply@anthropic.com>
120 lines
5.0 KiB
Python
120 lines
5.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""培训业务落库 + 管理端打通 自动化验证。
|
|
|
|
覆盖用户侧（小程序/网页共用）测评/调研/政策/流程流程「真实写库」，以及
|
|
运营端 /admin/* 管理面能否读到这些落库结果、能否创建活动。
|
|
|
|
环境说明：
|
|
- /api/* 由 app.training 子应用提供（生产经 dispatcher /api 前缀路由），
|
|
lifespan 为空操作，建表由 alembic/seed 非运行态完成 —— 这里把训练库指向临时
|
|
文件并执行 SCHEMA，实现完全隔离。
|
|
- 平台库（app.main）在测试中预置为空（Database.initialize 不在运行时建表/种子，
|
|
平台 RBAC 测试本身依赖外部迁移+种子，非本次范围）。因此管理端验证分两层：
|
|
1) 桥接函数 list_* / create_activity 直接对同一训练库读回，证明「落库→管理可见」；
|
|
2) 匿名访问 /admin/* 应 401/403 而非 404，证明路由已注册且受 RBAC 门卫。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.services import training_admin_bridge as training
|
|
from app.training import db as tdb
|
|
from app.training.main import app as training_app
|
|
|
|
|
|
@pytest.fixture()
|
|
def training_db(tmp_path, monkeypatch, client):
|
|
"""训练库指向临时文件并建表；client 为 conftest 平台端（仅用于路由门卫校验）。"""
|
|
tmp_db = tmp_path / "train.db"
|
|
monkeypatch.setattr(tdb, "DB_PATH", str(tmp_db))
|
|
conn = tdb.get_conn()
|
|
conn.executescript(tdb.SCHEMA)
|
|
conn.commit()
|
|
conn.close()
|
|
with TestClient(training_app) as mp:
|
|
yield mp, client
|
|
|
|
|
|
def _assert_physical_row(table: str, rid: str):
|
|
row = tdb.fetch_by_id(table, rid)
|
|
assert row is not None, f"{table} 未落库"
|
|
|
|
|
|
def test_opc_test_persisted(training_db):
|
|
mp, _ = training_db
|
|
calc = mp.post("/api/tests/opc/calculate", json={"version": "quick", "answers": {}})
|
|
assert calc.status_code == 200, calc.text
|
|
r = calc.json()["result"]
|
|
report = mp.post("/api/tests", json={
|
|
"typeCode": r["typeCode"], "persona": (r["persona"] or {}).get("name", ""),
|
|
"adaptIndex": r["adaptIndex"], "adaptLevel": (r["adaptLevel"] or {}).get("level", ""),
|
|
"tracks": [t["code"] for t in (r["tracks"] or [])], "version": "quick", "username": "u_test",
|
|
})
|
|
assert report.status_code == 200, report.text
|
|
tid = report.json()["id"]
|
|
_assert_physical_row("tests", tid)
|
|
# 管理面（桥接）可读回
|
|
assert any(x.get("id") == tid for x in training.list_tests())
|
|
|
|
|
|
def test_survey_persisted(training_db):
|
|
mp, _ = training_db
|
|
res = mp.post("/api/survey/submit", json={"username": "u_survey", "source": "mp", "answers": {"q1": "A"}})
|
|
assert res.status_code == 200, res.text
|
|
sid = res.json()["id"]
|
|
_assert_physical_row("survey_logs", sid)
|
|
assert any(x.get("id") == sid for x in training.list_surveys())
|
|
|
|
|
|
def test_policy_persisted(training_db):
|
|
mp, _ = training_db
|
|
calc = mp.post("/api/policy/calculate", json={
|
|
"answers": {"status": "base", "region": "yn", "entity": "individual",
|
|
"capital": "self", "industry": ["tourism"]}})
|
|
assert calc.status_code == 200, calc.text
|
|
rr = calc.json()["result"]
|
|
res = mp.post("/api/policy-logs", json={
|
|
"username": "u_pol", "answers": {"status": "base"},
|
|
"policiesCount": len(rr["policies"]), "subsidiesCount": len(rr["subsidies"]),
|
|
"loansCount": len(rr["loans"]), "summary": "测试摘要",
|
|
})
|
|
assert res.status_code == 200, res.text
|
|
pid = res.json()["id"]
|
|
_assert_physical_row("policy_logs", pid)
|
|
assert any(x.get("id") == pid for x in training.list_policies())
|
|
|
|
|
|
def test_plan_persisted(training_db):
|
|
mp, _ = training_db
|
|
gen = mp.post("/api/plan/generate", json={"needPark": True, "needRegister": False, "hasStaff": False})
|
|
assert gen.status_code == 200, gen.text
|
|
steps = gen.json()["plan"]
|
|
res = mp.post("/api/plan-logs", json={
|
|
"username": "u_plan", "region": "yn", "status": "base",
|
|
"needPark": True, "hasStaff": False, "stepsCount": len(steps),
|
|
})
|
|
assert res.status_code == 200, res.text
|
|
lid = res.json()["id"]
|
|
_assert_physical_row("plan_logs", lid)
|
|
assert any(x.get("id") == lid for x in training.list_plans())
|
|
|
|
|
|
def test_admin_create_activity(training_db):
|
|
_mp, _ = training_db
|
|
item = training.create_activity({
|
|
"title": "新增公益课", "type": "free", "mode": "online",
|
|
"start_at": "2026-09-20T19:30:00+08:00", "capacity": 50,
|
|
})
|
|
assert item["id"]
|
|
_assert_physical_row("events", item["id"])
|
|
assert any(x.get("id") == item["id"] for x in training.list_activities())
|
|
|
|
|
|
def test_admin_routes_gated(training_db):
|
|
# /admin/surveys|policies|plans 应已注册且受 RBAC 门卫（匿名非 404）
|
|
_mp, admin = training_db
|
|
for path in ("/admin/surveys", "/admin/policies", "/admin/plans", "/admin/activities"):
|
|
res = admin.get(path)
|
|
assert res.status_code in (401, 403), f"{path} 未注册或无门卫: {res.status_code}"
|