237899745/zcglxt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e71181f0a3069cb59cf3846e062a8c5544eb1374
zcglxt/tests/services/test_asset_service.py
Claude e71181f0a3 fix: 修复多个关键问题 
- 修复前端路由守卫：未登录时不显示提示，直接跳转登录页
- 修复API拦截器：401错误不显示提示，直接跳转
- 增强验证码显示：图片尺寸从120x40增加到200x80
- 增大验证码字体：从28号增加到48号
- 优化验证码字符：排除易混淆的0和1
- 减少干扰线：从5条减少到3条，添加背景色优化
- 增强登录API日志：添加详细的调试日志
- 增强验证码生成和验证日志
- 优化异常处理和错误追踪

影响文件：
- src/router/index.ts
- src/api/request.ts
- app/services/auth_service.py
- app/api/v1/auth.py
- app/schemas/user.py

测试状态：
- 前端构建通过
- 后端语法检查通过
- 验证码显示效果优化完成

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-25 00:26:21 +08:00

475 lines
16 KiB
Python
Raw Blame History

"""
资产管理服务层测试
测试内容:
- 资产创建业务逻辑
- 资产分配业务逻辑
- 状态机转换
- 权限验证
- 异常处理
"""
import pytest
from datetime import datetime
# from app.services.asset_service import AssetService
# from app.services.state_machine_service import StateMachineService
# from app.core.exceptions import BusinessException, NotFoundException
# class TestAssetService:
# """测试资产管理服务"""
#
# @pytest.fixture
# def asset_service(self):
# """创建AssetService实例"""
# return AssetService()
#
# def test_create_asset_generates_code(self, db, asset_service):
# """测试创建资产时自动生成编码"""
# asset_data = {
# "asset_name": "测试资产",
# "device_type_id": 1,
# "organization_id": 1
# }
#
# asset = asset_service.create_asset(
# db=db,
# asset_in=AssetCreate(**asset_data),
# creator_id=1
# )
#
# assert asset.asset_code is not None
# assert asset.asset_code.startswith("ASSET-")
# assert len(asset.asset_code) == 19 # ASSET-YYYYMMDD-XXXX
#
# def test_create_asset_records_status_history(self, db, asset_service):
# """测试创建资产时记录状态历史"""
# asset_data = AssetCreate(
# asset_name="测试资产",
# device_type_id=1,
# organization_id=1
# )
#
# asset = asset_service.create_asset(
# db=db,
# asset_in=asset_data,
# creator_id=1
# )
#
# # 验证状态历史
# history = db.query(AssetStatusHistory).filter(
# AssetStatusHistory.asset_id == asset.id
# ).all()
#
# assert len(history) == 1
# assert history[0].old_status is None
# assert history[0].new_status == "pending"
# assert history[0].operation_type == "create"
#
# def test_create_asset_with_invalid_device_type(self, db, asset_service):
# """测试使用无效设备类型创建资产"""
# asset_data = AssetCreate(
# asset_name="测试资产",
# device_type_id=999999, # 不存在的设备类型
# organization_id=1
# )
#
# with pytest.raises(NotFoundException):
# asset_service.create_asset(
# db=db,
# asset_in=asset_data,
# creator_id=1
# )
#
# def test_create_asset_with_invalid_organization(self, db, asset_service):
# """测试使用无效网点创建资产"""
# asset_data = AssetCreate(
# asset_name="测试资产",
# device_type_id=1,
# organization_id=999999 # 不存在的网点
# )
#
# with pytest.raises(NotFoundException):
# asset_service.create_asset(
# db=db,
# asset_in=asset_data,
# creator_id=1
# )
#
# def test_create_asset_validates_required_dynamic_fields(self, db, asset_service):
# """测试验证必填的动态字段"""
# # 假设计算机类型要求CPU和内存必填
# asset_data = AssetCreate(
# asset_name="测试计算机",
# device_type_id=1, # 计算机类型
# organization_id=1,
# dynamic_attributes={
# # 缺少必填的cpu和memory字段
# }
# )
#
# with pytest.raises(BusinessException):
# asset_service.create_asset(
# db=db,
# asset_in=asset_data,
# creator_id=1
# )
# class TestAssetAllocation:
# """测试资产分配"""
#
# @pytest.fixture
# def asset_service(self):
# return AssetService()
#
# def test_allocate_assets_success(self, db, asset_service, test_asset):
# """测试成功分配资产"""
# allocation_order = asset_service.allocate_assets(
# db=db,
# asset_ids=[test_asset.id],
# organization_id=2,
# operator_id=1
# )
#
# assert allocation_order is not None
# assert allocation_order.order_type == "allocation"
# assert allocation_order.asset_count == 1
#
# # 验证资产状态未改变(等待审批)
# db.refresh(test_asset)
# assert test_asset.status == "in_stock"
#
# def test_allocate_assets_invalid_status(self, db, asset_service):
# """测试分配状态不正确的资产"""
# # 创建一个使用中的资产
# in_use_asset = Asset(
# asset_code="ASSET-20250124-0002",
# asset_name="使用中的资产",
# device_type_id=1,
# organization_id=1,
# status="in_use"
# )
# db.add(in_use_asset)
# db.commit()
#
# with pytest.raises(BusinessException) as exc:
# asset_service.allocate_assets(
# db=db,
# asset_ids=[in_use_asset.id],
# organization_id=2,
# operator_id=1
# )
#
# assert "当前状态不允许分配" in str(exc.value)
#
# def test_allocate_assets_batch(self, db, asset_service):
# """测试批量分配资产"""
# # 创建多个资产
# assets = []
# for i in range(5):
# asset = Asset(
# asset_code=f"ASSET-20250124-{i:04d}",
# asset_name=f"测试资产{i}",
# device_type_id=1,
# organization_id=1,
# status="in_stock"
# )
# db.add(asset)
# assets.append(asset)
# db.commit()
#
# asset_ids = [a.id for a in assets]
#
# allocation_order = asset_service.allocate_assets(
# db=db,
# asset_ids=asset_ids,
# organization_id=2,
# operator_id=1
# )
#
# assert allocation_order.asset_count == 5
#
# def test_allocate_assets_to_same_organization(self, db, asset_service, test_asset):
# """测试分配到当前所在网点"""
# with pytest.raises(BusinessException):
# asset_service.allocate_assets(
# db=db,
# asset_ids=[test_asset.id],
# organization_id=test_asset.organization_id, # 相同网点
# operator_id=1
# )
#
# def test_allocate_duplicate_assets(self, db, asset_service):
# """测试分配时包含重复资产"""
# with pytest.raises(BusinessException):
# asset_service.allocate_assets(
# db=db,
# asset_ids=[1, 1, 2], # 资产ID重复
# organization_id=2,
# operator_id=1
# )
#
# def test_approve_allocation_order(self, db, asset_service):
# """测试审批分配单"""
# # 创建分配单
# allocation_order = asset_service.allocate_assets(
# db=db,
# asset_ids=[1],
# organization_id=2,
# operator_id=1
# )
#
# # 审批通过
# asset_service.approve_allocation_order(
# db=db,
# order_id=allocation_order.id,
# approval_status="approved",
# approver_id=2,
# remark="同意"
# )
#
# # 验证资产状态已更新
# asset = db.query(Asset).filter(Asset.id == 1).first()
# assert asset.status == "in_use"
#
# # 验证分配单执行状态
# db.refresh(allocation_order)
# assert allocation_order.approval_status == "approved"
# assert allocation_order.execute_status == "completed"
#
# def test_reject_allocation_order(self, db, asset_service):
# """测试拒绝分配单"""
# allocation_order = asset_service.allocate_assets(
# db=db,
# asset_ids=[1],
# organization_id=2,
# operator_id=1
# )
#
# # 审批拒绝
# asset_service.approve_allocation_order(
# db=db,
# order_id=allocation_order.id,
# approval_status="rejected",
# approver_id=2,
# remark="不符合条件"
# )
#
# # 验证资产状态未改变
# asset = db.query(Asset).filter(Asset.id == 1).first()
# assert asset.status == "in_stock"
#
# db.refresh(allocation_order)
# assert allocation_order.approval_status == "rejected"
# assert allocation_order.execute_status == "cancelled"
# class TestStateMachine:
# """测试状态机"""
#
# @pytest.fixture
# def state_machine(self):
# return StateMachineService()
#
# def test_valid_state_transitions(self, state_machine):
# """测试有效的状态转换"""
# valid_transitions = [
# ("pending", "in_stock"),
# ("in_stock", "in_use"),
# ("in_stock", "maintenance"),
# ("in_use", "transferring"),
# ("in_use", "maintenance"),
# ("maintenance", "in_stock"),
# ("transferring", "in_use"),
# ("in_use", "pending_scrap"),
# ("pending_scrap", "scrapped"),
# ]
#
# for old_status, new_status in valid_transitions:
# assert state_machine.can_transition(old_status, new_status) is True
#
# def test_invalid_state_transitions(self, state_machine):
# """测试无效的状态转换"""
# invalid_transitions = [
# ("pending", "in_use"), # pending不能直接到in_use
# ("in_stock", "pending"), # 不能回退到pending
# ("scrapped", "in_stock"), # 报废后不能恢复
# ("in_use", "pending_scrap"), # 应该先transferring
# ]
#
# for old_status, new_status in invalid_transitions:
# assert state_machine.can_transition(old_status, new_status) is False
#
# def test_record_state_change(self, db, state_machine, test_asset):
# """测试记录状态变更"""
# state_machine.record_state_change(
# db=db,
# asset_id=test_asset.id,
# old_status="in_stock",
# new_status="in_use",
# operator_id=1,
# operation_type="allocate",
# remark="资产分配"
# )
#
# history = db.query(AssetStatusHistory).filter(
# AssetStatusHistory.asset_id == test_asset.id
# ).first()
#
# assert history is not None
# assert history.old_status == "in_stock"
# assert history.new_status == "in_use"
# assert history.operation_type == "allocate"
# assert history.remark == "资产分配"
#
# def test_get_available_transitions(self, state_machine):
# """测试获取可用的状态转换"""
# transitions = state_machine.get_available_transitions("in_stock")
#
# assert "in_use" in transitions
# assert "maintenance" in transitions
# assert "pending_scrap" not in transitions
#
# def test_state_transition_with_invalid_permission(self, db, state_machine, test_asset):
# """测试无权限的状态转换"""
# # 普通用户不能直接报废资产
# with pytest.raises(PermissionDeniedException):
# state_machine.transition_state(
# db=db,
# asset_id=test_asset.id,
# new_status="scrapped",
# operator_id=999, # 无权限的用户
# operation_type="scrap"
# )
# class TestAssetStatistics:
# """测试资产统计"""
#
# @pytest.fixture
# def asset_service(self):
# return AssetService()
#
# def test_get_asset_overview(self, db, asset_service):
# """测试获取资产概览统计"""
# # 创建测试数据
# # ... 创建不同状态的资产
#
# stats = asset_service.get_asset_overview(db)
#
# assert stats["total_assets"] > 0
# assert stats["total_value"] > 0
# assert "assets_in_stock" in stats
# assert "assets_in_use" in stats
# assert "assets_maintenance" in stats
# assert "assets_scrapped" in stats
#
# def test_get_organization_distribution(self, db, asset_service):
# """测试获取网点分布统计"""
# distribution = asset_service.get_organization_distribution(db)
#
# assert isinstance(distribution, list)
# if len(distribution) > 0:
# assert "org_name" in distribution[0]
# assert "count" in distribution[0]
# assert "value" in distribution[0]
#
# def test_get_device_type_distribution(self, db, asset_service):
# """测试获取设备类型分布统计"""
# distribution = asset_service.get_device_type_distribution(db)
#
# assert isinstance(distribution, list)
# if len(distribution) > 0:
# assert "type_name" in distribution[0]
# assert "count" in distribution[0]
#
# def test_get_value_trend(self, db, asset_service):
# """测试获取价值趋势"""
# trend = asset_service.get_value_trend(
# db=db,
# start_date="2024-01-01",
# end_date="2024-12-31",
# group_by="month"
# )
#
# assert isinstance(trend, list)
# if len(trend) > 0:
# assert "date" in trend[0]
# assert "count" in trend[0]
# assert "value" in trend[0]
# 性能测试
# class TestAssetServicePerformance:
# """测试资产管理服务性能"""
#
# @pytest.fixture
# def asset_service(self):
# return AssetService()
#
# @pytest.mark.slow
# def test_large_asset_list_query_performance(self, db, asset_service):
# """测试大量资产查询性能"""
# # 创建1000个资产
# assets = []
# for i in range(1000):
# asset = Asset(
# asset_code=f"ASSET-20250124-{i:04d}",
# asset_name=f"测试资产{i}",
# device_type_id=1,
# organization_id=1,
# status="in_stock"
# )
# assets.append(asset)
# db.bulk_save_objects(assets)
# db.commit()
#
# import time
# start_time = time.time()
#
# items, total = asset_service.get_assets(
# db=db,
# skip=0,
# limit=20
# )
#
# elapsed_time = time.time() - start_time
#
# assert len(items) == 20
# assert total == 1000
# assert elapsed_time < 0.5 # 查询应该在500ms内完成
#
# @pytest.mark.slow
# def test_batch_allocation_performance(self, db, asset_service):
# """测试批量分配性能"""
# # 创建100个资产
# asset_ids = []
# for i in range(100):
# asset = Asset(
# asset_code=f"ASSET-20250124-{i:04d}",
# asset_name=f"测试资产{i}",
# device_type_id=1,
# organization_id=1,
# status="in_stock"
# )
# db.add(asset)
# db.flush()
# asset_ids.append(asset.id)
# db.commit()
#
# import time
# start_time = time.time()
#
# allocation_order = asset_service.allocate_assets(
# db=db,
# asset_ids=asset_ids,
# organization_id=2,
# operator_id=1
# )
#
# elapsed_time = time.time() - start_time
#
# assert allocation_order.asset_count == 100
# assert elapsed_time < 2.0 # 批量分配应该在2秒内完成
Reference in New Issue View Git Blame Copy Permalink