#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 微智电子 - 自动化测试机器人 功能：自动测试所有系统，生成可视化报告 作者：AI助手 日期：2026-02-28 """ import requests import json import time import sqlite3 from datetime import datetime from pathlib import Path import sys # 配置 BASE_URL = "http://127.0.0.1" TEST_USER = {"phone": "13981734846", "password": "123456"} REPORT_DIR = "/root/.openclaw/workspace/monitor/reports" DB_PATH = "/root/.openclaw/workspace/erp-backend/database/erp.db" class AutoTestRobot: """自动化测试机器人""" def __init__(self): self.results = [] self.token = None self.session = requests.Session() self.report = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "summary": {"total": 0, "passed": 0, "failed": 0, "warnings": 0}, "tests": [] } # 创建报告目录 Path(REPORT_DIR).mkdir(parents=True, exist_ok=True) def test_page_access(self, name, url, expected_status=200): """测试页面访问""" try: response = self.session.get(f"{BASE_URL}{url}", timeout=5) status = response.status_code passed = status == expected_status result = { "category": "页面访问", "name": name, "url": url, "status": status, "passed": passed, "response_time": f"{response.elapsed.total_seconds()*1000:.0f}ms", "message": "正常" if passed else f"期望{expected_status}, 实际{status}" } self.report["tests"].append(result) self._update_summary(passed) print(f"{'✓' if passed else '✗'} {name}: {status} ({result['response_time']})") return passed except Exception as e: result = { "category": "页面访问", "name": name, "url": url, "status": 0, "passed": False, "response_time": "0ms", "message": f"异常: {str(e)}" } self.report["tests"].append(result) self._update_summary(False) print(f"✗ {name}: 异常 - {str(e)}") return False def test_api_endpoint(self, name, url, method="GET", data=None): """测试API接口""" try: headers = {} if self.token: headers["Authorization"] = f"Bearer {self.token}" if method == "POST": response = self.session.post(f"{BASE_URL}{url}", json=data, headers=headers, timeout=5) else: response = self.session.get(f"{BASE_URL}{url}", headers=headers, timeout=5) status = response.status_code passed = status == 200 # 尝试解析JSON try: json_data = response.json() has_data = json_data.get("code") == 200 if isinstance(json_data, dict) else True except: has_data = False result = { "category": "API接口", "name": name, "url": url, "method": method, "status": status, "passed": passed and has_data, "has_data": has_data, "response_time": f"{response.elapsed.total_seconds()*1000:.0f}ms", "message": "正常" if (passed and has_data) else "数据异常" } self.report["tests"].append(result) self._update_summary(passed and has_data) print(f"{'✓' if (passed and has_data) else '✗'} {name}: {status}") return passed and has_data except Exception as e: result = { "category": "API接口", "name": name, "url": url, "status": 0, "passed": False, "response_time": "0ms", "message": f"异常: {str(e)}" } self.report["tests"].append(result) self._update_summary(False) print(f"✗ {name}: 异常 - {str(e)}") return False def test_login(self): """测试登录功能""" try: response = self.session.post( f"{BASE_URL}/api/login", json=TEST_USER, timeout=5 ) data = response.json() if data.get("code") == 200: self.token = data["data"]["token"] user = data["data"]["user"] result = { "category": "用户认证", "name": "用户登录", "url": "/api/login", "status": 200, "passed": True, "user": user["name"], "role": user["role"], "response_time": f"{response.elapsed.total_seconds()*1000:.0f}ms", "message": f"登录成功: {user['name']} ({user['role']})" } self.report["tests"].append(result) self._update_summary(True) print(f"✓ 用户登录: 成功 ({user['name']})") return True else: result = { "category": "用户认证", "name": "用户登录", "url": "/api/login", "status": response.status_code, "passed": False, "message": f"登录失败: {data.get('message', '未知错误')}" } self.report["tests"].append(result) self._update_summary(False) print(f"✗ 用户登录: 失败") return False except Exception as e: result = { "category": "用户认证", "name": "用户登录", "url": "/api/login", "status": 0, "passed": False, "message": f"异常: {str(e)}" } self.report["tests"].append(result) self._update_summary(False) print(f"✗ 用户登录: 异常 - {str(e)}") return False def test_database(self): """测试数据库连接""" try: conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() # 测试查询 cursor.execute("SELECT COUNT(*) FROM users") user_count = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM inventory") inventory_count = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM orders") order_count = cursor.fetchone()[0] conn.close() result = { "category": "数据库", "name": "数据库连接与查询", "passed": True, "stats": { "users": user_count, "inventory": inventory_count, "orders": order_count }, "message": f"正常: {user_count}用户, {inventory_count}库存, {order_count}订单" } self.report["tests"].append(result) self._update_summary(True) print(f"✓ 数据库: {user_count}用户, {inventory_count}库存, {order_count}订单") return True except Exception as e: result = { "category": "数据库", "name": "数据库连接与查询", "passed": False, "message": f"异常: {str(e)}" } self.report["tests"].append(result) self._update_summary(False) print(f"✗ 数据库: 异常 - {str(e)}") return False def _update_summary(self, passed): """更新统计""" self.report["summary"]["total"] += 1 if passed: self.report["summary"]["passed"] += 1 else: self.report["summary"]["failed"] += 1 def run_all_tests(self): """运行所有测试""" print("\n" + "="*60) print("🤖 自动化测试机器人启动") print("="*60) print(f"测试时间: {self.report['timestamp']}") print("-"*60 + "\n") # 1. 页面访问测试 print("📄 [1/5] 页面访问测试...") pages = [ ("统一登录中心", "/login.html"), ("ERP登录页", "/erp/login.html"), ("ERP首页", "/erp/index.html"), ("文件中心", "/external/index.html"), ("公司官网", "/company/index.html"), ("手机测试页", "/mobile-test.html"), ("版本管理", "/versions/index.html"), ("系统架构图", "/erp/system-architecture.html"), ] for name, url in pages: self.test_page_access(name, url) print() # 2. API接口测试 print("🔌 [2/5] API接口测试...") self.test_api_endpoint("健康检查", "/api/health") print() # 3. 用户认证测试 print("🔐 [3/5] 用户认证测试...") self.test_login() if self.token: self.test_api_endpoint("仪表盘统计", "/api/dashboard/stats") self.test_api_endpoint("库存列表", "/api/inventory") self.test_api_endpoint("订单列表", "/api/orders") self.test_api_endpoint("项目列表", "/api/projects") print() # 4. 数据库测试 print("💾 [4/5] 数据库测试...") self.test_database() print() # 5. 性能测试 print("⚡ [5/5] 性能测试...") start = time.time() for _ in range(5): self.session.get(f"{BASE_URL}/api/health", timeout=5) avg_time = (time.time() - start) / 5 * 1000 result = { "category": "性能测试", "name": "API平均响应时间", "passed": avg_time < 1000, "response_time": f"{avg_time:.0f}ms", "message": f"平均响应时间: {avg_time:.0f}ms" + (" (优秀)" if avg_time < 200 else " (正常)" if avg_time < 1000 else " (较慢)") } self.report["tests"].append(result) self._update_summary(avg_time < 1000) print(f"{'✓' if avg_time < 1000 else '⚠'} API平均响应: {avg_time:.0f}ms") print() # 生成报告 self.generate_report() def generate_report(self): """生成测试报告""" summary = self.report["summary"] pass_rate = (summary["passed"] / summary["total"] * 100) if summary["total"] > 0 else 0 # 生成HTML报告 html_content = f""" 自动化测试报告 - {self.report['timestamp']}

🤖 自动化测试报告

测试时间: {self.report['timestamp']}
{'✅ 全部通过' if summary['failed'] == 0 else '⚠️ 发现问题' if pass_rate >= 70 else '❌ 需要处理'}
{summary['total']}
总测试项
{summary['passed']}
通过
{summary['failed']}
失败
{pass_rate:.0f}%
通过率

📋 详细测试结果

""" # 按分类组织测试结果 categories = {} for test in self.report["tests"]: cat = test.get("category", "其他") if cat not in categories: categories[cat] = [] categories[cat].append(test) for cat, tests in categories.items(): html_content += f'
\n' html_content += f'
{cat}
\n' for test in tests: status_class = "pass" if test["passed"] else "fail" status_icon = "✓" if test["passed"] else "✗" meta = test.get("response_time", "") or test.get("status", "") html_content += f"""
{status_icon}
{test['name']}
{test.get('message', '')}
{meta}
""" html_content += '
\n' html_content += """

微智电子数字化平台 - 自动化测试机器人

此报告由AI自动生成

""" # 保存报告 report_file = f"{REPORT_DIR}/test-report-{datetime.now().strftime('%Y%m%d-%H%M%S')}.html" with open(report_file, 'w', encoding='utf-8') as f: f.write(html_content) # 同时保存最新报告 latest_file = f"{REPORT_DIR}/latest.html" with open(latest_file, 'w', encoding='utf-8') as f: f.write(html_content) # 打印汇总 print("="*60) print("📊 测试完成汇总") print("="*60) print(f"总测试项: {summary['total']}") print(f"✓ 通过: {summary['passed']}") print(f"✗ 失败: {summary['failed']}") print(f"📈 通过率: {pass_rate:.1f}%") print() if summary['failed'] == 0: print("🎉 恭喜！所有测试全部通过！") elif pass_rate >= 80: print("⚠️ 警告：部分测试未通过，建议检查") else: print("❌ 严重：大量测试失败，需要立即处理！") print() print(f"📄 完整报告: {report_file}") print(f"🌐 在线查看: http://118.126.91.196/test-report/latest.html") print("="*60) return report_file if __name__ == "__main__": robot = AutoTestRobot() robot.run_all_tests()