Pine
e346d952c1
- update_booking：审核 rejected → status=rejected（终态），而非 pending - _count_enrolled：排除 rejected（被驳回不计入已报名人数） - 仪表盘 stats 补 rejected 分类 - 历史数据：1 条 rejected+pending 已同步为 rejected - 冒烟测试同步更新
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""admin 报名管理状态一致性 · 冒烟测试（临时 sqlite）。
|
|
|
|
验证：审核（audit_status）与生命周期状态（status）联动一致——
|
|
- update_booking 审核 approved → status 自动置 confirmed
|
|
- update_booking 驳回 rejected → status 置 pending
|
|
- 显式传 status 时不覆盖
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
_tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
|
|
_tmp.close()
|
|
os.environ["PINEAGENTS_DEMO_DATABASE_URL"] = f"sqlite+aiosqlite:///{_tmp.name}"
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.services import training_admin_bridge as bridge # noqa: E402
|
|
from app.training import db as tdb # noqa: E402
|
|
|
|
# 建最小 bookings 表
|
|
conn = sqlite3.connect(_tmp.name)
|
|
conn.execute("""CREATE TABLE bookings (
|
|
id TEXT PRIMARY KEY, created_at TEXT, status TEXT, username TEXT, name TEXT,
|
|
contact TEXT, status_label TEXT, want TEXT, event_id TEXT, event_title TEXT,
|
|
event_start TEXT, topics TEXT, question TEXT, source TEXT, audit_status TEXT,
|
|
pay_status TEXT, checkin_at TEXT, appeal_status TEXT, appeal_reason TEXT,
|
|
appeal_at TEXT, order_no TEXT)""")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# 重置 training 模块引擎，使其使用临时库
|
|
tdb.reset_engine()
|
|
|
|
|
|
def mk(id_, status, audit):
|
|
conn = sqlite3.connect(_tmp.name)
|
|
conn.execute("INSERT INTO bookings (id, created_at, status, username, audit_status) VALUES (?,?,?,?,?)",
|
|
(id_, "2026-09-02T00:00:00Z", status, "u_" + id_, audit))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
mk("B1", "pending", "pending") # 人工审核中
|
|
mk("B2", "confirmed", "approved") # 已通过（历史修正后）
|
|
|
|
# ① 审核通过 → status 联动 confirmed
|
|
r = bridge.update_booking("B1", {"audit_status": "approved"})
|
|
print("① 审核通过:", r["status"], "/", r["audit_status"])
|
|
assert r["status"] == "confirmed" and r["audit_status"] == "approved"
|
|
|
|
# ② 驳回 → status 联动 rejected
|
|
r = bridge.update_booking("B2", {"audit_status": "rejected"})
|
|
print("② 驳回:", r["status"], "/", r["audit_status"])
|
|
assert r["status"] == "rejected" and r["audit_status"] == "rejected"
|
|
|
|
# ③ 显式 status 不被联动覆盖
|
|
r = bridge.update_booking("B1", {"status": "arrived"})
|
|
print("③ 显式 status:", r["status"], "/", r["audit_status"])
|
|
assert r["status"] == "arrived" and r["audit_status"] == "approved"
|
|
|
|
# ④ 仅改 note 不动状态
|
|
r = bridge.update_booking("B1", {"note": "备注"})
|
|
print("④ 仅备注:", r["status"], "/", r["audit_status"])
|
|
assert r["status"] == "arrived"
|
|
|
|
os.unlink(_tmp.name)
|
|
print("\n✅ 报名审核状态联动冒烟全部通过")
|