AcidMap/tests/integration/test_unit_grouping.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
单元分类功能集成测试

该测试文件用于验证单元分类功能是否正确集成到项目中
"""
import requests
import json
from typing import Dict, Any

def test_unit_grouping_api():
    """
    测试单元分类API接口
    """
    base_url = "http://localhost:8000"
    
    print("=" * 60)
    print("单元分类功能集成测试")
    print("=" * 60)
    
    # 测试1: 获取单元h_xtfx分类结果
    print("\n1. 测试获取单元h_xtfx分类结果")
    print("-" * 40)
    
    try:
        response = requests.get(f"{base_url}/api/unit-grouping/h_xtfx")
        
        if response.status_code == 200:
            data = response.json()
            print(f"✓ 请求成功")
            print(f"✓ 成功状态: {data.get('success', False)}")
            
            if data.get('success', False):
                statistics = data.get('statistics', {})
                print(f"✓ 总单元数: {statistics.get('total_units', 0)}")
                print(f"✓ 有数据的单元数: {statistics.get('units_with_data', 0)}")
                print(f"✓ 无数据的单元数: {statistics.get('units_without_data', 0)}")
                
                category_dist = statistics.get('category_distribution', {})
                print(f"✓ 类别分布:")
                for category, count in category_dist.items():
                    print(f"  - {category}: {count}")
                    
                # 显示前5个单元的结果
                unit_data = data.get('data', {})
                if unit_data:
                    print(f"✓ 前5个单元结果样例:")
                    for i, (unit_id, h_xtfx) in enumerate(list(unit_data.items())[:5]):
                        print(f"  - 单元 {unit_id}: {h_xtfx}")
            else:
                print(f"✗ 服务返回失败: {data.get('error', 'Unknown error')}")
        else:
            print(f"✗ 请求失败，状态码: {response.status_code}")
            print(f"✗ 错误信息: {response.text}")
            
    except requests.exceptions.ConnectionError:
        print("✗ 连接失败，请确保服务正在运行 (uvicorn main:app --reload)")
    except Exception as e:
        print(f"✗ 测试异常: {str(e)}")
    
    # 测试2: 获取统计信息
    print("\n2. 测试获取统计信息")
    print("-" * 40)
    
    try:
        response = requests.get(f"{base_url}/api/unit-grouping/statistics")
        
        if response.status_code == 200:
            data = response.json()
            print(f"✓ 请求成功")
            print(f"✓ 成功状态: {data.get('success', False)}")
            
            if data.get('success', False):
                statistics = data.get('statistics', {})
                print(f"✓ 统计信息获取成功")
                print(f"  - 总单元数: {statistics.get('total_units', 0)}")
                print(f"  - 有数据的单元数: {statistics.get('units_with_data', 0)}")
                print(f"  - 无数据的单元数: {statistics.get('units_without_data', 0)}")
        else:
            print(f"✗ 请求失败，状态码: {response.status_code}")
            print(f"✗ 错误信息: {response.text}")
            
    except requests.exceptions.ConnectionError:
        print("✗ 连接失败，请确保服务正在运行")
    except Exception as e:
        print(f"✗ 测试异常: {str(e)}")
    
    # 测试3: 获取特定单元的h_xtfx值
    print("\n3. 测试获取特定单元的h_xtfx值")
    print("-" * 40)
    
    try:
        # 首先获取一个存在的单元ID
        response = requests.get(f"{base_url}/api/unit-grouping/h_xtfx")
        if response.status_code == 200:
            data = response.json()
            if data.get('success', False):
                unit_data = data.get('data', {})
                if unit_data:
                    # 取第一个单元进行测试
                    test_unit_id = list(unit_data.keys())[0]
                    
                    # 测试获取特定单元
                    response = requests.get(f"{base_url}/api/unit-grouping/unit/{test_unit_id}")
                    
                    if response.status_code == 200:
                        unit_result = response.json()
                        print(f"✓ 请求成功")
                        print(f"✓ 单元 {test_unit_id} 的h_xtfx值: {unit_result.get('h_xtfx')}")
                    else:
                        print(f"✗ 请求失败，状态码: {response.status_code}")
                else:
                    print("✗ 没有可用的单元数据进行测试")
            else:
                print("✗ 无法获取单元数据进行测试")
        else:
            print("✗ 无法获取单元数据进行测试")
            
    except requests.exceptions.ConnectionError:
        print("✗ 连接失败，请确保服务正在运行")
    except Exception as e:
        print(f"✗ 测试异常: {str(e)}")
    
    print("\n" + "=" * 60)
    print("测试完成")
    print("=" * 60)
    
    # 测试4: 测试新的ORM接口
    print("\n4. 测试新的ORM接口功能")
    print("-" * 40)
    
    # 测试点位统计信息
    try:
        response = requests.get(f"{base_url}/api/unit-grouping/points/statistics")
        
        if response.status_code == 200:
            data = response.json()
            print(f"✓ 点位统计信息获取成功")
            distribution = data.get('distribution', {})
            print(f"✓ 总点位数: {data.get('total_points', 0)}")
            for category, stats in distribution.items():
                print(f"  - {category}: {stats['count']} ({stats['percentage']}%)")
        else:
            print(f"✗ 获取点位统计信息失败，状态码: {response.status_code}")
    except Exception as e:
        print(f"✗ 测试点位统计信息异常: {str(e)}")
    
    # 测试数据库摘要信息
    try:
        response = requests.get(f"{base_url}/api/unit-grouping/database/summary")
        
        if response.status_code == 200:
            data = response.json()
            print(f"✓ 数据库摘要信息获取成功")
            summary = data.get('summary', {})
            print(f"  - 总单元数: {summary.get('total_units', 0)}")
            print(f"  - 总点位数: {summary.get('total_points', 0)}")
            print(f"  - h_xtfx分类数: {summary.get('h_xtfx_categories', 0)}")
        else:
            print(f"✗ 获取数据库摘要信息失败，状态码: {response.status_code}")
    except Exception as e:
        print(f"✗ 测试数据库摘要信息异常: {str(e)}")
    
    # 测试批量获取单元信息
    try:
        response = requests.get(f"{base_url}/api/unit-grouping/units/batch?unit_ids=1&unit_ids=2&unit_ids=3")
        
        if response.status_code == 200:
            data = response.json()
            print(f"✓ 批量获取单元信息成功")
            print(f"  - 请求数量: {data.get('total_requested', 0)}")
            print(f"  - 找到数量: {data.get('total_found', 0)}")
        else:
            print(f"✗ 批量获取单元信息失败，状态码: {response.status_code}")
    except Exception as e:
        print(f"✗ 测试批量获取单元信息异常: {str(e)}")
    
    # 输出使用说明
    print("\n使用说明:")
    print("1. 启动服务: uvicorn main:app --reload")
    print("2. 访问API文档: http://localhost:8000/docs")
    print("3. 主要接口:")
    print("   - GET /api/unit-grouping/h_xtfx - 获取所有单元的h_xtfx分类结果")
    print("   - GET /api/unit-grouping/statistics - 获取统计信息")
    print("   - GET /api/unit-grouping/unit/{unit_id} - 获取特定单元的h_xtfx值")
    print("   - GET /api/unit-grouping/points/statistics - 获取点位统计信息（ORM）")
    print("   - GET /api/unit-grouping/database/summary - 获取数据库摘要信息（ORM）")
    print("   - GET /api/unit-grouping/units/batch - 批量获取单元信息（ORM）")
    print("   - GET /api/unit-grouping/points/by-area - 按区域获取点位数据（ORM）")

if __name__ == "__main__":
    test_unit_grouping_api()