DataOps-platform/test_check_interface_only.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
独立测试 /api/meta/check 接口
直接通过 HTTP 请求测试，不依赖 Flask test_client
"""

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

# 配置
BASE_URL = "http://192.168.3.143:5000"
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 字段为空")
                else:
                    print(f"\n❌ 测试 1 失败: code={data.get('code')}, msg={data.get('msg')}")
            else:
                print(f"\n❌ 测试 1 失败: HTTP 状态码 {response.status_code}")
                
        except ValueError as e:
            print(f"❌ 响应不是有效的 JSON: {str(e)}")
            print(f"原始响应: {response.text}")
            
    except requests.Timeout:
        print("❌ 请求超时（超过10秒）")
    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❌ 测试 2 失败: 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}")
        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')}")
        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"\n接口地址: {BASE_URL}{CHECK_ENDPOINT}")
    print("状态: 可用")

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()