#!/usr/bin/env python3 """ DMS服务连接测试脚本 专门用于测试DMS API的连接和SSL配置 """ import requests import urllib3 import json import sys import time from urllib.parse import urljoin def test_dms_connection(): """测试DMS服务连接""" # 配置 base_url = "https://www.dev.ideas.cnpc" api_endpoint = "/api/schema/manage/schema" full_url = urljoin(base_url, api_endpoint) print("🔧 DMS服务连接测试") print("=" * 60) print(f"目标服务器: {base_url}") print(f"API端点: {api_endpoint}") print(f"完整URL: {full_url}") print() # 禁用SSL警告（用于测试） urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # 测试1: 忽略SSL证书验证 print("📡 测试1: 忽略SSL证书验证") print("-" * 40) try: start_time = time.time() response = requests.get( full_url, verify=False, timeout=30, headers={ 'User-Agent': 'DMS-Compliance-Test/1.0', 'Accept': 'application/json' } ) end_time = time.time() print(f"✅ 请求成功!") print(f"⏱️ 响应时间: {end_time - start_time:.2f}秒") print(f"📊 HTTP状态码: {response.status_code}") print(f"📋 响应头:") for key, value in response.headers.items(): print(f" {key}: {value}") print(f"\n📄 响应内容:") if response.status_code == 200: try: data = response.json() print(f"✅ JSON解析成功") print(f"📊 数据类型: {type(data)}") if isinstance(data, dict): print(f"🔑 响应键: {list(data.keys())}") # 检查DMS特定的响应结构 if 'code' in data: print(f"📈 业务代码: {data.get('code')}") if 'message' in data: print(f"💬 消息: {data.get('message')}") if 'data' in data: data_content = data.get('data') print(f"📦 数据内容类型: {type(data_content)}") if isinstance(data_content, dict) and 'records' in data_content: records = data_content.get('records', []) print(f"📝 记录数量: {len(records)}") if records: print(f"📋 第一条记录示例: {records[0] if len(records) > 0 else 'N/A'}") # 显示响应内容的前500字符 response_text = json.dumps(data, ensure_ascii=False, indent=2) if len(response_text) > 500: print(f"📄 响应内容（前500字符）:") print(response_text[:500] + "...") else: print(f"📄 完整响应内容:") print(response_text) except json.JSONDecodeError as e: print(f"❌ JSON解析失败: {e}") print(f"📄 原始响应内容（前500字符）:") print(response.text[:500]) else: print(f"⚠️ HTTP状态码异常: {response.status_code}") print(f"📄 响应内容: {response.text[:500]}") return True except requests.exceptions.SSLError as e: print(f"❌ SSL错误: {e}") print("💡 建议: 使用 --ignore-ssl 参数") return False except requests.exceptions.ConnectionError as e: print(f"❌ 连接错误: {e}") print("💡 建议: 检查网络连接和服务器地址") return False except requests.exceptions.Timeout as e: print(f"❌ 超时错误: {e}") print("💡 建议: 增加超时时间或检查网络延迟") return False except Exception as e: print(f"❌ 其他错误: {e}") print(f"错误类型: {type(e).__name__}") return False def test_ssl_verification(): """测试SSL证书验证""" print("\n📡 测试2: 启用SSL证书验证") print("-" * 40) base_url = "https://www.dev.ideas.cnpc" api_endpoint = "/api/schema/manage/schema" full_url = urljoin(base_url, api_endpoint) try: start_time = time.time() response = requests.get( full_url, verify=True, # 启用SSL验证 timeout=30, headers={ 'User-Agent': 'DMS-Compliance-Test/1.0', 'Accept': 'application/json' } ) end_time = time.time() print(f"✅ SSL验证通过!") print(f"⏱️ 响应时间: {end_time - start_time:.2f}秒") print(f"📊 HTTP状态码: {response.status_code}") return True except requests.exceptions.SSLError as e: print(f"❌ SSL验证失败（这是预期的）: {e}") print("💡 这证明需要使用 --ignore-ssl 参数") return False except Exception as e: print(f"❌ 其他错误: {e}") return False def test_basic_connectivity(): """测试基础网络连接""" print("\n📡 测试3: 基础网络连接") print("-" * 40) base_url = "https://www.dev.ideas.cnpc" try: # 测试根路径 print(f"🔍 测试根路径: {base_url}") response = requests.get(base_url, verify=False, timeout=10) print(f"✅ 根路径连接成功: {response.status_code}") # 测试其他可能的端点 test_endpoints = [ "/", "/api", "/api/health", "/health" ] for endpoint in test_endpoints: try: test_url = urljoin(base_url, endpoint) response = requests.get(test_url, verify=False, timeout=5) print(f"✅ {endpoint}: {response.status_code}") except: print(f"❌ {endpoint}: 连接失败") return True except Exception as e: print(f"❌ 基础连接失败: {e}") return False def test_domain_mapping(): """测试域映射文件""" print("\n📁 测试4: 域映射文件") print("-" * 40) domain_file = "./assets/doc/dms/domain.json" try: with open(domain_file, 'r', encoding='utf-8') as f: domain_data = json.load(f) print(f"✅ 域映射文件读取成功") print(f"📁 文件路径: {domain_file}") print(f"📊 域映射数据: {json.dumps(domain_data, ensure_ascii=False, indent=2)}") return True except FileNotFoundError: print(f"❌ 域映射文件不存在: {domain_file}") return False except json.JSONDecodeError as e: print(f"❌ 域映射文件JSON格式错误: {e}") return False except Exception as e: print(f"❌ 读取域映射文件出错: {e}") return False def generate_curl_command(): """生成等效的curl命令""" print("\n🔧 等效的curl命令") print("-" * 40) base_url = "https://www.dev.ideas.cnpc" api_endpoint = "/api/schema/manage/schema" full_url = urljoin(base_url, api_endpoint) curl_commands = [ f"# 忽略SSL证书验证:", f"curl -k -v '{full_url}'", f"", f"# 带请求头:", f"curl -k -v -H 'Accept: application/json' -H 'User-Agent: DMS-Compliance-Test/1.0' '{full_url}'", f"", f"# 启用SSL验证:", f"curl -v '{full_url}'", f"", f"# 测试连接（仅获取响应头）:", f"curl -k -I '{full_url}'" ] for cmd in curl_commands: print(cmd) def main(): """主函数""" print("🚀 DMS服务连接诊断工具") print("=" * 80) print("此工具将测试DMS服务的连接性和SSL配置") print("=" * 80) results = [] # 执行所有测试 results.append(("DMS API连接（忽略SSL）", test_dms_connection())) results.append(("SSL证书验证", test_ssl_verification())) results.append(("基础网络连接", test_basic_connectivity())) results.append(("域映射文件", test_domain_mapping())) # 生成curl命令 generate_curl_command() # 总结结果 print("\n" + "=" * 80) print("📋 测试结果总结") print("=" * 80) passed = 0 for test_name, result in results: status = "✅ 通过" if result else "❌ 失败" print(f"{test_name:25} : {status}") if result: passed += 1 print(f"\n📊 总计: {passed}/{len(results)} 个测试通过") # 给出建议 print("\n💡 建议和解决方案:") print("-" * 40) if results[0][1]: # DMS API连接成功 print("✅ DMS API连接正常，建议在主程序中使用 --ignore-ssl 参数") print(" 命令示例:") print(" python run_api_tests.py --dms ./assets/doc/dms/domain.json --ignore-ssl") else: print("❌ DMS API连接失败，请检查:") print(" 1. 网络连接是否正常") print(" 2. 服务器地址是否正确") print(" 3. 防火墙设置") print(" 4. 服务器是否在线") if not results[1][1]: # SSL验证失败 print("⚠️ SSL证书验证失败，这是正常的，使用 --ignore-ssl 参数即可") if not results[3][1]: # 域映射文件问题 print("❌ 域映射文件有问题，请检查文件是否存在且格式正确") return passed == len(results) if __name__ == "__main__": try: success = main() sys.exit(0 if success else 1) except KeyboardInterrupt: print("\n\n⚠️ 测试被用户中断") sys.exit(1) except Exception as e: print(f"\n\n❌ 测试过程中发生未预期的错误: {e}") import traceback traceback.print_exc() sys.exit(1)