Pine
ac4c1aabe4
- scripts/migrate_sqlite_to_mysql.py：离线迁移（MySQL 建表→alembic stamp head→ 逐表拷贝[NOT NULL 列类型化兜底]→自增回填→行数校验），支持 --drop-existing 重建与增量续跑 - tests/test_training_persist.py：夹具改 monkeypatch DATABASE_URL + reset_engine， 建表统一走 Base.metadata（原 DB_PATH/SCHEMA 随 sqlite3 直连移除）
124 lines
5.2 KiB
Python
124 lines
5.2 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):
|
|
"""训练库指向临时 SQLite 并建表（统一 Base.metadata）；client 为 conftest 平台端。"""
|
|
from app import config
|
|
from app.infrastructure import models # noqa: F401 注册全部 ORM 到 metadata
|
|
from app.infrastructure.db import Base
|
|
|
|
monkeypatch.setattr(config, "DATABASE_URL", f"sqlite+aiosqlite:///{tmp_path / 'train.db'}")
|
|
tdb.reset_engine()
|
|
Base.metadata.create_all(tdb._get_engine())
|
|
try:
|
|
with TestClient(training_app) as mp:
|
|
yield mp, client
|
|
finally:
|
|
tdb.reset_engine()
|
|
|
|
|
|
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}"
|