DataOps-platform/test_check_218.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试 /api/meta/check 接口 - 使用新服务器地址
服务器: http://192.168.3.218:18183
"""

import requests
import json
from datetime import datetime
import urllib.parse

# 配置
BASE_URL = "http://192.168.3.218:18183"
CHECK_ENDPOINT = "/api/meta/check"

def print_section(title, char="="):
    """打印分节标题"""
    print("\n" + char*70)
    print(f"  {title}")
    print(char*70 + "\n")

def test_check_interface():
    """测试检查接口"""
    
    print_section("测试 /api/meta/check 接口", "=")
    
    print(f"⏰ 测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"🌐 目标服务器: {BASE_URL}")
    print(f"📡 接口路径: {CHECK_ENDPOINT}")
    print(f"🔗 完整URL: {BASE_URL}{CHECK_ENDPOINT}")
    
    # 测试 1: 检查已存在的元数据
    print_section("测试 1: 检查已存在的元数据", "-")
    
    test_name_1 = "其他费用定额"
    url_1 = f"{BASE_URL}{CHECK_ENDPOINT}?name_zh={urllib.parse.quote(test_name_1)}"
    
    print(f"📝 测试元数据名: {test_name_1}")
    print(f"🔗 请求URL: {url_1}")
    print(f"📤 请求方法: GET")
    
    try:
        print("\n⏳ 发送请求...")
        response = requests.get(url_1, timeout=10)
        
        print(f"✅ 响应状态码: {response.status_code}")
        print(f"⚡ 响应时间: {response.elapsed.total_seconds():.3f} 秒")
        print(f"\n📦 响应内容:")
        
        try:
            data = response.json()
            print(json.dumps(data, indent=2, ensure_ascii=False))
            
            # 验证响应格式
            if response.status_code == 200:
                if data.get('code') == 200:
                    exists = data.get('data', {}).get('exists')
                    name_zh = data.get('data', {}).get('name_zh')
                    msg = data.get('msg')
                    
                    print("\n✅ 响应格式正确")
                    print(f"📊 返回字段:")
                    print(f"   ├─ code: {data.get('code')}")
                    print(f"   ├─ msg: {msg}")
                    print(f"   ├─ data.exists: {exists}")
                    print(f"   └─ data.name_zh: {name_zh}")
                    
                    if exists is not None:
                        print(f"\n🎯 结果: 元数据 '{test_name_1}' {'✅ 存在' if exists else '❌ 不存在'}")
                        print("🎉 测试 1 通过！")
                    else:
                        print("\n❌ 测试 1 失败: exists 字段为空")
                elif data.get('code') == 500:
                    print(f"\n⚠️  接口返回错误: {data.get('msg')}")
                    print("这可能是正常的（例如：元数据不存在）")
                else:
                    print(f"\n❌ 测试 1 失败: code={data.get('code')}, msg={data.get('msg')}")
            else:
                print(f"\n❌ 测试 1 失败: HTTP 状态码 {response.status_code}")
                print(f"响应内容: {response.text}")
                
        except ValueError as e:
            print(f"❌ 响应不是有效的 JSON: {str(e)}")
            print(f"原始响应: {response.text[:200]}")
            
    except requests.Timeout:
        print("❌ 请求超时（超过10秒）")
        print("💡 可能原因: 服务器响应慢或网络延迟")
    except requests.ConnectionError as e:
        print(f"❌ 连接失败: {str(e)}")
        print("\n💡 可能的原因:")
        print("  1. 服务器未启动")
        print("  2. 网络不可达")
        print("  3. 防火墙阻止")
        print("  4. IP 地址或端口错误")
    except Exception as e:
        print(f"❌ 请求失败: {str(e)}")
    
    # 测试 2: 检查不存在的元数据
    print_section("测试 2: 检查不存在的元数据", "-")
    
    import time
    test_name_2 = f"测试元数据_不存在_{int(time.time())}"
    url_2 = f"{BASE_URL}{CHECK_ENDPOINT}?name_zh={urllib.parse.quote(test_name_2)}"
    
    print(f"📝 测试元数据名: {test_name_2}")
    print(f"🔗 请求URL: {url_2}")
    print(f"📤 请求方法: GET")
    
    try:
        print("\n⏳ 发送请求...")
        response = requests.get(url_2, timeout=10)
        
        print(f"✅ 响应状态码: {response.status_code}")
        print(f"⚡ 响应时间: {response.elapsed.total_seconds():.3f} 秒")
        print(f"\n📦 响应内容:")
        
        data = response.json()
        print(json.dumps(data, indent=2, ensure_ascii=False))
        
        if response.status_code == 200:
            if data.get('code') == 200:
                exists = data.get('data', {}).get('exists')
                
                if exists == False:
                    print(f"\n🎯 结果: 元数据 '{test_name_2}' 不存在（预期）")
                    print("🎉 测试 2 通过！")
                elif exists == True:
                    print(f"\n⚠️  意外结果: 新创建的测试名称竟然已存在")
                else:
                    print(f"\n❌ 测试 2 失败: exists={exists}")
            else:
                print(f"\n⚠️  接口返回 code={data.get('code')}")
        else:
            print(f"\n❌ 测试 2 失败: HTTP 状态码 {response.status_code}")
            
    except Exception as e:
        print(f"❌ 请求失败: {str(e)}")
    
    # 测试 3: 缺少参数
    print_section("测试 3: 参数验证（缺少 name_zh）", "-")
    
    url_3 = f"{BASE_URL}{CHECK_ENDPOINT}"
    
    print(f"🔗 请求URL: {url_3}")
    print(f"📤 请求方法: GET")
    print("📝 参数: 无（故意不传 name_zh）")
    
    try:
        print("\n⏳ 发送请求...")
        response = requests.get(url_3, timeout=10)
        
        print(f"✅ 响应状态码: {response.status_code}")
        print(f"⚡ 响应时间: {response.elapsed.total_seconds():.3f} 秒")
        print(f"\n📦 响应内容:")
        
        data = response.json()
        print(json.dumps(data, indent=2, ensure_ascii=False))
        
        # 应该返回错误
        if data.get('code') != 200:
            msg = data.get('msg', '')
            if '缺少' in msg or 'name_zh' in msg:
                print(f"\n✅ 正确返回参数错误: {msg}")
                print("🎉 测试 3 通过！")
            else:
                print(f"\n⚠️  返回了错误但错误信息不明确: {msg}")
                print("🎉 测试 3 通过（接口有参数验证）")
        else:
            print("\n❌ 测试 3 失败: 应该返回错误但返回了成功")
            
    except Exception as e:
        print(f"❌ 请求失败: {str(e)}")
    
    # 测试 4: 特殊字符处理
    print_section("测试 4: 特殊字符处理", "-")
    
    test_name_4 = "测试@#$%&*()中文名称"
    url_4 = f"{BASE_URL}{CHECK_ENDPOINT}?name_zh={urllib.parse.quote(test_name_4)}"
    
    print(f"📝 测试元数据名: {test_name_4}")
    print(f"🔐 URL编码后: {urllib.parse.quote(test_name_4)}")
    print(f"🔗 请求URL: {url_4}")
    
    try:
        print("\n⏳ 发送请求...")
        response = requests.get(url_4, timeout=10)
        
        print(f"✅ 响应状态码: {response.status_code}")
        print(f"⚡ 响应时间: {response.elapsed.total_seconds():.3f} 秒")
        print(f"\n📦 响应内容:")
        
        data = response.json()
        print(json.dumps(data, indent=2, ensure_ascii=False))
        
        if response.status_code == 200:
            if data.get('code') == 200:
                returned_name = data.get('data', {}).get('name_zh')
                if returned_name == test_name_4:
                    print(f"\n✅ 特殊字符处理正确")
                    print("🎉 测试 4 通过！")
                else:
                    print(f"\n⚠️  返回的名称不匹配:")
                    print(f"   期望: {test_name_4}")
                    print(f"   实际: {returned_name}")
            else:
                print(f"\n接口返回错误: {data.get('msg')}")
                print("（这可能是正常的）")
        else:
            print(f"\n❌ HTTP 状态码: {response.status_code}")
            
    except Exception as e:
        print(f"❌ 请求失败: {str(e)}")
    
    # 测试总结
    print_section("测试总结", "=")
    
    print("✅ 接口测试完成！")
    print("\n📊 测试覆盖:")
    print("  ✅ 测试 1: 检查已存在的元数据")
    print("  ✅ 测试 2: 检查不存在的元数据")
    print("  ✅ 测试 3: 参数验证")
    print("  ✅ 测试 4: 特殊字符处理")
    
    print("\n🎯 接口信息:")
    print(f"  🌐 服务器: {BASE_URL}")
    print(f"  📡 接口: {CHECK_ENDPOINT}")
    print(f"  🔗 完整地址: {BASE_URL}{CHECK_ENDPOINT}")
    
    print("\n💡 使用示例:")
    print(f"  curl \"{BASE_URL}{CHECK_ENDPOINT}?name_zh=元数据名称\"")

if __name__ == "__main__":
    try:
        test_check_interface()
    except KeyboardInterrupt:
        print("\n\n⚠️  测试被用户中断")
    except Exception as e:
        print(f"\n\n❌ 测试过程中发生错误: {str(e)}")
        import traceback
        traceback.print_exc()