913 lines
31 KiB
Python
913 lines
31 KiB
Python
"""
|
|
统计分析 API 测试
|
|
|
|
测试范围:
|
|
- 资产统计测试 (20+用例)
|
|
- 分布统计测试 (15+用例)
|
|
- 趋势统计测试 (10+用例)
|
|
- 缓存测试 (10+用例)
|
|
- 导出测试 (5+用例)
|
|
|
|
总计: 60+ 用例
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.asset import Asset
|
|
from app.models.organization import Organization
|
|
from app.models.maintenance import Maintenance
|
|
|
|
|
|
# ================================
|
|
# Fixtures
|
|
# ================================
|
|
|
|
@pytest.fixture
|
|
def test_assets_for_statistics(db: Session) -> list:
|
|
"""创建用于统计的测试资产"""
|
|
assets = []
|
|
|
|
# 不同状态的资产
|
|
statuses = ["in_stock", "in_use", "maintenance", "scrapped"]
|
|
for i, status in enumerate(statuses):
|
|
for j in range(3):
|
|
asset = Asset(
|
|
asset_code=f"STAT-{status[:3].upper()}-{j+1:03d}",
|
|
asset_name=f"统计测试资产{i}-{j}",
|
|
device_type_id=1,
|
|
organization_id=1,
|
|
status=status,
|
|
purchase_price=Decimal(str(10000 * (i + 1))),
|
|
purchase_date=datetime.now() - timedelta(days=30 * (i + 1))
|
|
)
|
|
db.add(asset)
|
|
assets.append(asset)
|
|
|
|
db.commit()
|
|
for asset in assets:
|
|
db.refresh(asset)
|
|
return assets
|
|
|
|
|
|
@pytest.fixture
|
|
def test_orgs_for_statistics(db: Session) -> list:
|
|
"""创建用于统计的测试组织"""
|
|
orgs = []
|
|
for i in range(3):
|
|
org = Organization(
|
|
org_code=f"STAT-ORG-{i+1:03d}",
|
|
org_name=f"统计测试组织{i+1}",
|
|
org_type="department",
|
|
status="active"
|
|
)
|
|
db.add(org)
|
|
orgs.append(org)
|
|
db.commit()
|
|
for org in orgs:
|
|
db.refresh(org)
|
|
return orgs
|
|
|
|
|
|
# ================================
|
|
# 资产统计测试 (20+用例)
|
|
# ================================
|
|
|
|
class TestAssetStatistics:
|
|
"""资产统计测试"""
|
|
|
|
def test_get_total_asset_count(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产总数"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_count" in data
|
|
assert data["total_count"] >= len(test_assets_for_statistics)
|
|
|
|
def test_get_asset_count_by_status(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试按状态统计资产数量"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/by-status",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) > 0
|
|
assert all("status" in item and "count" in item for item in data)
|
|
|
|
def test_get_asset_count_by_type(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试按类型统计资产数量"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/by-type",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert all("device_type" in item and "count" in item for item in data)
|
|
|
|
def test_get_asset_count_by_organization(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试按组织统计资产数量"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/by-organization",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_total_asset_value(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产总价值"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/total-value",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_value" in data
|
|
assert isinstance(data["total_value"], (int, float, str))
|
|
|
|
def test_get_asset_value_by_status(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试按状态统计资产价值"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/value-by-status",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert all("status" in item and "total_value" in item for item in data)
|
|
|
|
def test_get_asset_value_by_type(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试按类型统计资产价值"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/value-by-type",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_asset_purchase_statistics(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产采购统计"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/purchase-statistics",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_purchase_count" in data
|
|
assert "total_purchase_value" in data
|
|
|
|
def test_get_asset_purchase_by_month(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试按月统计资产采购"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/purchase-by-month",
|
|
params={"year": 2025},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_asset_depreciation_summary(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产折旧汇总"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/depreciation-summary",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_depreciation" in data
|
|
|
|
def test_get_asset_age_distribution(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产使用年限分布"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/age-distribution",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert all("age_range" in item and "count" in item for item in data)
|
|
|
|
def test_get_new_asset_statistics(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取新增资产统计"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/new-assets",
|
|
params={"days": 30},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "count" in data
|
|
assert "total_value" in data
|
|
|
|
def test_get_scrapped_asset_statistics(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取报废资产统计"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/scrapped-assets",
|
|
params={"start_date": "2025-01-01", "end_date": "2025-12-31"},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "count" in data
|
|
|
|
def test_get_asset_utilization_rate(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产利用率"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/utilization-rate",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "utilization_rate" in data
|
|
assert "in_use_count" in data
|
|
assert "total_count" in data
|
|
|
|
def test_get_asset_maintenance_rate(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产维修率"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/maintenance-rate",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "maintenance_rate" in data
|
|
|
|
def test_get_asset_summary_dashboard(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产汇总仪表盘数据"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/summary-dashboard",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_assets" in data
|
|
assert "total_value" in data
|
|
assert "utilization_rate" in data
|
|
assert "maintenance_rate" in data
|
|
|
|
def test_search_statistics(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试搜索统计"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/search",
|
|
params={"keyword": "统计"},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "count" in data
|
|
|
|
def test_get_asset_top_list_by_value(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取价值最高的资产列表"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/top-by-value",
|
|
params={"limit": 10},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_statistics_by_custom_field(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试按自定义字段统计"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/by-custom-field",
|
|
params={"field_name": "manufacturer"},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code in [200, 400] # 可能不支持该字段
|
|
|
|
def test_get_multi_dimension_statistics(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试多维度统计"""
|
|
response = client.post(
|
|
"/api/v1/statistics/assets/multi-dimension",
|
|
json={
|
|
"dimensions": ["status", "device_type"],
|
|
"metrics": ["count", "total_value"]
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "data" in data
|
|
|
|
|
|
# ================================
|
|
# 分布统计测试 (15+用例)
|
|
# ================================
|
|
|
|
class TestDistributionStatistics:
|
|
"""分布统计测试"""
|
|
|
|
def test_get_geographic_distribution(self, client, auth_headers, test_orgs_for_statistics):
|
|
"""测试获取地理分布统计"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/geographic",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_organization_hierarchy_distribution(self, client, auth_headers, test_orgs_for_statistics):
|
|
"""测试获取组织层级分布"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/organization-hierarchy",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_department_distribution(self, client, auth_headers, test_orgs_for_statistics):
|
|
"""测试获取部门分布"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/by-department",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_asset_category_distribution(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产类别分布"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/by-category",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_asset_value_distribution(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产价值分布"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/value-ranges",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert all("range" in item and "count" in item for item in data)
|
|
|
|
def test_get_asset_location_distribution(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产位置分布"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/by-location",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_asset_brand_distribution(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产品牌分布"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/by-brand",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_asset_supplier_distribution(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产供应商分布"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/by-supplier",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_get_asset_status_distribution_pie_chart(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产状态分布饼图数据"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/status-pie-chart",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "labels" in data
|
|
assert "data" in data
|
|
assert isinstance(data["labels"], list)
|
|
assert isinstance(data["data"], list)
|
|
|
|
def test_get_organization_asset_tree(self, client, auth_headers, test_orgs_for_statistics):
|
|
"""测试获取组织资产树"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/org-asset-tree",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "tree" in data
|
|
|
|
def test_get_cross_tabulation(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试交叉统计表"""
|
|
response = client.post(
|
|
"/api/v1/statistics/distribution/cross-tabulation",
|
|
json={
|
|
"row_field": "status",
|
|
"column_field": "device_type_id"
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "rows" in data
|
|
assert "columns" in data
|
|
assert "data" in data
|
|
|
|
def test_get_distribution_heatmap_data(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取分布热力图数据"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/heatmap",
|
|
params={"dimension": "organization_asset"},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "heatmap_data" in data
|
|
|
|
def test_get_asset_concentration_index(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产集中度指数"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/concentration-index",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "index" in data
|
|
|
|
def test_get_distribution_comparison(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试分布对比分析"""
|
|
response = client.post(
|
|
"/api/v1/statistics/distribution/comparison",
|
|
json={
|
|
"dimension": "status",
|
|
"period1": {"start": "2025-01-01", "end": "2025-06-30"},
|
|
"period2": {"start": "2024-01-01", "end": "2024-06-30"}
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "period1" in data
|
|
assert "period2" in data
|
|
|
|
def test_get_distribution_trend(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试分布趋势"""
|
|
response = client.get(
|
|
"/api/v1/statistics/distribution/trend",
|
|
params={"dimension": "status", "months": 12},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "trend_data" in data
|
|
|
|
|
|
# ================================
|
|
# 趋势统计测试 (10+用例)
|
|
# ================================
|
|
|
|
class TestTrendStatistics:
|
|
"""趋势统计测试"""
|
|
|
|
def test_get_asset_growth_trend(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取资产增长趋势"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/asset-growth",
|
|
params={"period": "monthly", "months": 12},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "trend" in data
|
|
assert isinstance(data["trend"], list)
|
|
|
|
def test_get_value_change_trend(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取价值变化趋势"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/value-change",
|
|
params={"period": "monthly", "months": 12},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "trend" in data
|
|
|
|
def test_get_utilization_trend(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取利用率趋势"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/utilization",
|
|
params={"period": "weekly", "weeks": 12},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "trend" in data
|
|
|
|
def test_get_maintenance_cost_trend(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取维修费用趋势"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/maintenance-cost",
|
|
params={"period": "monthly", "months": 12},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "trend" in data
|
|
|
|
def test_get_allocation_trend(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取分配趋势"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/allocation",
|
|
params={"period": "monthly", "months": 12},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_get_transfer_trend(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取调拨趋势"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/transfer",
|
|
params={"period": "monthly", "months": 12},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_get_scrap_rate_trend(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取报废率趋势"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/scrap-rate",
|
|
params={"period": "monthly", "months": 12},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "trend" in data
|
|
|
|
def test_get_forecast_data(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取预测数据"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/forecast",
|
|
params={
|
|
"metric": "asset_count",
|
|
"method": "linear_regression",
|
|
"forecast_periods": 6
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "forecast" in data
|
|
assert "confidence_interval" in data
|
|
|
|
def test_get_year_over_year_comparison(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取同比数据"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/year-over-year",
|
|
params={"metric": "total_value"},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "current_year" in data
|
|
assert "previous_year" in data
|
|
assert "growth_rate" in data
|
|
|
|
def test_get_moving_average(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试获取移动平均"""
|
|
response = client.get(
|
|
"/api/v1/statistics/trends/moving-average",
|
|
params={"metric": "asset_count", "window": 7},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "moving_average" in data
|
|
|
|
|
|
# ================================
|
|
# 缓存测试 (10+用例)
|
|
# ================================
|
|
|
|
class TestStatisticsCache:
|
|
"""统计缓存测试"""
|
|
|
|
def test_cache_is_working(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试缓存是否生效"""
|
|
# 第一次请求
|
|
response1 = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
assert response1.status_code == 200
|
|
|
|
# 第二次请求应该从缓存读取
|
|
response2 = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
assert response2.status_code == 200
|
|
|
|
def test_cache_key_generation(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试缓存键生成"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/by-status",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_cache_invalidation_on_asset_change(self, client, auth_headers, db: Session, test_assets_for_statistics):
|
|
"""测试资产变更时缓存失效"""
|
|
# 获取初始统计
|
|
response1 = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
count1 = response1.json()["total_count"]
|
|
|
|
# 创建新资产
|
|
new_asset = Asset(
|
|
asset_code="CACHE-TEST-001",
|
|
asset_name="缓存测试资产",
|
|
device_type_id=1,
|
|
organization_id=1,
|
|
status="in_stock"
|
|
)
|
|
db.add(new_asset)
|
|
db.commit()
|
|
|
|
# 再次获取统计
|
|
response2 = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
count2 = response2.json()["total_count"]
|
|
|
|
# 验证缓存已更新
|
|
assert count2 == count1 + 1
|
|
|
|
def test_cache_expiration(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试缓存过期"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_clear_cache(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试清除缓存"""
|
|
response = client.post(
|
|
"/api/v1/statistics/cache/clear",
|
|
json={"cache_keys": ["assets:total-count"]},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_cache_statistics(self, client, auth_headers):
|
|
"""测试获取缓存统计"""
|
|
response = client.get(
|
|
"/api/v1/statistics/cache/stats",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "hit_count" in data
|
|
assert "miss_count" in data
|
|
|
|
def test_warm_up_cache(self, client, auth_headers):
|
|
"""测试缓存预热"""
|
|
response = client.post(
|
|
"/api/v1/statistics/cache/warm-up",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "warmed_up_keys" in data
|
|
|
|
def test_cache_with_different_parameters(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试不同参数使用不同缓存"""
|
|
response1 = client.get(
|
|
"/api/v1/statistics/assets/purchase-by-month?year=2024",
|
|
headers=auth_headers
|
|
)
|
|
response2 = client.get(
|
|
"/api/v1/statistics/assets/purchase-by-month?year=2025",
|
|
headers=auth_headers
|
|
)
|
|
assert response1.status_code == 200
|
|
assert response2.status_code == 200
|
|
|
|
def test_distributed_cache_consistency(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试分布式缓存一致性"""
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
def test_cache_performance(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试缓存性能"""
|
|
import time
|
|
|
|
# 未缓存请求
|
|
start = time.time()
|
|
response1 = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
uncached_time = time.time() - start
|
|
|
|
# 缓存请求
|
|
start = time.time()
|
|
response2 = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
cached_time = time.time() - start
|
|
|
|
# 缓存请求应该更快
|
|
# 注意: 这个断言可能因为网络延迟等因素不稳定
|
|
# assert cached_time < uncached_time
|
|
|
|
|
|
# ================================
|
|
# 导出测试 (5+用例)
|
|
# ================================
|
|
|
|
class TestStatisticsExport:
|
|
"""统计导出测试"""
|
|
|
|
def test_export_statistics_to_excel(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试导出统计数据到Excel"""
|
|
response = client.post(
|
|
"/api/v1/statistics/export/excel",
|
|
json={
|
|
"report_type": "asset_summary",
|
|
"filters": {"status": "in_use"},
|
|
"columns": ["asset_code", "asset_name", "purchase_price"]
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "download_url" in data
|
|
|
|
def test_export_statistics_to_pdf(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试导出统计数据到PDF"""
|
|
response = client.post(
|
|
"/api/v1/statistics/export/pdf",
|
|
json={
|
|
"report_type": "asset_distribution",
|
|
"include_charts": True
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "download_url" in data
|
|
|
|
def test_export_statistics_to_csv(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试导出统计数据到CSV"""
|
|
response = client.post(
|
|
"/api/v1/statistics/export/csv",
|
|
json={
|
|
"query": "assets_by_status",
|
|
"parameters": {}
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code in [200, 202] # 可能异步处理
|
|
|
|
def test_scheduled_export(self, client, auth_headers):
|
|
"""测试定时导出"""
|
|
response = client.post(
|
|
"/api/v1/statistics/export/schedule",
|
|
json={
|
|
"report_type": "monthly_report",
|
|
"schedule": "0 0 1 * *", # 每月1号
|
|
"recipients": ["admin@example.com"],
|
|
"format": "excel"
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "schedule_id" in data
|
|
|
|
def test_get_export_history(self, client, auth_headers):
|
|
"""测试获取导出历史"""
|
|
response = client.get(
|
|
"/api/v1/statistics/export/history",
|
|
headers=auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
|
|
# ================================
|
|
# 测试标记
|
|
# ================================
|
|
|
|
@pytest.mark.unit
|
|
class TestStatisticsUnit:
|
|
"""单元测试标记"""
|
|
|
|
def test_calculation_accuracy(self):
|
|
"""测试计算准确性"""
|
|
pass
|
|
|
|
def test_rounding_rules(self):
|
|
"""测试舍入规则"""
|
|
pass
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestStatisticsIntegration:
|
|
"""集成测试标记"""
|
|
|
|
def test_full_statistics_workflow(self, client, auth_headers, test_assets_for_statistics):
|
|
"""测试完整统计流程"""
|
|
# 1. 获取基础统计
|
|
response1 = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
assert response1.status_code == 200
|
|
|
|
# 2. 获取详细统计
|
|
response2 = client.get(
|
|
"/api/v1/statistics/assets/by-status",
|
|
headers=auth_headers
|
|
)
|
|
assert response2.status_code == 200
|
|
|
|
# 3. 导出报告
|
|
response3 = client.post(
|
|
"/api/v1/statistics/export/excel",
|
|
json={"report_type": "asset_summary"},
|
|
headers=auth_headers
|
|
)
|
|
assert response3.status_code == 200
|
|
|
|
|
|
@pytest.mark.slow
|
|
class TestStatisticsSlowTests:
|
|
"""慢速测试标记"""
|
|
|
|
def test_large_dataset_statistics(self, client, auth_headers):
|
|
"""测试大数据集统计"""
|
|
pass
|
|
|
|
|
|
@pytest.mark.smoke
|
|
class TestStatisticsSmoke:
|
|
"""冒烟测试标记"""
|
|
|
|
def test_basic_statistics_endpoints(self, client, auth_headers):
|
|
"""冒烟测试: 基础统计接口"""
|
|
endpoints = [
|
|
"/api/v1/statistics/assets/total-count",
|
|
"/api/v1/statistics/assets/by-status",
|
|
"/api/v1/statistics/assets/total-value"
|
|
]
|
|
|
|
for endpoint in endpoints:
|
|
response = client.get(endpoint, headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.performance
|
|
class TestStatisticsPerformance:
|
|
"""性能测试标记"""
|
|
|
|
def test_query_response_time(self, client, auth_headers):
|
|
"""测试查询响应时间"""
|
|
import time
|
|
|
|
start = time.time()
|
|
response = client.get(
|
|
"/api/v1/statistics/assets/total-count",
|
|
headers=auth_headers
|
|
)
|
|
elapsed = time.time() - start
|
|
|
|
assert response.status_code == 200
|
|
assert elapsed < 1.0 # 响应时间应小于1秒
|
|
|
|
def test_concurrent_statistics_requests(self, client, auth_headers):
|
|
"""测试并发统计请求"""
|
|
pass
|