一人控制16个工站做测试：主从架构自动化测试辅助系统实战

一人控制16个工站做测试主从架构自动化测试辅助系统实战一、项目背景在制造业产线测试场景中经常需要同时控制多个测试工站如16个执行相同或不同的测试任务。人工逐个操作不仅效率低下还容易出现人为失误。如何实现一人管控多工站、自动化分配任务、实时监控状态成为了待解决的问题。核心痛点人工操作效率低16个工站逐个配置测试参数耗时耗力状态监控困难无法实时掌握每个工站的测试进度和结果数据管理混乱测试数据分散在各个工站难以集中管理任务分配无序缺乏统一的任务调度机制资源利用率低最痛点科室人手不够拿我开刀了。只能开始我的拿手好戏。二、技术方案设计2.1 架构选型经过研究我选择了**主从架构Master-Slave**作为解决方案┌─────────────────────────────────────────────────────────────┐ │ 主节点 (Master) │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 任务调度器 │ │ 状态监控 │ │ 数据中心 │ │ │ │ Task Scheduler│ │ Status Monitor│ │ Data Center │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ └─────────┼─────────────────┼─────────────────┼──────────────┘ │ │ │ │ HTTP API │ HTTP API │ HTTP API ▼ ▼ ▼ ┌─────────┴─────────────────┴─────────────────┴──────────────┐ │ 局域网 (LAN) │ └─────────┬─────────────────┬─────────────────┬──────────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ 从节点 01 │ │ 从节点 02 │ │ ... │ │ Slave_001 │ │ Slave_002 │ │ Slave_016 │ │ ┌───────────┐ │ │ ┌───────────┐ │ │ ┌───────────┐ │ │ │测试执行器│ │ │ │测试执行器│ │ │ │测试执行器│ │ │ │Test Exec │ │ │ │Test Exec │ │ │ │Test Exec │ │ │ └───────────┘ │ │ └───────────┘ │ │ └───────────┘ │ └─────────────────┘ └─────────────────┘ └─────────────────┘2.2 技术栈选择组件技术选型理由语言Python 3.x语法简洁生态丰富跨平台Web框架FastAPI高性能异步框架自动生成API文档网络通信HTTP/REST内网环境稳定实现简单Excel处理openpyxl支持xlsx格式读写方便并发处理threading轻量级线程适合IO密集型场景2.3 核心设计思想2.3.1 从节点主动轮询模式从节点采用主动轮询方式与主节点通信而非主节点主动推送优点无需配置端口映射网络穿透简单流程从节点定时发送心跳 → 请求任务 → 执行任务 → 回传数据2.3.2 流式文件传输针对测试数据量大的特点采用流式分块传输defstream_file_generator(file_path,chunk_size4194304):文件分块生成器每次只读4MB防止内存爆掉withopen(file_path,rb)asf:whileTrue:chunkf.read(chunk_size)ifnotchunk:breakyieldchunk优势内存占用恒定在几MB支持任意大小文件传输。三、项目开发历程3.1 需求分析与功能规划根据业务需求我规划了以下核心功能主节点功能监测在线从节点数量和状态记录已完成任务节点名称、时间、数据目录从Excel读取任务配置自动分配给空闲从节点实时显示节点状态和任务分配信息从节点功能接收主节点发送的任务和文件将文件移动到指定位置先清空目录执行测试任务测试完成后将测试数据回传至主节点3.2 配置文件设计3.2.1 config.xlsx - 任务配置表列名说明示例testlist_name测试文件名含扩展名test_case_001.txttestlist_path源文件目录路径./testlistsauto_test目标文件目录路径./auto_test备注任务状态空/已完成/执行失败设计决策将路径和文件名分离提高配置灵活性。3.2.2 config.txt - 节点配置主节点配置role0 port8000从节点配置role1 master_ip192.168.1.100 master_port8000 slave_idslave_0013.3 API接口设计接口方法功能/api/heartbeatGET从节点心跳检测/api/get_slavesGET获取所有节点状态/api/get_taskGET从节点请求任务/api/download_fileGET从节点下载测试文件/api/upload_stream_filePOST从节点上传测试数据/api/task_completePOST任务完成通知/api/task_failedPOST任务失败通知四、遇到的问题与解决方案4.1 Bug 1任务名称传递错误导致状态更新失败问题描述从节点完成任务后主节点无法正确更新config.xlsx中的状态。根因分析# 错误代码 - 发送了task_id而非task_nameresponserequests.post(f{master_url}/api/task_complete,headers{task-name:task_id,# ❌ 错误应该是task_name...})解决方案修改return_test_data函数传递正确的任务名称defreturn_test_data(task_id,task_name,local_result_dir):# ...responserequests.post(f{master_url}/api/task_complete,headers{task-name:task_name,# ✅ 正确...})4.2 Bug 2从节点掉线后无法检测问题描述从节点断开连接后主节点仍显示为在线状态。根因分析原代码在get_slaves()接口中直接删除离线节点无法追踪历史连接记录。解决方案保留所有节点记录通过心跳时间判断状态master_app.get(/api/get_slaves)defget_slaves():nowdatetime.now()offline_threshold30# 30秒无心跳判定为离线forslave_id,infoinconnected_slaves.items():last_seendatetime.fromisoformat(info[last_seen])if(now-last_seen).secondsoffline_threshold:connected_slaves[slave_id][status]offlineelse:connected_slaves[slave_id][status]online# 返回详细状态信息return{online_slaves:online_slaves,offline_slaves:offline_slaves,...}4.3 Bug 3路径遍历安全漏洞问题描述/api/download_file接口未验证路径合法性存在路径遍历风险。根因分析# 错误代码 - 直接使用用户输入的路径defdownload_file(file_path:str):returnFileResponse(file_path)# ❌ 可被利用访问任意文件解决方案添加路径安全验证defdownload_file(file_path:str):safe_pathos.path.abspath(file_path)ifnotsafe_path.startswith(os.path.abspath(.)):raiseHTTPException(status_code403,detail文件路径不合法)ifnotos.path.exists(safe_path):raiseHTTPException(status_code404,detail文件不存在)returnFileResponse(safe_path)4.4 问题主节点无法实时显示状态问题描述主节点启动后被uvicorn阻塞无法输出状态信息。解决方案使用后台线程定时显示状态defmaster_display_loop():whileTrue:time.sleep(10)# 打印节点状态和任务信息print(\n*60)print(f[Master] 节点状态监控 -{now.strftime(%Y-%m-%d %H:%M:%S)})# ...defrun_as_master():display_threadthreading.Thread(targetmaster_display_loop,daemonTrue)display_thread.start()uvicorn.run(master_app,host0.0.0.0,portMASTER_PORT)4.5 问题任务路径格式设计优化问题描述最初设计中testlist_path和auto_test存储完整文件路径不够灵活。优化方案将路径和文件名分离主节点自动组合testlist_dirtask[testlist_path]# ./testliststestlist_filenametask[testlist_name]# test_case_001.txtfull_testlist_pathos.path.join(testlist_dir,testlist_filename)# ./testlists/test_case_001.txt五、核心代码实现5.1 主节点任务分配逻辑master_app.get(/api/get_task)defget_task(slave_id:strHeader(...)):# 检查是否有正在执行的任务ifslave_idinassigned_tasksandassigned_tasks[slave_id][status]in_progress:return{has_task:False,message:已有任务在执行中}# 获取待执行任务pending_tasksget_pending_tasks()ifnotpending_tasks:return{has_task:False,message:暂无待执行任务}taskpending_tasks[0]task_idftask_{int(time.time())}# 组合完整路径full_testlist_pathos.path.join(task[testlist_path],task[testlist_name])full_auto_test_pathos.path.join(task[auto_test],task[testlist_name])# 记录任务分配信息assigned_tasks[slave_id]{task_id:task_id,task_name:task[testlist_name],testlist_path:full_testlist_path,auto_test:full_auto_test_path,status:assigned}# 输出任务分配日志print(f\n[Master] 任务分配:)print(f 从节点:{slave_id})print(f 任务ID:{task_id})print(f 任务名称:{task[testlist_name]})return{has_task:True,task:{task_id:task_id,task_name:task[testlist_name],testlist_path:full_testlist_path,auto_test:full_auto_test_path}}5.2 从节点任务执行流程defrun_as_slave():whileTrue:try:# 1. 发送心跳requests.get(f{MASTER_URL}/api/heartbeat,headers{slave-id:SLAVE_ID})# 2. 请求任务responserequests.get(f{MASTER_URL}/api/get_task,headers{slave-id:SLAVE_ID})res_dataresponse.json()ifres_data.get(has_task):taskres_data[task]# 3. 清空目标目录local_testlist_diros.path.dirname(task[auto_test])# ... 删除目录内容# 4. 下载测试文件download_file_from_master(task[testlist_path],task[auto_test])# 5. 执行测试test_successexecute_test(task)# 6. 回传数据iftest_success:return_test_data(task[task_id],task[task_name],local_result_dir)else:# 通知任务失败requests.post(f{MASTER_URL}/api/task_failed,...)exceptrequests.exceptions.ConnectionError:print([Slave] 无法连接到主节点等待重试...)time.sleep(5)六、项目结构master_slave_auto/ ├── app.py # 主从节点统一程序 ├── config.txt # 节点配置文件 ├── config.xlsx # 任务配置表 ├── task_records.json # 任务记录文件 ├── testlists/ # 测试列表文件目录 │ ├── test_case_001.txt │ ├── test_case_002.txt │ └── test_case_003.txt └── auto_test/ # 测试执行目录 └── run_test.sh七、使用说明7.1 主节点部署修改config.txtrole0 port8000启动程序python3 app.py7.2 从节点部署修改config.txtrole1 master_ip192.168.1.100 master_port8000 slave_idslave_001启动程序python3 app.py7.3 配置任务在config.xlsx中添加测试任务testlist_nametestlist_pathauto_test备注test_case_001.txt./testlists./auto_testtest_case_002.txt./testlists./auto_test八、运行效果8.1 主节点监控界面 [Master] 节点状态监控 - 2026-07-10 15:35:00 [在线节点] (3个): ✓ slave_001 - 最后活跃: 2026-07-10T15:34:55 ✓ slave_002 - 最后活跃: 2026-07-10T15:34:54 ✓ slave_003 - 最后活跃: 2026-07-10T15:34:53 [离线节点] (1个): ✗ slave_004 - 最后活跃: 2026-07-10T15:30:00 ⚠️ 警告: 该节点已掉线请管理员检查 [进行中的任务] (2个): slave_001 - test_case_001.txt (assigned) slave_002 - test_case_002.txt (in_progress) 8.2 任务分配日志[Master] 任务分配: 从节点: slave_001 任务ID: task_1783668826 任务名称: test_case_001.txt 源文件路径: ./testlists/test_case_001.txt 目标路径: ./auto_test/test_case_001.txt 分配时间: 2026-07-10 15:33:46九、总结与展望9.1 项目总结架构简洁单文件实现主从逻辑部署方便实时监控后台线程定时显示节点状态和任务信息流式传输支持大文件传输内存占用低安全可靠路径验证、任务失败处理、节点掉线检测9.2 待优化项任务优先级当前采用FIFO策略可增加优先级调度负载均衡根据从节点负载智能分配任务Web管理界面提供可视化管理界面9.3 扩展Docker化使用容器化部署简化环境配置消息队列引入RabbitMQ/Kafka实现异步任务调度技术交流欢迎留言讨论共同优化改进