Compare commits
133 Commits
| Author | SHA1 | Date |
|---|---|---|
|
|
be0e0325bb | 3 days ago |
|
|
bee3c2a63c | 4 days ago |
|
|
32a473f920 | 4 days ago |
|
|
a074c0a699 | 4 days ago |
|
|
dc5a337fc6 | 4 days ago |
|
|
4bbb649cf8 | 4 days ago |
|
|
a842ffd407 | 4 days ago |
|
|
077dd17f36 | 4 days ago |
|
|
af563d255d | 4 days ago |
|
|
e7596f1d35 | 4 days ago |
|
|
74180f8c37 | 4 days ago |
|
|
ddcbe53434 | 4 days ago |
|
|
fa97759fcf | 4 days ago |
|
|
48ab0d957c | 4 days ago |
|
|
5c77721eb2 | 4 days ago |
|
|
5bdb4ae526 | 5 days ago |
|
|
cd598826f4 | 5 days ago |
|
|
66b9aa8233 | 5 days ago |
|
|
2b1a07702b | 6 days ago |
|
|
67596cf412 | 6 days ago |
|
|
e472927698 | 6 days ago |
|
|
cfc5df5c45 | 6 days ago |
|
|
288906d3e7 | 6 days ago |
|
|
262f968b7a | 6 days ago |
|
|
a03726e52b | 6 days ago |
|
|
a1869e0d7f | 6 days ago |
|
|
d49680d725 | 6 days ago |
|
|
bb9247547d | 1 week ago |
|
|
ac6f783642 | 1 week ago |
|
|
f15f216ad5 | 1 week ago |
|
|
adba670b68 | 1 week ago |
|
|
f58dc1a83d | 1 week ago |
|
|
871ad5855e | 1 week ago |
|
|
c9b76b9673 | 1 week ago |
|
|
5f1051fb83 | 1 week ago |
|
|
06a86a9738 | 1 week ago |
|
|
cb67dc73a3 | 1 week ago |
|
|
e137505127 | 1 week ago |
|
|
8627af3e8f | 1 week ago |
|
|
e447ec0984 | 1 week ago |
|
|
c60c66cc13 | 1 week ago |
|
|
785360884d | 1 week ago |
|
|
b6e65cab47 | 1 week ago |
|
|
adca87d8f5 | 1 week ago |
|
|
df1e43ec3a | 1 week ago |
|
|
b5af0d22ab | 1 week ago |
|
|
fe90dc173e | 1 week ago |
|
|
54ddaed884 | 1 week ago |
|
|
e220d7d853 | 1 week ago |
|
|
c9e9790dee | 1 week ago |
|
|
0b4c141347 | 1 week ago |
|
|
b41601226c | 1 week ago |
|
|
ffe14d8c8c | 1 week ago |
|
|
f41785ade7 | 1 week ago |
|
|
203aff6acc | 1 week ago |
|
|
26c04b1271 | 1 week ago |
|
|
fa6c1a00fd | 1 week ago |
|
|
14138bbf67 | 1 week ago |
|
|
0b0889654d | 1 week ago |
|
|
ab8660b1a5 | 1 week ago |
|
|
0285bd87af | 1 week ago |
|
|
dab0f638f9 | 1 week ago |
|
|
9ff6cf0ebe | 1 week ago |
|
|
3193a98809 | 1 week ago |
|
|
ede170b310 | 1 week ago |
|
|
792245b059 | 1 week ago |
|
|
9b4ddd90d5 | 2 weeks ago |
|
|
52d4b8fbb2 | 2 weeks ago |
|
|
6ad854f436 | 2 weeks ago |
|
|
8fd6c8f023 | 2 weeks ago |
|
|
4ad327aa8d | 2 weeks ago |
|
|
1c4eb8a5ad | 2 weeks ago |
|
|
7838bcbd51 | 2 weeks ago |
|
|
92348a1055 | 2 weeks ago |
|
|
7befcacf25 | 2 weeks ago |
|
|
bc2470c0e5 | 2 weeks ago |
|
|
d590f380f8 | 2 weeks ago |
|
|
deebc98ac4 | 2 weeks ago |
|
|
922956b531 | 2 weeks ago |
|
|
7f5594b201 | 2 weeks ago |
|
|
7b2358cfaa | 2 weeks ago |
|
|
3761898828 | 2 weeks ago |
|
|
352a650b5e | 2 weeks ago |
|
|
eef08f0bb1 | 2 weeks ago |
|
|
d7e37eddbd | 2 weeks ago |
|
|
f97e66e36f | 2 weeks ago |
|
|
14de420214 | 2 weeks ago |
|
|
14f785932f | 2 weeks ago |
|
|
b8271a003c | 2 weeks ago |
|
|
d44156ba85 | 2 weeks ago |
|
|
71db922c56 | 2 weeks ago |
|
|
0f7e9d761c | 2 weeks ago |
|
|
f30e59cd8d | 2 weeks ago |
|
|
ee7f9c1225 | 2 weeks ago |
|
|
a71c0a4cf9 | 2 weeks ago |
|
|
7cc6ddc573 | 2 weeks ago |
|
|
c55301bc14 | 2 weeks ago |
|
|
1ca3892115 | 2 weeks ago |
|
|
2485d10735 | 2 weeks ago |
|
|
7fa9d3e62f | 2 weeks ago |
|
|
4f724b791d | 2 weeks ago |
|
|
f3ed994d4b | 2 weeks ago |
|
|
09d5b433dc | 2 weeks ago |
|
|
64838e0583 | 2 weeks ago |
|
|
732d0df542 | 2 weeks ago |
|
|
1ea95c3251 | 2 weeks ago |
|
|
dc1c95f4a6 | 2 weeks ago |
|
|
47f70ab6f5 | 2 weeks ago |
|
|
31bb750499 | 2 weeks ago |
|
|
4704e890aa | 2 weeks ago |
|
|
c4c9382028 | 2 weeks ago |
|
|
f43c9e6371 | 2 weeks ago |
|
|
5b7149b489 | 2 weeks ago |
|
|
825f88252b | 2 weeks ago |
|
|
65637f91d5 | 2 weeks ago |
|
|
77c51e04b4 | 2 weeks ago |
|
|
751f85f4d0 | 2 weeks ago |
|
|
52539efac5 | 2 weeks ago |
|
|
3c475894b9 | 2 weeks ago |
|
|
4038f551f9 | 3 weeks ago |
|
|
21643d5430 | 3 weeks ago |
|
|
7467e6d51d | 3 weeks ago |
|
|
d97d14f4eb | 3 weeks ago |
|
|
8eb4281da5 | 3 weeks ago |
|
|
6c0b64cb8c | 3 weeks ago |
|
|
adaa6e702f | 3 weeks ago |
|
|
2aadd03459 | 3 weeks ago |
|
|
5c66f8562c | 3 weeks ago |
|
|
e53fae1367 | 3 weeks ago |
|
|
3d30f1134e | 3 weeks ago |
|
|
ce11c50bd2 | 3 weeks ago |
|
|
7983a5b055 | 3 weeks ago |
|
|
5b9e732e2a | 3 weeks ago |
|
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 |
@ -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)
|
||||||