import os import sys import json import logging import argparse import traceback # 用于更详细的错误日志 from pathlib import Path from flask import Flask, request, jsonify, send_from_directory from flask_cors import CORS # 用于处理跨域请求 # 将ddms_compliance_suite的父目录添加到sys.path # 假设flask_app.py与ddms_compliance_suite目录在同一级别，或者ddms_compliance_suite在其PYTHONPATH中 # 如果 ddms_compliance_suite 是一个已安装的包，则不需要这个 # current_dir = os.path.dirname(os.path.abspath(__file__)) # project_root = os.path.dirname(current_dir) # 假设项目根目录是上一级 # sys.path.insert(0, project_root) # 或者更具体地添加包含ddms_compliance_suite的目录 # sys.path.insert(0, os.path.join(project_root, 'ddms_compliance_suite')) from ddms_compliance_suite.test_orchestrator import APITestOrchestrator, TestSummary from ddms_compliance_suite.input_parser.parser import InputParser, ParsedYAPISpec, ParsedSwaggerSpec # 从 run_api_tests.py 导入辅助函数 (如果它们被重构为可导入的) # 为了简单起见，我们可能会直接在 flask_app.py 中重新实现一些逻辑或直接调用Orchestrator app = Flask(__name__, static_folder='static', static_url_path='') CORS(app) # 允许所有来源的跨域请求，生产环境中应配置更严格的规则 # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # --- 辅助函数 --- def get_orchestrator_from_config(config: dict) -> APITestOrchestrator: """根据配置字典实例化APITestOrchestrator""" return APITestOrchestrator( base_url=config.get('base_url', ''), custom_test_cases_dir=config.get('custom_test_cases_dir'), llm_api_key=config.get('llm_api_key'), llm_base_url=config.get('llm_base_url'), llm_model_name=config.get('llm_model_name'), use_llm_for_request_body=config.get('use_llm_for_request_body', False), use_llm_for_path_params=config.get('use_llm_for_path_params', False), use_llm_for_query_params=config.get('use_llm_for_query_params', False), use_llm_for_headers=config.get('use_llm_for_headers', False), output_dir=config.get('output_dir') # 虽然Orchestrator内部可能不直接用它保存，但可以传入 ) # --- API 端点 --- @app.route('/') def serve_index(): return send_from_directory(app.static_folder, 'index.html') @app.route('/run-tests', methods=['POST']) def run_tests_endpoint(): try: config_data = request.json if not config_data: return jsonify({"error": "Request body must be JSON"}), 400 logger.info(f"接收到测试运行请求: {config_data}") # 校验必需参数 if not config_data.get('base_url'): return jsonify({"error": "'base_url' is required"}), 400 if not config_data.get('yapi_file_path') and not config_data.get('swagger_file_path'): return jsonify({"error": "Either 'yapi_file_path' or 'swagger_file_path' is required"}), 400 orchestrator = get_orchestrator_from_config(config_data) summary = TestSummary() # 为本次运行创建新的摘要 parsed_spec = None api_spec_type = "" if config_data.get('yapi_file_path'): api_spec_type = "YAPI" yapi_path = config_data['yapi_file_path'] if not os.path.isabs(yapi_path): yapi_path = os.path.join(os.getcwd(), yapi_path) # 假设相对路径相对于服务器工作目录 if not os.path.exists(yapi_path): return jsonify({"error": f"YAPI file not found: {yapi_path}"}), 400 logger.info(f"解析YAPI文件: {yapi_path}") parsed_spec = orchestrator.parser.parse_yapi_spec(yapi_path) if not parsed_spec: logger.error(f"解析YAPI文件失败: {yapi_path}") return jsonify({"error": f"Failed to parse YAPI file: {yapi_path}"}), 500 elif config_data.get('swagger_file_path'): api_spec_type = "Swagger/OpenAPI" swagger_path = config_data['swagger_file_path'] if not os.path.isabs(swagger_path): swagger_path = os.path.join(os.getcwd(), swagger_path) # 假设相对路径 if not os.path.exists(swagger_path): return jsonify({"error": f"Swagger file not found: {swagger_path}"}), 400 logger.info(f"解析Swagger/OpenAPI文件: {swagger_path}") parsed_spec = orchestrator.parser.parse_swagger_spec(swagger_path) if not parsed_spec: logger.error(f"解析Swagger文件失败: {swagger_path}") return jsonify({"error": f"Failed to parse Swagger file: {swagger_path}"}), 500 # 执行测试用例 logger.info(f"开始从已解析的 {api_spec_type} 规范执行测试用例...") summary = orchestrator._execute_tests_from_parsed_spec( parsed_spec=parsed_spec, summary=summary, categories=config_data.get('categories'), # 逗号分隔的字符串，需要转为列表 tags=config_data.get('tags'), # 同上 custom_test_cases_dir=config_data.get('custom_test_cases_dir') ) logger.info("测试用例执行完成。") # 执行场景测试 (如果指定了目录) scenarios_dir = config_data.get('scenarios_dir') if scenarios_dir and parsed_spec: if not os.path.isabs(scenarios_dir): scenarios_dir = os.path.join(os.getcwd(), scenarios_dir) logger.info(f"开始执行API场景测试，目录: {scenarios_dir}") orchestrator.run_scenarios_from_spec( scenarios_dir=scenarios_dir, parsed_spec=parsed_spec, summary=summary ) logger.info("API场景测试执行完毕。") summary.finalize_summary() # 最终确定摘要，计算总时长等 # summary.print_summary_to_console() # 后端服务通常不直接打印到控制台 # 可以在这里决定如何保存报告，例如保存到 output_dir (如果提供) output_dir_path_str = config_data.get('output_dir') main_report_file_path_str = "" api_calls_output_path_str = "" api_calls_filename = "api_call_details.md" if output_dir_path_str: output_path = Path(output_dir_path_str) output_path.mkdir(parents=True, exist_ok=True) main_report_file_path = output_path / f"summary_report.json" # 默认保存为json main_report_file_path_str = str(main_report_file_path) with open(main_report_file_path, 'w', encoding='utf-8') as f: f.write(summary.to_json(pretty=True)) logger.info(f"主测试报告已保存到: {main_report_file_path}") # 保存API调用详情 api_calls_output_path_str = str(output_path) # (需要从 run_api_tests.py 移植 save_api_call_details_to_file 或类似功能) # 暂时跳过保存 api_call_details 文件，因为 orchestrator.get_api_call_details() 需要被调用 # 并且保存逻辑也需要移植。 # save_api_call_details_to_file(orchestrator.get_api_call_details(), api_calls_output_path_str, api_calls_filename) return jsonify({ "message": "测试执行完成。", "summary": summary.to_dict(), "report_file": main_report_file_path_str, # 报告文件路径（如果保存了） # "api_calls_file": api_calls_output_path_str + "/" + api_calls_filename # API调用详情文件路径 }), 200 except Exception as e: logger.error(f"执行测试时发生错误: {e}\n{traceback.format_exc()}") return jsonify({"error": f"执行测试时发生内部错误: {str(e)}"}), 500 @app.route('/list-yapi-categories', methods=['POST']) def list_yapi_categories_endpoint(): try: data = request.json yapi_file = data.get('yapi_file_path') if not yapi_file: return jsonify({"error": "'yapi_file_path' is required"}), 400 if not os.path.isabs(yapi_file): yapi_file = os.path.join(os.getcwd(), yapi_file) if not os.path.exists(yapi_file): return jsonify({"error": f"YAPI file not found: {yapi_file}"}), 400 parser = InputParser() parsed_yapi = parser.parse_yapi_spec(yapi_file) if not parsed_yapi or not parsed_yapi.categories: return jsonify({"error": "Failed to parse YAPI categories or no categories found"}), 500 categories_list = [ {"name": cat.get('name', '未命名'), "description": cat.get('desc', '无描述')} for cat in parsed_yapi.categories ] return jsonify(categories_list), 200 except Exception as e: logger.error(f"列出YAPI分类时出错: {e}\n{traceback.format_exc()}") return jsonify({"error": f"处理YAPI分类列表时出错: {str(e)}"}), 500 @app.route('/list-swagger-tags', methods=['POST']) def list_swagger_tags_endpoint(): try: data = request.json swagger_file = data.get('swagger_file_path') if not swagger_file: return jsonify({"error": "'swagger_file_path' is required"}), 400 if not os.path.isabs(swagger_file): swagger_file = os.path.join(os.getcwd(), swagger_file) if not os.path.exists(swagger_file): return jsonify({"error": f"Swagger file not found: {swagger_file}"}), 400 parser = InputParser() parsed_swagger = parser.parse_swagger_spec(swagger_file) if not parsed_swagger or not parsed_swagger.tags: return jsonify({"error": "Failed to parse Swagger tags or no tags found"}), 500 tags_list = [ {"name": tag.get('name', '未命名'), "description": tag.get('description', '无描述')} for tag in parsed_swagger.tags ] return jsonify(tags_list), 200 except Exception as e: logger.error(f"列出Swagger标签时出错: {e}\n{traceback.format_exc()}") return jsonify({"error": f"处理Swagger标签列表时出错: {str(e)}"}), 500 if __name__ == '__main__': # 注意：在生产环境中，应使用Gunicorn或uWSGI等WSGI服务器运行Flask应用 app.run(debug=True, host='0.0.0.0', port=5050)