#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 功能测试脚本（基于API和页面结构验证） 不依赖浏览器自动化，通过API和HTML结构验证功能 """ import requests import json import re from datetime import datetime BASE_URL = "http://127.0.0.1" TEST_USER = {"phone": "13981734846", "password": "123456"} class FunctionTester: """功能测试器""" def __init__(self): self.session = requests.Session() self.token = None self.results = [] def add_result(self, category, name, passed, details=""): """添加测试结果""" result = { "timestamp": datetime.now().strftime("%H:%M:%S"), "category": category, "name": name, "passed": passed, "details": details } self.results.append(result) status = "✅" if passed else "❌" print(f"{status} [{category}] {name}") if details: print(f" {details}") return passed def test_login_flow(self): """测试完整登录流程""" print("\n🔐 测试登录流程...") # 1. 获取登录页面 try: resp = self.session.get(f"{BASE_URL}/erp/login.html", timeout=5) has_form = 'id="loginForm"' in resp.text or 'id="phone"' in resp.text self.add_result("登录", "登录页包含表单", has_form, "找到手机号输入框" if has_form else "未找到表单") except Exception as e: self.add_result("登录", "登录页访问", False, str(e)) return False # 2. 提交登录 try: resp = self.session.post( f"{BASE_URL}/api/login", json=TEST_USER, timeout=5 ) data = resp.json() if data.get("code") == 200: self.token = data["data"]["token"] user = data["data"]["user"] self.add_result("登录", "API登录成功", True, f"用户: {user['name']}, 角色: {user['role']}") return True else: self.add_result("登录", "API登录", False, data.get("message", "未知错误")) return False except Exception as e: self.add_result("登录", "API登录", False, str(e)) return False def test_dashboard_data(self): """测试仪表盘数据""" print("\n📊 测试仪表盘数据...") if not self.token: self.add_result("仪表盘", "数据加载", False, "未登录") return try: resp = self.session.get( f"{BASE_URL}/api/dashboard/stats", headers={"Authorization": f"Bearer {self.token}"}, timeout=5 ) data = resp.json() if data.get("code") == 200: stats = data.get("data", {}) checks = [ ("revenue" in stats, "营收数据"), ("order_count" in stats, "订单数量"), ("total_inventory" in stats, "库存总量"), ("warning_count" in stats, "预警数量"), ] all_present = all(check[0] for check in checks) details = ", ".join([c[1] for c in checks if c[0]]) self.add_result("仪表盘", "统计数据完整性", all_present, details) # 验证数据合理性 revenue = stats.get("revenue", 0) valid_revenue = isinstance(revenue, (int, float)) and revenue >= 0 self.add_result("仪表盘", "营收数据合理性", valid_revenue, f"营收: ¥{revenue:,.2f}") else: self.add_result("仪表盘", "数据加载", False, "API返回错误") except Exception as e: self.add_result("仪表盘", "数据加载", False, str(e)) def test_inventory_operations(self): """测试库存操作""" print("\n📦 测试库存功能...") if not self.token: self.add_result("库存", "列表加载", False, "未登录") return # 1. 获取库存列表 try: resp = self.session.get( f"{BASE_URL}/api/inventory", headers={"Authorization": f"Bearer {self.token}"}, timeout=5 ) data = resp.json() if data.get("code") == 200: items = data.get("data", []) has_data = len(items) > 0 self.add_result("库存", "列表加载", has_data, f"返回 {len(items)} 条记录") if has_data: # 验证数据结构 first = items[0] required_fields = ["id", "material_code", "name", "quantity"] has_fields = all(f in first for f in required_fields) self.add_result("库存", "数据字段完整性", has_fields, "包含必要字段" if has_fields else "字段缺失") else: self.add_result("库存", "列表加载", False, "API错误") except Exception as e: self.add_result("库存", "列表加载", False, str(e)) def test_order_operations(self): """测试订单操作""" print("\n📋 测试订单功能...") if not self.token: self.add_result("订单", "列表加载", False, "未登录") return try: resp = self.session.get( f"{BASE_URL}/api/orders", headers={"Authorization": f"Bearer {self.token}"}, timeout=5 ) data = resp.json() if data.get("code") == 200: orders = data.get("data", []) self.add_result("订单", "列表加载", True, f"返回 {len(orders)} 条订单") # 验证状态字段 if orders: statuses = [o.get("status") for o in orders] valid_statuses = all(s in ["pending", "processing", "completed", "cancelled"] for s in statuses) self.add_result("订单", "状态字段有效性", valid_statuses) else: self.add_result("订单", "列表加载", False, "API错误") except Exception as e: self.add_result("订单", "列表加载", False, str(e)) def test_user_permissions(self): """测试用户权限""" print("\n🔑 测试权限控制...") if not self.token: return try: resp = self.session.get( f"{BASE_URL}/api/user/info", headers={"Authorization": f"Bearer {self.token}"}, timeout=5 ) data = resp.json() if data.get("code") == 200: user = data.get("data", {}) has_permissions = "permissions" in user is_ceo = user.get("role") == "ceo" self.add_result("权限", "用户角色识别", True, f"角色: {user.get('role', 'unknown')}, CEO权限: {is_ceo}") else: self.add_result("权限", "用户信息获取", False) except Exception as e: self.add_result("权限", "用户信息获取", False, str(e)) def test_page_functionality(self): """测试页面功能（通过HTML结构）""" print("\n🌐 测试页面功能...") pages_to_test = [ ("/erp/login.html", ["loginForm", "phone", "password", "submit"], "登录表单"), ("/erp/index.html", ["statRevenue", "statOrders", "revenueChart"], "仪表盘元素"), ("/external/index.html", ["研发部", "生产部", "销售部"], "部门列表"), ] for url, elements, desc in pages_to_test: try: resp = self.session.get(f"{BASE_URL}{url}", timeout=5) html = resp.text found_elements = [] for elem in elements: if elem in html: found_elements.append(elem) coverage = len(found_elements) / len(elements) * 100 passed = coverage >= 80 # 80%以上元素存在就认为通过 self.add_result("页面", f"{desc}完整性", passed, f"找到 {len(found_elements)}/{len(elements)} 个关键元素") except Exception as e: self.add_result("页面", f"{desc}检查", False, str(e)) def generate_report(self): """生成测试报告""" print("\n" + "="*60) print("📊 功能测试报告") print("="*60) total = len(self.results) passed = sum(1 for r in self.results if r["passed"]) failed = total - passed print(f"\n总测试项: {total}") print(f"✅ 通过: {passed}") print(f"❌ 失败: {failed}") print(f"📈 通过率: {passed/total*100:.1f}%" if total > 0 else "N/A") if failed > 0: print("\n⚠️ 失败项目：") for r in self.results: if not r["passed"]: print(f" ❌ [{r['category']}] {r['name']}") if r.get("details"): print(f" {r['details']}") print("\n" + "="*60) # 保存详细报告 report_file = f"/root/.openclaw/workspace/monitor/reports/function-test-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json" with open(report_file, 'w') as f: json.dump({ "timestamp": datetime.now().isoformat(), "summary": {"total": total, "passed": passed, "failed": failed}, "results": self.results }, f, indent=2, ensure_ascii=False) print(f"📄 详细报告: {report_file}") return passed == total # 主程序 if __name__ == "__main__": tester = FunctionTester() print("="*60) print("🤖 功能测试机器人启动") print("="*60) print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("="*60) # 执行测试 tester.test_login_flow() tester.test_dashboard_data() tester.test_inventory_operations() tester.test_order_operations() tester.test_user_permissions() tester.test_page_functionality() # 生成报告 all_passed = tester.generate_report() exit(0 if all_passed else 1)