You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
578 lines
24 KiB
578 lines
24 KiB
"""
|
|
面试模块黑盒验收脚本
|
|
使用新账号，验证码从数据库读取（模拟后端日志行为）
|
|
"""
|
|
import requests
|
|
import json
|
|
import time
|
|
import sys
|
|
import pymysql
|
|
import datetime
|
|
import os
|
|
|
|
BASE = "http://127.0.0.1:8080/api/v1"
|
|
HEADERS = {"Content-Type": "application/json"}
|
|
|
|
# 数据库连接
|
|
def get_db():
|
|
return pymysql.connect(host='127.0.0.1', port=3307,
|
|
user=os.environ.get('DB_USERNAME', 'mydreamjob'),
|
|
password=os.environ['DB_PASSWORD'],
|
|
database='mydreamjob')
|
|
|
|
def get_db_code(phone, purpose="login"):
|
|
conn = get_db()
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"SELECT code FROM sms_verification_codes WHERE phone=%s AND purpose=%s AND used=0 ORDER BY created_at DESC LIMIT 1",
|
|
(phone, purpose))
|
|
row = cur.fetchone()
|
|
conn.close()
|
|
return row[0] if row else None
|
|
|
|
results = []
|
|
passed = 0
|
|
failed = 0
|
|
|
|
def test(name, fn):
|
|
global passed, failed
|
|
try:
|
|
fn()
|
|
print(f" PASS: {name}")
|
|
results.append((name, "PASS", ""))
|
|
passed += 1
|
|
except AssertionError as e:
|
|
print(f" FAIL: {name} - {e}")
|
|
results.append((name, "FAIL", str(e)))
|
|
failed += 1
|
|
except Exception as e:
|
|
print(f" FAIL: {name} - {type(e).__name__}: {e}")
|
|
results.append((name, "FAIL", f"{type(e).__name__}: {e}"))
|
|
failed += 1
|
|
|
|
def send_code(phone):
|
|
r = requests.post(f"{BASE}/auth/sms/send?phone={phone}&purpose=login", headers=HEADERS)
|
|
if r.status_code != 200:
|
|
raise Exception(f"Send code failed: {r.status_code} {r.text}")
|
|
time.sleep(0.5)
|
|
code = get_db_code(phone, "login")
|
|
if not code:
|
|
raise Exception(f"No code found for {phone}")
|
|
return code
|
|
|
|
def login(phone, role):
|
|
code = send_code(phone)
|
|
r = requests.post(f"{BASE}/auth/login", json={
|
|
"phone": phone, "code": code
|
|
}, headers=HEADERS)
|
|
if r.status_code != 200:
|
|
raise Exception(f"Login failed: {r.status_code} {r.text}")
|
|
user_id = r.json()["data"]["userId"]
|
|
# Set role in DB (mock login always creates STUDENT)
|
|
if role == "COMPANY":
|
|
conn = get_db()
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO user_roles (user_id, role) VALUES (%s, 'COMPANY') ON DUPLICATE KEY UPDATE role=role", (user_id,))
|
|
cur.execute(
|
|
"UPDATE users SET role = 'COMPANY' WHERE id = %s", (user_id,))
|
|
conn.commit()
|
|
conn.close()
|
|
print(f" Set user {user_id} to COMPANY role")
|
|
return r.json()["data"]["accessToken"]
|
|
|
|
def auth_header(token):
|
|
return {"Content-Type": "application/json", "Authorization": f"Bearer {token}"}
|
|
|
|
# ── Setup: 创建新账号 ──
|
|
print("=== Setup: New Accounts ===")
|
|
ts = int(time.time())
|
|
|
|
# 企业账号
|
|
company_phone = f"138{ts % 100000000:08d}"
|
|
print(f" Company: {company_phone}")
|
|
company_token = login(company_phone, "COMPANY")
|
|
print(f" Company token: {company_token[:20]}...")
|
|
|
|
# 求职者账号
|
|
student_phone = f"139{ts % 100000000:08d}"
|
|
print(f" Student: {student_phone}")
|
|
student_token = login(student_phone, "STUDENT")
|
|
print(f" Student token: {student_token[:20]}...")
|
|
|
|
# 第三方账号（非参与者）
|
|
third_phone = f"150{ts % 100000000:08d}"
|
|
print(f" Third party: {third_phone}")
|
|
third_token = login(third_phone, "STUDENT")
|
|
print(f" Third token: {third_token[:20]}...")
|
|
|
|
# 其他企业
|
|
other_company_phone = f"151{ts % 100000000:08d}"
|
|
print(f" Other company: {other_company_phone}")
|
|
other_company_token = login(other_company_phone, "COMPANY")
|
|
print(f" Other company token: {other_company_token[:20]}...")
|
|
|
|
# ── 创建公司资料 ──
|
|
print("=== Company Profile ===")
|
|
r = requests.post(f"{BASE}/companies/me", json={
|
|
"name": "黑盒测试科技有限公司",
|
|
"description": "面试模块黑盒验收测试企业",
|
|
"industry": "互联网"
|
|
}, headers=auth_header(company_token))
|
|
if r.status_code != 200:
|
|
print(f" Create company warning: {r.status_code} {r.text}")
|
|
else:
|
|
print(f" Company created: {r.json()['data']['name']}")
|
|
|
|
# ── 创建其他企业资料 ──
|
|
r = requests.post(f"{BASE}/companies/me", json={
|
|
"name": "其他黑盒测试企业",
|
|
"description": "其他企业",
|
|
"industry": "金融"
|
|
}, headers=auth_header(other_company_token))
|
|
print(f" Other company profile: {r.status_code}")
|
|
|
|
# ── 创建学生资料 ──
|
|
print("=== Student Profile ===")
|
|
r = requests.put(f"{BASE}/profile/me", json={
|
|
"realName": "黑盒测试学生",
|
|
"education": "本科"
|
|
}, headers=auth_header(student_token))
|
|
print(f" Student profile: {r.status_code}")
|
|
|
|
# ── 创建岗位并投递 ──
|
|
print("=== Create Job ===")
|
|
r = requests.post(f"{BASE}/jobs", json={
|
|
"title": "黑盒测试岗位",
|
|
"description": "面试模块黑盒验收测试岗位",
|
|
"city": "北京",
|
|
"employmentType": "FULL_TIME",
|
|
"salaryMin": 10000,
|
|
"salaryMax": 20000
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Create job failed: {r.status_code} {r.text}"
|
|
job_id = r.json()["data"]["id"]
|
|
print(f" Job ID: {job_id}")
|
|
|
|
# 发布岗位 (DRAFT -> PUBLISHED)
|
|
print("=== Publish Job ===")
|
|
r = requests.patch(f"{BASE}/jobs/{job_id}/status", json={"status": "PUBLISHED"},
|
|
headers=auth_header(company_token))
|
|
if r.status_code != 200:
|
|
print(f" Publish warning: {r.status_code} {r.text}")
|
|
else:
|
|
print(f" Job published: {r.json()['data']['status']}")
|
|
|
|
# 投递
|
|
print("=== Create Application ===")
|
|
r = requests.post(f"{BASE}/applications", json={"jobId": job_id},
|
|
headers=auth_header(student_token))
|
|
if r.status_code != 200:
|
|
print(f" Create application failed: {r.status_code} {r.text}")
|
|
# Try listing applications to see if there's already one
|
|
r2 = requests.get(f"{BASE}/applications/me", headers=auth_header(student_token))
|
|
print(f" My apps: {r2.status_code}")
|
|
if r2.status_code == 200:
|
|
data = r2.json()["data"]
|
|
if data["total"] > 0:
|
|
application_id = data["items"][0]["id"]
|
|
print(f" Using existing application: {application_id}")
|
|
else:
|
|
print(" No existing application, exiting")
|
|
sys.exit(1)
|
|
else:
|
|
sys.exit(1)
|
|
else:
|
|
application_id = r.json()["data"]["id"]
|
|
print(f" Application ID: {application_id}")
|
|
|
|
# ── 黑盒测试 ──
|
|
print("\n=== Black-box Tests ===")
|
|
|
|
interview_id = None
|
|
|
|
# 1. 企业创建面试
|
|
def test_01():
|
|
global interview_id
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=7)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=7)).strftime("%Y-%m-%dT15:00:00")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "技术面试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/blackbox"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
|
|
d = r.json()
|
|
assert d["data"]["status"] == "SCHEDULED", f"Expected SCHEDULED, got {d['data']['status']}"
|
|
interview_id = d["data"]["id"]
|
|
print(f" Interview ID: {interview_id}")
|
|
|
|
test("1. 企业创建面试", test_01)
|
|
|
|
# 2. 企业查询该投递的面试列表
|
|
def test_02():
|
|
r = requests.get(f"{BASE}/applications/{application_id}/interviews",
|
|
headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
|
|
data = r.json()["data"]
|
|
assert len(data) > 0, "Expected at least 1 interview"
|
|
assert data[0]["id"] == interview_id
|
|
|
|
test("2. 企业查询投递面试列表", test_02)
|
|
|
|
# 3. 求职者查询自己的面试
|
|
def test_03():
|
|
r = requests.get(f"{BASE}/interviews/me", headers=auth_header(student_token))
|
|
assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
|
|
assert r.json()["data"]["total"] > 0, "Expected at least 1 interview"
|
|
|
|
test("3. 求职者查询自己的面试", test_03)
|
|
|
|
# 4. 双方查看面试详情
|
|
def test_04():
|
|
r1 = requests.get(f"{BASE}/interviews/{interview_id}", headers=auth_header(company_token))
|
|
assert r1.status_code == 200, f"Company view failed: {r1.status_code}"
|
|
r2 = requests.get(f"{BASE}/interviews/{interview_id}", headers=auth_header(student_token))
|
|
assert r2.status_code == 200, f"Student view failed: {r2.status_code}"
|
|
assert r1.json()["data"]["status"] == "SCHEDULED"
|
|
|
|
test("4. 双方查看面试详情", test_04)
|
|
|
|
# 5. 企业修改 SCHEDULED 面试
|
|
def test_05():
|
|
new_future = (datetime.datetime.now() + datetime.timedelta(days=14)).strftime("%Y-%m-%dT10:00:00")
|
|
new_end = (datetime.datetime.now() + datetime.timedelta(days=14)).strftime("%Y-%m-%dT11:00:00")
|
|
r = requests.put(f"{BASE}/interviews/{interview_id}", json={
|
|
"title": "技术面试(已修改)",
|
|
"startAt": new_future,
|
|
"endAt": new_end
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
|
|
assert r.json()["data"]["title"] == "技术面试(已修改)"
|
|
|
|
test("5. 企业修改SCHEDULED面试", test_05)
|
|
|
|
# 6. 求职者 CONFIRMED
|
|
def test_06():
|
|
r = requests.post(f"{BASE}/interviews/{interview_id}/respond", json={
|
|
"action": "CONFIRM", "reason": "时间可以接受"
|
|
}, headers=auth_header(student_token))
|
|
assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
|
|
assert r.json()["data"]["status"] == "CONFIRMED"
|
|
|
|
test("6. 求职者CONFIRMED", test_06)
|
|
|
|
# 7. 求职者 DECLINED (需要新面试)
|
|
def test_07():
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=21)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=21)).strftime("%Y-%m-%dT15:00:00")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "拒绝测试面试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/decline"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Create failed: {r.status_code}: {r.text}"
|
|
iid = r.json()["data"]["id"]
|
|
r2 = requests.post(f"{BASE}/interviews/{iid}/respond", json={
|
|
"action": "DECLINE", "reason": "时间不合适"
|
|
}, headers=auth_header(student_token))
|
|
assert r2.status_code == 200, f"Expected 200, got {r2.status_code}: {r2.text}"
|
|
assert r2.json()["data"]["status"] == "DECLINED"
|
|
|
|
test("7. 求职者DECLINED", test_07)
|
|
|
|
# 8. 求职者 RESCHEDULE_REQUESTED (需要新面试)
|
|
def test_08():
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=28)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=28)).strftime("%Y-%m-%dT15:00:00")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "改期请求测试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/reschedule"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Create failed: {r.status_code}: {r.text}"
|
|
iid = r.json()["data"]["id"]
|
|
r2 = requests.post(f"{BASE}/interviews/{iid}/respond", json={
|
|
"action": "REQUEST_RESCHEDULE", "reason": "希望改到下周"
|
|
}, headers=auth_header(student_token))
|
|
assert r2.status_code == 200, f"Expected 200, got {r2.status_code}: {r2.text}"
|
|
assert r2.json()["data"]["status"] == "RESCHEDULE_REQUESTED"
|
|
|
|
test("8. 求职者RESCHEDULE_REQUESTED", test_08)
|
|
|
|
# 9. 企业 CANCEL (需要新面试)
|
|
def test_09():
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=35)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=35)).strftime("%Y-%m-%dT15:00:00")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "取消测试面试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/cancel"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Create failed: {r.status_code}: {r.text}"
|
|
iid = r.json()["data"]["id"]
|
|
r2 = requests.post(f"{BASE}/interviews/{iid}/cancel", json={
|
|
"reason": "岗位已招满"
|
|
}, headers=auth_header(company_token))
|
|
assert r2.status_code == 200, f"Expected 200, got {r2.status_code}: {r2.text}"
|
|
assert r2.json()["data"]["status"] == "CANCELLED"
|
|
|
|
test("9. 企业CANCEL", test_09)
|
|
|
|
# 10. 企业 COMPLETE - 需要 CONFIRMED 且已过 endAt
|
|
def test_10():
|
|
# 创建一个很快结束的面试
|
|
soon_start = (datetime.datetime.now() + datetime.timedelta(minutes=1)).strftime("%Y-%m-%dT%H:%M:%S")
|
|
soon_end = (datetime.datetime.now() + datetime.timedelta(minutes=2)).strftime("%Y-%m-%dT%H:%M:%S")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "完成测试面试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": soon_start,
|
|
"endAt": soon_end,
|
|
"meetingUrl": "https://meet.example.com/complete"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Create failed: {r.status_code}: {r.text}"
|
|
iid = r.json()["data"]["id"]
|
|
# 确认
|
|
r2 = requests.post(f"{BASE}/interviews/{iid}/respond", json={
|
|
"action": "CONFIRM"
|
|
}, headers=auth_header(student_token))
|
|
assert r2.status_code == 200, f"Confirm failed: {r2.status_code}: {r2.text}"
|
|
# 等待面试结束
|
|
print(f" Waiting for interview to end...")
|
|
time.sleep(130) # wait 130 seconds (2+ minutes)
|
|
# 完成
|
|
r3 = requests.post(f"{BASE}/interviews/{iid}/complete", json={
|
|
"notes": "面试表现优秀"
|
|
}, headers=auth_header(company_token))
|
|
assert r3.status_code == 200, f"Expected 200, got {r3.status_code}: {r3.text}"
|
|
assert r3.json()["data"]["status"] == "COMPLETED"
|
|
|
|
test("10. 企业COMPLETE", test_10)
|
|
|
|
# 11. 查询 timeline
|
|
def test_11():
|
|
r = requests.get(f"{BASE}/interviews/{interview_id}/timeline",
|
|
headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
|
|
data = r.json()["data"]
|
|
assert len(data) > 0, "Expected at least 1 timeline entry"
|
|
# Check history entries
|
|
statuses = [e["toStatus"] for e in data]
|
|
assert "SCHEDULED" in statuses, "Timeline should contain SCHEDULED"
|
|
assert "CONFIRMED" in statuses, "Timeline should contain CONFIRMED"
|
|
|
|
test("11. 查询timeline", test_11)
|
|
|
|
# 12. 确认通知发送 - check notification queue
|
|
def test_12():
|
|
# 检查 notification 队列是否有消息（通过 RabbitMQ API）
|
|
# 注意：消息可能已被消费，所以这里只检查队列存在
|
|
r = requests.get("http://127.0.0.1:15672/api/queues/%2F/queue.notification",
|
|
auth=("mydreamjob_dev", "Ch8mP@ssw0rd_2024!"))
|
|
assert r.status_code == 200, f"Queue check failed: {r.status_code}"
|
|
print(f" Notification queue messages: {r.json().get('messages', 'N/A')}")
|
|
|
|
test("12. 通知队列存在", test_12)
|
|
|
|
# 13. 幂等测试 - 重复操作
|
|
def test_13():
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=42)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=42)).strftime("%Y-%m-%dT15:00:00")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "幂等测试面试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/idempotent"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Create failed: {r.status_code}: {r.text}"
|
|
iid = r.json()["data"]["id"]
|
|
|
|
# 重复取消
|
|
r2 = requests.post(f"{BASE}/interviews/{iid}/cancel", json={"reason": "第一次取消"},
|
|
headers=auth_header(company_token))
|
|
assert r2.status_code == 200, f"First cancel: {r2.status_code}: {r2.text}"
|
|
r3 = requests.post(f"{BASE}/interviews/{iid}/cancel", json={"reason": "第二次取消"},
|
|
headers=auth_header(company_token))
|
|
assert r3.status_code == 200, f"Idempotent cancel: {r3.status_code}: {r3.text}"
|
|
assert r3.json()["data"]["status"] == "CANCELLED"
|
|
|
|
# 重复确认
|
|
future2 = (datetime.datetime.now() + datetime.timedelta(days=43)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end2 = (datetime.datetime.now() + datetime.timedelta(days=43)).strftime("%Y-%m-%dT15:00:00")
|
|
r4 = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "幂等确认测试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future2,
|
|
"endAt": future_end2,
|
|
"meetingUrl": "https://meet.example.com/idem2"
|
|
}, headers=auth_header(company_token))
|
|
assert r4.status_code == 200
|
|
iid2 = r4.json()["data"]["id"]
|
|
r5 = requests.post(f"{BASE}/interviews/{iid2}/respond", json={"action": "CONFIRM"},
|
|
headers=auth_header(student_token))
|
|
assert r5.status_code == 200
|
|
r6 = requests.post(f"{BASE}/interviews/{iid2}/respond", json={"action": "CONFIRM"},
|
|
headers=auth_header(student_token))
|
|
assert r6.status_code == 200, f"Idempotent confirm: {r6.status_code}: {r6.text}"
|
|
assert r6.json()["data"]["status"] == "CONFIRMED"
|
|
|
|
test("13. 幂等测试(重复取消/确认返回200)", test_13)
|
|
|
|
# 14. 非参与者访问 403
|
|
def test_14():
|
|
r = requests.get(f"{BASE}/interviews/{interview_id}", headers=auth_header(third_token))
|
|
assert r.status_code == 403, f"Expected 403, got {r.status_code}: {r.text}"
|
|
|
|
test("14. 非参与者访问返回403", test_14)
|
|
|
|
# 15. 其他企业修改 403
|
|
def test_15():
|
|
# 其他企业尝试创建面试（不属于他的投递）
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=50)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=50)).strftime("%Y-%m-%dT15:00:00")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "越权测试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/unauth"
|
|
}, headers=auth_header(other_company_token))
|
|
assert r.status_code == 403, f"Expected 403, got {r.status_code}: {r.text}"
|
|
|
|
test("15. 其他企业创建返回403", test_15)
|
|
|
|
# 16. 非法状态流转 409
|
|
def test_16():
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=56)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=56)).strftime("%Y-%m-%dT15:00:00")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "状态流转测试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/flow"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Create failed: {r.status_code}: {r.text}"
|
|
iid = r.json()["data"]["id"]
|
|
# 确认
|
|
requests.post(f"{BASE}/interviews/{iid}/respond", json={"action": "CONFIRM"},
|
|
headers=auth_header(student_token))
|
|
# 再次确认 - 应该返回 200 (幂等) 或 409
|
|
r2 = requests.post(f"{BASE}/interviews/{iid}/respond", json={"action": "DECLINE"},
|
|
headers=auth_header(student_token))
|
|
# CONFIRMED -> DECLINE 是非法流转，预期 409
|
|
assert r2.status_code == 409, f"Expected 409, got {r2.status_code}: {r2.text}"
|
|
|
|
test("16. 非法状态流转返回409", test_16)
|
|
|
|
# 17. 已取消的面试再次修改
|
|
def test_17():
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=60)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=60)).strftime("%Y-%m-%dT15:00:00")
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "修改已取消测试",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/cancelled"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 200, f"Create failed: {r.status_code}: {r.text}"
|
|
iid = r.json()["data"]["id"]
|
|
requests.post(f"{BASE}/interviews/{iid}/cancel", json={}, headers=auth_header(company_token))
|
|
r2 = requests.put(f"{BASE}/interviews/{iid}", json={"title": "修改已取消"},
|
|
headers=auth_header(company_token))
|
|
assert r2.status_code != 200, f"Expected error, got {r2.status_code}: {r2.text}"
|
|
|
|
test("17. 已取消面试不能修改", test_17)
|
|
|
|
# 18. 缺失字段/过去时间/结束时间早于开始时间
|
|
def test_18():
|
|
future = (datetime.datetime.now() + datetime.timedelta(days=65)).strftime("%Y-%m-%dT14:00:00")
|
|
future_end = (datetime.datetime.now() + datetime.timedelta(days=65)).strftime("%Y-%m-%dT15:00:00")
|
|
|
|
# 空标题
|
|
r = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end,
|
|
"meetingUrl": "https://meet.example.com/empty"
|
|
}, headers=auth_header(company_token))
|
|
assert r.status_code == 400, f"Empty title: expected 400, got {r.status_code}: {r.text}"
|
|
|
|
# 过去时间
|
|
r2 = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "过去时间",
|
|
"interviewType": "ONLINE",
|
|
"startAt": "2020-01-01T14:00:00",
|
|
"endAt": "2020-01-01T15:00:00",
|
|
"meetingUrl": "https://meet.example.com/past"
|
|
}, headers=auth_header(company_token))
|
|
assert r2.status_code != 200, f"Past time: expected error, got {r2.status_code}: {r2.text}"
|
|
|
|
# 结束时间早于开始时间
|
|
r3 = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "时间错误",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future_end,
|
|
"endAt": future,
|
|
"meetingUrl": "https://meet.example.com/timeerr"
|
|
}, headers=auth_header(company_token))
|
|
assert r3.status_code != 200, f"End before start: expected error, got {r3.status_code}: {r3.text}"
|
|
|
|
# 缺失必填字段
|
|
r4 = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "缺字段"
|
|
}, headers=auth_header(company_token))
|
|
assert r4.status_code == 400, f"Missing fields: expected 400, got {r4.status_code}: {r4.text}"
|
|
|
|
# ONLINE 缺少 meetingUrl
|
|
r5 = requests.post(f"{BASE}/applications/{application_id}/interviews", json={
|
|
"title": "缺链接",
|
|
"interviewType": "ONLINE",
|
|
"startAt": future,
|
|
"endAt": future_end
|
|
}, headers=auth_header(company_token))
|
|
assert r5.status_code == 400, f"Missing meetingUrl: expected 400, got {r5.status_code}: {r5.text}"
|
|
|
|
test("18. 字段校验(缺失/过去/倒序)", test_18)
|
|
|
|
# 19. 并发更新不能覆盖状态历史
|
|
def test_19():
|
|
# 验证 version 乐观锁
|
|
# 先获取当前 interview 的 version
|
|
r = requests.get(f"{BASE}/interviews/{interview_id}", headers=auth_header(company_token))
|
|
assert r.status_code == 200
|
|
version = r.json()["data"]["version"]
|
|
print(f" Current version: {version}")
|
|
|
|
# 查看 timeline 记录数量
|
|
r2 = requests.get(f"{BASE}/interviews/{interview_id}/timeline",
|
|
headers=auth_header(company_token))
|
|
assert r2.status_code == 200
|
|
timeline_count = len(r2.json()["data"])
|
|
print(f" Timeline entries: {timeline_count}")
|
|
|
|
# 验证 version 随着操作递增
|
|
# 状态历史记录数应该 >= 状态变更次数
|
|
assert timeline_count >= 2, f"Expected at least 2 timeline entries, got {timeline_count}"
|
|
|
|
test("19. 状态历史记录完整", test_19)
|
|
|
|
# ── Summary ──
|
|
print(f"\n=== Results: {passed}/{passed+failed} passed ===")
|
|
for name, status, detail in results:
|
|
if status == "FAIL":
|
|
print(f" FAIL: {name} - {detail}")
|
|
|
|
sys.exit(0 if failed == 0 else 1)
|
|
|