AI_product_Description/test_api.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
API 测试脚本
用于测试 /api/description 和 /api/title 两个端点
"""
import requests
import json
import sys
from typing import Dict, Any

# API 基础配置
BASE_URL = "http://60.165.238.181:5555"  # 修改为你的服务地址

def test_api_description():
    """测试商品描述生成接口"""
    print("=" * 80)
    print("测试 /api/description 接口")
    print("=" * 80)
    
    url = f"{BASE_URL}/api/description"
    data = {'img': 'https://img2.goelia.com.au/prod/product/1ENC6E220/material/main/Shopify/-1/72736752b0ad405382d5ed277dabc660.jpg', 
        'graphic_label': ['-100% Merino wool', '-With pockets', '-H-line fit'], 
        'spu': '1ENC6E220', 
        'plm_info': '1、手工流苏边设计   2、贴袋设计   3、金属纽扣'}
    
    print(f"\n请求 URL: {url}")
    print(f"请求数据: {json.dumps(data, ensure_ascii=False, indent=2)}")
    
    try:
        response = requests.post(url, json=data, timeout=120)
        
        print(f"\n响应状态码: {response.status_code}")
        print(f"响应内容:")
        print(json.dumps(response.json(), ensure_ascii=False, indent=2))
        
        if response.status_code == 200:
            print("\n✅ 测试通过")
            return True
        else:
            print("\n❌ 测试失败")
            return False
            
    except requests.exceptions.ConnectionError:
        print("\n❌ 连接错误: 无法连接到服务器，请确保服务已启动")
        return False
    except requests.exceptions.Timeout:
        print("\n❌ 超时错误: 请求超时")
        return False
    except Exception as e:
        print(f"\n❌ 发生错误: {str(e)}")
        return False


def test_api_title():
    """测试商品标题生成接口"""
    print("\n" + "=" * 80)
    print("测试 /api/title 接口")
    print("=" * 80)
    
    url = f"{BASE_URL}/api/title"
    data = {
        "spu": "TEST_SPU_002",
        "desc": "This maxi dress features unparalleled comfort and a unique texture with its tencel blend fabric. The square neckline and smocked bodice create a flattering silhouette, while the layered skirt adds romantic flair.",
        "tags": ["舒适", "优雅", "修身"],
        "reference_title": "Elegant Maxi Dress with Floral Pattern"
    }
    
    print(f"\n请求 URL: {url}")
    print(f"请求数据: {json.dumps(data, ensure_ascii=False, indent=2)}")
    
    try:
        response = requests.post(url, json=data, timeout=120)
        
        print(f"\n响应状态码: {response.status_code}")
        print(f"响应内容:")
        print(json.dumps(response.json(), ensure_ascii=False, indent=2))
        
        if response.status_code == 200:
            print("\n✅ 测试通过")
            return True
        else:
            print("\n❌ 测试失败")
            return False
            
    except requests.exceptions.ConnectionError:
        print("\n❌ 连接错误: 无法连接到服务器，请确保服务已启动")
        return False
    except requests.exceptions.Timeout:
        print("\n❌ 超时错误: 请求超时")
        return False
    except Exception as e:
        print(f"\n❌ 发生错误: {str(e)}")
        return False


def test_error_cases():
    """测试错误情况"""
    print("\n" + "=" * 80)
    print("测试错误情况处理")
    print("=" * 80)
    
    # 测试缺少必需字段
    print("\n1. 测试缺少必需字段 (description)")
    url = f"{BASE_URL}/api/description"
    data = {
        "spu": "TEST_SPU_003",
        # 缺少 plm_info, img, graphic_label
    }
    
    try:
        response = requests.post(url, json=data, timeout=10)
        print(f"响应状态码: {response.status_code}")
        print(f"响应内容: {json.dumps(response.json(), ensure_ascii=False, indent=2)}")
    except Exception as e:
        print(f"发生错误: {str(e)}")
    
    # 测试无效的图片URL
    print("\n2. 测试无效的图片URL")
    data = {
        "spu": "TEST_SPU_004",
        "plm_info": "测试商品信息",
        "img": "https://invalid-url-test.com/nonexistent.jpg",
        "graphic_label": ["测试标签"]
    }
    
    try:
        response = requests.post(url, json=data, timeout=30)
        print(f"响应状态码: {response.status_code}")
        print(f"响应内容: {json.dumps(response.json(), ensure_ascii=False, indent=2)}")
    except Exception as e:
        print(f"发生错误: {str(e)}")


def main():
    """主函数"""
    print("\n" + "=" * 80)
    print("开始 API 测试")
    print("=" * 80)
    
    # 检查服务器是否运行
    try:
        response = requests.get(f"{BASE_URL}/", timeout=5)
        print("✅ 服务器正在运行")
    except requests.exceptions.ConnectionError:
        print("⚠️  警告: 无法连接到服务器")
        print("请确保 Flask 应用已启动 (app_v2.py)")
        response = input("\n是否继续测试? (y/n): ")
        if response.lower() != 'y':
            sys.exit(0)
    except Exception as e:
        print(f"⚠️  警告: {str(e)}")
    
    results = []
    
    # 测试描述生成接口
    results.append(("Description API", test_api_description()))
    
    # 测试标题生成接口
    # results.append(("Title API", test_api_title()))
    
    # 测试错误情况
    # test_error_cases()
    
    # 打印测试总结
    print("\n" + "=" * 80)
    print("测试总结")
    print("=" * 80)
    for test_name, result in results:
        status = "✅ 通过" if result else "❌ 失败"
        print(f"{test_name}: {status}")
    
    passed = sum(1 for _, result in results if result)
    total = len(results)
    print(f"\n总计: {passed}/{total} 个测试通过")


if __name__ == "__main__":
    main()