Compare commits
133 Commits
杨博文
Ryan
|
After Width: | Height: | Size: 18 KiB |
|
After Width: | Height: | Size: 30 KiB |
|
After Width: | Height: | Size: 22 KiB |
|
After Width: | Height: | Size: 27 KiB |
|
After Width: | Height: | Size: 41 KiB |
|
Before Width: | Height: | Size: 81 KiB |
|
After Width: | Height: | Size: 817 KiB |
|
After Width: | Height: | Size: 47 KiB |
|
Before Width: | Height: | Size: 149 KiB |
|
Before Width: | Height: | Size: 124 KiB |
|
After Width: | Height: | Size: 2.0 MiB |
|
After Width: | Height: | Size: 682 KiB |
|
Before Width: | Height: | Size: 125 KiB |
|
After Width: | Height: | Size: 368 KiB |
Binary file not shown.
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,5 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
MuseGuard 测试包
|
||||
包含单元测试和集成测试
|
||||
"""
|
||||
@ -0,0 +1,258 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
测试数据工厂
|
||||
使用 factory_boy 生成测试数据
|
||||
"""
|
||||
|
||||
import factory
|
||||
from factory.alchemy import SQLAlchemyModelFactory
|
||||
from faker import Faker
|
||||
from datetime import datetime
|
||||
|
||||
from app import db
|
||||
from app.database import (
|
||||
User, Role, Task, TaskType, TaskStatus, Image, ImageType,
|
||||
UserConfig, PerturbationConfig, FinetuneConfig, DataType,
|
||||
Perturbation, Finetune, Heatmap, Evaluate, EvaluationResult
|
||||
)
|
||||
|
||||
fake = Faker('zh_CN')
|
||||
|
||||
|
||||
class BaseFactory(SQLAlchemyModelFactory):
|
||||
"""基础工厂类"""
|
||||
|
||||
class Meta:
|
||||
abstract = True
|
||||
sqlalchemy_session = db.session
|
||||
sqlalchemy_session_persistence = 'commit'
|
||||
|
||||
|
||||
class RoleFactory(BaseFactory):
|
||||
"""角色工厂"""
|
||||
|
||||
class Meta:
|
||||
model = Role
|
||||
|
||||
role_code = factory.Sequence(lambda n: f'role_{n}')
|
||||
name = factory.LazyAttribute(lambda obj: f'角色_{obj.role_code}')
|
||||
max_concurrent_tasks = factory.Faker('random_int', min=1, max=10)
|
||||
description = factory.Faker('sentence', locale='zh_CN')
|
||||
|
||||
|
||||
class UserFactory(BaseFactory):
|
||||
"""用户工厂"""
|
||||
|
||||
class Meta:
|
||||
model = User
|
||||
|
||||
username = factory.Sequence(lambda n: f'user_{n}')
|
||||
email = factory.LazyAttribute(lambda obj: f'{obj.username}@example.com')
|
||||
role_id = 3 # 默认普通用户
|
||||
is_active = True
|
||||
|
||||
@factory.lazy_attribute
|
||||
def password_hash(self):
|
||||
from werkzeug.security import generate_password_hash
|
||||
return generate_password_hash('defaultpassword123')
|
||||
|
||||
@classmethod
|
||||
def _create(cls, model_class, *args, **kwargs):
|
||||
"""创建用户时设置密码"""
|
||||
password = kwargs.pop('password', 'defaultpassword123')
|
||||
obj = super()._create(model_class, *args, **kwargs)
|
||||
obj.set_password(password)
|
||||
return obj
|
||||
|
||||
|
||||
class AdminUserFactory(UserFactory):
|
||||
"""管理员用户工厂"""
|
||||
|
||||
username = factory.Sequence(lambda n: f'admin_{n}')
|
||||
role_id = 1
|
||||
|
||||
|
||||
class VipUserFactory(UserFactory):
|
||||
"""VIP用户工厂"""
|
||||
|
||||
username = factory.Sequence(lambda n: f'vip_{n}')
|
||||
role_id = 2
|
||||
|
||||
|
||||
class TaskTypeFactory(BaseFactory):
|
||||
"""任务类型工厂"""
|
||||
|
||||
class Meta:
|
||||
model = TaskType
|
||||
|
||||
task_type_code = factory.Sequence(lambda n: f'type_{n}')
|
||||
task_type_name = factory.LazyAttribute(lambda obj: f'任务类型_{obj.task_type_code}')
|
||||
description = factory.Faker('sentence', locale='zh_CN')
|
||||
|
||||
|
||||
class TaskStatusFactory(BaseFactory):
|
||||
"""任务状态工厂"""
|
||||
|
||||
class Meta:
|
||||
model = TaskStatus
|
||||
|
||||
task_status_code = factory.Sequence(lambda n: f'status_{n}')
|
||||
task_status_name = factory.LazyAttribute(lambda obj: f'状态_{obj.task_status_code}')
|
||||
description = factory.Faker('sentence', locale='zh_CN')
|
||||
|
||||
|
||||
class ImageTypeFactory(BaseFactory):
|
||||
"""图片类型工厂"""
|
||||
|
||||
class Meta:
|
||||
model = ImageType
|
||||
|
||||
image_code = factory.Sequence(lambda n: f'img_type_{n}')
|
||||
image_name = factory.LazyAttribute(lambda obj: f'图片类型_{obj.image_code}')
|
||||
description = factory.Faker('sentence', locale='zh_CN')
|
||||
|
||||
|
||||
class DataTypeFactory(BaseFactory):
|
||||
"""数据类型工厂"""
|
||||
|
||||
class Meta:
|
||||
model = DataType
|
||||
|
||||
data_type_code = factory.Sequence(lambda n: f'data_{n}')
|
||||
instance_prompt = factory.Faker('sentence', locale='zh_CN')
|
||||
class_prompt = factory.Faker('sentence', locale='zh_CN')
|
||||
description = factory.Faker('sentence', locale='zh_CN')
|
||||
|
||||
|
||||
class PerturbationConfigFactory(BaseFactory):
|
||||
"""加噪配置工厂"""
|
||||
|
||||
class Meta:
|
||||
model = PerturbationConfig
|
||||
|
||||
perturbation_code = factory.Sequence(lambda n: f'pert_{n}')
|
||||
perturbation_name = factory.LazyAttribute(lambda obj: f'加噪算法_{obj.perturbation_code}')
|
||||
description = factory.Faker('sentence', locale='zh_CN')
|
||||
|
||||
|
||||
class FinetuneConfigFactory(BaseFactory):
|
||||
"""微调配置工厂"""
|
||||
|
||||
class Meta:
|
||||
model = FinetuneConfig
|
||||
|
||||
finetune_code = factory.Sequence(lambda n: f'ft_{n}')
|
||||
finetune_name = factory.LazyAttribute(lambda obj: f'微调方式_{obj.finetune_code}')
|
||||
description = factory.Faker('sentence', locale='zh_CN')
|
||||
|
||||
|
||||
class TaskFactory(BaseFactory):
|
||||
"""任务工厂"""
|
||||
|
||||
class Meta:
|
||||
model = Task
|
||||
|
||||
flow_id = factory.Sequence(lambda n: 1000000 + n)
|
||||
tasks_type_id = 1 # perturbation
|
||||
user_id = factory.LazyAttribute(lambda obj: UserFactory().user_id)
|
||||
tasks_status_id = 1 # waiting
|
||||
description = factory.Faker('sentence', locale='zh_CN')
|
||||
created_at = factory.LazyFunction(datetime.now)
|
||||
|
||||
|
||||
class PerturbationTaskFactory(TaskFactory):
|
||||
"""加噪任务工厂"""
|
||||
|
||||
tasks_type_id = 1
|
||||
|
||||
@factory.post_generation
|
||||
def perturbation(obj, create, extracted, **kwargs):
|
||||
if not create:
|
||||
return
|
||||
|
||||
if extracted:
|
||||
return extracted
|
||||
|
||||
return PerturbationFactory(tasks_id=obj.tasks_id)
|
||||
|
||||
|
||||
class FinetuneTaskFactory(TaskFactory):
|
||||
"""微调任务工厂"""
|
||||
|
||||
tasks_type_id = 2
|
||||
|
||||
@factory.post_generation
|
||||
def finetune(obj, create, extracted, **kwargs):
|
||||
if not create:
|
||||
return
|
||||
|
||||
if extracted:
|
||||
return extracted
|
||||
|
||||
return FinetuneFactory(tasks_id=obj.tasks_id)
|
||||
|
||||
|
||||
class PerturbationFactory(BaseFactory):
|
||||
"""加噪详情工厂"""
|
||||
|
||||
class Meta:
|
||||
model = Perturbation
|
||||
|
||||
tasks_id = factory.LazyAttribute(lambda obj: TaskFactory(tasks_type_id=1).tasks_id)
|
||||
data_type_id = 1
|
||||
perturbation_configs_id = 1
|
||||
perturbation_intensity = factory.Faker('pyfloat', min_value=0.1, max_value=1.0)
|
||||
perturbation_name = factory.Faker('word', locale='zh_CN')
|
||||
|
||||
|
||||
class FinetuneFactory(BaseFactory):
|
||||
"""微调详情工厂"""
|
||||
|
||||
class Meta:
|
||||
model = Finetune
|
||||
|
||||
tasks_id = factory.LazyAttribute(lambda obj: TaskFactory(tasks_type_id=2).tasks_id)
|
||||
finetune_configs_id = 1
|
||||
data_type_id = 1
|
||||
finetune_name = factory.Faker('word', locale='zh_CN')
|
||||
custom_prompt = factory.Faker('sentence', locale='zh_CN')
|
||||
|
||||
|
||||
class ImageFactory(BaseFactory):
|
||||
"""图片工厂"""
|
||||
|
||||
class Meta:
|
||||
model = Image
|
||||
|
||||
task_id = factory.LazyAttribute(lambda obj: TaskFactory().tasks_id)
|
||||
image_types_id = 1 # original
|
||||
stored_filename = factory.Sequence(lambda n: f'image_{n}.png')
|
||||
file_path = factory.LazyAttribute(lambda obj: f'/tmp/{obj.stored_filename}')
|
||||
file_size = factory.Faker('random_int', min=1024, max=1024*1024)
|
||||
width = factory.Faker('random_int', min=256, max=2048)
|
||||
height = factory.Faker('random_int', min=256, max=2048)
|
||||
|
||||
|
||||
class UserConfigFactory(BaseFactory):
|
||||
"""用户配置工厂"""
|
||||
|
||||
class Meta:
|
||||
model = UserConfig
|
||||
|
||||
user_id = factory.LazyAttribute(lambda obj: UserFactory().user_id)
|
||||
data_type_id = 1
|
||||
perturbation_configs_id = 1
|
||||
perturbation_intensity = factory.Faker('pyfloat', min_value=0.1, max_value=1.0)
|
||||
finetune_configs_id = 1
|
||||
|
||||
|
||||
class EvaluationResultFactory(BaseFactory):
|
||||
"""评估结果工厂"""
|
||||
|
||||
class Meta:
|
||||
model = EvaluationResult
|
||||
|
||||
fid_score = factory.Faker('pyfloat', min_value=0.0, max_value=100.0)
|
||||
lpips_score = factory.Faker('pyfloat', min_value=0.0, max_value=1.0)
|
||||
ssim_score = factory.Faker('pyfloat', min_value=0.0, max_value=1.0)
|
||||
psnr_score = factory.Faker('pyfloat', min_value=10.0, max_value=50.0)
|
||||
@ -0,0 +1,5 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
集成测试包
|
||||
测试API端点和组件间的交互
|
||||
"""
|
||||
@ -0,0 +1,251 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
认证API集成测试
|
||||
测试用户注册、登录、密码修改等接口
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
|
||||
class TestAuthRegister:
|
||||
"""用户注册接口测试"""
|
||||
|
||||
@patch('app.controllers.auth_controller.VerificationService')
|
||||
def test_register_成功注册(self, mock_verification, client, init_database):
|
||||
"""测试成功注册新用户"""
|
||||
# Mock 验证码服务
|
||||
mock_service = MagicMock()
|
||||
mock_service.verify_code.return_value = True
|
||||
mock_verification.return_value = mock_service
|
||||
|
||||
response = client.post('/api/auth/register', json={
|
||||
'username': 'newuser',
|
||||
'password': 'newpassword123',
|
||||
'email': 'newuser@example.com',
|
||||
'code': '123456'
|
||||
})
|
||||
|
||||
assert response.status_code == 201
|
||||
data = response.get_json()
|
||||
assert 'message' in data
|
||||
assert data['message'] == '注册成功'
|
||||
assert 'user' in data
|
||||
assert data['user']['username'] == 'newuser'
|
||||
|
||||
def test_register_缺少必填字段(self, client, init_database):
|
||||
"""测试注册时缺少必填字段"""
|
||||
response = client.post('/api/auth/register', json={
|
||||
'username': 'newuser'
|
||||
# 缺少 password 和 email
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
|
||||
@patch('app.controllers.auth_controller.VerificationService')
|
||||
def test_register_用户名已存在(self, mock_verification, client, init_database, test_user):
|
||||
"""测试注册已存在的用户名"""
|
||||
mock_service = MagicMock()
|
||||
mock_service.verify_code.return_value = True
|
||||
mock_verification.return_value = mock_service
|
||||
|
||||
response = client.post('/api/auth/register', json={
|
||||
'username': 'testuser', # 已存在
|
||||
'password': 'password123',
|
||||
'email': 'another@example.com',
|
||||
'code': '123456'
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert '用户名已存在' in data['error']
|
||||
|
||||
@patch('app.controllers.auth_controller.VerificationService')
|
||||
def test_register_邮箱已存在(self, mock_verification, client, init_database, test_user):
|
||||
"""测试注册已存在的邮箱"""
|
||||
mock_service = MagicMock()
|
||||
mock_service.verify_code.return_value = True
|
||||
mock_verification.return_value = mock_service
|
||||
|
||||
response = client.post('/api/auth/register', json={
|
||||
'username': 'anotheruser',
|
||||
'password': 'password123',
|
||||
'email': 'test@example.com', # 已存在
|
||||
'code': '123456'
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert '邮箱' in data['error']
|
||||
|
||||
def test_register_邮箱格式错误(self, client, init_database):
|
||||
"""测试注册时邮箱格式错误"""
|
||||
response = client.post('/api/auth/register', json={
|
||||
'username': 'newuser',
|
||||
'password': 'password123',
|
||||
'email': 'invalid-email',
|
||||
'code': '123456'
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert '邮箱格式' in data['error']
|
||||
|
||||
|
||||
class TestAuthLogin:
|
||||
"""用户登录接口测试"""
|
||||
|
||||
def test_login_成功登录(self, client, init_database, test_user):
|
||||
"""测试成功登录"""
|
||||
response = client.post('/api/auth/login', json={
|
||||
'username': 'testuser',
|
||||
'password': 'testpassword123'
|
||||
})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert 'access_token' in data
|
||||
assert 'user' in data
|
||||
assert data['user']['username'] == 'testuser'
|
||||
|
||||
def test_login_错误密码(self, client, init_database, test_user):
|
||||
"""测试使用错误密码登录"""
|
||||
response = client.post('/api/auth/login', json={
|
||||
'username': 'testuser',
|
||||
'password': 'wrongpassword'
|
||||
})
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert '用户名或密码错误' in data['error']
|
||||
|
||||
def test_login_不存在的用户(self, client, init_database):
|
||||
"""测试登录不存在的用户"""
|
||||
response = client.post('/api/auth/login', json={
|
||||
'username': 'nonexistent',
|
||||
'password': 'password123'
|
||||
})
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert '用户名或密码错误' in data['error']
|
||||
|
||||
def test_login_缺少字段(self, client, init_database):
|
||||
"""测试登录时缺少字段"""
|
||||
response = client.post('/api/auth/login', json={
|
||||
'username': 'testuser'
|
||||
# 缺少 password
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert 'error' in data
|
||||
|
||||
def test_login_禁用账户(self, init_database, client):
|
||||
"""测试登录被禁用的账户"""
|
||||
from app.database import User
|
||||
from app import db
|
||||
|
||||
# 创建禁用用户
|
||||
user = User(
|
||||
username='disableduser',
|
||||
email='disabled@example.com',
|
||||
role_id=3,
|
||||
is_active=False
|
||||
)
|
||||
user.set_password('password123')
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
|
||||
response = client.post('/api/auth/login', json={
|
||||
'username': 'disableduser',
|
||||
'password': 'password123'
|
||||
})
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert '禁用' in data['error']
|
||||
|
||||
|
||||
class TestAuthProfile:
|
||||
"""用户信息接口测试"""
|
||||
|
||||
def test_get_profile_获取用户信息(self, client, auth_headers):
|
||||
"""测试获取当前用户信息"""
|
||||
response = client.get('/api/auth/profile', headers=auth_headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert 'user' in data
|
||||
assert data['user']['username'] == 'testuser'
|
||||
|
||||
def test_get_profile_未认证(self, client, init_database):
|
||||
"""测试未认证时获取用户信息"""
|
||||
response = client.get('/api/auth/profile')
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestAuthChangePassword:
|
||||
"""修改密码接口测试"""
|
||||
|
||||
def test_change_password_成功修改(self, client, auth_headers, test_user):
|
||||
"""测试成功修改密码"""
|
||||
response = client.post('/api/auth/change-password',
|
||||
headers=auth_headers,
|
||||
json={
|
||||
'old_password': 'testpassword123',
|
||||
'new_password': 'newpassword456'
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert '成功' in data['message']
|
||||
|
||||
# 验证新密码可以登录
|
||||
login_response = client.post('/api/auth/login', json={
|
||||
'username': 'testuser',
|
||||
'password': 'newpassword456'
|
||||
})
|
||||
assert login_response.status_code == 200
|
||||
|
||||
def test_change_password_旧密码错误(self, client, auth_headers):
|
||||
"""测试使用错误的旧密码修改"""
|
||||
response = client.post('/api/auth/change-password',
|
||||
headers=auth_headers,
|
||||
json={
|
||||
'old_password': 'wrongoldpassword',
|
||||
'new_password': 'newpassword456'
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.get_json()
|
||||
assert '旧密码错误' in data['error']
|
||||
|
||||
def test_change_password_缺少字段(self, client, auth_headers):
|
||||
"""测试修改密码时缺少字段"""
|
||||
response = client.post('/api/auth/change-password',
|
||||
headers=auth_headers,
|
||||
json={
|
||||
'old_password': 'testpassword123'
|
||||
# 缺少 new_password
|
||||
}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
class TestAuthLogout:
|
||||
"""用户登出接口测试"""
|
||||
|
||||
def test_logout_成功登出(self, client, auth_headers):
|
||||
"""测试成功登出"""
|
||||
response = client.post('/api/auth/logout', headers=auth_headers)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert '登出成功' in data['message']
|
||||
@ -0,0 +1,5 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
单元测试包
|
||||
测试独立的函数和类方法
|
||||
"""
|
||||
@ -0,0 +1,232 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Repository 层单元测试
|
||||
测试数据访问层的 CRUD 操作
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from app.database import User, Task, Image, UserConfig
|
||||
from app.repositories.user_repository import UserRepository, UserConfigRepository
|
||||
from app.repositories.task_repository import TaskRepository, PerturbationRepository
|
||||
from app.repositories.image_repository import ImageRepository
|
||||
|
||||
|
||||
class TestUserRepository:
|
||||
"""用户 Repository 单元测试"""
|
||||
|
||||
def test_get_by_username_存在的用户(self, init_database, test_user):
|
||||
"""测试根据用户名获取存在的用户"""
|
||||
repo = UserRepository()
|
||||
user = repo.get_by_username('testuser')
|
||||
|
||||
assert user is not None
|
||||
assert user.username == 'testuser'
|
||||
assert user.email == 'test@example.com'
|
||||
|
||||
def test_get_by_username_不存在的用户(self, init_database):
|
||||
"""测试根据用户名获取不存在的用户"""
|
||||
repo = UserRepository()
|
||||
user = repo.get_by_username('nonexistent')
|
||||
|
||||
assert user is None
|
||||
|
||||
def test_get_by_email_存在的邮箱(self, init_database, test_user):
|
||||
"""测试根据邮箱获取用户"""
|
||||
repo = UserRepository()
|
||||
user = repo.get_by_email('test@example.com')
|
||||
|
||||
assert user is not None
|
||||
assert user.username == 'testuser'
|
||||
|
||||
def test_username_exists_检查用户名是否存在(self, init_database, test_user):
|
||||
"""测试检查用户名是否存在"""
|
||||
repo = UserRepository()
|
||||
|
||||
assert repo.username_exists('testuser') is True
|
||||
assert repo.username_exists('nonexistent') is False
|
||||
|
||||
def test_email_exists_检查邮箱是否存在(self, init_database, test_user):
|
||||
"""测试检查邮箱是否存在"""
|
||||
repo = UserRepository()
|
||||
|
||||
assert repo.email_exists('test@example.com') is True
|
||||
assert repo.email_exists('nonexistent@example.com') is False
|
||||
|
||||
def test_authenticate_正确凭据(self, init_database, test_user):
|
||||
"""测试使用正确凭据认证"""
|
||||
repo = UserRepository()
|
||||
user = repo.authenticate('testuser', 'testpassword123')
|
||||
|
||||
assert user is not None
|
||||
assert user.username == 'testuser'
|
||||
|
||||
def test_authenticate_错误密码(self, init_database, test_user):
|
||||
"""测试使用错误密码认证"""
|
||||
repo = UserRepository()
|
||||
user = repo.authenticate('testuser', 'wrongpassword')
|
||||
|
||||
assert user is None
|
||||
|
||||
def test_authenticate_不存在的用户(self, init_database):
|
||||
"""测试认证不存在的用户"""
|
||||
repo = UserRepository()
|
||||
user = repo.authenticate('nonexistent', 'password')
|
||||
|
||||
assert user is None
|
||||
|
||||
def test_is_admin_管理员用户(self, init_database, admin_user):
|
||||
"""测试判断管理员用户"""
|
||||
repo = UserRepository()
|
||||
|
||||
assert repo.is_admin(admin_user) is True
|
||||
|
||||
def test_is_admin_普通用户(self, init_database, test_user):
|
||||
"""测试判断普通用户不是管理员"""
|
||||
repo = UserRepository()
|
||||
|
||||
assert repo.is_admin(test_user) is False
|
||||
|
||||
def test_is_vip_VIP用户(self, init_database, vip_user):
|
||||
"""测试判断VIP用户"""
|
||||
repo = UserRepository()
|
||||
|
||||
assert repo.is_vip(vip_user) is True
|
||||
|
||||
def test_get_max_concurrent_tasks_不同角色(self, init_database, test_user, admin_user, vip_user):
|
||||
"""测试获取不同角色的最大并发任务数"""
|
||||
repo = UserRepository()
|
||||
|
||||
assert repo.get_max_concurrent_tasks(admin_user) == 10
|
||||
assert repo.get_max_concurrent_tasks(vip_user) == 5
|
||||
assert repo.get_max_concurrent_tasks(test_user) == 2
|
||||
|
||||
|
||||
class TestTaskRepository:
|
||||
"""任务 Repository 单元测试"""
|
||||
|
||||
def test_get_by_user_获取用户任务(self, init_database, sample_task, test_user):
|
||||
"""测试获取用户的所有任务"""
|
||||
repo = TaskRepository()
|
||||
tasks = repo.get_by_user(test_user.user_id)
|
||||
|
||||
assert len(tasks) == 1
|
||||
assert tasks[0].tasks_id == sample_task.tasks_id
|
||||
|
||||
def test_get_by_user_无任务用户(self, init_database, admin_user):
|
||||
"""测试获取无任务用户的任务列表"""
|
||||
repo = TaskRepository()
|
||||
tasks = repo.get_by_user(admin_user.user_id)
|
||||
|
||||
assert len(tasks) == 0
|
||||
|
||||
def test_is_owner_验证任务归属(self, init_database, sample_task, test_user, admin_user):
|
||||
"""测试验证任务归属"""
|
||||
repo = TaskRepository()
|
||||
|
||||
assert repo.is_owner(sample_task, test_user.user_id) is True
|
||||
assert repo.is_owner(sample_task, admin_user.user_id) is False
|
||||
|
||||
def test_get_for_user_获取用户任务带权限验证(self, init_database, sample_task, test_user, admin_user):
|
||||
"""测试获取用户任务（带权限验证）"""
|
||||
repo = TaskRepository()
|
||||
|
||||
# 任务所有者可以获取
|
||||
task = repo.get_for_user(sample_task.tasks_id, test_user.user_id)
|
||||
assert task is not None
|
||||
|
||||
# 非所有者无法获取
|
||||
task = repo.get_for_user(sample_task.tasks_id, admin_user.user_id)
|
||||
assert task is None
|
||||
|
||||
def test_get_type_code_获取任务类型代码(self, init_database, sample_task):
|
||||
"""测试获取任务类型代码"""
|
||||
repo = TaskRepository()
|
||||
type_code = repo.get_type_code(sample_task)
|
||||
|
||||
assert type_code == 'perturbation'
|
||||
|
||||
def test_is_type_判断任务类型(self, init_database, sample_task):
|
||||
"""测试判断任务类型"""
|
||||
repo = TaskRepository()
|
||||
|
||||
assert repo.is_type(sample_task, 'perturbation') is True
|
||||
assert repo.is_type(sample_task, 'finetune') is False
|
||||
|
||||
|
||||
class TestImageRepository:
|
||||
"""图片 Repository 单元测试"""
|
||||
|
||||
def test_get_by_task_获取任务图片(self, init_database, sample_image, sample_task):
|
||||
"""测试获取任务的所有图片"""
|
||||
repo = ImageRepository()
|
||||
images = repo.get_by_task(sample_task.tasks_id)
|
||||
|
||||
assert len(images) == 1
|
||||
assert images[0].images_id == sample_image.images_id
|
||||
|
||||
def test_count_by_task_统计任务图片数量(self, init_database, sample_image, sample_task):
|
||||
"""测试统计任务的图片数量"""
|
||||
repo = ImageRepository()
|
||||
count = repo.count_by_task(sample_task.tasks_id)
|
||||
|
||||
assert count == 1
|
||||
|
||||
def test_is_owner_验证图片归属(self, init_database, sample_image, test_user, admin_user):
|
||||
"""测试验证图片归属（通过任务）"""
|
||||
repo = ImageRepository()
|
||||
|
||||
assert repo.is_owner(sample_image, test_user.user_id) is True
|
||||
assert repo.is_owner(sample_image, admin_user.user_id) is False
|
||||
|
||||
def test_get_by_task_and_type_获取指定类型图片(self, init_database, sample_image, sample_task):
|
||||
"""测试获取任务指定类型的图片"""
|
||||
repo = ImageRepository()
|
||||
|
||||
# 获取原图类型
|
||||
images = repo.get_by_task_and_type(sample_task.tasks_id, 'original')
|
||||
assert len(images) == 1
|
||||
|
||||
# 获取不存在的类型
|
||||
images = repo.get_by_task_and_type(sample_task.tasks_id, 'perturbed')
|
||||
assert len(images) == 0
|
||||
|
||||
|
||||
class TestBaseRepository:
|
||||
"""基础 Repository 单元测试"""
|
||||
|
||||
def test_get_by_id_获取实体(self, init_database, test_user):
|
||||
"""测试根据ID获取实体"""
|
||||
repo = UserRepository()
|
||||
user = repo.get_by_id(test_user.user_id)
|
||||
|
||||
assert user is not None
|
||||
assert user.user_id == test_user.user_id
|
||||
|
||||
def test_get_by_id_不存在的实体(self, init_database):
|
||||
"""测试获取不存在的实体"""
|
||||
repo = UserRepository()
|
||||
user = repo.get_by_id(99999)
|
||||
|
||||
assert user is None
|
||||
|
||||
def test_exists_检查实体存在(self, init_database, test_user):
|
||||
"""测试检查实体是否存在"""
|
||||
repo = UserRepository()
|
||||
|
||||
assert repo.exists(test_user.user_id) is True
|
||||
assert repo.exists(99999) is False
|
||||
|
||||
def test_count_统计数量(self, init_database, test_user, admin_user):
|
||||
"""测试统计实体数量"""
|
||||
repo = UserRepository()
|
||||
count = repo.count()
|
||||
|
||||
assert count == 2 # test_user 和 admin_user
|
||||
|
||||
def test_find_by_条件查询(self, init_database, test_user):
|
||||
"""测试根据条件查询"""
|
||||
repo = UserRepository()
|
||||
users = repo.find_by(is_active=True)
|
||||
|
||||
assert len(users) >= 1
|
||||
assert all(u.is_active for u in users)
|
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue