138 lines
8.2 KiB
Python
138 lines
8.2 KiB
Python
"""使用独立内存数据库验证页面调用的真实 HTTP API，不修改本机业务数据。"""
|
|
import json
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime, date
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import uvicorn
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
from database import Base, get_db
|
|
from main import app
|
|
from model import Student, Classes, Advisor, Teacher, Score, Employment
|
|
from dao.workspace_dao import execute_query
|
|
from schema.workspace_schema import DataQuery
|
|
|
|
|
|
class WorkspaceTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(cls.engine)
|
|
cls.Session = sessionmaker(bind=cls.engine)
|
|
def test_db():
|
|
with cls.Session() as db:
|
|
yield db
|
|
app.dependency_overrides[get_db] = test_db
|
|
cls.server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=18044, log_level="error"))
|
|
cls.thread = threading.Thread(target=cls.server.run, daemon=True)
|
|
cls.thread.start()
|
|
for _ in range(100):
|
|
if cls.server.started:
|
|
break
|
|
time.sleep(.05)
|
|
assert cls.server.started, "Test server failed to start"
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.server.should_exit = True
|
|
cls.thread.join(5)
|
|
app.dependency_overrides.clear()
|
|
cls.engine.dispose()
|
|
|
|
def http(self, path, method="GET", body=None, expected=200):
|
|
req = urllib.request.Request("http://127.0.0.1:18044" + path,
|
|
data=json.dumps(body).encode() if body is not None else None,
|
|
method=method, headers={"Content-Type": "application/json"})
|
|
try:
|
|
response = urllib.request.urlopen(req, timeout=10)
|
|
except urllib.error.HTTPError as error:
|
|
response = error
|
|
raw = response.read()
|
|
self.assertEqual(response.status, expected, raw.decode(errors="replace"))
|
|
return json.loads(raw) if response.headers.get("content-type", "").startswith("application/json") else raw
|
|
|
|
def api(self, path, *args, **kwargs):
|
|
return self.http("/api/studentsManagement"+path, *args, **kwargs)
|
|
|
|
def test_01_crud_all_modules(self):
|
|
advisor = self.api("/createAdvisor", "POST", {"advisor_name":"测试顾问","phone":"13800009999","gender":"男"})
|
|
classroom = self.api("/add_class", "POST", {"class_name":"测试一班","start_time":"2026-09-01T08:00:00","head_teacher":"甲老师","teacher":"乙老师"})
|
|
self.api(f"/get_one_class/{classroom['cid']}") # 新增后必须可查，验证is_del默认值
|
|
teacher = self.api("/add/teacher", "POST", {"t_name":"测试教师","phone":"13800008888","subject":"Python","entry_time":"2026-09-01"})
|
|
student_body={"student_no":"TEST0001","student_name":"测试学生","class_id":classroom['cid'],"advisor_id":advisor['id'],"flag":1,"gender":"女","age":20,"enrollment_time":"2026-09-01","state":"在读"}
|
|
student=self.api("/api/create/student","POST",student_body)['data'];sid=student['s_id']
|
|
self.api("/api/create/student","POST",student_body,expected=409)
|
|
score=self.api("/score/addScore","POST",{"student_id":sid,"exam_id":1,"score":78})
|
|
self.api("/addEmployment","POST",{"student_id":sid,"employment_start_time":"2026-09-11T00:00:00","offer_time":"2026-09-01T00:00:00"},expected=422)
|
|
job=self.api("/addEmployment","POST",{"student_id":sid,"company_name":"测试公司","salary":10000,"employment_start_time":"2026-09-01T00:00:00","offer_time":"2026-09-11T00:00:00"})
|
|
# 单独修改一个日期时，必须与另一个原有日期合并校验，失败不能改动薪资。
|
|
self.api(f"/modifyEmployment/{sid}","PUT",{"offer_time":"2026-08-31T23:59:59","salary":999},expected=422)
|
|
self.api(f"/modifyEmployment/{sid}","PUT",{"employment_start_time":"2026-09-11T00:00:01"},expected=422)
|
|
current=self.api('/workspace/query','POST',{'module':'employment'})['items'][0]
|
|
self.assertEqual(current['salary'],10000)
|
|
self.assertEqual(current['offer_time'],'2026-09-11T00:00:00')
|
|
# 相同日期合法，统计应为0天，而非空值。
|
|
self.api(f"/modifyEmployment/{sid}","PUT",{"employment_start_time":"2026-09-11T00:00:00"})
|
|
self.assertEqual(self.api('/workspace/query','POST',{'module':'employment'})['items'][0]['duration_days'],0)
|
|
self.api(f"/modifyEmployment/{sid}","PUT",{"employment_start_time":"2026-09-01T00:00:00"})
|
|
updates=[(f"/updateAdvisor/{advisor['id']}",{"advisor_name":"更新顾问"}),
|
|
(f"/update_class/{classroom['cid']}",{"class_name":"更新班级","start_time":"2026-09-01T08:00:00","head_teacher":"甲老师","teacher":"丙老师"}),
|
|
(f"/update/{teacher['tid']}",{"subject":"数据库"}),
|
|
(f"/api/update/student/{sid}",{**student_body,"student_name":"更新学生"}),
|
|
(f"/score/modifyScore/{sid}",{"student_id":sid,"exam_id":1,"score":88}),
|
|
(f"/modifyEmployment/{sid}",{"salary":12000})]
|
|
for path,body in updates:
|
|
self.api(path,"PUT",body)
|
|
for module in ['students','classes','teachers','advisors','scores','employment']:
|
|
result=self.api('/workspace/query','POST',{'module':module})
|
|
self.assertEqual(result['total'],1,module)
|
|
self.assertEqual(self.api('/workspace/query','POST',{'module':'advisors'})['items'][0]['phone'],'138****9999')
|
|
self.assertAlmostEqual(self.api('/workspace/query','POST',{'module':'employment'})['items'][0]['duration_days'],10)
|
|
deleted=[f"/score/removeScore/{sid}/1",f"/removeEmployment/{sid}",f"/delete/{teacher['tid']}",f"/api/delete/student/{sid}",f"/delete_class/{classroom['cid']}"]
|
|
for path in deleted:self.api(path,'DELETE')
|
|
self.api(f"/changeAdvisorStatus/{advisor['id']}/status",'PUT',{'flag':0})
|
|
for module in ['students','classes','teachers','advisors','scores','employment']:
|
|
self.assertEqual(self.api('/workspace/query','POST',{'module':module})['total'],0,module)
|
|
|
|
def test_02_validation_and_read_only(self):
|
|
self.api('/workspace/query','POST',{'module':'students','filters':[{'field':'password','value':'x'}]},expected=422)
|
|
self.api('/workspace/query','POST',{'module':'students','sql':'DELETE FROM student'},expected=422)
|
|
self.api('/workspace/query','POST',{'module':'students','aggregate':'avg','aggregate_field':'student_name'},expected=422)
|
|
self.api('/workspace/query','POST',{'module':'students','page_size':101},expected=422)
|
|
result=self.api('/workspace/query','POST',{'module':'students','filters':[{'field':'student_name','op':'contains','value':"' OR 1=1 --"}]})
|
|
self.assertEqual(result['total'],0)
|
|
with self.Session() as db:
|
|
self.assertEqual(db.query(Student).count(),1) # 逻辑删除的记录仍在
|
|
|
|
def test_03_page_assets(self):
|
|
self.assertIn('沃林'.encode(),self.http('/'))
|
|
for path in ['/static/app.js','/static/styles.css','/static/ai.js','/static/floating-assistant.js','/static/assets/assistant-avatar.png','/docs','/openapi.json']:
|
|
self.http(path)
|
|
self.api('/ai/chat','POST',{'question':' '},expected=422)
|
|
|
|
def test_04_aggregate_full_scope_and_nulls(self):
|
|
with self.Session() as db:
|
|
for i,score in enumerate([10,50,90]):
|
|
student=Student(student_no=f'AG{i}',student_name=f'统计{i}',class_id=1,advisor_id=1,flag=1)
|
|
db.add(student);db.flush()
|
|
db.add(Score(student_id=student.sid,exam_id=1,score=score,flag=1))
|
|
db.commit()
|
|
q=DataQuery(module='scores',aggregate='avg',aggregate_field='score',page_size=1)
|
|
result=execute_query(db,q)
|
|
self.assertEqual(result['items'][0]['value'],50)
|
|
page=execute_query(db,DataQuery(module='students',page_size=1))
|
|
self.assertTrue(page['truncated'])
|
|
self.assertEqual(page['total'],3)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|