392 lines
16 KiB
Python
392 lines
16 KiB
Python
|
import pytest
|
||
|
|
import json
|
||
|
|
from app.models import User, TeamMember, File, DownloadRecord
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
|
||
|
|
|
||
|
|
class TestTeamLeaderManagement:
|
||
|
|
"""团队长管理功能测试"""
|
||
|
|
|
||
|
|
def test_get_team_leaders_admin_only(self, client, auth_headers):
|
||
|
|
"""测试只有管理员可以获取团队长列表"""
|
||
|
|
# 普通用户尝试访问
|
||
|
|
response = client.get('/api/admin/team-leaders', headers=auth_headers['user'])
|
||
|
|
assert response.status_code == 403
|
||
|
|
|
||
|
|
# 团队长尝试访问
|
||
|
|
response = client.get('/api/admin/team-leaders', headers=auth_headers['leader'])
|
||
|
|
assert response.status_code == 403
|
||
|
|
|
||
|
|
# 管理员访问
|
||
|
|
response = client.get('/api/admin/team-leaders', headers=auth_headers['admin'])
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert 'team_leaders' in response.json
|
||
|
|
assert 'total' in response.json
|
||
|
|
assert 'pages' in response.json
|
||
|
|
|
||
|
|
def test_set_team_leader_permission(self, client, auth_headers, app):
|
||
|
|
"""测试设置团队长权限"""
|
||
|
|
with app.app_context():
|
||
|
|
# 创建一个普通用户
|
||
|
|
from app.models import User
|
||
|
|
from werkzeug.security import generate_password_hash
|
||
|
|
|
||
|
|
test_user = User(
|
||
|
|
username='testuser',
|
||
|
|
email='test@test.com',
|
||
|
|
password_hash=generate_password_hash('test123'),
|
||
|
|
is_admin=False,
|
||
|
|
is_team_leader=False,
|
||
|
|
is_active=True,
|
||
|
|
created_at=datetime.utcnow()
|
||
|
|
)
|
||
|
|
from app import db
|
||
|
|
db.session.add(test_user)
|
||
|
|
db.session.commit()
|
||
|
|
user_id = test_user.id
|
||
|
|
|
||
|
|
# 管理员设置团队长权限
|
||
|
|
response = client.put(f'/api/admin/users/{user_id}/team-leader',
|
||
|
|
headers=auth_headers['admin'],
|
||
|
|
json={'is_team_leader': True})
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.json['message'] == '团队长权限已设置'
|
||
|
|
|
||
|
|
# 验证用户已成为团队长
|
||
|
|
user = User.query.get(user_id)
|
||
|
|
assert user.is_team_leader is True
|
||
|
|
|
||
|
|
# 取消团队长权限
|
||
|
|
response = client.put(f'/api/admin/users/{user_id}/team-leader',
|
||
|
|
headers=auth_headers['admin'],
|
||
|
|
json={'is_team_leader': False})
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.json['message'] == '团队长权限已取消'
|
||
|
|
|
||
|
|
# 验证权限已取消
|
||
|
|
user = User.query.get(user_id)
|
||
|
|
assert user.is_team_leader is False
|
||
|
|
|
||
|
|
def test_add_team_member(self, client, auth_headers, app):
|
||
|
|
"""测试添加团队成员"""
|
||
|
|
with app.app_context():
|
||
|
|
from app.models import User
|
||
|
|
from werkzeug.security import generate_password_hash
|
||
|
|
from app import db
|
||
|
|
|
||
|
|
# 创建团队成员用户
|
||
|
|
member_user = User(
|
||
|
|
username='member',
|
||
|
|
email='member@test.com',
|
||
|
|
password_hash=generate_password_hash('member123'),
|
||
|
|
is_admin=False,
|
||
|
|
is_team_leader=False,
|
||
|
|
is_active=True,
|
||
|
|
created_at=datetime.utcnow()
|
||
|
|
)
|
||
|
|
db.session.add(member_user)
|
||
|
|
db.session.commit()
|
||
|
|
member_id = member_user.id
|
||
|
|
|
||
|
|
# 获取团队长ID
|
||
|
|
leader = User.query.filter_by(username='leader').first()
|
||
|
|
leader_id = leader.id
|
||
|
|
|
||
|
|
# 管理员添加团队成员
|
||
|
|
response = client.post('/api/admin/team-members',
|
||
|
|
headers=auth_headers['admin'],
|
||
|
|
json={'leader_id': leader_id, 'member_id': member_id})
|
||
|
|
assert response.status_code == 201
|
||
|
|
assert response.json['message'] == '团队成员添加成功'
|
||
|
|
|
||
|
|
# 验证团队成员关系已创建
|
||
|
|
team_member = TeamMember.query.filter_by(
|
||
|
|
leader_id=leader_id, member_id=member_id).first()
|
||
|
|
assert team_member is not None
|
||
|
|
|
||
|
|
def test_remove_team_member(self, client, auth_headers, app):
|
||
|
|
"""测试移除团队成员"""
|
||
|
|
with app.app_context():
|
||
|
|
from app.models import User
|
||
|
|
from werkzeug.security import generate_password_hash
|
||
|
|
from app import db
|
||
|
|
|
||
|
|
# 创建团队成员用户
|
||
|
|
member_user = User(
|
||
|
|
username='member2',
|
||
|
|
email='member2@test.com',
|
||
|
|
password_hash=generate_password_hash('member123'),
|
||
|
|
is_admin=False,
|
||
|
|
is_team_leader=False,
|
||
|
|
is_active=True,
|
||
|
|
created_at=datetime.utcnow()
|
||
|
|
)
|
||
|
|
db.session.add(member_user)
|
||
|
|
db.session.commit()
|
||
|
|
member_id = member_user.id
|
||
|
|
|
||
|
|
# 获取团队长ID
|
||
|
|
leader = User.query.filter_by(username='leader').first()
|
||
|
|
leader_id = leader.id
|
||
|
|
|
||
|
|
# 先添加团队成员
|
||
|
|
team_member = TeamMember(leader_id=leader_id, member_id=member_id)
|
||
|
|
db.session.add(team_member)
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
# 管理员移除团队成员
|
||
|
|
response = client.delete(f'/api/admin/team-members/{member_id}',
|
||
|
|
headers=auth_headers['admin'])
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.json['message'] == '团队成员已移除'
|
||
|
|
|
||
|
|
# 验证团队成员关系已删除
|
||
|
|
team_member = TeamMember.query.filter_by(
|
||
|
|
leader_id=leader_id, member_id=member_id).first()
|
||
|
|
assert team_member is None
|
||
|
|
|
||
|
|
def test_get_users_without_team(self, client, auth_headers):
|
||
|
|
"""测试获取没有团队的用户"""
|
||
|
|
response = client.get('/api/admin/users/without-team',
|
||
|
|
headers=auth_headers['admin'])
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert 'users' in response.json
|
||
|
|
assert 'total' in response.json
|
||
|
|
assert isinstance(response.json['users'], list)
|
||
|
|
|
||
|
|
|
||
|
|
class TestTeamLeaderFunctions:
|
||
|
|
"""团队长功能测试"""
|
||
|
|
|
||
|
|
def test_get_team_members_leader_only(self, client, auth_headers):
|
||
|
|
"""测试只有团队长可以获取团队成员列表"""
|
||
|
|
# 普通用户尝试访问
|
||
|
|
response = client.get('/api/team/members', headers=auth_headers['user'])
|
||
|
|
assert response.status_code == 403
|
||
|
|
|
||
|
|
# 管理员尝试访问
|
||
|
|
response = client.get('/api/team/members', headers=auth_headers['admin'])
|
||
|
|
assert response.status_code == 403
|
||
|
|
|
||
|
|
# 团队长访问
|
||
|
|
response = client.get('/api/team/members', headers=auth_headers['leader'])
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert 'members' in response.json
|
||
|
|
assert 'total' in response.json
|
||
|
|
|
||
|
|
def test_get_team_statistics_leader_only(self, client, auth_headers):
|
||
|
|
"""测试只有团队长可以获取团队统计信息"""
|
||
|
|
# 普通用户尝试访问
|
||
|
|
response = client.get('/api/team/statistics', headers=auth_headers['user'])
|
||
|
|
assert response.status_code == 403
|
||
|
|
|
||
|
|
# 团队长访问
|
||
|
|
response = client.get('/api/team/statistics', headers=auth_headers['leader'])
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert 'total_members' in response.json
|
||
|
|
assert 'total_files' in response.json
|
||
|
|
assert 'total_downloads' in response.json
|
||
|
|
|
||
|
|
def test_update_member_quota_leader_only(self, client, auth_headers, app):
|
||
|
|
"""测试只有团队长可以更新成员配额"""
|
||
|
|
with app.app_context():
|
||
|
|
from app.models import User
|
||
|
|
from werkzeug.security import generate_password_hash
|
||
|
|
from app import db
|
||
|
|
|
||
|
|
# 创建团队成员
|
||
|
|
member_user = User(
|
||
|
|
username='teammember',
|
||
|
|
email='teammember@test.com',
|
||
|
|
password_hash=generate_password_hash('member123'),
|
||
|
|
is_admin=False,
|
||
|
|
is_team_leader=False,
|
||
|
|
is_active=True,
|
||
|
|
daily_download_limit=5,
|
||
|
|
created_at=datetime.utcnow()
|
||
|
|
)
|
||
|
|
db.session.add(member_user)
|
||
|
|
db.session.commit()
|
||
|
|
member_id = member_user.id
|
||
|
|
|
||
|
|
# 添加团队成员关系
|
||
|
|
leader = User.query.filter_by(username='leader').first()
|
||
|
|
team_member = TeamMember(leader_id=leader.id, member_id=member_id)
|
||
|
|
db.session.add(team_member)
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
# 普通用户尝试更新配额
|
||
|
|
response = client.put(f'/api/team/members/{member_id}/quota',
|
||
|
|
headers=auth_headers['user'],
|
||
|
|
json={'daily_download_limit': 10})
|
||
|
|
assert response.status_code == 403
|
||
|
|
|
||
|
|
# 团队长更新配额
|
||
|
|
response = client.put(f'/api/team/members/{member_id}/quota',
|
||
|
|
headers=auth_headers['leader'],
|
||
|
|
json={'daily_download_limit': 10})
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.json['message'] == '成员配额已更新'
|
||
|
|
|
||
|
|
# 验证配额已更新
|
||
|
|
member = User.query.get(member_id)
|
||
|
|
assert member.daily_download_limit == 10
|
||
|
|
|
||
|
|
def test_toggle_member_status_leader_only(self, client, auth_headers, app):
|
||
|
|
"""测试只有团队长可以启用/禁用成员"""
|
||
|
|
with app.app_context():
|
||
|
|
from app.models import User
|
||
|
|
from werkzeug.security import generate_password_hash
|
||
|
|
from app import db
|
||
|
|
|
||
|
|
# 创建团队成员
|
||
|
|
member_user = User(
|
||
|
|
username='teammember2',
|
||
|
|
email='teammember2@test.com',
|
||
|
|
password_hash=generate_password_hash('member123'),
|
||
|
|
is_admin=False,
|
||
|
|
is_team_leader=False,
|
||
|
|
is_active=True,
|
||
|
|
created_at=datetime.utcnow()
|
||
|
|
)
|
||
|
|
db.session.add(member_user)
|
||
|
|
db.session.commit()
|
||
|
|
member_id = member_user.id
|
||
|
|
|
||
|
|
# 添加团队成员关系
|
||
|
|
leader = User.query.filter_by(username='leader').first()
|
||
|
|
team_member = TeamMember(leader_id=leader.id, member_id=member_id)
|
||
|
|
db.session.add(team_member)
|
||
|
|
db.session.commit()
|
||
|
|
|
||
|
|
# 普通用户尝试禁用成员
|
||
|
|
response = client.put(f'/api/team/members/{member_id}/status',
|
||
|
|
headers=auth_headers['user'],
|
||
|
|
json={'is_active': False})
|
||
|
|
assert response.status_code == 403
|
||
|
|
|
||
|
|
# 团队长禁用成员
|
||
|
|
response = client.put(f'/api/team/members/{member_id}/status',
|
||
|
|
headers=auth_headers['leader'],
|
||
|
|
json={'is_active': False})
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.json['message'] == '成员状态已更新'
|
||
|
|
|
||
|
|
# 验证成员已被禁用
|
||
|
|
member = User.query.get(member_id)
|
||
|
|
assert member.is_active is False
|
||
|
|
|
||
|
|
def test_get_available_categories_leader_only(self, client, auth_headers):
|
||
|
|
"""测试只有团队长可以获取可分配分类"""
|
||
|
|
# 普通用户尝试访问
|
||
|
|
response = client.get('/api/team/available-categories',
|
||
|
|
headers=auth_headers['user'])
|
||
|
|
assert response.status_code == 403
|
||
|
|
|
||
|
|
# 团队长访问
|
||
|
|
response = client.get('/api/team/available-categories',
|
||
|
|
headers=auth_headers['leader'])
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert 'categories' in response.json
|
||
|
|
assert isinstance(response.json['categories'], list)
|
||
|
|
|
||
|
|
|
||
|
|
class TestTeamLeaderValidation:
|
||
|
|
"""团队长功能验证测试"""
|
||
|
|
|
||
|
|
def test_cannot_add_self_as_member(self, client, auth_headers, app):
|
||
|
|
"""测试团队长不能添加自己为成员"""
|
||
|
|
with app.app_context():
|
||
|
|
from app.models import User
|
||
|
|
leader = User.query.filter_by(username='leader').first()
|
||
|
|
leader_id = leader.id
|
||
|
|
|
||
|
|
# 尝试添加自己为成员
|
||
|
|
response = client.post('/api/admin/team-members',
|
||
|
|
headers=auth_headers['admin'],
|
||
|
|
json={'leader_id': leader_id, 'member_id': leader_id})
|
||
|
|
assert response.status_code == 400
|
||
|
|
assert '不能添加自己为团队成员' in response.json['message']
|
||
|
|
|
||
|
|
def test_cannot_add_admin_as_member(self, client, auth_headers, app):
|
||
|
|
"""测试不能添加管理员为成员"""
|
||
|
|
with app.app_context():
|
||
|
|
from app.models import User
|
||
|
|
admin = User.query.filter_by(username='admin').first()
|
||
|
|
admin_id = admin.id
|
||
|
|
|
||
|
|
leader = User.query.filter_by(username='leader').first()
|
||
|
|
leader_id = leader.id
|
||
|
|
|
||
|
|
# 尝试添加管理员为成员
|
||
|
|
response = client.post('/api/admin/team-members',
|
||
|
|
headers=auth_headers['admin'],
|
||
|
|
json={'leader_id': leader_id, 'member_id': admin_id})
|
||
|
|
assert response.status_code == 400
|
||
|
|
assert '不能添加管理员为团队成员' in response.json['message']
|
||
|
|
|
||
|
|
def test_cannot_add_existing_member(self, client, auth_headers, app):
|
||
|
|
"""测试不能重复添加团队成员"""
|
||
|
|
with app.app_context():
|
||
|
|
from app.models import User
|
||
|
|
from werkzeug.security import generate_password_hash
|
||
|
|
from app import db
|
||
|
|
|
||
|
|
# 创建团队成员
|
||
|
|
member_user = User(
|
||
|
|
username='duplicatemember',
|
||
|
|
email='duplicate@test.com',
|
||
|
|
password_hash=generate_password_hash('member123'),
|
||
|
|
is_admin=False,
|
||
|
|
is_team_leader=False,
|
||
|
|
is_active=True,
|
||
|
|
created_at=datetime.utcnow()
|
||
|
|
)
|
||
|
|
db.session.add(member_user)
|
||
|
|
db.session.commit()
|
||
|
|
member_id = member_user.id
|
||
|
|
|
||
|
|
# 获取团队长ID
|
||
|
|
leader = User.query.filter_by(username='leader').first()
|
||
|
|
leader_id = leader.id
|
||
|
|
|
||
|
|
# 第一次添加成功
|
||
|
|
response = client.post('/api/admin/team-members',
|
||
|
|
headers=auth_headers['admin'],
|
||
|
|
json={'leader_id': leader_id, 'member_id': member_id})
|
||
|
|
assert response.status_code == 201
|
||
|
|
|
||
|
|
# 第二次添加应该失败
|
||
|
|
response = client.post('/api/admin/team-members',
|
||
|
|
headers=auth_headers['admin'],
|
||
|
|
json={'leader_id': leader_id, 'member_id': member_id})
|
||
|
|
assert response.status_code == 400
|
||
|
|
assert '该用户已经是团队成员' in response.json['message']
|
||
|
|
|
||
|
|
def test_cannot_update_non_team_member_quota(self, client, auth_headers, app):
|
||
|
|
"""测试不能更新非团队成员的配额"""
|
||
|
|
with app.app_context():
|
||
|
|
from app.models import User
|
||
|
|
from werkzeug.security import generate_password_hash
|
||
|
|
from app import db
|
||
|
|
|
||
|
|
# 创建非团队成员用户
|
||
|
|
non_member = User(
|
||
|
|
username='nonmember',
|
||
|
|
email='nonmember@test.com',
|
||
|
|
password_hash=generate_password_hash('member123'),
|
||
|
|
is_admin=False,
|
||
|
|
is_team_leader=False,
|
||
|
|
is_active=True,
|
||
|
|
created_at=datetime.utcnow()
|
||
|
|
)
|
||
|
|
db.session.add(non_member)
|
||
|
|
db.session.commit()
|
||
|
|
non_member_id = non_member.id
|
||
|
|
|
||
|
|
# 团队长尝试更新非成员配额
|
||
|
|
response = client.put(f'/api/team/members/{non_member_id}/quota',
|
||
|
|
headers=auth_headers['leader'],
|
||
|
|
json={'daily_download_limit': 10})
|
||
|
|
assert response.status_code == 404
|
||
|
|
assert '该用户不是您的团队成员' in response.json['message']
|