460 lines
17 KiB
Python
460 lines
17 KiB
Python
"""
|
|
资产管理模块API测试
|
|
|
|
测试内容:
|
|
- 资产列表查询
|
|
- 资产详情查询
|
|
- 创建资产
|
|
- 更新资产
|
|
- 删除资产
|
|
- 批量导入
|
|
- 扫码查询
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import date
|
|
|
|
|
|
# class TestAssetList:
|
|
# """测试资产列表"""
|
|
#
|
|
# def test_get_assets_success(self, client: TestClient, auth_headers):
|
|
# """测试获取资产列表成功"""
|
|
# response = client.get(
|
|
# "/api/v1/assets",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
# data = response.json()
|
|
# assert data["code"] == 200
|
|
# assert "items" in data["data"]
|
|
# assert "total" in data["data"]
|
|
# assert "page" in data["data"]
|
|
#
|
|
# def test_get_assets_with_pagination(self, client: TestClient, auth_headers):
|
|
# """测试分页查询"""
|
|
# response = client.get(
|
|
# "/api/v1/assets?page=1&page_size=10",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
# data = response.json()
|
|
# assert data["data"]["page"] == 1
|
|
# assert data["data"]["page_size"] == 10
|
|
# assert len(data["data"]["items"]) <= 10
|
|
#
|
|
# def test_get_assets_with_keyword(self, client: TestClient, auth_headers, test_asset):
|
|
# """测试关键词搜索"""
|
|
# response = client.get(
|
|
# f"/api/v1/assets?keyword={test_asset.asset_name}",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
# data = response.json()
|
|
# assert len(data["data"]["items"]) > 0
|
|
#
|
|
# def test_get_assets_with_device_type_filter(self, client: TestClient, auth_headers):
|
|
# """测试按设备类型筛选"""
|
|
# response = client.get(
|
|
# "/api/v1/assets?device_type_id=1",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
#
|
|
# def test_get_assets_with_status_filter(self, client: TestClient, auth_headers):
|
|
# """测试按状态筛选"""
|
|
# response = client.get(
|
|
# "/api/v1/assets?status=in_stock",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
#
|
|
# def test_get_assets_with_organization_filter(self, client: TestClient, auth_headers):
|
|
# """测试按网点筛选"""
|
|
# response = client.get(
|
|
# "/api/v1/assets?organization_id=1",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
#
|
|
# def test_get_assets_with_date_range(self, client: TestClient, auth_headers):
|
|
# """测试按采购日期范围筛选"""
|
|
# response = client.get(
|
|
# "/api/v1/assets?purchase_date_start=2024-01-01&purchase_date_end=2024-12-31",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
#
|
|
# def test_get_assets_with_sorting(self, client: TestClient, auth_headers):
|
|
# """测试排序"""
|
|
# response = client.get(
|
|
# "/api/v1/assets?sort_by=purchase_date&sort_order=desc",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
#
|
|
# def test_get_assets_unauthorized(self, client: TestClient):
|
|
# """测试未授权访问"""
|
|
# response = client.get("/api/v1/assets")
|
|
# assert response.status_code == 401
|
|
#
|
|
# @pytest.mark.parametrize("page,page_size", [
|
|
# (0, 20), # 页码从0开始
|
|
# (1, 0), # 每页0条
|
|
# (-1, 20), # 负页码
|
|
# (1, 1000), # 超大页码
|
|
# ])
|
|
# def test_get_assets_invalid_pagination(self, client: TestClient, auth_headers, page, page_size):
|
|
# """测试无效分页参数"""
|
|
# response = client.get(
|
|
# f"/api/v1/assets?page={page}&page_size={page_size}",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 422
|
|
|
|
|
|
# class TestAssetDetail:
|
|
# """测试资产详情"""
|
|
#
|
|
# def test_get_asset_detail_success(self, client: TestClient, auth_headers, test_asset):
|
|
# """测试获取资产详情成功"""
|
|
# response = client.get(
|
|
# f"/api/v1/assets/{test_asset.id}",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
# data = response.json()
|
|
# assert data["code"] == 200
|
|
# assert data["data"]["id"] == test_asset.id
|
|
# assert data["data"]["asset_code"] == test_asset.asset_code
|
|
# assert "status_history" in data["data"]
|
|
#
|
|
# def test_get_asset_detail_not_found(self, client: TestClient, auth_headers):
|
|
# """测试获取不存在的资产"""
|
|
# response = client.get(
|
|
# "/api/v1/assets/999999",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 404
|
|
# data = response.json()
|
|
# assert data["code"] == 30002 # 资产不存在
|
|
#
|
|
# def test_get_asset_detail_unauthorized(self, client: TestClient, test_asset):
|
|
# """测试未授权访问"""
|
|
# response = client.get(f"/api/v1/assets/{test_asset.id}")
|
|
# assert response.status_code == 401
|
|
|
|
|
|
# class TestCreateAsset:
|
|
# """测试创建资产"""
|
|
#
|
|
# def test_create_asset_success(self, client: TestClient, auth_headers, sample_asset_data):
|
|
# """测试创建资产成功"""
|
|
# response = client.post(
|
|
# "/api/v1/assets",
|
|
# headers=auth_headers,
|
|
# json=sample_asset_data
|
|
# )
|
|
# assert response.status_code == 201
|
|
# data = response.json()
|
|
# assert data["code"] == 200
|
|
# assert "asset_code" in data["data"]
|
|
# assert data["data"]["asset_code"].startswith("ASSET-")
|
|
# assert data["data"]["status"] == "pending"
|
|
#
|
|
# def test_create_asset_without_auth(self, client: TestClient, sample_asset_data):
|
|
# """测试未认证创建"""
|
|
# response = client.post("/api/v1/assets", json=sample_asset_data)
|
|
# assert response.status_code == 401
|
|
#
|
|
# def test_create_asset_missing_required_fields(self, client: TestClient, auth_headers):
|
|
# """测试缺少必填字段"""
|
|
# response = client.post(
|
|
# "/api/v1/assets",
|
|
# headers=auth_headers,
|
|
# json={"asset_name": "测试资产"} # 缺少device_type_id等必填字段
|
|
# )
|
|
# assert response.status_code == 422
|
|
#
|
|
# @pytest.mark.parametrize("field,value,error_msg", [
|
|
# ("asset_name", "", "资产名称不能为空"),
|
|
# ("asset_name", "a" * 201, "资产名称过长"),
|
|
# ("device_type_id", 0, "设备类型ID无效"),
|
|
# ("device_type_id", -1, "设备类型ID无效"),
|
|
# ("purchase_price", -100, "采购价格不能为负数"),
|
|
# ])
|
|
# def test_create_asset_invalid_field(self, client: TestClient, auth_headers, field, value, error_msg):
|
|
# """测试无效字段值"""
|
|
# data = {
|
|
# "asset_name": "测试资产",
|
|
# "device_type_id": 1,
|
|
# "organization_id": 1
|
|
# }
|
|
# data[field] = value
|
|
#
|
|
# response = client.post(
|
|
# "/api/v1/assets",
|
|
# headers=auth_headers,
|
|
# json=data
|
|
# )
|
|
# assert response.status_code in [400, 422]
|
|
#
|
|
# def test_create_asset_duplicate_serial_number(self, client: TestClient, auth_headers, sample_asset_data):
|
|
# """测试序列号重复"""
|
|
# # 第一次创建
|
|
# client.post("/api/v1/assets", headers=auth_headers, json=sample_asset_data)
|
|
#
|
|
# # 第二次使用相同序列号创建
|
|
# response = client.post("/api/v1/assets", headers=auth_headers, json=sample_asset_data)
|
|
# assert response.status_code == 409 # Conflict
|
|
#
|
|
# def test_create_asset_with_dynamic_attributes(self, client: TestClient, auth_headers):
|
|
# """测试带动态字段创建"""
|
|
# data = {
|
|
# "asset_name": "测试资产",
|
|
# "device_type_id": 1,
|
|
# "organization_id": 1,
|
|
# "dynamic_attributes": {
|
|
# "cpu": "Intel i5-10400",
|
|
# "memory": "16GB",
|
|
# "disk": "512GB SSD",
|
|
# "gpu": "GTX 1660Ti"
|
|
# }
|
|
# }
|
|
#
|
|
# response = client.post(
|
|
# "/api/v1/assets",
|
|
# headers=auth_headers,
|
|
# json=data
|
|
# )
|
|
# assert response.status_code == 201
|
|
#
|
|
# def test_create_asset_invalid_device_type(self, client: TestClient, auth_headers, sample_asset_data):
|
|
# """测试无效的设备类型"""
|
|
# sample_asset_data["device_type_id"] = 999999
|
|
# response = client.post(
|
|
# "/api/v1/assets",
|
|
# headers=auth_headers,
|
|
# json=sample_asset_data
|
|
# )
|
|
# assert response.status_code == 400
|
|
#
|
|
# def test_create_asset_invalid_organization(self, client: TestClient, auth_headers, sample_asset_data):
|
|
# """测试无效的网点"""
|
|
# sample_asset_data["organization_id"] = 999999
|
|
# response = client.post(
|
|
# "/api/v1/assets",
|
|
# headers=auth_headers,
|
|
# json=sample_asset_data
|
|
# )
|
|
# assert response.status_code == 400
|
|
|
|
|
|
# class TestUpdateAsset:
|
|
# """测试更新资产"""
|
|
#
|
|
# def test_update_asset_success(self, client: TestClient, auth_headers, test_asset):
|
|
# """测试更新资产成功"""
|
|
# response = client.put(
|
|
# f"/api/v1/assets/{test_asset.id}",
|
|
# headers=auth_headers,
|
|
# json={
|
|
# "asset_name": "更新后的资产名称",
|
|
# "location": "新位置"
|
|
# }
|
|
# )
|
|
# assert response.status_code == 200
|
|
#
|
|
# def test_update_asset_partial_fields(self, client: TestClient, auth_headers, test_asset):
|
|
# """测试部分字段更新"""
|
|
# response = client.put(
|
|
# f"/api/v1/assets/{test_asset.id}",
|
|
# headers=auth_headers,
|
|
# json={"location": "只更新位置"}
|
|
# )
|
|
# assert response.status_code == 200
|
|
#
|
|
# def test_update_asset_not_found(self, client: TestClient, auth_headers):
|
|
# """测试更新不存在的资产"""
|
|
# response = client.put(
|
|
# "/api/v1/assets/999999",
|
|
# headers=auth_headers,
|
|
# json={"asset_name": "新名称"}
|
|
# )
|
|
# assert response.status_code == 404
|
|
#
|
|
# def test_update_asset_status_forbidden(self, client: TestClient, auth_headers, test_asset):
|
|
# """测试禁止直接修改状态"""
|
|
# response = client.put(
|
|
# f"/api/v1/assets/{test_asset.id}",
|
|
# headers=auth_headers,
|
|
# json={"status": "in_use"} # 状态应该通过分配单修改
|
|
# )
|
|
# # 状态字段应该被忽略或返回错误
|
|
# assert response.status_code in [200, 400]
|
|
#
|
|
# def test_update_asset_unauthorized(self, client: TestClient, test_asset):
|
|
# """测试未授权更新"""
|
|
# response = client.put(
|
|
# f"/api/v1/assets/{test_asset.id}",
|
|
# json={"asset_name": "新名称"}
|
|
# )
|
|
# assert response.status_code == 401
|
|
|
|
|
|
# class TestDeleteAsset:
|
|
# """测试删除资产"""
|
|
#
|
|
# def test_delete_asset_success(self, client: TestClient, auth_headers, test_asset):
|
|
# """测试删除资产成功"""
|
|
# response = client.delete(
|
|
# f"/api/v1/assets/{test_asset.id}",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
#
|
|
# # 验证删除
|
|
# get_response = client.get(
|
|
# f"/api/v1/assets/{test_asset.id}",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert get_response.status_code == 404
|
|
#
|
|
# def test_delete_asset_not_found(self, client: TestClient, auth_headers):
|
|
# """测试删除不存在的资产"""
|
|
# response = client.delete(
|
|
# "/api/v1/assets/999999",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 404
|
|
#
|
|
# def test_delete_asset_in_use(self, client: TestClient, auth_headers):
|
|
# """测试删除使用中的资产"""
|
|
# # 创建使用中的资产
|
|
# # ... 创建in_use状态的资产
|
|
#
|
|
# response = client.delete(
|
|
# "/api/v1/assets/1",
|
|
# headers=auth_headers
|
|
# )
|
|
# # 使用中的资产不能删除
|
|
# assert response.status_code == 400
|
|
#
|
|
# def test_delete_asset_without_permission(self, client: TestClient, auth_headers):
|
|
# """测试无权限删除"""
|
|
# # 使用普通用户token而非管理员
|
|
# response = client.delete(
|
|
# "/api/v1/assets/1",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 403
|
|
|
|
|
|
# class TestAssetImport:
|
|
# """测试批量导入资产"""
|
|
#
|
|
# def test_import_assets_success(self, client: TestClient, auth_headers):
|
|
# """测试导入成功"""
|
|
# # 准备测试Excel文件
|
|
# # ... 创建临时Excel文件
|
|
#
|
|
# with open("test_import.xlsx", "rb") as f:
|
|
# files = {"file": ("test_import.xlsx", f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
|
|
# response = client.post(
|
|
# "/api/v1/assets/import",
|
|
# headers=auth_headers,
|
|
# files=files
|
|
# )
|
|
#
|
|
# assert response.status_code == 200
|
|
# data = response.json()
|
|
# assert data["data"]["total"] > 0
|
|
# assert data["data"]["success"] > 0
|
|
#
|
|
# def test_import_assets_partial_failure(self, client: TestClient, auth_headers):
|
|
# """测试部分失败"""
|
|
# # 准备包含错误数据的Excel文件
|
|
#
|
|
# with open("test_import_partial_fail.xlsx", "rb") as f:
|
|
# files = {"file": ("test_import.xlsx", f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
|
|
# response = client.post(
|
|
# "/api/v1/assets/import",
|
|
# headers=auth_headers,
|
|
# files=files
|
|
# )
|
|
#
|
|
# assert response.status_code == 200
|
|
# data = response.json()
|
|
# assert data["data"]["failed"] > 0
|
|
# assert len(data["data"]["errors"]) > 0
|
|
#
|
|
# def test_import_assets_invalid_file_format(self, client: TestClient, auth_headers):
|
|
# """测试无效文件格式"""
|
|
# with open("test.txt", "rb") as f:
|
|
# files = {"file": ("test.txt", f, "text/plain")}
|
|
# response = client.post(
|
|
# "/api/v1/assets/import",
|
|
# headers=auth_headers,
|
|
# files=files
|
|
# )
|
|
#
|
|
# assert response.status_code == 400
|
|
#
|
|
# def test_import_assets_missing_columns(self, client: TestClient, auth_headers):
|
|
# """测试缺少必填列"""
|
|
# # 准备缺少必填列的Excel文件
|
|
#
|
|
# with open("test_missing_columns.xlsx", "rb") as f:
|
|
# files = {"file": ("test.xlsx", f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
|
|
# response = client.post(
|
|
# "/api/v1/assets/import",
|
|
# headers=auth_headers,
|
|
# files=files
|
|
# )
|
|
#
|
|
# assert response.status_code == 400
|
|
|
|
|
|
# class TestAssetScan:
|
|
# """测试扫码查询"""
|
|
#
|
|
# def test_scan_asset_success(self, client: TestClient, auth_headers, test_asset):
|
|
# """测试扫码查询成功"""
|
|
# response = client.get(
|
|
# f"/api/v1/assets/scan/{test_asset.asset_code}",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
# data = response.json()
|
|
# assert data["data"]["asset_code"] == test_asset.asset_code
|
|
#
|
|
# def test_scan_asset_invalid_code(self, client: TestClient, auth_headers):
|
|
# """测试无效的资产编码"""
|
|
# response = client.get(
|
|
# "/api/v1/assets/scan/INVALID-CODE",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 404
|
|
#
|
|
# def test_scan_asset_without_auth(self, client: TestClient, test_asset):
|
|
# """测试未认证扫码"""
|
|
# response = client.get(f"/api/v1/assets/scan/{test_asset.asset_code}")
|
|
# assert response.status_code == 401
|
|
|
|
|
|
# class TestAssetStatistics:
|
|
# """测试资产统计"""
|
|
#
|
|
# def test_get_asset_summary(self, client: TestClient, auth_headers):
|
|
# """测试获取资产汇总"""
|
|
# response = client.get(
|
|
# "/api/v1/assets",
|
|
# headers=auth_headers
|
|
# )
|
|
# assert response.status_code == 200
|
|
# data = response.json()
|
|
# assert "summary" in data["data"]
|
|
# assert "total_count" in data["data"]["summary"]
|
|
# assert "total_value" in data["data"]["summary"]
|
|
# assert "status_distribution" in data["data"]["summary"]
|