"""使用独立内存数据库验证页面调用的真实 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}) 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"}) 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)