double_work/backend/pick_student/test_main.py

import io
import unittest

import pandas as pd
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from pick_student import crud, schemas
from pick_student.database import Base
from pick_student.login import get_password_hash
from pick_student.main import get_db
from run import app  # 从 run.py 导入主 app
from pick_student.crud import create_teacher, create_class

# 设置测试数据库
SQLALCHEMY_DATABASE_URL = "sqlite:///./test_test.db"  # 使用不同的数据库文件名
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建所有表
Base.metadata.create_all(bind=engine)

# 重写 get_db 函数用于测试
def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    except Exception as e:
        db.close()
        raise e  # 重新抛出异常
    finally:
        db.close()  # 确保数据库连接被正确关闭

# 覆盖 app 中的依赖项
app.dependency_overrides[get_db] = override_get_db

# 创建测试客户端
client = TestClient(app)
class TestMainRoutes(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        # 在测试开始时设置初始数据库状态
        db = TestingSessionLocal()
        hashed_password = get_password_hash("password")  # 假设已经生成一个哈希密码
        create_teacher(db, user_name="test_teacher", password_hash=hashed_password)
        crud.create_class(db, cclass=schemas.create_class(class_name="2班", user_name="test_teacher",class_time="周三 5-6节"))
        crud.create_class(db, cclass=schemas.create_class(class_name="3班", user_name="test_teacher",class_time="周三 5-6节"))
        crud.create_student(db, student=schemas.create_student(class_name="2班", name="Alice", student_id=101, user_name="test_teacher"))
        crud.create_student(db, student=schemas.create_student(class_name="2班", name="Bob", student_id=102,user_name="test_teacher"))
        crud.create_student(db, student=schemas.create_student(class_name="2班", name="Charlie", student_id=103, user_name="test_teacher"))
        crud.create_student(db, student=schemas.create_student(class_name="3班", name="David", student_id=104, user_name="test_teacher"))
        db.close()

    def test_create_classroom(self):
        # 测试创建班级的成功情况
        response = client.post(
            "/teacher/pick_student/create_class/",
            params={"class_name": "1班","class_time":"周三 5-6节"},
            headers={"Authorization": "Bearer " + self.get_access_token()}
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["message"], "班级创建成功")

    def test_upload_students(self):
        # 测试上传学生信息的成功情况
        class_name = "1班"
        data = {
            'name': ['Alice', 'Bob'],
            'student_id': [101, 102]
            , 'user_name': ['test_teacher', 'test_teacher']
        }
        df = pd.DataFrame(data)
        excel_buffer = io.BytesIO()
        df.to_excel(excel_buffer, index=False)
        excel_buffer.seek(0)

        response = client.post(
            "/teacher/pick_student/upload_students/",
            files={"file": ("students.xlsx", excel_buffer, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
            params={"class_name": class_name},
            headers={"Authorization": "Bearer " + self.get_access_token()}
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["message"], "Student list uploaded successfully.")
    def test_get_students(self):
        response = client.get(
            "/teacher/pick_student/get_students/",
            params={"class_name": "2班"},
            headers={"Authorization": "Bearer " + self.get_access_token()}
        )
        self.assertEqual(response.status_code, 200)
    def test_update_score(self):
        response = client.put(
            "/teacher/pick_student/update_score/",
            params={"class_name": "2班", "student_id": 101, "arrival": True, "question_repeated": False, "question_correct": 2.5},
            headers={"Authorization": "Bearer " + self.get_access_token()}
        )
        print(response.json())
        self.assertEqual(response.status_code, 200)
        self.assertIn("scores", response.json())
    def test_is_reverse_day(self):
        response = client.put(
            "/teacher/pick_student/is_reverse_day/",
            params={"flag": True},
            headers={"Authorization": "Bearer " + self.get_access_token()}
        )
        print(response.json())
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json(), {"message": "Flag has been set True"})
    def test_random_pick(self):
        # 测试随机点名的功能
        response = client.get(
            "/teacher/pick_student/random_pick/",
            params={"class_name": "2班"},
            headers={"Authorization": "Bearer " + self.get_access_token()}
        )
        print(response.json())
        self.assertEqual(response.status_code, 200)
        data = response.json()
        self.assertIn("name", data)
        self.assertIn("student_id", data)

    def test_get_classes(self):
        response = client.get(
            "/teacher/pick_student/getclasses/",
            headers={"Authorization": "Bearer " + self.get_access_token()}
        )
        self.assertEqual(response.status_code, 200)
        data = response.json()
    def test_delete_class(self):
        response = client.delete(
            "/teacher/pick_student/delete_class/",
            params={"class_name": "3班"},
            headers={"Authorization": "Bearer " + self.get_access_token()}
        )
        self.assertEqual(response.status_code, 200)
    def get_access_token(self):
        # 获取访问令牌
        response = client.post(
            "/teacher/jwt/token",
            data={"username": "test_teacher", "password": "password"},
        )
        return response.json()["access_token"]

    @classmethod
    def tearDownClass(cls):
        # 清理测试数据库
        Base.metadata.drop_all(bind=engine)

if __name__ == "__main__":
    unittest.main()