998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
285 lines
9.2 KiB
Python
285 lines
9.2 KiB
Python
"""
|
|
课程模块测试
|
|
"""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.course import Course, CourseStatus, CourseCategory
|
|
from app.services.course_service import course_service
|
|
|
|
|
|
class TestCourseAPI:
|
|
"""课程API测试类"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_course_success(self, client: AsyncClient, admin_headers: dict):
|
|
"""测试成功创建课程"""
|
|
course_data = {
|
|
"name": "测试课程",
|
|
"description": "这是一个测试课程",
|
|
"category": "technology",
|
|
"difficulty_level": 3,
|
|
"tags": ["Python", "测试"]
|
|
}
|
|
|
|
response = await client.post(
|
|
"/api/v1/courses",
|
|
json=course_data,
|
|
headers=admin_headers
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["code"] == 200
|
|
assert data["message"] == "创建课程成功"
|
|
assert data["data"]["name"] == course_data["name"]
|
|
assert data["data"]["status"] == "draft"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_course_unauthorized(self, client: AsyncClient, user_headers: dict):
|
|
"""测试非管理员创建课程失败"""
|
|
course_data = {
|
|
"name": "测试课程",
|
|
"description": "这是一个测试课程"
|
|
}
|
|
|
|
response = await client.post(
|
|
"/api/v1/courses",
|
|
json=course_data,
|
|
headers=user_headers
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_courses_list(self, client: AsyncClient, user_headers: dict, db_session: AsyncSession):
|
|
"""测试获取课程列表"""
|
|
# 先创建几个测试课程
|
|
courses = [
|
|
Course(
|
|
name=f"测试课程{i}",
|
|
description=f"描述{i}",
|
|
category=CourseCategory.TECHNOLOGY if i % 2 == 0 else CourseCategory.BUSINESS,
|
|
status=CourseStatus.PUBLISHED if i < 2 else CourseStatus.DRAFT,
|
|
is_featured=i == 0
|
|
)
|
|
for i in range(3)
|
|
]
|
|
|
|
for course in courses:
|
|
db_session.add(course)
|
|
await db_session.commit()
|
|
|
|
# 测试获取所有课程
|
|
response = await client.get(
|
|
"/api/v1/courses",
|
|
headers=user_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["code"] == 200
|
|
assert len(data["data"]["items"]) == 3
|
|
assert data["data"]["total"] == 3
|
|
|
|
# 测试筛选已发布课程
|
|
response = await client.get(
|
|
"/api/v1/courses?status=published",
|
|
headers=user_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["data"]["items"]) == 2
|
|
|
|
# 测试分类筛选
|
|
response = await client.get(
|
|
"/api/v1/courses?category=technology",
|
|
headers=user_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["data"]["items"]) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_course_detail(self, client: AsyncClient, user_headers: dict, db_session: AsyncSession):
|
|
"""测试获取课程详情"""
|
|
# 创建测试课程
|
|
course = Course(
|
|
name="测试课程详情",
|
|
description="详细描述",
|
|
category=CourseCategory.TECHNOLOGY,
|
|
status=CourseStatus.PUBLISHED
|
|
)
|
|
db_session.add(course)
|
|
await db_session.commit()
|
|
await db_session.refresh(course)
|
|
|
|
# 获取课程详情
|
|
response = await client.get(
|
|
f"/api/v1/courses/{course.id}",
|
|
headers=user_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["data"]["id"] == course.id
|
|
assert data["data"]["name"] == course.name
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_course_not_found(self, client: AsyncClient, user_headers: dict):
|
|
"""测试获取不存在的课程"""
|
|
response = await client.get(
|
|
"/api/v1/courses/99999",
|
|
headers=user_headers
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_course(self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession):
|
|
"""测试更新课程"""
|
|
# 创建测试课程
|
|
course = Course(
|
|
name="原始课程名",
|
|
description="原始描述",
|
|
category=CourseCategory.TECHNOLOGY,
|
|
status=CourseStatus.DRAFT
|
|
)
|
|
db_session.add(course)
|
|
await db_session.commit()
|
|
await db_session.refresh(course)
|
|
|
|
# 更新课程
|
|
update_data = {
|
|
"name": "更新后的课程名",
|
|
"status": "published"
|
|
}
|
|
|
|
response = await client.put(
|
|
f"/api/v1/courses/{course.id}",
|
|
json=update_data,
|
|
headers=admin_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["data"]["name"] == update_data["name"]
|
|
assert data["data"]["status"] == "published"
|
|
assert data["data"]["published_at"] is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_course(self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession):
|
|
"""测试删除课程"""
|
|
# 创建测试课程
|
|
course = Course(
|
|
name="待删除课程",
|
|
description="这个课程将被删除",
|
|
category=CourseCategory.GENERAL,
|
|
status=CourseStatus.DRAFT
|
|
)
|
|
db_session.add(course)
|
|
await db_session.commit()
|
|
await db_session.refresh(course)
|
|
|
|
# 删除课程
|
|
response = await client.delete(
|
|
f"/api/v1/courses/{course.id}",
|
|
headers=admin_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["data"] is True
|
|
|
|
# 验证软删除
|
|
deleted_course = await course_service.get_by_id(db_session, course.id)
|
|
assert deleted_course is None # 因为get_by_id会过滤掉软删除的记录
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_published_course_fail(self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession):
|
|
"""测试删除已发布课程失败"""
|
|
# 创建已发布课程
|
|
course = Course(
|
|
name="已发布课程",
|
|
description="这是已发布的课程",
|
|
category=CourseCategory.TECHNOLOGY,
|
|
status=CourseStatus.PUBLISHED
|
|
)
|
|
db_session.add(course)
|
|
await db_session.commit()
|
|
await db_session.refresh(course)
|
|
|
|
# 尝试删除
|
|
response = await client.delete(
|
|
f"/api/v1/courses/{course.id}",
|
|
headers=admin_headers
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
|
|
class TestKnowledgePointAPI:
|
|
"""知识点API测试类"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_knowledge_points(self, client: AsyncClient, user_headers: dict, db_session: AsyncSession):
|
|
"""测试获取知识点列表"""
|
|
# 创建测试课程
|
|
course = Course(
|
|
name="测试课程",
|
|
description="包含知识点的课程",
|
|
category=CourseCategory.TECHNOLOGY,
|
|
status=CourseStatus.PUBLISHED
|
|
)
|
|
db_session.add(course)
|
|
await db_session.commit()
|
|
await db_session.refresh(course)
|
|
|
|
# 获取知识点（应该为空）
|
|
response = await client.get(
|
|
f"/api/v1/courses/{course.id}/knowledge-points",
|
|
headers=user_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["data"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_knowledge_point(self, client: AsyncClient, admin_headers: dict, db_session: AsyncSession):
|
|
"""测试创建知识点"""
|
|
# 创建测试课程
|
|
course = Course(
|
|
name="测试课程",
|
|
description="用于测试知识点",
|
|
category=CourseCategory.TECHNOLOGY,
|
|
status=CourseStatus.DRAFT
|
|
)
|
|
db_session.add(course)
|
|
await db_session.commit()
|
|
await db_session.refresh(course)
|
|
|
|
# 创建知识点
|
|
point_data = {
|
|
"name": "Python基础",
|
|
"description": "学习Python基础知识",
|
|
"weight": 2.0,
|
|
"estimated_hours": 10
|
|
}
|
|
|
|
response = await client.post(
|
|
f"/api/v1/courses/{course.id}/knowledge-points",
|
|
json=point_data,
|
|
headers=admin_headers
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["data"]["name"] == point_data["name"]
|
|
assert data["data"]["course_id"] == course.id
|
|
assert data["data"]["level"] == 1
|
|
assert data["data"]["parent_id"] is None
|