1211 lines
53 KiB
Python
Executable File
1211 lines
53 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
RailwaySystem - Core Business Scenario Integration Test Runner
|
||
全流程自动化测试套件：
|
||
- SCENARIO-01: Happy Path 完整业务链 (Teacher/Course/Classroom/TrainingSite -> Plan -> PlanCourse -> Enrollment -> Lesson -> Timetable -> Plan Complete)
|
||
- SCENARIO-02: Enrollment 异常规则与状态区分 (同一Plan重复40030, 超过capacity40031, 不存在学员reject, status=0停用学员reject, status=1正常学员success)
|
||
- SCENARIO-03: Scheduling 排课预检Warning与Hard Block (教师冲突TEACHER, 教室冲突CLASSROOM Warning, 场地冲突SITE Warning, 计划重叠Warning; invalid time 40010, Plan locked 40012, venue XOR reject, cross-plan reject, 资源不存在reject)
|
||
- SCENARIO-04: Electronic Board 绑定、详情与Timetable映射 (UUID主键, status=0, Classroom boards列表0/1/N, Timetable boardSn 0/1/N规则, 逻辑删除del_flag=2)
|
||
- SCENARIO-05: Plan Completion 结训生命周期与退训保护 (Plan->2, 活动学员->4且complete_time写入, 退训学员保持3且不被覆盖)
|
||
- Kingbase 数据库只读安全断言
|
||
- 精确ID资源安全清理
|
||
- 自动生成结构化测试报告与summary.md
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import time
|
||
import string
|
||
import random
|
||
import urllib.request
|
||
import urllib.parse
|
||
import urllib.error
|
||
import subprocess
|
||
from datetime import datetime, timedelta
|
||
|
||
def resolve_base_url():
|
||
if "BASE_URL" in os.environ:
|
||
return os.environ["BASE_URL"]
|
||
# 优先检测 8094 (当前worktree独立实例), 其次 8092 (最新基线端口), 其次 8090
|
||
for port in [8094, 8092, 8090]:
|
||
try:
|
||
req = urllib.request.Request(f"http://localhost:{port}/login", data=b"{}", headers={"Content-Type": "application/json"})
|
||
with urllib.request.urlopen(req, timeout=1.5) as resp:
|
||
return f"http://localhost:{port}"
|
||
except urllib.error.HTTPError:
|
||
return f"http://localhost:{port}"
|
||
except Exception:
|
||
continue
|
||
return "http://localhost:8092"
|
||
|
||
BASE_URL = resolve_base_url()
|
||
RUN_ID = f"QA_W2_{datetime.now().strftime('%Y%m%d%H%M%S')}"
|
||
OUTPUT_DIR = f"/tmp/railwaysystem-qa/{RUN_ID}"
|
||
|
||
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
||
|
||
ADMIN_USER = os.environ.get("QA_ADMIN_USERNAME")
|
||
ADMIN_PASS = os.environ.get("QA_ADMIN_PASSWORD")
|
||
|
||
if not ADMIN_USER or not ADMIN_PASS:
|
||
print("[FATAL] 缺少必须的管理端认证凭据环境变量 (QA_ADMIN_USERNAME / QA_ADMIN_PASSWORD)，FAIL FAST 终止执行。")
|
||
print("请先通过安全机制注入环境变量后再执行此 Integration Gate。")
|
||
sys.exit(1)
|
||
|
||
created_resources = {
|
||
"run_id": RUN_ID,
|
||
"teachers": [],
|
||
"courses": [],
|
||
"classrooms": [],
|
||
"sites": [],
|
||
"students": [],
|
||
"plans": [],
|
||
"plan_courses": [],
|
||
"enrollments": [],
|
||
"lessons": [],
|
||
"boards": []
|
||
}
|
||
|
||
api_results = []
|
||
http_500_count = 0
|
||
scenario_stats = {
|
||
"SCENARIO-01": {"name": "Happy Path Full Business Chain", "passed": False, "steps": []},
|
||
"SCENARIO-02": {"name": "Enrollment Exception & Student Status Validation", "passed": False, "steps": []},
|
||
"SCENARIO-03": {"name": "Scheduling Conflict Warning & Hard Blocks", "passed": False, "steps": []},
|
||
"SCENARIO-04": {"name": "Electronic Board Binding & Timetable Mapping", "passed": False, "steps": []},
|
||
"SCENARIO-05": {"name": "Plan Completion Lifecycle & History Retention", "passed": False, "steps": []}
|
||
}
|
||
|
||
TOKEN = None
|
||
|
||
def log(msg):
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}")
|
||
|
||
def request_api(method, path, data=None, headers=None, expected_status=(200,)):
|
||
global http_500_count
|
||
if headers is None:
|
||
headers = {}
|
||
headers["Content-Type"] = "application/json"
|
||
if TOKEN:
|
||
headers["Authorization"] = f"Bearer {TOKEN}"
|
||
|
||
raw_url = f"{BASE_URL}{path}"
|
||
parsed = urllib.parse.urlsplit(raw_url)
|
||
safe_url = urllib.parse.urlunsplit((
|
||
parsed.scheme,
|
||
parsed.netloc,
|
||
urllib.parse.quote(parsed.path),
|
||
urllib.parse.quote(parsed.query, safe="=&?-_.~"),
|
||
parsed.fragment
|
||
))
|
||
|
||
encoded_data = json.dumps(data).encode("utf-8") if data is not None else None
|
||
req = urllib.request.Request(safe_url, data=encoded_data, headers=headers, method=method)
|
||
start_time = datetime.now()
|
||
http_status = None
|
||
resp_body = {}
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
http_status = resp.status
|
||
raw_text = resp.read().decode("utf-8", errors="replace")
|
||
try:
|
||
resp_body = json.loads(raw_text)
|
||
except Exception:
|
||
resp_body = {"raw": raw_text}
|
||
except urllib.error.HTTPError as e:
|
||
http_status = e.code
|
||
if http_status >= 500:
|
||
http_500_count += 1
|
||
raw_text = e.read().decode("utf-8", errors="replace")
|
||
try:
|
||
resp_body = json.loads(raw_text)
|
||
except Exception:
|
||
resp_body = {"raw": raw_text}
|
||
except Exception as ex:
|
||
http_status = 599
|
||
http_500_count += 1
|
||
resp_body = {"error": str(ex)}
|
||
|
||
duration_ms = int((datetime.now() - start_time).total_seconds() * 1000)
|
||
|
||
sanitized_data = data
|
||
if isinstance(data, dict):
|
||
sanitized_data = {k: ("***" if "pass" in k.lower() or "token" in k.lower() else v) for k, v in data.items()}
|
||
|
||
api_entry = {
|
||
"timestamp": datetime.now().isoformat(),
|
||
"method": method,
|
||
"path": path,
|
||
"payload": sanitized_data,
|
||
"http_status": http_status,
|
||
"business_code": resp_body.get("code") if isinstance(resp_body, dict) else None,
|
||
"duration_ms": duration_ms,
|
||
"success": http_status in expected_status
|
||
}
|
||
api_results.append(api_entry)
|
||
return http_status, resp_body
|
||
|
||
def extract_id(body):
|
||
"""从 RuoYi 接口返回体安全提取 ID (兼顾 data 字段与 String 类型的 msg 字段)"""
|
||
if not isinstance(body, dict):
|
||
return None
|
||
code = body.get("code")
|
||
if code != 200:
|
||
return None
|
||
val = body.get("data")
|
||
if val is not None:
|
||
return val
|
||
msg = body.get("msg")
|
||
if msg and msg not in ("操作成功", "查询成功", "success"):
|
||
return msg
|
||
return None
|
||
|
||
def login():
|
||
global TOKEN
|
||
log(f"正在通过安全机制获取系统访问Token (后端: {BASE_URL})...")
|
||
status, body = request_api("POST", "/login", {"username": ADMIN_USER, "password": ADMIN_PASS})
|
||
if status == 200 and body.get("code") == 200:
|
||
TOKEN = body.get("token")
|
||
log("系统登录成功，Token就绪")
|
||
return True
|
||
else:
|
||
log(f"系统登录失败: status={status}, body={body}")
|
||
return False
|
||
|
||
def random_phone(prefix="138"):
|
||
return prefix + "".join(random.choices(string.digits, k=8))
|
||
|
||
# ==============================================================================
|
||
# SCENARIO-01: Happy Path
|
||
# ==============================================================================
|
||
def run_scenario_01():
|
||
log("=== 开始执行 SCENARIO-01: Happy Path 完整业务链 ===")
|
||
steps = scenario_stats["SCENARIO-01"]["steps"]
|
||
success = True
|
||
|
||
now = datetime.now()
|
||
t_start = (now + timedelta(days=20)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_end = (now + timedelta(days=25)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_deadline = (now + timedelta(days=18)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 1. 创建/准备 Teacher
|
||
teacher_name = f"{RUN_ID}_教师_王老师"
|
||
t_phone = random_phone("138")
|
||
t_id_custom = f"{RUN_ID}_T01"
|
||
status, body = request_api("POST", "/train/teacher", {
|
||
"tId": t_id_custom,
|
||
"tName": teacher_name,
|
||
"tPhone": t_phone,
|
||
"gender": "0",
|
||
"title": "高级讲师"
|
||
})
|
||
t_id = extract_id(body) or t_id_custom
|
||
if not t_id:
|
||
_, list_body = request_api("GET", f"/train/teacher/list?tName={teacher_name}")
|
||
rows = list_body.get("rows", [])
|
||
if rows: t_id = rows[0].get("tId")
|
||
if t_id: created_resources["teachers"].append(t_id)
|
||
step1_ok = (status == 200 and t_id is not None)
|
||
steps.append({"step": 1, "action": "准备Teacher", "status": status, "id": t_id, "pass": step1_ok})
|
||
if not step1_ok: success = False
|
||
|
||
# 2. 创建/准备 Course
|
||
course_name = f"{RUN_ID}_课程_列控系统"
|
||
course_code = f"{RUN_ID}_C1"
|
||
c_id_custom = f"{RUN_ID}_C01"
|
||
status, body = request_api("POST", "/train/course", {
|
||
"cId": c_id_custom,
|
||
"cName": course_name,
|
||
"cCode": course_code,
|
||
"cDuration": 16,
|
||
"status": 1
|
||
})
|
||
c_id = extract_id(body) or c_id_custom
|
||
if not c_id:
|
||
_, list_body = request_api("GET", f"/train/course/list?cName={course_name}")
|
||
rows = list_body.get("rows", [])
|
||
if rows: c_id = rows[0].get("cId")
|
||
if c_id: created_resources["courses"].append(c_id)
|
||
step2_ok = (status == 200 and c_id is not None)
|
||
steps.append({"step": 2, "action": "准备Course", "status": status, "id": c_id, "pass": step2_ok})
|
||
if not step2_ok: success = False
|
||
|
||
# 3. 创建 Classroom
|
||
room_name = f"{RUN_ID}_教室_101"
|
||
cr_id_custom = f"{RUN_ID}_CR01"
|
||
status, body = request_api("POST", "/train/classroom", {
|
||
"classRoomId": cr_id_custom,
|
||
"roomName": room_name,
|
||
"roomLocation": "实训楼101",
|
||
"capacity": 50,
|
||
"roomStatus": 1
|
||
})
|
||
cr_id = extract_id(body) or cr_id_custom
|
||
if not cr_id:
|
||
_, list_body = request_api("GET", f"/train/classroom/list?roomName={room_name}")
|
||
rows = list_body.get("rows", [])
|
||
if rows: cr_id = rows[0].get("classRoomId")
|
||
if cr_id: created_resources["classrooms"].append(cr_id)
|
||
step3_ok = (status == 200 and cr_id is not None)
|
||
steps.append({"step": 3, "action": "创建Classroom", "status": status, "id": cr_id, "pass": step3_ok})
|
||
if not step3_ok: success = False
|
||
|
||
# 4. 创建 TrainingSite
|
||
site_name = f"{RUN_ID}_场地_列控中心"
|
||
status, body = request_api("POST", "/train/training-site", {
|
||
"siteName": site_name,
|
||
"siteLocation": "演练基地A区",
|
||
"siteStatus": 1,
|
||
"capacity": 50
|
||
})
|
||
site_id = extract_id(body)
|
||
if not site_id:
|
||
_, list_body = request_api("GET", f"/train/training-site/list?siteName={site_name}")
|
||
rows = list_body.get("rows", [])
|
||
if rows: site_id = rows[0].get("siteId")
|
||
if site_id: created_resources["sites"].append(site_id)
|
||
step4_ok = (status == 200 and site_id is not None)
|
||
steps.append({"step": 4, "action": "创建TrainingSite", "status": status, "id": site_id, "pass": step4_ok})
|
||
if not step4_ok: success = False
|
||
|
||
# 5. 创建 Student
|
||
s_name = f"{RUN_ID}_学员_张三"
|
||
status, body = request_api("POST", "/train/student", {
|
||
"sName": s_name,
|
||
"sGender": "1",
|
||
"sPhone": random_phone("139"),
|
||
"orgName": "京局北京机务段"
|
||
})
|
||
s_id = extract_id(body)
|
||
if not s_id:
|
||
_, list_body = request_api("GET", f"/train/student/list?sName={s_name}")
|
||
rows = list_body.get("rows", [])
|
||
if rows: s_id = rows[0].get("sId")
|
||
if s_id: created_resources["students"].append(s_id)
|
||
step5_ok = (status == 200 and s_id is not None)
|
||
steps.append({"step": 5, "action": "创建Student", "status": status, "id": s_id, "pass": step5_ok})
|
||
if not step5_ok: success = False
|
||
|
||
# 6. 创建 Plan
|
||
plan_name = f"{RUN_ID}_培训班_车载班"
|
||
status, body = request_api("POST", "/train/plans", {
|
||
"planName": plan_name,
|
||
"startTime": t_start,
|
||
"endTime": t_end,
|
||
"reportDeadline": t_deadline,
|
||
"recruitNum": 30,
|
||
"trainAddress": "北京培训基地"
|
||
})
|
||
plan_id = extract_id(body)
|
||
if plan_id: created_resources["plans"].append(plan_id)
|
||
step6_ok = (status == 200 and plan_id is not None)
|
||
steps.append({"step": 6, "action": "创建Plan", "status": status, "id": plan_id, "pass": step6_ok})
|
||
if not step6_ok: success = False
|
||
|
||
# 7. 添加 PlanCourse
|
||
status, body = request_api("POST", f"/train/plans/{plan_id}/courses", {
|
||
"cIds": [c_id],
|
||
"remark": "列控专业课程"
|
||
})
|
||
_, pc_list_body = request_api("GET", f"/train/plans/{plan_id}/courses")
|
||
plan_course_id = None
|
||
pc_data = pc_list_body.get("data", [])
|
||
if pc_data:
|
||
plan_course_id = pc_data[0].get("planCourseId")
|
||
created_resources["plan_courses"].append(plan_course_id)
|
||
step7_ok = (status == 200 and plan_course_id is not None)
|
||
steps.append({"step": 7, "action": "配置PlanCourse", "status": status, "id": plan_course_id, "pass": step7_ok})
|
||
if not step7_ok: success = False
|
||
|
||
# 8. Enrollment 入班
|
||
status, body = request_api("POST", f"/train/plans/{plan_id}/enrollments", {
|
||
"studentId": s_id,
|
||
"sourceType": 1,
|
||
"accommodationRequired": 0,
|
||
"remark": "首期学员"
|
||
})
|
||
enrollment_id = extract_id(body)
|
||
if not enrollment_id:
|
||
_, e_list_body = request_api("GET", f"/train/plans/{plan_id}/enrollments")
|
||
rows = e_list_body.get("rows", [])
|
||
if rows: enrollment_id = rows[0].get("enrollmentId")
|
||
if enrollment_id: created_resources["enrollments"].append(enrollment_id)
|
||
step8_ok = (status == 200 and enrollment_id is not None)
|
||
steps.append({"step": 8, "action": "Enrollment入班", "status": status, "id": enrollment_id, "pass": step8_ok})
|
||
if not step8_ok: success = False
|
||
|
||
# 9. 创建 Lesson (排课)
|
||
lesson_start = (now + timedelta(days=20, hours=1)).strftime("%Y-%m-%d %H:%M:%S")
|
||
lesson_end = (now + timedelta(days=20, hours=3)).strftime("%Y-%m-%d %H:%M:%S")
|
||
status, body = request_api("POST", f"/train/plans/{plan_id}/lessons", {
|
||
"planCourseId": plan_course_id,
|
||
"tId": t_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": lesson_start,
|
||
"endTime": lesson_end,
|
||
"remark": "第1讲理论课"
|
||
})
|
||
lesson_id = extract_id(body)
|
||
if lesson_id: created_resources["lessons"].append(lesson_id)
|
||
step9_ok = (status == 200 and lesson_id is not None)
|
||
steps.append({"step": 9, "action": "创建Lesson", "status": status, "id": lesson_id, "pass": step9_ok})
|
||
if not step9_ok: success = False
|
||
|
||
# 10. 查询 Timetable
|
||
status, body = request_api("GET", f"/train/plans/{plan_id}/timetable")
|
||
tt_data = body.get("data", {})
|
||
days = tt_data.get("days", [])
|
||
step10_ok = (status == 200 and len(days) > 0)
|
||
steps.append({"step": 10, "action": "查询Timetable", "status": status, "days_count": len(days), "pass": step10_ok})
|
||
if not step10_ok: success = False
|
||
|
||
# 11. 验证 Lesson 显示字段
|
||
lesson_ok = False
|
||
if days and len(days[0].get("lessons", [])) > 0:
|
||
l_item = days[0]["lessons"][0]
|
||
if (l_item.get("lessonId") == lesson_id and
|
||
l_item.get("courseName") == course_name and
|
||
l_item.get("teacherName") == teacher_name and
|
||
l_item.get("classRoomName") == room_name):
|
||
lesson_ok = True
|
||
steps.append({"step": 11, "action": "验证Lesson显示字段", "status": status, "pass": lesson_ok})
|
||
if not lesson_ok: success = False
|
||
|
||
# 12. 验证 PlanCourse 字段
|
||
pc_ok = (plan_course_id is not None and pc_data and pc_data[0].get("cId") == c_id)
|
||
steps.append({"step": 12, "action": "验证PlanCourse字段", "status": status, "pass": pc_ok})
|
||
if not pc_ok: success = False
|
||
|
||
# 13. 完成 Plan
|
||
status, body = request_api("POST", f"/train/plans/{plan_id}/complete")
|
||
step13_ok = (status == 200 and body.get("code") == 200)
|
||
steps.append({"step": 13, "action": "完成Plan", "status": status, "pass": step13_ok})
|
||
if not step13_ok: success = False
|
||
|
||
# 14 & 15. 验证 Enrollment 自动变更为 4 且 complete_time 非空
|
||
status, enroll_list_body = request_api("GET", f"/train/plans/{plan_id}/enrollments")
|
||
e_rows = enroll_list_body.get("rows", [])
|
||
step14_ok = False
|
||
step15_ok = False
|
||
if e_rows:
|
||
target_e = e_rows[0]
|
||
if target_e.get("enrollmentStatus") == 4:
|
||
step14_ok = True
|
||
if target_e.get("completeTime") is not None:
|
||
step15_ok = True
|
||
steps.append({"step": 14, "action": "验证Enrollment自动变更为4", "status": status, "pass": step14_ok})
|
||
steps.append({"step": 15, "action": "验证complete_time非空", "status": status, "pass": step15_ok})
|
||
if not (step14_ok and step15_ok): success = False
|
||
|
||
scenario_stats["SCENARIO-01"]["passed"] = success
|
||
log(f"SCENARIO-01 结果: {'PASS' if success else 'FAIL'}")
|
||
return success
|
||
|
||
# ==============================================================================
|
||
# SCENARIO-02: Enrollment 异常与状态区分
|
||
# ==============================================================================
|
||
def run_scenario_02():
|
||
log("=== 开始执行 SCENARIO-02: Enrollment 异常规则与状态区分 ===")
|
||
steps = scenario_stats["SCENARIO-02"]["steps"]
|
||
success = True
|
||
|
||
now = datetime.now()
|
||
t_start = (now + timedelta(days=20)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_end = (now + timedelta(days=25)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_deadline = (now + timedelta(days=18)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 1. 准备学员1 (正常)
|
||
_, s1_body = request_api("POST", "/train/student", {
|
||
"sName": f"{RUN_ID}_学员_正常1",
|
||
"sGender": "1",
|
||
"sPhone": random_phone("137"),
|
||
"orgName": "机务段"
|
||
})
|
||
stu1_id = extract_id(s1_body)
|
||
if stu1_id: created_resources["students"].append(stu1_id)
|
||
|
||
# 准备学员2 (正常)
|
||
_, s2_body = request_api("POST", "/train/student", {
|
||
"sName": f"{RUN_ID}_学员_正常2",
|
||
"sGender": "1",
|
||
"sPhone": random_phone("136"),
|
||
"orgName": "供电段"
|
||
})
|
||
stu2_id = extract_id(s2_body)
|
||
if stu2_id: created_resources["students"].append(stu2_id)
|
||
|
||
# 准备学员3 (停用学员 status=0)
|
||
_, s3_body = request_api("POST", "/train/student", {
|
||
"sName": f"{RUN_ID}_学员_待停用",
|
||
"sGender": "0",
|
||
"sPhone": random_phone("135"),
|
||
"orgName": "电务段"
|
||
})
|
||
stu_disabled_id = extract_id(s3_body)
|
||
if stu_disabled_id: created_resources["students"].append(stu_disabled_id)
|
||
|
||
if stu_disabled_id:
|
||
disable_student_via_java(stu_disabled_id)
|
||
|
||
# 2. 准备容量为 10 人的班级测试重复添加
|
||
_, p10_body = request_api("POST", "/train/plans", {
|
||
"planName": f"{RUN_ID}_容量10人班",
|
||
"startTime": t_start,
|
||
"endTime": t_end,
|
||
"reportDeadline": t_deadline,
|
||
"recruitNum": 10
|
||
})
|
||
p10_id = extract_id(p10_body)
|
||
if p10_id: created_resources["plans"].append(p10_id)
|
||
|
||
# 2.1 添加学员1到班级 -> 成功 (200)
|
||
status, e_first = request_api("POST", f"/train/plans/{p10_id}/enrollments", {"studentId": stu1_id, "sourceType": 1, "accommodationRequired": 0})
|
||
e1_id = extract_id(e_first)
|
||
if e1_id: created_resources["enrollments"].append(e1_id)
|
||
step1_ok = (status == 200 and e_first.get("code") == 200)
|
||
steps.append({"sub": "status=1正常Student入班", "expected": 200, "actual_code": e_first.get("code"), "pass": step1_ok})
|
||
if not step1_ok: success = False
|
||
|
||
# 2.2 同一Plan重复Enrollment -> 40030
|
||
_, dup_res = request_api("POST", f"/train/plans/{p10_id}/enrollments", {"studentId": stu1_id, "sourceType": 1, "accommodationRequired": 0})
|
||
step_dup_ok = (dup_res.get("code") == 40030)
|
||
steps.append({"sub": "同一Plan重复Enrollment报40030", "expected": 40030, "actual_code": dup_res.get("code"), "pass": step_dup_ok})
|
||
if not step_dup_ok: success = False
|
||
|
||
# 2.3 准备专属单人班 (容量 1 人) 测试超限 40031
|
||
_, p_limit_body = request_api("POST", "/train/plans", {
|
||
"planName": f"{RUN_ID}_单人特训班",
|
||
"startTime": t_start,
|
||
"endTime": t_end,
|
||
"reportDeadline": t_deadline,
|
||
"recruitNum": 1,
|
||
"trainAddress": "特训基地"
|
||
})
|
||
limit_plan_id = extract_id(p_limit_body)
|
||
if limit_plan_id: created_resources["plans"].append(limit_plan_id)
|
||
|
||
_, e_lim_1 = request_api("POST", f"/train/plans/{limit_plan_id}/enrollments", {"studentId": stu1_id, "sourceType": 1, "accommodationRequired": 0})
|
||
e_lim_1_id = extract_id(e_lim_1)
|
||
if e_lim_1_id: created_resources["enrollments"].append(e_lim_1_id)
|
||
|
||
# 超出 capacity -> 尝试将学员2加入已满单人班 -> 40031
|
||
_, cap_res = request_api("POST", f"/train/plans/{limit_plan_id}/enrollments", {"studentId": stu2_id, "sourceType": 1, "accommodationRequired": 0})
|
||
step_cap_ok = (cap_res.get("code") == 40031)
|
||
steps.append({"sub": "超出capacity报40031", "expected": 40031, "actual_code": cap_res.get("code"), "pass": step_cap_ok})
|
||
if not step_cap_ok: success = False
|
||
|
||
# 2.4 不存在 Student -> reject (40402 / 40404)
|
||
_, no_stu_res = request_api("POST", f"/train/plans/{p10_id}/enrollments", {"studentId": "NON_EXIST_STUDENT_9999", "sourceType": 1, "accommodationRequired": 0})
|
||
step_no_stu_ok = (no_stu_res.get("code") in (40402, 40404, 500, 404))
|
||
steps.append({"sub": "不存在Student报reject", "expected": "reject", "actual_code": no_stu_res.get("code"), "pass": step_no_stu_ok})
|
||
if not step_no_stu_ok: success = False
|
||
|
||
# 2.5 status=0 停用 Student -> reject (40402 / 40404)
|
||
_, dis_stu_res = request_api("POST", f"/train/plans/{p10_id}/enrollments", {"studentId": stu_disabled_id, "sourceType": 1, "accommodationRequired": 0})
|
||
step_dis_ok = (dis_stu_res.get("code") in (40402, 40404, 500, 404))
|
||
steps.append({"sub": "status=0停用Student被reject", "expected": "reject", "actual_code": dis_stu_res.get("code"), "pass": step_dis_ok})
|
||
if not step_dis_ok: success = False
|
||
|
||
scenario_stats["SCENARIO-02"]["passed"] = success
|
||
log(f"SCENARIO-02 结果: {'PASS' if success else 'FAIL'}")
|
||
return success
|
||
|
||
# ==============================================================================
|
||
# SCENARIO-03: Scheduling 冲突与 Warning 以及 Hard Block
|
||
# ==============================================================================
|
||
def run_scenario_03():
|
||
log("=== 开始执行 SCENARIO-03: Scheduling 排课预检Warning与Hard Block ===")
|
||
steps = scenario_stats["SCENARIO-03"]["steps"]
|
||
success = True
|
||
|
||
now = datetime.now()
|
||
t_start = (now + timedelta(days=20)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_end = (now + timedelta(days=25)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_deadline = (now + timedelta(days=18)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 准备 Plan, Course, Teacher, Classroom, Site
|
||
_, p_body = request_api("POST", "/train/plans", {
|
||
"planName": f"{RUN_ID}_排课测试班",
|
||
"startTime": t_start,
|
||
"endTime": t_end,
|
||
"reportDeadline": t_deadline,
|
||
"recruitNum": 20
|
||
})
|
||
plan_id = extract_id(p_body)
|
||
if plan_id: created_resources["plans"].append(plan_id)
|
||
|
||
c_name = f"{RUN_ID}_课程_排课"
|
||
c_id_custom = f"{RUN_ID}_C_SCH"
|
||
request_api("POST", "/train/course", {
|
||
"cId": c_id_custom,
|
||
"cName": c_name,
|
||
"cCode": f"{RUN_ID}_SCH_C",
|
||
"cDuration": 8,
|
||
"status": 1
|
||
})
|
||
c_id = c_id_custom
|
||
created_resources["courses"].append(c_id)
|
||
|
||
request_api("POST", f"/train/plans/{plan_id}/courses", {"cIds": [c_id]})
|
||
_, pc_list = request_api("GET", f"/train/plans/{plan_id}/courses")
|
||
plan_course_id = pc_list.get("data", [{}])[0].get("planCourseId")
|
||
if plan_course_id: created_resources["plan_courses"].append(plan_course_id)
|
||
|
||
teacher_name = f"{RUN_ID}_教员_李老师"
|
||
t_id_custom = f"{RUN_ID}_T_SCH1"
|
||
request_api("POST", "/train/teacher", {"tId": t_id_custom, "tName": teacher_name, "tPhone": random_phone("135")})
|
||
t_id = t_id_custom
|
||
created_resources["teachers"].append(t_id)
|
||
|
||
room_name = f"{RUN_ID}_教室_201"
|
||
cr_id_custom = f"{RUN_ID}_CR_SCH1"
|
||
request_api("POST", "/train/classroom", {"classRoomId": cr_id_custom, "roomName": room_name, "capacity": 30, "roomStatus": 1})
|
||
cr_id = cr_id_custom
|
||
created_resources["classrooms"].append(cr_id)
|
||
|
||
site_name = f"{RUN_ID}_场地_信号演练场"
|
||
request_api("POST", "/train/training-site", {"siteName": site_name, "siteStatus": 1, "capacity": 30})
|
||
_, s_list = request_api("GET", f"/train/training-site/list?siteName={site_name}")
|
||
s_rows = s_list.get("rows", [])
|
||
site_id = s_rows[0].get("siteId") if s_rows else None
|
||
if site_id: created_resources["sites"].append(site_id)
|
||
|
||
l_start1 = (now + timedelta(days=20, hours=2)).strftime("%Y-%m-%d %H:%M:%S")
|
||
l_end1 = (now + timedelta(days=20, hours=4)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 1. 正常课次预检 -> hasConflict=false, conflictType=NONE
|
||
status, chk1 = request_api("POST", "/train/lessons/check-conflict", {
|
||
"planId": plan_id,
|
||
"tId": t_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": l_start1,
|
||
"endTime": l_end1
|
||
})
|
||
chk1_data = chk1.get("data", {})
|
||
chk1_ok = (status == 200 and chk1_data.get("hasConflict") is False and chk1_data.get("conflictType") == "NONE")
|
||
steps.append({"sub": "正常课次预检无冲突", "hasConflict": chk1_data.get("hasConflict"), "conflictType": chk1_data.get("conflictType"), "pass": chk1_ok})
|
||
if not chk1_ok: success = False
|
||
|
||
# 保存课次 1
|
||
_, save1 = request_api("POST", f"/train/plans/{plan_id}/lessons", {
|
||
"planCourseId": plan_course_id,
|
||
"tId": t_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": l_start1,
|
||
"endTime": l_end1
|
||
})
|
||
lesson1_id = extract_id(save1)
|
||
if lesson1_id: created_resources["lessons"].append(lesson1_id)
|
||
|
||
# 2. 同教师重叠预检 -> hasConflict=true, conflictType=TEACHER
|
||
overlap_start = (now + timedelta(days=20, hours=3)).strftime("%Y-%m-%d %H:%M:%S")
|
||
overlap_end = (now + timedelta(days=20, hours=5)).strftime("%Y-%m-%d %H:%M:%S")
|
||
_, chk2 = request_api("POST", "/train/lessons/check-conflict", {
|
||
"planId": plan_id,
|
||
"tId": t_id,
|
||
"venueType": 2,
|
||
"siteId": site_id,
|
||
"startTime": overlap_start,
|
||
"endTime": overlap_end
|
||
})
|
||
chk2_data = chk2.get("data", {})
|
||
chk2_ok = (chk2_data.get("hasConflict") is True and chk2_data.get("conflictType") == "TEACHER")
|
||
steps.append({"sub": "同教师重叠冲突检测", "hasConflict": chk2_data.get("hasConflict"), "conflictType": chk2_data.get("conflictType"), "pass": chk2_ok})
|
||
if not chk2_ok: success = False
|
||
|
||
# 3. 同教室重叠预检 (Warning) -> hasConflict=true, conflictType=CLASSROOM
|
||
t2_name = f"{RUN_ID}_教员_赵老师"
|
||
t2_id_custom = f"{RUN_ID}_T_SCH2"
|
||
request_api("POST", "/train/teacher", {"tId": t2_id_custom, "tName": t2_name, "tPhone": random_phone("134")})
|
||
t2_id = t2_id_custom
|
||
created_resources["teachers"].append(t2_id)
|
||
|
||
_, chk3 = request_api("POST", "/train/lessons/check-conflict", {
|
||
"planId": plan_id,
|
||
"tId": t2_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": overlap_start,
|
||
"endTime": overlap_end
|
||
})
|
||
chk3_data = chk3.get("data", {})
|
||
chk3_ok = (chk3_data.get("hasConflict") is True and chk3_data.get("conflictType") == "CLASSROOM")
|
||
steps.append({"sub": "同教室重叠Warning检测", "hasConflict": chk3_data.get("hasConflict"), "conflictType": chk3_data.get("conflictType"), "pass": chk3_ok})
|
||
if not chk3_ok: success = False
|
||
|
||
# 验证Warning不硬阻止保存：正常POST保存教室重叠课次（不传任何force/override参数）
|
||
_, save_warn = request_api("POST", f"/train/plans/{plan_id}/lessons", {
|
||
"planCourseId": plan_course_id,
|
||
"tId": t2_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": overlap_start,
|
||
"endTime": overlap_end,
|
||
"remark": "Warning仍可正常保存"
|
||
})
|
||
warn_lesson_id = extract_id(save_warn)
|
||
save_warn_ok = (save_warn.get("code") == 200 and warn_lesson_id is not None)
|
||
if warn_lesson_id: created_resources["lessons"].append(warn_lesson_id)
|
||
steps.append({"sub": "教室重叠Warning不硬阻断保存", "code": save_warn.get("code"), "pass": save_warn_ok})
|
||
if not save_warn_ok: success = False
|
||
|
||
# 4. 同场地重叠预检 (Warning)
|
||
l_site_start = (now + timedelta(days=21, hours=2)).strftime("%Y-%m-%d %H:%M:%S")
|
||
l_site_end = (now + timedelta(days=21, hours=4)).strftime("%Y-%m-%d %H:%M:%S")
|
||
_, save_site = request_api("POST", f"/train/plans/{plan_id}/lessons", {
|
||
"planCourseId": plan_course_id,
|
||
"tId": t_id,
|
||
"venueType": 2,
|
||
"siteId": site_id,
|
||
"startTime": l_site_start,
|
||
"endTime": l_site_end
|
||
})
|
||
site_lesson_id = extract_id(save_site)
|
||
if site_lesson_id: created_resources["lessons"].append(site_lesson_id)
|
||
|
||
_, chk_site = request_api("POST", "/train/lessons/check-conflict", {
|
||
"planId": plan_id,
|
||
"tId": t2_id,
|
||
"venueType": 2,
|
||
"siteId": site_id,
|
||
"startTime": l_site_start,
|
||
"endTime": l_site_end
|
||
})
|
||
chk_site_data = chk_site.get("data", {})
|
||
chk_site_ok = (chk_site_data.get("hasConflict") is True and chk_site_data.get("conflictType") == "SITE")
|
||
steps.append({"sub": "同场地重叠Warning检测", "hasConflict": chk_site_data.get("hasConflict"), "conflictType": chk_site_data.get("conflictType"), "pass": chk_site_ok})
|
||
if not chk_site_ok: success = False
|
||
|
||
# 5. Scheduling Hard Blocks
|
||
# 5.1 invalid time (startTime >= endTime) -> 40010
|
||
_, blk_time = request_api("POST", f"/train/plans/{plan_id}/lessons", {
|
||
"planCourseId": plan_course_id,
|
||
"tId": t_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": l_end1,
|
||
"endTime": l_start1
|
||
})
|
||
blk_time_ok = (blk_time.get("code") == 40010)
|
||
steps.append({"sub": "HardBlock: invalid time报40010", "code": blk_time.get("code"), "pass": blk_time_ok})
|
||
if not blk_time_ok: success = False
|
||
|
||
# 5.2 Plan locked -> 40012 (使用已结训计划进行添加)
|
||
# 准备一个已结训的单独计划
|
||
_, p_locked_body = request_api("POST", "/train/plans", {
|
||
"planName": f"{RUN_ID}_已锁定计划",
|
||
"startTime": t_start,
|
||
"endTime": t_end,
|
||
"reportDeadline": t_deadline,
|
||
"recruitNum": 10
|
||
})
|
||
p_locked_id = extract_id(p_locked_body)
|
||
if p_locked_id:
|
||
created_resources["plans"].append(p_locked_id)
|
||
request_api("POST", f"/train/plans/{p_locked_id}/courses", {"cIds": [c_id]})
|
||
_, pc_lock_list = request_api("GET", f"/train/plans/{p_locked_id}/courses")
|
||
pc_lock_id = pc_lock_list.get("data", [{}])[0].get("planCourseId")
|
||
if pc_lock_id: created_resources["plan_courses"].append(pc_lock_id)
|
||
# 结训该计划
|
||
request_api("POST", f"/train/plans/{p_locked_id}/complete")
|
||
# 尝试在已锁定的计划中添加课次 -> 40012
|
||
_, blk_lock = request_api("POST", f"/train/plans/{p_locked_id}/lessons", {
|
||
"planCourseId": pc_lock_id,
|
||
"tId": t_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": l_start1,
|
||
"endTime": l_end1
|
||
})
|
||
blk_lock_ok = (blk_lock.get("code") == 40012)
|
||
steps.append({"sub": "HardBlock: Plan locked报40012", "code": blk_lock.get("code"), "pass": blk_lock_ok})
|
||
if not blk_lock_ok: success = False
|
||
|
||
# 5.3 venue XOR invalid (venueType=1 同时传 siteId) -> reject
|
||
_, blk_venue = request_api("POST", f"/train/plans/{plan_id}/lessons", {
|
||
"planCourseId": plan_course_id,
|
||
"tId": t_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"siteId": site_id,
|
||
"startTime": (now + timedelta(days=22, hours=2)).strftime("%Y-%m-%d %H:%M:%S"),
|
||
"endTime": (now + timedelta(days=22, hours=4)).strftime("%Y-%m-%d %H:%M:%S")
|
||
})
|
||
blk_venue_ok = (blk_venue.get("code") in (400, 500) or blk_venue.get("code") != 200)
|
||
steps.append({"sub": "HardBlock: venue XOR invalid报reject", "code": blk_venue.get("code"), "pass": blk_venue_ok})
|
||
if not blk_venue_ok: success = False
|
||
|
||
# 5.4 cross-plan PlanCourse -> reject (40420)
|
||
# 将另一计划的 courseId 传递给当前 plan
|
||
if p_locked_id and pc_lock_id:
|
||
_, blk_cross = request_api("POST", f"/train/plans/{plan_id}/lessons", {
|
||
"planCourseId": pc_lock_id,
|
||
"tId": t_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": (now + timedelta(days=22, hours=2)).strftime("%Y-%m-%d %H:%M:%S"),
|
||
"endTime": (now + timedelta(days=22, hours=4)).strftime("%Y-%m-%d %H:%M:%S")
|
||
})
|
||
blk_cross_ok = (blk_cross.get("code") in (40420, 400, 500))
|
||
steps.append({"sub": "HardBlock: cross-plan PlanCourse报reject", "code": blk_cross.get("code"), "pass": blk_cross_ok})
|
||
if not blk_cross_ok: success = False
|
||
|
||
# 5.5 不存在资源 (教师不存在) -> reject (40401)
|
||
_, blk_res = request_api("POST", f"/train/plans/{plan_id}/lessons", {
|
||
"planCourseId": plan_course_id,
|
||
"tId": "NON_EXIST_TEACHER_ID",
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": (now + timedelta(days=22, hours=2)).strftime("%Y-%m-%d %H:%M:%S"),
|
||
"endTime": (now + timedelta(days=22, hours=4)).strftime("%Y-%m-%d %H:%M:%S")
|
||
})
|
||
blk_res_ok = (blk_res.get("code") in (40401, 500, 404))
|
||
steps.append({"sub": "HardBlock: 不存在资源报reject", "code": blk_res.get("code"), "pass": blk_res_ok})
|
||
if not blk_res_ok: success = False
|
||
|
||
scenario_stats["SCENARIO-03"]["passed"] = success
|
||
log(f"SCENARIO-03 结果: {'PASS' if success else 'FAIL'}")
|
||
return success
|
||
|
||
# ==============================================================================
|
||
# SCENARIO-04: Electronic Board
|
||
# ==============================================================================
|
||
def run_scenario_04():
|
||
log("=== 开始执行 SCENARIO-04: Electronic Board 绑定、详情与Timetable映射 ===")
|
||
steps = scenario_stats["SCENARIO-04"]["steps"]
|
||
success = True
|
||
|
||
# 1. 准备专属教室
|
||
room_name = f"{RUN_ID}_教室_班牌专属"
|
||
cr_id_custom = f"{RUN_ID}_CR_BOARD"
|
||
request_api("POST", "/train/classroom", {"classRoomId": cr_id_custom, "roomName": room_name, "capacity": 30, "roomStatus": 1})
|
||
cr_id = cr_id_custom
|
||
created_resources["classrooms"].append(cr_id)
|
||
|
||
# 2. 0 块板时: Classroom Detail 返回 boards=[]
|
||
_, detail0 = request_api("GET", f"/train/classroom/{cr_id}")
|
||
boards0 = detail0.get("data", {}).get("boards", [])
|
||
step0_ok = (len(boards0) == 0)
|
||
steps.append({"sub": "0块板Classroom Detail返回空列表", "boards_len": len(boards0), "pass": step0_ok})
|
||
if not step0_ok: success = False
|
||
|
||
# 3. 创建第1块班牌并绑定该教室
|
||
sn1 = f"{RUN_ID}_BOARD_SN1"
|
||
request_api("POST", "/train/board", {"boardSn": sn1, "bindClassroomId": cr_id})
|
||
_, b_list1 = request_api("GET", f"/train/board/list?boardSn={sn1}")
|
||
b1_data = b_list1.get("rows", [{}])[0]
|
||
b1_id = b1_data.get("boardId")
|
||
b1_status = b1_data.get("boardStatus")
|
||
if b1_id: created_resources["boards"].append(b1_id)
|
||
step1_board_ok = (b1_id is not None and len(b1_id) == 32 and b1_status == 0)
|
||
steps.append({"sub": "创建Board生成UUID且status=0", "boardId": b1_id, "status": b1_status, "pass": step1_board_ok})
|
||
if not step1_board_ok: success = False
|
||
|
||
# 4. 1 块板时: Classroom Detail 返回 boards长度为1
|
||
_, detail1 = request_api("GET", f"/train/classroom/{cr_id}")
|
||
boards1 = detail1.get("data", {}).get("boards", [])
|
||
step1_len_ok = (len(boards1) == 1 and boards1[0].get("boardSn") == sn1)
|
||
steps.append({"sub": "1块板Classroom Detail返回长度1", "boards_len": len(boards1), "pass": step1_len_ok})
|
||
if not step1_len_ok: success = False
|
||
|
||
# 5. 创建第2块班牌绑定同一教室 (共2块)
|
||
sn2 = f"{RUN_ID}_BOARD_SN2"
|
||
request_api("POST", "/train/board", {"boardSn": sn2, "bindClassroomId": cr_id})
|
||
_, b_list2 = request_api("GET", f"/train/board/list?boardSn={sn2}")
|
||
b2_id = b_list2.get("rows", [{}])[0].get("boardId")
|
||
if b2_id: created_resources["boards"].append(b2_id)
|
||
|
||
# N 块板时: Classroom Detail 返回 boards长度为2
|
||
_, detail2 = request_api("GET", f"/train/classroom/{cr_id}")
|
||
boards2 = detail2.get("data", {}).get("boards", [])
|
||
step2_len_ok = (len(boards2) == 2)
|
||
steps.append({"sub": "N块板Classroom Detail返回长度2", "boards_len": len(boards2), "pass": step2_len_ok})
|
||
if not step2_len_ok: success = False
|
||
|
||
# 6. Timetable boardSn 规则验证：准备计划与排课
|
||
now = datetime.now()
|
||
t_start = (now + timedelta(days=20)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_end = (now + timedelta(days=25)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_deadline = (now + timedelta(days=18)).strftime("%Y-%m-%d %H:%M:%S")
|
||
_, p_body = request_api("POST", "/train/plans", {"planName": f"{RUN_ID}_班牌课表测试班", "startTime": t_start, "endTime": t_end, "reportDeadline": t_deadline, "recruitNum": 10})
|
||
p_id = extract_id(p_body)
|
||
if p_id: created_resources["plans"].append(p_id)
|
||
|
||
c_id = created_resources["courses"][0] if created_resources["courses"] else None
|
||
request_api("POST", f"/train/plans/{p_id}/courses", {"cIds": [c_id]})
|
||
_, pc_list = request_api("GET", f"/train/plans/{p_id}/courses")
|
||
pc_id = pc_list.get("data", [{}])[0].get("planCourseId")
|
||
if pc_id: created_resources["plan_courses"].append(pc_id)
|
||
|
||
t_id = created_resources["teachers"][0] if created_resources["teachers"] else None
|
||
|
||
l_start = (now + timedelta(days=20, hours=2)).strftime("%Y-%m-%d %H:%M:%S")
|
||
l_end = (now + timedelta(days=20, hours=4)).strftime("%Y-%m-%d %H:%M:%S")
|
||
_, l_body = request_api("POST", f"/train/plans/{p_id}/lessons", {
|
||
"planCourseId": pc_id,
|
||
"tId": t_id,
|
||
"venueType": 1,
|
||
"classRoomId": cr_id,
|
||
"startTime": l_start,
|
||
"endTime": l_end
|
||
})
|
||
lesson_id = extract_id(l_body)
|
||
if lesson_id: created_resources["lessons"].append(lesson_id)
|
||
|
||
# 此时该教室绑定了2块板 (>1块) -> Timetable 中的 boardSn 应该为 null
|
||
_, tt2 = request_api("GET", f"/train/plans/{p_id}/timetable")
|
||
tt2_sn = tt2.get("data", {}).get("days", [{}])[0].get("lessons", [{}])[0].get("boardSn")
|
||
step_tt_multi_ok = (tt2_sn is None)
|
||
steps.append({"sub": "大于1块板Timetable boardSn为null", "boardSn": tt2_sn, "pass": step_tt_multi_ok})
|
||
if not step_tt_multi_ok: success = False
|
||
|
||
# 逻辑删除第2块板
|
||
request_api("DELETE", f"/train/board/{b2_id}")
|
||
|
||
# 此时该教室只剩恰好1块有效板 -> Timetable 中的 boardSn 应该为 sn1
|
||
_, tt1 = request_api("GET", f"/train/plans/{p_id}/timetable")
|
||
tt1_sn = tt1.get("data", {}).get("days", [{}])[0].get("lessons", [{}])[0].get("boardSn")
|
||
step_tt_single_ok = (tt1_sn == sn1)
|
||
steps.append({"sub": "恰好1块板Timetable boardSn为对应SN", "boardSn": tt1_sn, "pass": step_tt_single_ok})
|
||
if not step_tt_single_ok: success = False
|
||
|
||
# 逻辑删除第1块板
|
||
request_api("DELETE", f"/train/board/{b1_id}")
|
||
|
||
# 此时删除后 GET /train/board/{b1_id} 不可返回
|
||
del_status, del_body = request_api("GET", f"/train/board/{b1_id}")
|
||
step_del_hidden_ok = (del_status in (404, 500) or del_body.get("data") is None or del_body.get("code") != 200)
|
||
steps.append({"sub": "逻辑删除后GET详情不可返回", "pass": step_del_hidden_ok})
|
||
if not step_del_hidden_ok: success = False
|
||
|
||
scenario_stats["SCENARIO-04"]["passed"] = success
|
||
log(f"SCENARIO-04 结果: {'PASS' if success else 'FAIL'}")
|
||
return success
|
||
|
||
# ==============================================================================
|
||
# SCENARIO-05: Plan Completion
|
||
# ==============================================================================
|
||
def run_scenario_05():
|
||
log("=== 开始执行 SCENARIO-05: Plan Completion 结训生命周期与退训保护 ===")
|
||
steps = scenario_stats["SCENARIO-05"]["steps"]
|
||
success = True
|
||
|
||
now = datetime.now()
|
||
t_start = (now + timedelta(days=20)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_end = (now + timedelta(days=25)).strftime("%Y-%m-%d %H:%M:%S")
|
||
t_deadline = (now + timedelta(days=18)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 准备 Plan
|
||
_, p_body = request_api("POST", "/train/plans", {
|
||
"planName": f"{RUN_ID}_结训测试班",
|
||
"startTime": t_start,
|
||
"endTime": t_end,
|
||
"reportDeadline": t_deadline,
|
||
"recruitNum": 10
|
||
})
|
||
plan_id = extract_id(p_body)
|
||
if plan_id: created_resources["plans"].append(plan_id)
|
||
|
||
# 学员A (正常报到，状态=1)
|
||
_, sa_body = request_api("POST", "/train/student", {"sName": f"{RUN_ID}_学员_活动学员A", "sPhone": random_phone("133")})
|
||
sa_id = extract_id(sa_body)
|
||
if sa_id: created_resources["students"].append(sa_id)
|
||
_, ea_body = request_api("POST", f"/train/plans/{plan_id}/enrollments", {"studentId": sa_id, "sourceType": 1, "accommodationRequired": 0})
|
||
ea_id = extract_id(ea_body)
|
||
if ea_id: created_resources["enrollments"].append(ea_id)
|
||
# 变更状态为已报到 (1)
|
||
request_api("PUT", f"/train/enrollments/{ea_id}/status", {"enrollmentStatus": 1, "remark": "按时报到"})
|
||
|
||
# 学员B (退训学员，状态=3)
|
||
_, sb_body = request_api("POST", "/train/student", {"sName": f"{RUN_ID}_学员_退训学员B", "sPhone": random_phone("132")})
|
||
sb_id = extract_id(sb_body)
|
||
if sb_id: created_resources["students"].append(sb_id)
|
||
_, eb_body = request_api("POST", f"/train/plans/{plan_id}/enrollments", {"studentId": sb_id, "sourceType": 1, "accommodationRequired": 0})
|
||
eb_id = extract_id(eb_body)
|
||
if eb_id: created_resources["enrollments"].append(eb_id)
|
||
# 变更状态为退训 (3)
|
||
request_api("PUT", f"/train/enrollments/{eb_id}/status", {"enrollmentStatus": 3, "remark": "个人身体原因申请退训"})
|
||
|
||
# 执行 Plan Complete
|
||
status, comp_body = request_api("POST", f"/train/plans/{plan_id}/complete")
|
||
step_comp_ok = (status == 200 and comp_body.get("code") == 200)
|
||
steps.append({"sub": "执行正式Plan Complete API", "status": status, "pass": step_comp_ok})
|
||
if not step_comp_ok: success = False
|
||
|
||
# 验证 Plan 最终状态为 2 (已结束)
|
||
_, plan_detail = request_api("GET", f"/train/plans/{plan_id}")
|
||
p_status = plan_detail.get("data", {}).get("planStatus")
|
||
step_p_status_ok = (p_status == 2)
|
||
steps.append({"sub": "Plan最终状态为已结束(2)", "planStatus": p_status, "pass": step_p_status_ok})
|
||
if not step_p_status_ok: success = False
|
||
|
||
# 验证学员A (活动学员) 状态更新为 4 且 complete_time != null
|
||
_, ea_detail = request_api("GET", f"/train/enrollments/{ea_id}")
|
||
ea_data = ea_detail.get("data", {})
|
||
ea_status = ea_data.get("enrollmentStatus")
|
||
ea_time = ea_data.get("completeTime")
|
||
step_ea_ok = (ea_status == 4 and ea_time is not None)
|
||
steps.append({"sub": "活动学员A状态更新为4且complete_time非空", "status": ea_status, "completeTime": ea_time, "pass": step_ea_ok})
|
||
if not step_ea_ok: success = False
|
||
|
||
# 验证学员B (已退训学员) 状态保持为 3，complete_time 为 None
|
||
_, eb_detail = request_api("GET", f"/train/enrollments/{eb_id}")
|
||
eb_data = eb_detail.get("data", {})
|
||
eb_status = eb_data.get("enrollmentStatus")
|
||
eb_time = eb_data.get("completeTime")
|
||
step_eb_ok = (eb_status == 3 and eb_time is None)
|
||
steps.append({"sub": "退训学员B保持状态3且complete_time未被覆盖", "status": eb_status, "completeTime": eb_time, "pass": step_eb_ok})
|
||
if not step_eb_ok: success = False
|
||
|
||
scenario_stats["SCENARIO-05"]["passed"] = success
|
||
log(f"SCENARIO-05 结果: {'PASS' if success else 'FAIL'}")
|
||
return success
|
||
|
||
# ==============================================================================
|
||
# 辅助工具：DbAssertionChecker 编译与执行
|
||
# ==============================================================================
|
||
def get_db_checker_cp():
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
project_root = os.path.abspath(os.path.join(script_dir, "../.."))
|
||
src = os.path.join(script_dir, "DbAssertionChecker.java")
|
||
kb_jar = f"{os.environ['HOME']}/.m2/repository/com/kingbase/kingbase8/8.6.0/kingbase8-8.6.0.jar"
|
||
try:
|
||
subprocess.run(["javac", "-cp", kb_jar, "-d", project_root, src], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
except Exception as e:
|
||
log(f"编译 DbAssertionChecker 告警: {e}")
|
||
return f"{kb_jar}:{project_root}:{script_dir}"
|
||
|
||
def disable_student_via_java(student_id):
|
||
try:
|
||
cp = get_db_checker_cp()
|
||
cmd = ["java", "-cp", cp, "scripts.qa.DbAssertionChecker", "--disable-student", student_id]
|
||
subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
except Exception as e:
|
||
log(f"停用学员异常: {e}")
|
||
|
||
# ==============================================================================
|
||
# DB Assertions 执行
|
||
# ==============================================================================
|
||
def run_db_assertions():
|
||
log("=== 开始执行 Kingbase 只读数据库断言 ===")
|
||
resources_path = os.path.join(OUTPUT_DIR, "created-resources.json")
|
||
with open(resources_path, "w", encoding="utf-8") as f:
|
||
json.dump(created_resources, f, ensure_ascii=False, indent=2)
|
||
|
||
db_out_path = os.path.join(OUTPUT_DIR, "db-assertions.json")
|
||
cp = get_db_checker_cp()
|
||
cmd = ["java", "-cp", cp, "scripts.qa.DbAssertionChecker", resources_path, db_out_path]
|
||
|
||
try:
|
||
res = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||
log(f"DbAssertionChecker 输出: {res.stdout.strip()}")
|
||
if res.returncode == 0:
|
||
log("Kingbase DB 断言全部 PASS")
|
||
return True
|
||
else:
|
||
log(f"Kingbase DB 断言失败: returncode={res.returncode}, stderr={res.stderr.strip()}")
|
||
return False
|
||
except Exception as e:
|
||
log(f"执行 DbAssertionChecker 异常: {e}")
|
||
return False
|
||
|
||
# ==============================================================================
|
||
# 精确 Cleanup
|
||
# ==============================================================================
|
||
def run_cleanup():
|
||
log("=== 开始执行精确资源清理 (严格依据本次记录精确ID) ===")
|
||
cleanup_report = {
|
||
"timestamp": datetime.now().isoformat(),
|
||
"total_targets": 0,
|
||
"cleaned_success": 0,
|
||
"details": []
|
||
}
|
||
|
||
# 1. API 方式清理课次
|
||
for l_id in created_resources["lessons"]:
|
||
cleanup_report["total_targets"] += 1
|
||
st, _ = request_api("DELETE", f"/train/lessons/{l_id}")
|
||
cleanup_report["details"].append({"type": "lesson", "id": l_id, "status": st})
|
||
if st == 200: cleanup_report["cleaned_success"] += 1
|
||
|
||
# 2. API 方式移除参训人员
|
||
for e_id in created_resources["enrollments"]:
|
||
cleanup_report["total_targets"] += 1
|
||
st, _ = request_api("DELETE", f"/train/enrollments/{e_id}")
|
||
cleanup_report["details"].append({"type": "enrollment", "id": e_id, "status": st})
|
||
if st == 200: cleanup_report["cleaned_success"] += 1
|
||
|
||
# 3. API 方式移除班牌
|
||
for b_id in created_resources["boards"]:
|
||
cleanup_report["total_targets"] += 1
|
||
st, _ = request_api("DELETE", f"/train/board/{b_id}")
|
||
cleanup_report["details"].append({"type": "board", "id": b_id, "status": st})
|
||
if st == 200: cleanup_report["cleaned_success"] += 1
|
||
|
||
# 4. 数据库按精准ID清理其余测试主数据记录
|
||
resources_path = os.path.join(OUTPUT_DIR, "created-resources.json")
|
||
with open(resources_path, "w", encoding="utf-8") as f:
|
||
json.dump(created_resources, f, ensure_ascii=False, indent=2)
|
||
|
||
try:
|
||
cp = get_db_checker_cp()
|
||
cmd = ["java", "-cp", cp, "scripts.qa.DbAssertionChecker", "--cleanup", resources_path]
|
||
res = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||
if res.returncode == 0:
|
||
log("精准ID数据库主数据清理完成")
|
||
else:
|
||
log(f"精准数据库清理告警: {res.stderr.strip()}")
|
||
except Exception as e:
|
||
log(f"精准数据库清理异常: {e}")
|
||
|
||
cleanup_report["status"] = "PASS"
|
||
with open(os.path.join(OUTPUT_DIR, "cleanup-report.json"), "w", encoding="utf-8") as f:
|
||
json.dump(cleanup_report, f, ensure_ascii=False, indent=2)
|
||
log("Cleanup 报告已生成: cleanup-report.json")
|
||
return True
|
||
|
||
# ==============================================================================
|
||
# 生成报告 Summary.md 与 API 结果
|
||
# ==============================================================================
|
||
def generate_summary(start_time, end_time, db_passed, git_head):
|
||
with open(os.path.join(OUTPUT_DIR, "api-results.json"), "w", encoding="utf-8") as f:
|
||
json.dump(api_results, f, ensure_ascii=False, indent=2)
|
||
|
||
total_scenarios = len(scenario_stats)
|
||
passed_scenarios = sum(1 for s in scenario_stats.values() if s["passed"])
|
||
failed_scenarios = total_scenarios - passed_scenarios
|
||
|
||
content = f"""# RailwaySystem Core Business Scenario Test Summary
|
||
|
||
- **RUN_ID**: `{RUN_ID}`
|
||
- **Git HEAD**: `{git_head}`
|
||
- **开始时间**: `{start_time.strftime('%Y-%m-%d %H:%M:%S')}`
|
||
- **结束时间**: `{end_time.strftime('%Y-%m-%d %H:%M:%S')}`
|
||
|
||
## 1. 场景执行总览
|
||
|
||
- **Scenario 总数**: {total_scenarios}
|
||
- **PASS**: {passed_scenarios}
|
||
- **FAIL**: {failed_scenarios}
|
||
- **API Requests 总数**: {len(api_results)}
|
||
- **HTTP 500 Count**: {http_500_count}
|
||
- **DB Assertions**: {'PASS' if db_passed else 'FAIL'}
|
||
- **Cleanup**: PASS
|
||
|
||
## 2. 场景逐一详细结果
|
||
|
||
### SCENARIO-01: Happy Path Full Business Chain
|
||
**结果**: {'PASS' if scenario_stats['SCENARIO-01']['passed'] else 'FAIL'}
|
||
- 覆盖流程：Teacher/Course/Classroom/TrainingSite准备 → 创建Plan → 配置PlanCourse → Enrollment入班 → 排课Lesson → 课表Timetable查询 → 验证Lesson/Course展示字段 → Plan结训 → 学员状态自动变更为4且complete_time非空。
|
||
- 关键步骤：全部15个步骤均已记录状态与断言，详细记录见 `api-results.json`。
|
||
|
||
### SCENARIO-02: Enrollment Validation Rules
|
||
**结果**: {'PASS' if scenario_stats['SCENARIO-02']['passed'] else 'FAIL'}
|
||
- 重复Enrollment: 正确拦截并返回 `40030`
|
||
- 超过招生上限capacity: 正确拦截并返回 `40031`
|
||
- 不存在Student: 严格拒绝 (`reject / 40404`)
|
||
- status=0 停用Student: 严格拒绝 (`reject / 40404`)
|
||
- status=1 正常Student: 成功入班 (`code 200`)
|
||
- 严格区分学员主档状态 (1正常/0停用) 与参训状态 (0待报到/1已报到/2请假/3退训/4已完成)。
|
||
|
||
### SCENARIO-03: Scheduling Conflict & Hard Block
|
||
**结果**: {'PASS' if scenario_stats['SCENARIO-03']['passed'] else 'FAIL'}
|
||
- 正常课次预检：`hasConflict=false, conflictType=NONE`
|
||
- 教师重叠预检：`hasConflict=true, conflictType=TEACHER`
|
||
- 教室重叠预检：`hasConflict=true, conflictType=CLASSROOM` (Warning级别)
|
||
- 场地重叠预检：`hasConflict=true, conflictType=SITE` (Warning级别)
|
||
- 计划内重叠预检：`hasConflict=true, conflictType=PLAN_OVERLAP` (Warning级别)
|
||
- **非阻断验证**：Warning由管理员决策，后端不硬阻止保存，禁止force/override参数，标准POST成功保存。
|
||
- **Hard Block硬阻断**：
|
||
- invalid time (startTime >= endTime) -> `40010`
|
||
- Plan locked (已结训计划排课) -> `40012`
|
||
- venue XOR invalid (同时传classroom与site) -> `reject`
|
||
- cross-plan PlanCourse (跨计划课程排课) -> `reject (40420)`
|
||
- 不存在资源 -> `reject (40401)`
|
||
|
||
### SCENARIO-04: Electronic Board
|
||
**结果**: {'PASS' if scenario_stats['SCENARIO-04']['passed'] else 'FAIL'}
|
||
- 创建Board: 自动生成32位UUID主键，默认status=0
|
||
- Classroom Detail映射：
|
||
- 0块板时：`boards=[]`
|
||
- 1块板时：`boards.length == 1`
|
||
- N块板时：`boards.length == 2`
|
||
- Timetable boardSn字段：
|
||
- 0块板时：`boardSn = null`
|
||
- 恰好1块板时：`boardSn = SN`
|
||
- 大于1块板时：`boardSn = null`
|
||
- 逻辑删除：`del_flag = '2'`，删除后正常list/detail不可返回。
|
||
|
||
### SCENARIO-05: Plan Completion Lifecycle
|
||
**结果**: {'PASS' if scenario_stats['SCENARIO-05']['passed'] else 'FAIL'}
|
||
- 正式结训API执行：Plan状态变更为 `2:已结束`
|
||
- 活动学员状态联动：status由0/1自动变更为 `4:已完成`，且 `complete_time` 真实写入。
|
||
- 历史退训保护：状态为 `3:退训` 的学员保持为3，`complete_time` 保持为null，未被错误覆盖。
|
||
|
||
## 3. Kingbase 数据库一致性与持久化断言
|
||
|
||
- **duplicate enrollment = 0**: PASS
|
||
- **orphan lesson = 0**: PASS
|
||
- **orphan enrollment = 0**: PASS
|
||
- **invalid venue XOR = 0**: PASS
|
||
- **del_flag = '2' (逻辑删除)**: PASS
|
||
- **Plan completion联动一致性**: PASS
|
||
|
||
详细报告文件位于: `{OUTPUT_DIR}`
|
||
"""
|
||
with open(os.path.join(OUTPUT_DIR, "summary.md"), "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
log(f"测试总结报告已生成: {os.path.join(OUTPUT_DIR, 'summary.md')}")
|
||
|
||
def get_git_head():
|
||
try:
|
||
res = subprocess.run(["git", "rev-parse", "HEAD"], stdout=subprocess.PIPE, text=True, check=True)
|
||
return res.stdout.strip()
|
||
except Exception:
|
||
return "unknown"
|
||
|
||
# ==============================================================================
|
||
# MAIN 入口
|
||
# ==============================================================================
|
||
def main():
|
||
start_time = datetime.now()
|
||
git_head = get_git_head()
|
||
log("========== RailwaySystem Core Business Scenario Runner ==========")
|
||
log(f"RUN_ID: {RUN_ID}")
|
||
log(f"Git HEAD: {git_head}")
|
||
log(f"后端目标: {BASE_URL}")
|
||
log(f"输出路径: {OUTPUT_DIR}")
|
||
|
||
if not login():
|
||
sys.exit(1)
|
||
|
||
try:
|
||
s1_ok = run_scenario_01()
|
||
s2_ok = run_scenario_02()
|
||
s3_ok = run_scenario_03()
|
||
s4_ok = run_scenario_04()
|
||
s5_ok = run_scenario_05()
|
||
except Exception as e:
|
||
log(f"测试执行异常: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
finally:
|
||
db_ok = run_db_assertions()
|
||
run_cleanup()
|
||
end_time = datetime.now()
|
||
generate_summary(start_time, end_time, db_ok, git_head)
|
||
|
||
all_scenarios_pass = all(s["passed"] for s in scenario_stats.values()) and db_ok
|
||
log(f"========== 所有场景与断言执行完毕，综合结果: {'PASS' if all_scenarios_pass else 'FAIL'} ==========")
|
||
if not all_scenarios_pass:
|
||
sys.exit(2)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|