资讯动态

Django+Vue自动化测试平台:Celery调度与WebSocket日志流实战

发布时间:2026/9/12 4:51:01 来源:尧图企业网站定制
简介这是一套基于Python3、Django后端与Vue前端构建的自动化测试平台完整开发资料面向计算机相关专业在校生、教师及初级开发人员适用于毕业设计、课程设计、项目实训与技术进阶学习。资源包含2000个文件主体为1694个Python源码涵盖Django模型、视图、API接口及测试用例、123个HTML模板含管理后台与测试任务页面、92个JavaScript脚本实现Vue组件交互与动态渲染辅以CSS样式文件如responsive.css、select2.min.css、widgets.css等保障界面响应式与控件功能整体压缩包仅21.18MB轻量易部署。已有64人下载学习资源源自高分结题项目答辩95分所有代码经实测可运行附带详细开发文档与模块说明覆盖平台架构设计、前后端联调要点、测试用例编写规范及常见问题排错路径特别适合从零理解全栈测试平台工程化落地逻辑。1. 这不是又一个“DjangoVue”Demo而是一个能跑通CI/CD流水线的自动化测试平台骨架当你在招聘JD里看到“熟悉基于DjangoVue的测试平台开发”或在团队技术复盘中听到“手工回归太慢得搭个自动化测试平台”多数人第一反应是翻GitHub找现成项目——结果下载下来发现Django只写了用户登录、Vue页面全是静态mock数据、连最基础的用例执行日志都存不进数据库。这个标题里的“全部资料详细文档.zip”核心价值不在代码量而在它把自动化测试平台的真实交付链路拆解成了可验证的模块Django后端要承接Selenium/Playwright执行器的异步任务调度与结果归档Vue前端必须支持用例分组筛选、实时日志流式渲染、失败截图点击查看且整套系统能在无图形界面的Linux服务器上稳定运行。它面向的是有实际测试资产如已有的Pytest用例集、需要统一纳管执行环境、并对接Jenkins/GitLab CI的QA工程师或DevOps角色而不是刚学完Vue路由的前端新手。2. Django后端用CeleryRedis构建可伸缩的测试任务调度中枢自动化测试平台的后端绝非传统CRUD Web服务其核心瓶颈在于并发任务调度与执行状态持久化。若直接用Django同步视图调用subprocess.run()执行测试脚本不仅阻塞HTTP请求、无法查看实时进度更会在多用户同时触发时导致进程资源争抢。因此必须引入异步任务队列。当前生产环境最成熟的选择仍是Celery Redis组合——它比Django-Q更易监控比RQ更适合复杂任务依赖且与Django ORM天然兼容。2.1 为什么选Celery而非Django-Q或APSchedulerDjango-Q虽轻量但其内置的SQLite存储在高并发下易出现锁表APScheduler本质是单机定时器无法跨Worker节点协调任务。而Celery通过Redis作为消息中间件天然支持水平扩展你可以在测试机集群中部署多个Celery Worker每个Worker监听同一队列Django主进程只需将任务发往Redis由Celery自动负载均衡分发。更重要的是Celery Result Backend如Redis或Django-DB能持久化任务状态使Vue前端可通过AJAX轮询获取PENDING/STARTED/SUCCESS/FAILURE等精确状态这是实现“实时日志流”的前提。2.2 配置Celery与Django深度集成的关键步骤首先在Django项目根目录创建celery.py注意与settings.py同级# celery.py import os from celery import Celery os.environ.setdefault(DJANGO_SETTINGS_MODULE, autotest_platform.settings) app Celery(autotest_platform) app.config_from_object(django.conf:settings, namespaceCELERY) app.autodiscover_tasks()接着在settings.py中添加Celery配置务必替换为你的Redis地址# settings.py CELERY_BROKER_URL redis://127.0.0.1:6379/0 # Redis作为消息代理 CELERY_RESULT_BACKEND redis://127.0.0.1:6379/1 # Redis作为结果后端 CELERY_TASK_SERIALIZER json CELERY_RESULT_SERIALIZER json CELERY_ACCEPT_CONTENT [json] CELERY_TIMEZONE Asia/Shanghai CELERY_ENABLE_UTC False CELERY_TASK_ROUTES { tasks.run_test_suite: {queue: test_queue}, }提示CELERY_RESULT_BACKEND必须与CELERY_BROKER_URL使用不同Redis DB如/0和/1否则任务状态可能被消息代理覆盖。生产环境建议用独立Redis实例避免与缓存共用。2.3 编写可被Vue调用的测试任务执行器在Django App如test_engine中创建tasks.py定义核心任务# test_engine/tasks.py import subprocess import json import os from celery import shared_task from django.conf import settings from .models import TestExecution, TestCase shared_task(bindTrue, max_retries3) def run_test_suite(self, execution_id, test_path, pytest_argsNone): 执行Pytest测试套件并实时写入执行日志 :param execution_id: TestExecution模型ID用于状态更新 :param test_path: 测试文件路径如 tests/api/test_login.py::test_valid_login :param pytest_args: 额外参数列表如 [-v, --tbshort] try: execution TestExecution.objects.get(idexecution_id) execution.status RUNNING execution.save() # 构建pytest命令 cmd [pytest, test_path] if pytest_args: cmd.extend(pytest_args) # 在虚拟环境中执行关键避免全局Python污染 venv_python os.path.join(settings.BASE_DIR, .venv, bin, python3) cmd [venv_python, -m, pytest] cmd[1:] # 实时捕获stdout/stderr并写入数据库 process subprocess.Popen( cmd, stdoutsubprocess.PIPE, stderrsubprocess.STDOUT, cwdsettings.BASE_DIR, textTrue, encodingutf-8 ) log_lines [] for line in iter(process.stdout.readline, ): log_lines.append(line.strip()) # 每10行批量更新一次日志字段避免高频DB写入 if len(log_lines) 10: execution.log \n.join(log_lines) \n execution.save() log_lines [] # 获取最终返回码 return_code process.wait() execution.return_code return_code execution.log \n.join(log_lines) execution.status SUCCESS if return_code 0 else FAILED execution.save() return {status: execution.status, return_code: return_code} except TestExecution.DoesNotExist: raise self.retry(countdown60, max_retries3) except Exception as exc: execution.status ERROR execution.log f\nException: {str(exc)} execution.save() raise self.retry(excexc, countdown60, max_retries3)2.3.1 关键参数说明与安全约束test_path必须校验合法性禁止传入../../../etc/passwd类路径遍历。应在视图层用os.path.abspath()与settings.TEST_ROOT比对确保路径在允许目录内。venv_python硬编码为.venv/bin/python3强制所有测试在隔离环境中运行。这解决了“只有python3没有python”导致的兼容性问题也规避了python3离线安装第三方库时的环境混乱。subprocess.Popen的cwd设为settings.BASE_DIR确保pytest能正确加载conftest.py及fixture。max_retries3防止因临时网络抖动导致任务永久失败重试间隔countdown60秒避免雪崩。2.4 创建REST API供Vue调用任务提交与状态查询使用Django REST Framework暴露两个端点# test_engine/views.py from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from .models import TestExecution from .serializers import TestExecutionSerializer, TestRunSerializer from .tasks import run_test_suite class TestRunView(APIView): def post(self, request): serializer TestRunSerializer(datarequest.data) if serializer.is_valid(): # 创建执行记录 execution TestExecution.objects.create( test_nameserializer.validated_data[test_name], test_pathserializer.validated_data[test_path], statusPENDING, created_byrequest.user ) # 异步触发Celery任务 task run_test_suite.delay( execution_idexecution.id, test_pathserializer.validated_data[test_path], pytest_argsserializer.validated_data.get(pytest_args, []) ) execution.celery_task_id task.id execution.save() return Response({execution_id: execution.id, task_id: task.id}, statusstatus.HTTP_202_ACCEPTED) return Response(serializer.errors, statusstatus.HTTP_400_BAD_REQUEST) class TestStatusView(APIView): def get(self, request, execution_id): try: execution TestExecution.objects.get(idexecution_id) # 从Celery获取最新状态比DB更准 from celery import current_app task current_app.AsyncResult(execution.celery_task_id) return Response({ status: task.status, result: task.result, log: execution.log[-5000:], # 返回最后5KB日志防爆内存 return_code: execution.return_code }) except TestExecution.DoesNotExist: return Response({error: Execution not found}, statusstatus.HTTP_404_NOT_FOUND)对应URL配置# urls.py from django.urls import path from . import views urlpatterns [ path(api/run/, views.TestRunView.as_view(), namerun-test), path(api/status/int:execution_id/, views.TestStatusView.as_view(), nametest-status), ]注意TestStatusView中task.status返回Celery原生状态PENDING/STARTED/SUCCESS而execution.status是Django模型字段。两者需保持同步——run_test_suite任务成功后会更新DB但若任务异常退出DB状态可能滞后。因此前端应优先信任Celery状态DB字段仅作持久化备份。3. Vue前端用WebSocket实现实时日志流与失败截图联动Vue前端若仅靠轮询/api/status/接口获取日志会产生大量无效HTTP请求尤其当测试运行超10分钟时。更优方案是Django Channels WebSocket让后端在run_test_suite执行过程中主动推送日志片段。这解决了“vue打包后布局异常”之外更本质的体验问题用户无需反复点击“刷新日志”失败瞬间就能看到截图缩略图。3.1 集成Django Channels处理WebSocket连接安装Channels并配置ASGIpip install channels redis# settings.py INSTALLED_APPS [channels] ASGI_APPLICATION autotest_platform.asgi.application CHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], }, }, }创建asgi.py替代默认wsgi.py# asgi.py import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from test_engine import routing os.environ.setdefault(DJANGO_SETTINGS_MODULE, autotest_platform.settings) application ProtocolTypeRouter({ http: get_asgi_application(), websocket: AuthMiddlewareStack( URLRouter([ path(ws/logs/int:execution_id/, routing.LogConsumer.as_asgi()), ]) ), })3.2 编写LogConsumer推送实时日志# test_engine/consumers.py import json from channels.generic.websocket import AsyncWebsocketConsumer from channels.db import database_sync_to_async from .models import TestExecution class LogConsumer(AsyncWebsocketConsumer): async def connect(self): self.execution_id self.scope[url_route][kwargs][execution_id] self.group_name flogs_{self.execution_id} await self.channel_layer.group_add( self.group_name, self.channel_name ) await self.accept() async def disconnect(self, close_code): await self.channel_layer.group_discard( self.group_name, self.channel_name ) async def receive(self, text_data): pass # 此Consumer只接收服务端推送不处理客户端消息 # 服务端调用此方法推送日志 async def send_log(self, event): await self.send(text_datajson.dumps({ type: log_line, content: event[content], timestamp: event[timestamp] }))在test_engine/routing.py中注册# routing.py from django.urls import re_path from . import consumers websocket_urlpatterns [ re_path(rws/logs/(?Pexecution_id\w)/$, consumers.LogConsumer.as_asgi()), ]3.3 Vue组件中建立WebSocket连接并渲染日志!-- components/TestRunner.vue -- template div classtest-runner div classlog-container reflogContainer pre classlog-output{{ logContent }}/pre div v-ifisRunning classlog-loading正在执行... {{ progress }}%/div /div div v-ifscreenshotUrl classscreenshot-preview h3失败截图/h3 img :srcscreenshotUrl altFailure screenshot clickopenFullscreen / /div /div /template script export default { props: [executionId], data() { return { logContent: , isRunning: true, progress: 0, screenshotUrl: , ws: null } }, mounted() { this.connectWebSocket(); }, beforeUnmount() { if (this.ws) this.ws.close(); }, methods: { connectWebSocket() { const wsUrl ws://${window.location.host}/ws/logs/${this.executionId}/; this.ws new WebSocket(wsUrl); this.ws.onopen () { console.log(WebSocket connected); }; this.ws.onmessage (event) { const data JSON.parse(event.data); if (data.type log_line) { this.logContent data.content \n; // 自动滚动到底部 this.$nextTick(() { const container this.$refs.logContainer; container.scrollTop container.scrollHeight; }); } else if (data.type screenshot) { this.screenshotUrl data.url; } }; this.ws.onerror (error) { console.error(WebSocket error:, error); }; }, openFullscreen() { const img new Image(); img.src this.screenshotUrl; document.body.appendChild(img); img.style.position fixed; img.style.top 50%; img.style.left 50%; img.style.transform translate(-50%, -50%); img.style.maxWidth 90vw; img.style.maxHeight 90vh; img.style.zIndex 9999; img.onclick () document.body.removeChild(img); } } } /script3.3.1 修改Celery任务以推送WebSocket消息在run_test_suite任务中当捕获到新日志行时调用Channels发送# test_engine/tasks.py续 from channels.layers import get_channel_layer from asgiref.sync import async_to_sync shared_task(bindTrue, max_retries3) def run_test_suite(self, execution_id, test_path, pytest_argsNone): # ... 前置逻辑不变 ... channel_layer get_channel_layer() for line in iter(process.stdout.readline, ): log_lines.append(line.strip()) # 推送单行日志到WebSocket async_to_sync(channel_layer.group_send)( flogs_{execution_id}, { type: send_log, content: line.strip(), timestamp: timezone.now().isoformat() } ) # ... 其余逻辑 ...提示async_to_sync是关键——Celery Worker运行在同步线程中必须将其转为同步调用才能与Channels交互。若忽略此转换会报RuntimeError: You cannot use AsyncToSync in the same thread as an async function。3.4 处理Vue环境配置与依赖安装常见陷阱根据热词vue安装依赖和vue安装及环境配置必须明确本平台要求Node.js ≥ 16.xVue 3推荐版本且package.json中需锁定关键依赖{ dependencies: { axios: ^1.6.0, element-plus: ^2.7.0, vue: ^3.4.0 }, devDependencies: { vitejs/plugin-vue: ^4.5.0, vite: ^5.0.0 } }执行npm install时若遇error: command [/opt/driver-monitor/.venv/bin/python3, -m, ensurepip类错误本质是Node.js尝试调用Python构建某些C扩展如node-sass已废弃但旧项目残留。解决方案是移除所有含原生编译的依赖改用纯JS替代品sass→sassDart Sass纯JS实现npm install -D sasscanvas→ 若测试报告需图表改用chart.js纯JSnpm uninstall node-sass npm install -D sass4. 生产部署宝塔面板一键部署DjangoVue分离架构的实操细节宝塔部署django是高频搜索词但多数教程止步于“上传代码、配置Python项目”忽略了前后端分离场景下的Nginx反向代理与静态资源路径冲突。本节给出在宝塔面板v8.0中部署该平台的完整路径覆盖linux 系统离线安装python3与django install mysqlclient等真实痛点。4.1 后端部署解决离线环境Python3与MySQL驱动安装若服务器无外网需提前准备离线包Python3.10源码包Python-3.10.12.tgzmysqlclientwheel包从https://pypi.org/project/mysqlclient/#files 下载对应平台的.whl如mysqlclient-2.2.4-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whlDjango、celery、channels等wheel包用pip download -d ./wheels --no-deps --platform manylinux_2_17_x86_64 --python-version 310 --only-binary:all: django celery channels生成在宝塔面板中进入“软件商店” → “Python项目” → 点击“添加项目”选择Python版本为3.10若未安装先在“运行环境”中安装Python3.10项目路径填/www/wwwroot/autotest-backend在“项目管理” → “依赖管理”中点击“上传本地包”选择所有.whl文件上传并安装关键配置在“项目设置”中将“启动文件”设为asgi.py“启动命令”填daphne -b 127.0.0.1:8000 autotest_platform.asgi:applicationDaphne是Channels推荐的ASGI服务器比Uvicorn更稳定。4.2 前端部署Nginx反向代理与静态资源路径修正Vue构建产物默认输出到dist/但Django的STATIC_ROOT通常设为/www/wwwroot/autotest-backend/static/。若将Vue dist直接放此处会导致Django Admin静态文件被覆盖。正确做法是前后端完全分离Vue构建输出到/www/wwwroot/autotest-frontend独立站点Nginx配置两套location# 宝塔面板 → 网站 → 设置 → 配置文件 upstream django_backend { server 127.0.0.1:8000; } server { listen 80; server_name test.yourdomain.com; # 前端静态资源 location / { root /www/wwwroot/autotest-frontend; try_files $uri $uri/ /index.html; } # API接口代理 location /api/ { proxy_pass http://django_backend/; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } # WebSocket代理关键否则日志不推送 location /ws/ { proxy_pass http://django_backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; } }4.3 数据库与Celery Worker守护进程配置在宝塔“数据库”中创建MySQL数据库autotest_db然后在Django项目中执行cd /www/wwwroot/autotest-backend source .venv/bin/activate python manage.py migrate python manage.py createsuperuserCelery Worker需常驻运行进入“宝塔” → “计划任务” → “添加Linux脚本”任务名称Start Celery Worker执行周期每分钟脚本内容#!/bin/bash cd /www/wwwroot/autotest-backend source .venv/bin/activate pgrep -f celery -A autotest_platform worker /dev/null || celery -A autotest_platform worker -l info -Q test_queue 验证是否生效执行ps aux | grep celery应看到类似/www/wwwroot/autotest-backend/.venv/bin/python3 /www/wwwroot/autotest-backend/.venv/bin/celery -A autotest_platform worker -l info -Q test_queue的进程。5. 故障排查从Django Admin界面美化到Vue路由参数失效的典型修复路径当平台部署后出现功能异常按以下顺序排查覆盖django admin界面美化、vue路由参数、django执行查询-删除对象等高频问题。5.1 Django Admin界面无法加载静态文件CSS/JS空白现象登录Admin后页面无样式浏览器Console报404admin/css/base.css。原因Django未收集静态文件或Nginx未配置静态路径。修复步骤在Django项目根目录执行python manage.py collectstatic --noinput确认settings.py中STATIC_ROOT /www/wwwroot/autotest-backend/static/在Nginx配置中添加静态文件location在server块内location /static/ { alias /www/wwwroot/autotest-backend/static/; }5.2 Vue路由参数丢失导致用例详情页404现象点击用例列表项跳转/case/123页面空白Vue Router报No match found for location with path /case/123。原因Nginx未配置try_files兜底导致HTML5 History模式下子路由被当作真实路径请求。修复步骤确认Vue Router使用history模式createWebHistory()在Nginx的location /块中try_files指令必须包含/index.htmllocation / { root /www/wwwroot/autotest-frontend; try_files $uri $uri/ /index.html; # 关键缺此行则子路由404 }5.3 Django执行delete()后数据库未生效但日志显示成功现象在Admin中删除测试执行记录页面提示成功但数据库表中记录仍在。原因Django默认开启事务若删除操作在事务中未提交或模型设置了on_deletemodels.SET_NULL但外键字段允许NULL。诊断命令# 进入Django shell python manage.py shell from test_engine.models import TestExecution obj TestExecution.objects.get(id123) obj.delete() # 观察返回值(1, {test_engine.TestExecution: 1}) 表示删除成功1条 TestExecution.objects.filter(id123).exists() # 应返回False若仍存在检查模型定义# models.py class TestExecution(models.Model): # 错误示例未设on_delete # test_case models.ForeignKey(TestCase) # 正确写法 test_case models.ForeignKey(TestCase, on_deletemodels.CASCADE) # 级联删除5.4 Celery Worker日志显示Connection refused连接Redis失败现象Worker进程启动即退出日志报redis.exceptions.ConnectionError: Error 111 connecting to 127.0.0.1:6379.原因Redis服务未运行或防火墙阻止6379端口。快速验证# 检查Redis状态 systemctl status redis # 若未运行启动 systemctl start redis # 检查端口监听 netstat -tuln | grep 6379 # 若无输出修改Redis配置 vim /etc/redis/redis.conf # 确保 bind 127.0.0.1非0.0.0.0安全起见 # 确保 protected-mode no仅内网环境可关最后执行systemctl restart redis并重启Celery Worker。终极技巧当所有配置看似正确却仍失败时在Django Shell中手动触发任务绕过HTTP层直查Celery from test_engine.tasks import run_test_suite result run_test_suite.apply_async(args[1, tests/demo.py], countdown1) result.get(timeout10) # 若此处卡住问题必在Celery或Redis本文还有配套的精品资源点击获取

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价