diff --git a/Autotestplat/celery.py b/Autotestplat/celery.py index 17c581b7..c95fa578 100644 --- a/Autotestplat/celery.py +++ b/Autotestplat/celery.py @@ -10,7 +10,7 @@ from django.conf import settings os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Autotestplat.settings') -django.setup() + app = Celery('Autotestplat') @@ -20,6 +20,8 @@ app.conf.enable_utc = True app.conf.timezone = "Asia/Shanghai" +django.setup() + @app.task(bind=True) def debug_task(self): print('Request: {0!r}'.format(self.request)) \ No newline at end of file diff --git a/Autotestplat/settings.py b/Autotestplat/settings.py index 7e962b67..d70c7ec5 100644 --- a/Autotestplat/settings.py +++ b/Autotestplat/settings.py @@ -73,7 +73,6 @@ WSGI_APPLICATION = 'Autotestplat.wsgi.application' - DATABASES = { 'default': { 'ENGINE': 'django.db.backends.mysql', @@ -81,7 +80,12 @@ 'HOST': '127.0.0.1', 'PORT': 3306, 'USER': 'root', - 'PASSWORD': 'test123456'} + 'PASSWORD': 'Liuzhichao@123', + 'OPTIONS': { + 'charset': 'utf8mb4', + 'init_command': "SET sql_mode='STRICT_TRANS_TABLES'", + }, + } } AUTH_PASSWORD_VALIDATORS = [ @@ -104,14 +108,14 @@ TIME_ZONE = 'Asia/Shanghai' USE_I18N = True USE_L10N = True -USE_TZ = True +USE_TZ = False STATIC_URL = '/static/' STATICFILES_DIRS = [ - os.path.join(BASE_DIR, "autotest").join("static") + os.path.join(BASE_DIR, "autotest", "static") ] - +STATIC_ROOT = os.path.join(BASE_DIR, 'staticfiles') MEDIA_URL = "/upload/" MEDIA_ROOT = os.path.join(BASE_DIR,'upload') @@ -125,9 +129,18 @@ CUBES_REDIS_TIMEOUT=60*60 NEVER_REDIS_TIMEOUT=365*24*60*60 - +# 根据 settings.py 配置，项目使用了 3 个 Redis 数据库： +# Redis 缓存（CACHES） +# 地址：redis://127.0.0.1:6379/1 +# 用途：Django 缓存后端 +# Celery Broker +# 地址：redis://127.0.0.1:6379/0 +# 用途：Celery 任务队列消息代理 +# Celery 结果后端 +# 地址：redis://127.0.0.1:6379/1 +# 用途：存储 Celery 任务执行结果 CELERY_TIMEZONE = 'Asia/Shanghai' -CELERY_ENABLE_UTC=True +CELERY_ENABLE_UTC=False # BROKER_URL= 'amqp://root:root@127.0.0.1:5672/of1' #rabbitmq BROKER_URL = 'redis://127.0.0.1:6379/0' BROKER_TRANSPORT = 'redis' @@ -135,4 +148,16 @@ CELERY_ACCEPT_CONTENT = ['json'] CELERY_TASK_SERIALIZER = 'json' CELERY_RESULT_SERIALIZER = 'json' -CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' \ No newline at end of file +CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' + +# 设置 Celery Beat 自动检测数据库变化的间隔（秒） +# 默认是 5 分钟，设置为 10 秒可以更快生效 +# CELERYBEAT_MAX_LOOP_INTERVAL = 10 + +LOGIN_URL = '/autotest/login/' +LOGIN_REDIRECT_URL = '/autotest/index/' +LOGOUT_REDIRECT_URL = '/autotest/login/' + +SESSION_COOKIE_AGE = 3600 # 1 小时（60 * 60 = 3600 秒） +SESSION_EXPIRE_AT_BROWSER_CLOSE = True # 关闭浏览器后失效 +SESSION_SAVE_EVERY_REQUEST = False # 不刷新过期时间（固定窗口） \ No newline at end of file diff --git a/Redis-x64-3.2.100/dump.rdb b/Redis-x64-3.2.100/dump.rdb index cda7b72c..694d8464 100644 Binary files a/Redis-x64-3.2.100/dump.rdb and b/Redis-x64-3.2.100/dump.rdb differ diff --git a/autotest/models.py b/autotest/models.py index fac32d19..464d04c8 100644 --- a/autotest/models.py +++ b/autotest/models.py @@ -216,15 +216,15 @@ class Meta: db_table = 'autotestplat_testplan_interface' class AutotestplatTestplanInterfaceResult(models.Model): - report_id = models.CharField(max_length=20,null=True) + report_id = models.CharField(max_length=50,null=True) product_id = models.IntegerField(null=True) product_name = models.CharField(max_length=255, blank=True, null=True) suit_id = models.IntegerField(null=True) suit_name = models.CharField(max_length=100, null=True) suit_interface_id = models.IntegerField(null=True) interface_name = models.CharField(max_length=200,null=True) - url = models.CharField(max_length=100,null=True) - body = models.CharField(max_length=200,null=True) + url = models.CharField(max_length=500,null=True) + body = models.TextField(null=True, blank=True) mode = models.CharField(max_length=20,null=True) assert_keywords_old = models.CharField(max_length=20,null=True) task_mode = models.CharField(max_length=20,null=True) @@ -232,7 +232,7 @@ class AutotestplatTestplanInterfaceResult(models.Model): response_time = models.TextField(null=True) result = models.CharField(max_length=20,null=True) pass_pers = models.CharField(max_length=20, null=True) - date_time = models.CharField(max_length=20,null=True) + date_time = models.CharField(max_length=50,null=True) class Meta: managed = True diff --git a/autotest/static/js/performance.js b/autotest/static/js/performance.js index faf91dd6..7b91458c 100644 --- a/autotest/static/js/performance.js +++ b/autotest/static/js/performance.js @@ -1,4 +1,5 @@ setProgressURL = 'apiperformance/progress/' +checkStatusURL = 'apiperformance/status/' function search_performance_interface() { @@ -145,14 +146,14 @@ function search_performance_interface() { return; } else if(num_bf == ''){ - alert('请输入“并发数”') + alert('请输入"并发数"') return; } else if(num_sj == ''){ - alert('请输入“持续运行时间”') + alert('请输入"持续运行时间"') return; } - + $('.jmeter_status').empty().append('状态：准备执行'); $.ajax({ url: "/autotest/apiperformance/prepare/", @@ -163,34 +164,82 @@ function search_performance_interface() { success: function(result) { console.log(result); if(result == 'success'){ - + $('.jmeter_status').empty().append('') var progressBar = document.getElementById('progressBar'); - var duration = document.getElementById('num_sj').value; - var increment = parseInt(duration) * 60; + var duration = parseInt(document.getElementById('num_sj').value); + var increment = duration * 60; progressBar.max = increment; - let startTime = new Date(); - - var intervalId = setInterval(function() { - let now = new Date(); - let elapsed = now - startTime; - let seconds = Math.floor(elapsed / 1000); - progressBar.value = seconds % 60; - $.ajax({ - url: appURL+setProgressURL, - type: 'POST', - data: JSON.stringify({ - progress: progressBar.value, - progress_total: increment, - }), - success: (rst) => { - console.log("success") + + // 【关键修改】：启动 JMeter 并立即开始轮询真实状态 + $.ajax({ + url: "/autotest/apiperformance/start/", + headers:{'X-CSRFToken': '{{ csrf_token }}'}, + contentType: 'application/json;charset=utf-8', + type: "POST", + traditional: true, + success: function(result) { + if(result == 'success'){ + $('.jmeter_status').empty().append('状态：测试进行中'); + + // 启动状态轮询 + var statusCheckInterval = setInterval(function() { + $.ajax({ + url: appURL + checkStatusURL, + type: 'GET', + success: function(statusData) { + if(statusData.running) { + // JMeter 仍在运行，更新进度条 + progressBar.value = statusData.progress; + + // 更新数据库中的进度 + $.ajax({ + url: appURL + setProgressURL, + type: 'POST', + data: JSON.stringify({ + progress: statusData.progress, + progress_total: increment, + }), + success: (rst) => { + console.log("进度更新成功") + } + }); + } else { + // JMeter 已结束 + clearInterval(statusCheckInterval); + $('.jmeter_status').empty().append('状态：测试完成'); + progressBar.value = increment; + + // 更新最终状态 + $.ajax({ + url: appURL + setProgressURL, + type: 'POST', + data: JSON.stringify({ + progress: increment, + progress_total: increment, + }), + success: (rst) => { + console.log("最终状态更新成功") + } + }); + } + }, + error: function(err) { + console.error("状态检查失败:", err); + clearInterval(statusCheckInterval); + $('.jmeter_status').empty().append('状态：状态检查异常'); + } + }); + }, 500); // 每2秒轮询一次 } - }); - if (progressBar.value === progressBar.max-1) { - clearInterval(intervalId); - } - }, 100); + else { + $('.jmeter_status').empty().append('状态：启动失败'); + } + }, + fail: function(result) { + debugger + } + }); } else { $('.jmeter_status').empty().append('状态：出现异常'); @@ -201,49 +250,26 @@ function search_performance_interface() { debugger } }); - - $.ajax({ - url: "/autotest/apiperformance/start/", - headers:{'X-CSRFToken': '{{ csrf_token }}'}, - contentType: 'application/json;charset=utf-8', - type: "POST", - traditional: true, - success: function(result) { - console.log(result); - if(result == 'success'){ - $('.jmeter_status').empty().append('状态：测试完成'); - $.ajax({ - url: appURL+setProgressURL, - type: 'POST', - data: JSON.stringify({ - progress: 0, - progress_total: 1, - }), - success: (rst) => { - console.log("success") - } - }); - } - else { - $('.jmeter_status').empty().append('状态：测试失败'); - } - }, - fail: function(result) { - debugger - } - }); } function select_login(id1,name1,module1) { $('#login_panel').append(''+name1+''); + +} + +function view_test_result() { + window.open('/autotest/apiperformance/report/', '_blank'); } + function select_not_login(id1,name1,module1) { $('#not_login_panel').append(''+name1+''); } + function remove_login_selected(ele) { var id1 = '#'+ele; $('#login_panel').find(id1).remove(); } + function remove_not_login_selected(ele) { var id1 = '#'+ele; $('#not_login_panel').find(id1).remove(); diff --git a/autotest/static/js/webtestcase.js b/autotest/static/js/webtestcase.js index 6dad5fff..890eb440 100644 --- a/autotest/static/js/webtestcase.js +++ b/autotest/static/js/webtestcase.js @@ -563,7 +563,13 @@ function run_webtestcase_bycode(e,tips="运行成功"){ $("#addModal").modal('hide') } else{ - return alert(rst) + // 解析 JSON 响应 + var response = typeof rst === 'string' ? JSON.parse(rst) : rst; + if(response.message) { + alert(response.message); + } else { + alert('运行完成。该用例为：' + (response.web_testcase_code || web_testcase_code)); + } } }, error: (rst) =>{ @@ -585,7 +591,13 @@ function run_webtestcase_byproduct(e,tips="运行成功"){ $("#addModal").modal('hide') } else{ - return alert(rst) + // 解析 JSON 响应 + var response = typeof rst === 'string' ? JSON.parse(rst) : rst; + if(response.message) { + alert(response.message); + } else { + alert('运行完成'); + } } }, error: (rst) =>{ diff --git a/autotest/tasks.py b/autotest/tasks.py index 85be720f..56c2e0f4 100644 --- a/autotest/tasks.py +++ b/autotest/tasks.py @@ -1,4 +1,4 @@ -# -*- coding:utf-8 -*- +# -*- coding:utf-8 -*- from Autotestplat.celery import app @app.task(bind=True) diff --git a/autotest/templates/interface_performance.html b/autotest/templates/interface_performance.html index 704c7613..3ffe67da 100644 --- a/autotest/templates/interface_performance.html +++ b/autotest/templates/interface_performance.html @@ -1,5 +1,4 @@ - - + + + + + + + + + +
+ + 快运交接单服务系统 +
+ +
+
+
+
+
+
+

登录：

+ + + +
+ +
+ +
+
+
+ +
+
+ 上海乾臻信息科技有限公司 版权所有 +
+ + + + + + + + + + + + + + +【响应时间】：0.666862 秒【自动保存Cookie】：已从响应头捕获并保存 1 个Cookie到Redis + +【自动保存Cookie】：已更新 ohmsSessionID +【自动保存Cookie】：已更新 ohmsSessionID +【关联参数】： {ohmsSessionID} 在响应数据中未匹配到，请检测前置接口关键字配置，以及Redis是否已启动 +【自动保存Cookie】：已动态更新 Cookie_ohmsSessionID = 8004dd1d-8047-4d20-b... +【自动保存Cookie】：已动态更新 Cookie_ohmsSessionID = d1f72227-2d14-4711-b... + +【断言】： 【测试结果】： 测试通过 diff --git a/autotest/urls.py b/autotest/urls.py index cf2f78ba..b41783eb 100644 --- a/autotest/urls.py +++ b/autotest/urls.py @@ -3,7 +3,9 @@ #Auther:：Fin #Version：Autotestplat-V4.0 ############################################ -from django.urls import path +from django.urls import path, re_path # 新增 re_path +from django.views.static import serve # 新增 serve +import os # 新增 os from . import views, views_index from . import views_user from . import views_product @@ -15,6 +17,10 @@ from . import views_apptestcase from . import views_webtestcase +# 定义报告目录路径 (BASE_DIR 是 Autotestplat 文件夹) +BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +REPORT_DIR = os.path.join(BASE_DIR, 'autotest', 'static', 'output') + urlpatterns = [ path('index/', views_index.indexView, name='index'), path('user/', views_user.userView, name='user'), @@ -82,6 +88,7 @@ path('apiperformance/prepare/', views_performance.prepareJmeter), path('apiperformance/start/', views_performance.startTestJmeter), path('apiperformance/progress/', views_performance.showProgress), + path('apiperformance/status/', views_performance.getJmeterStatus), path('apptestcase/', views_apptestcase.getAppView, name='apptestcase'), path('apptestcase/getSearchSelect/', views_apptestcase.loadAppOptions), path('apptestcase/getTableData/', views_apptestcase.loadAppTestcaseTable), @@ -105,4 +112,12 @@ path('webtestcase/copyWebtestcase/', views_webtestcase.copyWebtestcase), path('webtestcase/run_webtestcase/', views_webtestcase.runWebtestcase), path('webtestcase/runAllTestcase/', views_webtestcase.runAllTestcase), + + # === 新增这两行路由 === + # 1. 处理主页请求：调用 Python 函数读取 index.html + path('apiperformance/report/', views_performance.report), + + # 2. 处理静态资源请求：使用 serve 视图加载 css/js/images + # 注意：这里用 .+ 而不是 .*，确保不匹配空路径（即不匹配 /report/ 本身） + re_path(r'^apiperformance/report/(?P.+)$', serve, {'document_root': REPORT_DIR}), ] \ No newline at end of file diff --git a/autotest/views_apptestcase.py b/autotest/views_apptestcase.py index 48d67532..6c30d7ec 100644 --- a/autotest/views_apptestcase.py +++ b/autotest/views_apptestcase.py @@ -6,7 +6,7 @@ import time,json,pymysql from datetime import datetime from django.views.decorators.csrf import csrf_exempt -from django.shortcuts import render +from django.shortcuts import render, redirect from django.http import JsonResponse, HttpResponse from django.db.models import Count from django.db.models import Q @@ -23,14 +23,46 @@ appium_version=4 +# def getAppView(request): +# user_name = request.session.get('user', '') +# product_all = AutotestplatProduct.objects.filter(delete_flag='N') +# product_id = AuthUser.objects.filter(username=user_name).first().last_name +# product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name +# c = csrf(request) +# c.update({"product_name":product_name,"product_alls":product_all}) +# return render_to_response("app_testcase.html",c) + def getAppView(request): user_name = request.session.get('user', '') + + # 检查用户是否登录 + if not user_name: + # 用户未登录，重定向到登录页面或返回错误 + return redirect('/autotest/login/') + + # 获取用户对象并进行空值检查 + user = AuthUser.objects.filter(username=user_name).first() + if user is None: + # 用户不存在，返回错误 + return HttpResponse('用户不存在') + product_all = AutotestplatProduct.objects.filter(delete_flag='N') - product_id = AuthUser.objects.filter(username=user_name).first().last_name - product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name + product_id = user.last_name + + # 检查 product_id 是否存在 + if not product_id: + # 用户没有关联产品，可能需要处理 + product_name = "未关联产品" + else: + product = AutotestplatProduct.objects.filter(id=product_id).first() + if product is None: + product_name = "产品不存在" + else: + product_name = product.product_name + c = csrf(request) - c.update({"product_name":product_name,"product_alls":product_all}) - return render_to_response("app_testcase.html",c) + c.update({"product_name": product_name, "product_alls": product_all}) + return render_to_response("app_testcase.html", c) @csrf_exempt def loadAppTestcaseTable(request): diff --git a/autotest/views_interface.py b/autotest/views_interface.py index 6c38283c..3a5c334d 100644 --- a/autotest/views_interface.py +++ b/autotest/views_interface.py @@ -8,12 +8,12 @@ from django.shortcuts import render_to_response from django.db import connection from django.db.models import Q -from django.db.models import Max +from django.db.models import Max,Count from django.core.cache import cache from django.template.context_processors import csrf from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger from .models import * -from datetime import datetime +from datetime import datetime,timedelta from django.db.models import Count from django.http import HttpResponse,JsonResponse from django.contrib.auth.decorators import login_required @@ -23,6 +23,7 @@ import requests import hashlib,string + current_dir = os.getcwd() logfile = os.path.join(current_dir, 'autotest', 'test_out.log') codefile= os.path.join(current_dir, 'autotest', 'code.jpg') @@ -68,6 +69,7 @@ def getApiView(req): except EmptyPage: page_list = paginator.page(paginator.num_pages) c = csrf(req) + # print(f"DEBUG: product_all query result: {list(product_all.values('id', 'product_name'))}") c.update({'page_list': page_list, 'interfaces': interfaces, 'type': type,'num':num,"product_name":product_name,"product_alls":product_all}) return render_to_response("interface_testcase.html",c) @@ -79,10 +81,10 @@ def loadApiTestcaseTable(request,menu_module_id): items = AutotestplatInterfaceTestcase.objects.all().values_list('id','name','mode','url', 'charger','product_id','menu_module_id','create_time').annotate(Count('id')).order_by('-id') else: user_product_id = AuthUser.objects.filter(username=username).first().last_name - if user_product_id: - items = AutotestplatInterfaceTestcase.objects.filter(Q(product_id=user_product_id)).values_list('id','name','mode','url', 'charger','product_id','menu_module_id','create_time').annotate(Count('id')).order_by('-id') - else: - items = AutotestplatInterfaceTestcase.objects.all().values_list('id','name','mode','url', 'charger','product_id','menu_module_id').annotate(Count('id')).order_by('-id') + if user_product_id: + items = AutotestplatInterfaceTestcase.objects.filter(Q(product_id=user_product_id)).values_list('id','name','mode','url', 'charger','product_id','menu_module_id','create_time').annotate(Count('id')).order_by('-id') + else: + items = AutotestplatInterfaceTestcase.objects.all().values_list('id','name','mode','url', 'charger','product_id','menu_module_id').annotate(Count('id')).order_by('-id') else: if AuthUser.objects.filter(username=username).first().is_superuser == 1: items = AutotestplatInterfaceTestcase.objects.filter(menu_module_id=menu_module_id).values_list('id','name','mode','url', 'charger','product_id','menu_module_id','create_time').annotate(Count('id')).order_by('-id') @@ -142,11 +144,26 @@ def addModule(request): module_name=request.POST.get('module_name') parent_module_id = request.POST.get('parent_module_id') type = 'api' + # if parent_module_id: + # parent_module_id=AutotestplatModule.objects.filter(module_name=parent_module_id).first().id + # else: + # parent_module_id = '0' + # AutotestplatModule.objects.create(module_name=module_name,parent_module_id=parent_module_id,product_id=product_id,charger=username,type=type) + # return HttpResponse('200') if parent_module_id: - parent_module_id=AutotestplatModule.objects.filter(module_name=parent_module_id).first().id + if product_id: + parent_module = AutotestplatModule.objects.filter(module_name=parent_module_id, + product_id=product_id).first() + else: + parent_module = AutotestplatModule.objects.filter(module_name=parent_module_id).first() + if parent_module: + parent_module_id = parent_module.id + else: + return HttpResponse(f'父模块"{parent_module_id}"不存在，请重试') else: parent_module_id = '0' - AutotestplatModule.objects.create(module_name=module_name,parent_module_id=parent_module_id,product_id=product_id,charger=username,type=type) + AutotestplatModule.objects.create(module_name=module_name, parent_module_id=parent_module_id, product_id=product_id, + charger=username, type=type) return HttpResponse('200') @@ -157,11 +174,27 @@ def modModule(request): return HttpResponse('200') +# def deleteModule(request): +# module_id = request.POST.get('id') +# module_name = request.POST.get('module_name') +# child_module=AutotestplatModule.objects.filter(parent_module_id=module_id).first() +# module_case = AutotestplatInterfaceTestcase.objects.filter(menu_module_id=module_id).first() +# if child_module: +# return HttpResponse(f'模块"{module_name}"中存在子模块，不能删除') +# elif module_case: +# return HttpResponse(f'模块"{module_name}"中存在用例，不能删除') +# else: +# AutotestplatModule.objects.filter(id=module_id).delete() +# return HttpResponse('200') def deleteModule(request): module_id = request.POST.get('id') module_name = request.POST.get('module_name') + module = AutotestplatModule.objects.filter(id=module_id).first() child_module=AutotestplatModule.objects.filter(parent_module_id=module_id).first() - module_case = AutotestplatInterfaceTestcase.objects.filter(menu_module_id=module_id).first() + if module and module.product_id: + module_case = AutotestplatInterfaceTestcase.objects.filter(menu_module_id=module_id, product_id=module.product_id).first() + else: + module_case = AutotestplatInterfaceTestcase.objects.filter(menu_module_id=module_id).first() if child_module: return HttpResponse(f'模块"{module_name}"中存在子模块，不能删除') elif module_case: @@ -170,7 +203,6 @@ def deleteModule(request): AutotestplatModule.objects.filter(id=module_id).delete() return HttpResponse('200') - def showAddWindow(req,menu_module_id): head_dict = {'Cookie':'','Accept':'','Content-Type':'application/json; charset=utf-8'} head_json = json.dumps(head_dict, sort_keys=True, indent=8).encode().decode('raw_unicode_escape') @@ -506,6 +538,16 @@ def startInterfaceSend(req): if req.method == "POST": raw_data = req.body raw_data = json.loads(raw_data) + # === 新增代码：发送请求前，先从 Redis 恢复 Cookie === + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + try: + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + session.cookies.update(cookie_dict) + print_log('【自动加载Cookie】：已从 Redis 恢复 Cookie 到当前会话') + except: + pass + # =================================================== interface_name = raw_data['interface_name'] public_list = AutotestplatParameter.objects.filter() keyword_list = ["{"+rec.keywords+"}" for rec in public_list] @@ -609,6 +651,25 @@ def startInterfaceSend(req): error_info = traceback.format_exc() print(error_info) return HttpResponse('【ERROR】：参数 '+head[rec]+' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 '+head[rec]+' 的前置接口，以及确认Redis是否已启动') + # === 新增：自动从 Redis 加载 Cookie 到请求头 === + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + try: + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + cookie_string = '; '.join([f"{k}={v}" for k, v in cookie_dict.items()]) + # 如果请求头中已有Cookie字段，追加；否则新建 + if 'Cookie' in head: + if head['Cookie']: + head['Cookie'] = head['Cookie'] + '; ' + cookie_string + else: + head['Cookie'] = cookie_string + else: + head['Cookie'] = cookie_string + print_log(f'【自动注入Cookie】：已将 {len(cookie_dict)} 个Cookie注入请求头') + except Exception as e: + print_log(f'【WARNING】加载Cookie失败：{str(e)}') + # ================================================ is_login_api = False n = 0 while (n < 5): @@ -620,71 +681,161 @@ def startInterfaceSend(req): if raw_data['interface_body'] != '': try: body = eval(raw_data['interface_body']) - if not isinstance(body, dict): + if not isinstance(body, (dict, list)): return HttpResponse('【ERROR】：' + raw_data['interface_body'] + ' 接口录入信息有误，请重新修改') except: return HttpResponse('【ERROR】：' + raw_data['interface_body'] + ' 接口录入信息有误，请重新修改') else: body = {} - for rec in body.keys(): - if(isinstance(body[rec],str) or isinstance(body[rec],list) or isinstance(body[rec],dict)): - for rec1 in keyword_list1: - if(rec1 in body[rec]): - if('captcha' not in rec1): - body[rec] = body[rec].replace(rec1, public_dict[rec1.replace('{','').replace('}','')]) - else: - is_login_api = True - yanzheng_url = public_dict[rec1.replace('{','').replace('}','')] - haha = request_get(yanzheng_url,{},{},0) - with open(codefile,'wb') as f: - f.write(haha.content) - yanzheng = getcaptcha() - body[rec] = yanzheng - print_log('【登录验证码】：',','),print_log(yanzheng) - for rec5 in keyword_list5: - if (rec5 in str(body[rec])): - try: - var_name = public_dict[rec5.replace('{', '').replace('}', '')] - var_value = str(eval(var_name)) - body = str(body).replace(rec5, var_name) - body = str(body).replace(var_name,var_value) - body=ast.literal_eval(body) - except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse('【ERROR】：参数 ' + rec5 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec5 + ' 的前置接口，以及确认Redis是否已启动') - for rec2 in keyword_list2: - if(rec2 in body[rec]): - try: - body[rec] = body[rec].replace(rec2, cache.get(rec2.replace('{','').replace('}','')).decode('utf-8')) - except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse('【ERROR】：参数 '+rec2+' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 '+rec2+' 的前置接口，以及确认Redis是否已启动') - for rec3 in keyword_list3: - if(rec3 in body[rec]): + if isinstance(body, dict): + for rec in body.keys(): + if(isinstance(body[rec],str) or isinstance(body[rec],list) or isinstance(body[rec],dict)): + for rec1 in keyword_list1: + if(rec1 in body[rec]): + if('captcha' not in rec1): + body[rec] = body[rec].replace(rec1, public_dict[rec1.replace('{','').replace('}','')]) + else: + is_login_api = True + yanzheng_url = public_dict[rec1.replace('{','').replace('}','')] + haha = request_get(yanzheng_url,{},{},0) + with open(codefile,'wb') as f: + f.write(haha.content) + yanzheng = getcaptcha() + body[rec] = yanzheng + print_log('【登录验证码】：',','),print_log(yanzheng) + + + + + + for rec5 in keyword_list5: + print_log('【DEBUG】检查 rec5: ' + rec5 + ' 是否在 ' + str(body[rec]) + ' 中') + if (rec5 in str(body[rec]) or '{{' + rec5[1:-1] + '}}' in str(body[rec])): + print_log('【DEBUG】匹配成功，rec5: ' + rec5) + try: + var_name = public_dict[rec5.replace('{', '').replace('}', '')] + var_value = str(eval(var_name)) + print_log('【DEBUG】var_name: ' + var_name + ', var_value: ' + var_value) + # 如果是字符串类型，自动添加引号 + # if isinstance(eval(var_name), str): + # var_value = '"' + var_value + '"' + # print_log('【DEBUG】添加引号后的 var_value: ' + var_value) + body_str = str(body) + print_log('【DEBUG】原始 body_str: ' + body_str) + # body = str(body).replace(rec5, var_name) + body_str = body_str.replace('{{' + rec5[1:-1] + '}}', var_value) + body_str = body_str.replace(rec5, var_value) + print_log('【DEBUG】替换后的 body_str: ' + body_str) + body = ast.literal_eval(body_str) + print_log('【DEBUG】解析后的 body: ' + str(body)) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：参数 ' + rec5 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec5 + ' 的前置接口，以及确认Redis是否已启动') + for rec2 in keyword_list2: + if(rec2 in body[rec]): + try: + body[rec] = body[rec].replace(rec2, cache.get(rec2.replace('{','').replace('}','')).decode('utf-8')) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：参数 '+rec2+' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 '+rec2+' 的前置接口，以及确认Redis是否已启动') + for rec3 in keyword_list3: + if(rec3 in body[rec]): + try: + body[rec] = body[rec].replace(rec3,cache.get(rec3.replace('{', '').replace('}', '')).decode('utf-8')) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec3 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec3 + ' 的前置接口，以及确认Redis是否已启动') + if('select' in body[rec]): try: - body[rec] = body[rec].replace(rec3,cache.get(rec3.replace('{', '').replace('}', '')).decode('utf-8')) + sql = body[rec] + cursor = connection.cursor() + cursor.execute(sql) + data = cursor.fetchall() + print(u'查询的结果为： ',data[0][0]) + body[rec] = data[0][0] except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse( - '【ERROR】：参数 ' + rec3 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec3 + ' 的前置接口，以及确认Redis是否已启动') - if('select' in body[rec]): + body[rec] = '【ERROR】：查询结果为空！' + # 处理 JSON 数组类型的请求体 + elif isinstance(body, list): + print_log('【DEBUG】检测到 JSON 数组，开始处理...') + body_str = str(body) + print_log('【DEBUG】原始 body_str: ' + body_str) + + # 处理变量类型 (var) + for rec5 in keyword_list5: + if (rec5 in body_str or '{{' + rec5[1:-1] + '}}' in body_str): + print_log('【DEBUG】在 JSON 数组中匹配到 rec5: ' + rec5) + try: + var_name = public_dict[rec5.replace('{', '').replace('}', '')] + var_value = str(eval(var_name)) + print_log('【DEBUG】var_name: ' + var_name + ', var_value: ' + var_value) + body_str = body_str.replace('{{' + rec5[1:-1] + '}}', var_value) + body_str = body_str.replace(rec5, var_value) + print_log('【DEBUG】替换变量后的 body_str: ' + body_str) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec5 + ' 没有参数值，请确认系统参数设置是否正确') + + # 处理常量类型 (con) + for rec1 in keyword_list1: + if rec1 in body_str: + try: + body_str = body_str.replace(rec1, + public_dict[rec1.replace('{', '').replace('}', '')]) + print_log('【DEBUG】替换常量后的 body_str: ' + body_str) + except Exception: + error_info = traceback.format_exc() + print(error_info) + + # 处理环境设置 (env) + for rec4 in keyword_list4: + if rec4 in body_str: + try: + body_str = body_str.replace(rec4, + public_dict[rec4.replace('{', '').replace('}', '')]) + print_log('【DEBUG】替换环境设置后的 body_str: ' + body_str) + except Exception: + error_info = traceback.format_exc() + print(error_info) + + # 处理关联参数 (res) + for rec2 in keyword_list2: + if rec2 in body_str: try: - sql = body[rec] - cursor = connection.cursor() - cursor.execute(sql) - data = cursor.fetchall() - print(u'查询的结果为： ',data[0][0]) - body[rec] = data[0][0] + body_str = body_str.replace(rec2, cache.get( + rec2.replace('{', '').replace('}', '')).decode('utf-8')) + print_log('【DEBUG】替换关联参数后的 body_str: ' + body_str) except Exception: - body[rec] = '【ERROR】：查询结果为空！' + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认是否已执行返回 ' + rec2 + ' 的接口') + + # 将字符串解析回 Python 对象 + try: + body = ast.literal_eval(body_str) + print_log('【DEBUG】解析后的 body: ' + str(body)) + except Exception as e: + print_log('【ERROR】解析 JSON 数组失败：' + str(e)) + return HttpResponse('【ERROR】：JSON 数组格式有误，请检查请求体格式') mode = raw_data['interface_mode'] try: response,cookie = interface_test_start(url,body,head,mode,body_format,True) except: return HttpResponse('【ERROR】：' + url + ' 接口录入信息有误，请重新修改') + # === 新增代码：无论是否配置正则，每次请求后都自动保存 Cookie === + if session.cookies: + for cookie_name, cookie_value in session.cookies.items(): + # 统一命名为 Cookie_名称，防止冲突 + cache.set(f"Cookie_{cookie_name}", cookie_value) + print_log(f'【自动保存Cookie】：已更新 {cookie_name}') + # ========================================================= if(raw_data['interface_id']!=''): id1=raw_data['interface_id'] public_resp = AutotestplatParameter.objects.filter(module_id=int(id1)).exclude(product_id='testplan') @@ -707,6 +858,15 @@ def startInterfaceSend(req): error_info = traceback.format_exc() print(error_info) print_log('【关联参数】： {' + rec.keywords +'} 在响应数据中未匹配到，请检测前置接口关键字配置，以及Redis是否已启动') + # === 新增通用代码开始 === + # 动态保存当前请求产生的所有 Cookie 到 Redis + if session.cookies: + for cookie_name, cookie_value in session.cookies.items(): + # 存入 Redis 的 key 命名为 "Cookie_name"，避免与用户自定义参数冲突 + redis_key = f"Cookie_{cookie_name}" + cache.set(redis_key, cookie_value) + print_log(f'【自动保存Cookie】：已动态更新 {redis_key} = {cookie_value[:20]}...') + # === 新增通用代码结束 === assert_keywords_old = raw_data['interface_assert'] for rec1 in keyword_list1: if(rec1 in assert_keywords_old): @@ -759,6 +919,15 @@ def interface_test_start(url,body,head,mode,body_format,is_out): response, cookie = Method_HEAD(url, body, head, body_format, is_out) elif (mode == 'OPTIONS' or mode == 'options'): response, cookie = Method_OPTIONS(url, body, head, body_format, is_out) + # === 新增代码：自动把 Session 里的所有 Cookie 保存到 Redis === + import requests.utils + # 提取当前 Session 中的所有 Cookie + cookie_dict = requests.utils.dict_from_cookiejar(session.cookies) + if cookie_dict: + # 存入 Redis，key 命名为 'auto_cookies_all' + cache.set('auto_cookies_all', json.dumps(cookie_dict)) + print_log(f'【自动保存Cookie】：已从响应头捕获并保存 {len(cookie_dict)} 个Cookie到Redis') + # =========================================================== print_log('') return response,cookie except Exception: @@ -768,49 +937,147 @@ def interface_test_start(url,body,head,mode,body_format,is_out): def request_post(url,body,head,body_format): if(body_format == 'JSON'): - body = json.dumps(body) + # 检测是否为 JSON 数组 + if isinstance(body, list): + body = json.dumps(body) # 序列化 JSON 数组 + else: + body = json.dumps(body) # 序列化 JSON 对象 else: keys = body.keys() for rec in keys: - if('[{' in str(body[rec]) and '}]' in str(body[rec])): + if'[{' in str(body[rec]) and '}]' in str(body[rec]): body = json.dumps(body) break + + # === 新增：请求前，直接从 Redis 同步最新 Cookie 到 session === + try: + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + # 先清空 session 中的旧 Cookie + session.cookies.clear() + # 从 Redis 获取最新 Cookie + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + session.cookies.update(cookie_dict) + print_log(f'【请求前同步Cookie】：已从 Redis 同步 {len(cookie_dict)} 个Cookie到 session') + except Exception as e: + print_log(f'【WARNING】请求前同步Cookie失败：{str(e)}') + # ==================================================================== + r = session.post(url,body,headers = head) return r def request_get(url,body,head,body_format): if(body_format == 'JSON'): body = json.dumps(body) + + # === 新增：请求前，直接从 Redis 同步最新 Cookie 到 session === + try: + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + session.cookies.clear() + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + session.cookies.update(cookie_dict) + except Exception as e: + print_log(f'【WARNING】请求前同步Cookie失败：{str(e)}') + # ==================================================================== + r = session.get(url,params = body, headers=head) return r def request_put(url,body,head,body_format): if(body_format == 'JSON'): body = json.dumps(body) + + # === 新增：请求前，直接从 Redis 同步最新 Cookie 到 session === + try: + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + session.cookies.clear() + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + session.cookies.update(cookie_dict) + except Exception as e: + print_log(f'【WARNING】请求前同步Cookie失败：{str(e)}') + # ==================================================================== + r = session.put(url,params = body, headers=head) return r def request_delete(url,body,head,body_format): if(body_format == 'JSON'): body = json.dumps(body) + + # === 新增：请求前，直接从 Redis 同步最新 Cookie 到 session === + try: + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + session.cookies.clear() + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + session.cookies.update(cookie_dict) + except Exception as e: + print_log(f'【WARNING】请求前同步Cookie失败：{str(e)}') + # ==================================================================== + r = session.delete(url,json = body, headers=head) return r def request_patch(url,body,head,body_format): if(body_format == 'JSON'): body = json.dumps(body) + + # === 新增：请求前，直接从 Redis 同步最新 Cookie 到 session === + try: + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + session.cookies.clear() + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + session.cookies.update(cookie_dict) + except Exception as e: + print_log(f'【WARNING】请求前同步Cookie失败：{str(e)}') + # ==================================================================== + r = session.patch(url,json = body, headers=head) return r def request_head(url,body,head,body_format): if(body_format == 'JSON'): body = json.dumps(body) + + # === 新增：请求前，直接从 Redis 同步最新 Cookie 到 session === + try: + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + session.cookies.clear() + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + session.cookies.update(cookie_dict) + except Exception as e: + print_log(f'【WARNING】请求前同步Cookie失败：{str(e)}') + # ==================================================================== + r = session.head(url,json = body, headers=head) return r def request_options(url,body,head,body_format): if(body_format == 'JSON'): body = json.dumps(body) + + # === 新增：请求前，直接从 Redis 同步最新 Cookie 到 session === + try: + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + session.cookies.clear() + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + session.cookies.update(cookie_dict) + except Exception as e: + print_log(f'【WARNING】请求前同步Cookie失败：{str(e)}') + # ==================================================================== + r = session.options(url,json = body, headers=head) return r @@ -981,37 +1248,58 @@ def showRequestData(req): print(error_info) return HttpResponse('【ERROR】：参数 '+head[rec]+' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 '+head[rec]+' 的前置接口，以及确认Redis是否已启动') body = eval(interfaces.body) - for rec in body.keys(): - if(isinstance(body[rec],str)): - for rec1 in keyword_list1: - if(rec1 in body[rec]): - if('captcha' not in rec1): - body[rec] = body[rec].replace(rec1, public_dict[rec1.replace('{','').replace('}','')]) - else: - is_login_api = True - yanzheng_url = public_dict[rec1.replace('{','').replace('}','')] - haha = request_get(yanzheng_url,{},{},0) - with open(codefile,'wb') as f: - f.write(haha.content) - yanzheng = getcaptcha() - body[rec] = yanzheng - print_log('【登录验证码】：',','),print_log(yanzheng) - for rec2 in keyword_list2: - if(rec2 in body[rec]): - try: - body[rec] = body[rec].replace(rec2, cache.get(rec2.replace('{','').replace('}',''))) - except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse('【ERROR】：参数 '+rec2+' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 '+rec2+' 的前置接口，以及确认Redis是否已启动') - if('select' in body[rec]): - sql = body[rec] - cursor = connection.cursor() - print(sql) - cursor.execute(sql) - data = cursor.fetchall() - print(u'查询的结果为： ',data[0][0]) - body[rec] = data[0][0] + if isinstance(body, dict): + for rec in body.keys(): + if (isinstance(body[rec], str)): + for rec1 in keyword_list1: + if (rec1 in body[rec]): + if ('captcha' not in rec1): + body[rec] = body[rec].replace(rec1, public_dict[rec1.replace('{', '').replace('}', '')]) + else: + is_login_api = True + yanzheng_url = public_dict[rec1.replace('{', '').replace('}', '')] + haha = request_get(yanzheng_url, {}, {}, 0) + with open(codefile, 'wb') as f: + f.write(haha.content) + yanzheng = getcaptcha() + body[rec] = yanzheng + print_log('【登录验证码】：', ','), print_log(yanzheng) + for rec2 in keyword_list2: + if (rec2 in body[rec]): + try: + body[rec] = body[rec].replace(rec2, cache.get(rec2.replace('{', '').replace('}', ''))) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec2 + ' 的前置接口，以及确认 Redis 是否已启动') + if ('select' in body[rec]): + sql = body[rec] + cursor = connection.cursor() + print(sql) + cursor.execute(sql) + data = cursor.fetchall() + print(u'查询的结果为： ', data[0][0]) + body[rec] = data[0][0] + elif isinstance(body, list): + body_str = str(body) + for rec2 in keyword_list2: + if (rec2 in body_str): + try: + body_str = body_str.replace(rec2, + cache.get(rec2.replace('{', '').replace('}', '')).decode('utf-8')) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：参数 ' + rec2 + ' 没有参数值') + try: + body = ast.literal_eval(body_str) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：JSON 数组格式有误') + else: + return HttpResponse('【ERROR】：body 参数格式错误，应为字典或列表格式') mode = interfaces.mode name = interfaces.name body_format = interfaces.body_format diff --git a/autotest/views_interfacereport.py b/autotest/views_interfacereport.py index a6bbaf66..7503a88a 100644 --- a/autotest/views_interfacereport.py +++ b/autotest/views_interfacereport.py @@ -19,14 +19,46 @@ from django.shortcuts import render_to_response from pdfkit import from_url import pdfkit +# def reportView(request): +# user_name = request.session.get('user', '') +# product_all = AutotestplatProduct.objects.filter(delete_flag='N') +# product_id = AuthUser.objects.filter(username=user_name).first().last_name +# product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name +# c = csrf(request) +# c.update({"product_name":product_name,"product_alls":product_all}) +# return render_to_response("interface_report.html",c) + def reportView(request): user_name = request.session.get('user', '') + + # 检查用户是否登录 + if not user_name: + # 用户未登录，重定向到登录页面或返回错误 + return redirect('/autotest/login/') + + # 获取用户对象并进行空值检查 + user = AuthUser.objects.filter(username=user_name).first() + if user is None: + # 用户不存在，返回错误 + return HttpResponse('用户不存在') + product_all = AutotestplatProduct.objects.filter(delete_flag='N') - product_id = AuthUser.objects.filter(username=user_name).first().last_name - product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name + product_id = user.last_name + + # 检查 product_id 是否存在 + if not product_id: + # 用户没有关联产品，可能需要处理 + product_name = "未关联产品" + else: + product = AutotestplatProduct.objects.filter(id=product_id).first() + if product is None: + product_name = "产品不存在" + else: + product_name = product.product_name + c = csrf(request) - c.update({"product_name":product_name,"product_alls":product_all}) - return render_to_response("interface_report.html",c) + c.update({"product_name": product_name, "product_alls": product_all}) + return render_to_response("interface_report.html", c) @csrf_exempt def loadReport(request): diff --git a/autotest/views_interfacetestplan.py b/autotest/views_interfacetestplan.py index 41b33344..25c6bc5d 100644 --- a/autotest/views_interfacetestplan.py +++ b/autotest/views_interfacetestplan.py @@ -12,7 +12,7 @@ from django.template.context_processors import csrf from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger from .models import * -from datetime import datetime +from datetime import datetime,timedelta from django.http import HttpResponse from .views_interface import cache,print_log,interface_test_start, assert_test_old from django.contrib.auth.decorators import login_required @@ -78,12 +78,13 @@ def interfaceTestplan(req): else: interval_id = None if crontab_id!=None and crontab_id!='': - runtime_year = datetime.now().year - runtime_month = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][5] - runtime_day = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][4] + # runtime_year = datetime.now().year + # runtime_month = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][5] + # runtime_day = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][4] runtime_hour = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][2] runtime_min = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][1] - runtime = str(runtime_year) + '-' + runtime_month + '-' + runtime_day + ' ' + runtime_hour + ':' + runtime_min + #runtime = str(runtime_year) + '-' + runtime_month + '-' + runtime_day + ' ' + runtime_hour + ':' + runtime_min + runtime = '每天 ' + runtime_hour + ':' + runtime_min + ' 执行' new_id.task_id = runtime elif interval_id!=None and interval_id!='': runtime_every = IntervalSchedule.objects.filter(id=interval_id).values_list()[0][1] @@ -136,7 +137,8 @@ def addTestplan(request): name_exist = AutotestplatTestplan.objects.filter(suit_name=suit_list[0]).first() if name_exist: return HttpResponse(f'测试计划[{suit_list[0]}]已存在，请重新填写') - suit_info1 = AutotestplatTestplan(suit_name=suit_list[0],url_host=suit_list[1],charger=suit_list[2],product_id=product_id,interface_name=interfaces,interface_name_display=interfaces1,task_progress=None,interface_num=L,run_time='') + run_time_value = suit_list[5] if suit_list[5] and suit_list[5] != '' else '' + suit_info1 = AutotestplatTestplan(suit_name=suit_list[0],url_host=suit_list[1],charger=suit_list[2],product_id=product_id,interface_name=interfaces,interface_name_display=interfaces1,task_progress=None,interface_num=L,run_time=run_time_value) suit_info1.save() suit_id1 = AutotestplatTestplan.objects.order_by('-id')[0].id suit_name1 = AutotestplatTestplan.objects.filter(id=suit_id1).first().suit_name @@ -145,19 +147,19 @@ def addTestplan(request): crontab = suit_list[5] interval = suit_list[6] if crontab !=None and crontab!='': - year = crontab.split('-')[0] - month = crontab.split('-')[1] - day = crontab.split('-')[2].split(' ')[0] - hour = crontab.split(':')[0].split(' ')[1] + # year = crontab.split('-')[0] + # month = crontab.split('-')[1] + # day = crontab.split('-')[2].split(' ')[0] + hour = crontab.split(':')[0] min = crontab.split(':')[1] - CrontabSchedule.objects.create(month_of_year=month, day_of_month=day, hour=hour, minute=min, day_of_week='*') + CrontabSchedule.objects.create(month_of_year='*', day_of_month='*', hour=hour, minute=min, day_of_week='*') crontab_id = CrontabSchedule.objects.order_by('-id').first().id args = '[' + str(suit_id1) + ']' description = request.session.get('user', '') enabled = '1' - PeriodicTask.objects.create(name=name, task=task, args=args, crontab_id=crontab_id, interval_id=None, + PeriodicTask.objects.create(name=str(suit_id1), task=task, args=args, crontab_id=crontab_id, interval_id=None, enabled=enabled, description=description) - max_task_id = PeriodicTask.objects.order_by('-id')[0].id + max_task_id = PeriodicTask.objects.filter(crontab_id=crontab_id).first().id elif interval !=None and interval!='': if interval == "每分钟1次": interval_id = '5' @@ -183,20 +185,37 @@ def addTestplan(request): interface_name1 = interface_list[i].split('_')[2] url1 = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().url assert_use_new1 = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().assert_use_new - assert_keywords_old1 = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().assert_keywords_old - assert_keywords_is_contain1 = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().assert_keywords_is_contain + assert_keywords_old1 = AutotestplatInterfaceTestcase.objects.filter( + id=interface_id).first().assert_keywords_old + assert_keywords_is_contain1 = AutotestplatInterfaceTestcase.objects.filter( + id=interface_id).first().assert_keywords_is_contain body_format = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().body_format, formated_dict = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().body, update_cookie1 = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().update_cookie, mode1 = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().mode, head1 = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().head, + # 确保 body 是 JSON 字符串格式（双引号） + try: + body_value = formated_dict[0] + if body_value: + # 尝试解析为 Python 对象 + body_obj = eval(body_value) if not body_value.startswith('{') and not body_value.startswith( + '[') else json.loads(body_value) + # 重新序列化为标准 JSON 格式 + if isinstance(body_obj, (dict, list)): + formated_dict = (json.dumps(body_obj, ensure_ascii=False),) + except Exception: + # 如果解析失败，保持原值 + pass para_suit_info = AutotestplatTestplanInterface(suit_id=suit_id1, suit_name=suit_name1, - interface_id=interface_id,interface_name=interface_name1, url=url1, - body_format=body_format[0], body=formated_dict[0], mode=mode1[0], - update_cookie=update_cookie1[0], head=head1[0], - assert_keywords_is_contain=assert_keywords_is_contain1, - assert_use_new=assert_use_new1, - assert_keywords_old=assert_keywords_old1) + interface_id=interface_id, interface_name=interface_name1, + url=url1, + body_format=body_format[0], body=formated_dict[0], + mode=mode1[0], + update_cookie=update_cookie1[0], head=head1[0], + assert_keywords_is_contain=assert_keywords_is_contain1, + assert_use_new=assert_use_new1, + assert_keywords_old=assert_keywords_old1) para_suit_info.save() print(para_suit_info, ' insert interface success!') max_id = AutotestplatTestplanInterface.objects.order_by('-id')[0].id @@ -246,21 +265,25 @@ def delTestplan(request): AutotestplatTestplanInterfaceOrder.objects.filter(suit_id=id1).delete() return HttpResponse('delete success!') + def searchTestplan(req): username = req.session.get('user', '') key_words_product_id = req.GET.get('key_words_product_id', '') key_words_suit = req.GET.get('key_words_suit', '') key_words_name = req.GET.get('key_words_name', '') key_words_charger = req.GET.get('key_words_charger', '') - interfaces_all = AutotestplatInterfaceTestcase.objects.all().order_by('-product_id','order') - if key_words_product_id =='': + interfaces_all = AutotestplatInterfaceTestcase.objects.all().order_by('-product_id', 'order') + if key_words_product_id == '': if AuthUser.objects.filter(username=username).first().is_superuser == 1: product_id = "" else: product_id = AuthUser.objects.filter(username=username).first().last_name else: product_id = AutotestplatProduct.objects.filter(product_name=key_words_product_id).first().id - suits = AutotestplatTestplan.objects.filter(Q(suit_name__icontains=key_words_suit),Q(product_id__icontains=product_id),Q(interface_name__icontains=key_words_name),Q(charger__icontains=key_words_charger)).order_by('id') + suits = AutotestplatTestplan.objects.filter(Q(suit_name__icontains=key_words_suit), + Q(product_id__icontains=product_id), + Q(interface_name__icontains=key_words_name), + Q(charger__icontains=key_words_charger)).order_by('id') new_suits = [] for id in suits: new_id = id @@ -274,13 +297,14 @@ def searchTestplan(req): else: interval_id = None if crontab_id != None and crontab_id != '': - runtime_year = datetime.now().year - runtime_month = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][5] - runtime_day = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][4] + # runtime_year = datetime.now().year + # runtime_month = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][5] + # runtime_day = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][4] runtime_hour = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][2] runtime_min = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][1] - runtime = str( - runtime_year) + '-' + runtime_month + '-' + runtime_day + ' ' + runtime_hour + ':' + runtime_min + # runtime = str( + # runtime_year) + '-' + runtime_month + '-' + runtime_day + ' ' + runtime_hour + ':' + runtime_min + runtime = '每天 ' + runtime_hour + ':' + runtime_min + ' 执行' new_id.task_id = runtime elif interval_id != None and interval_id != '': runtime_every = IntervalSchedule.objects.filter(id=interval_id).values_list()[0][1] @@ -307,9 +331,23 @@ def searchTestplan(req): page_list = paginator.page(1) except EmptyPage: page_list = paginator.page(paginator.num_pages) + + # 获取产品相关信息（与 interfaceTestplan 函数保持一致） + if AuthUser.objects.filter(username=username).first().is_superuser == 1: + product_name = '' + else: + product_id_user = AuthUser.objects.filter(username=username).first().last_name + product_name = AutotestplatProduct.objects.filter( + id=product_id_user).first().product_name if product_id_user else '' + product_all = AutotestplatProduct.objects.filter(delete_flag='N') + env_para = AutotestplatParameter.objects.filter(type="env") + interval = IntervalSchedule.objects.all().order_by('-id') + c = csrf(req) - c.update({'page_list': page_list, 'num': num, 'interfaces_all':interfaces_all}) - return render_to_response("interface_testplan.html",c) + c.update({'page_list': page_list, 'num': num, 'interfaces_all': interfaces_all, 'product_name': product_name, + 'product_alls': product_all, 'env_paras': env_para, 'intervals': interval}) + return render_to_response("interface_testplan.html", c) + def searchTestplanApiTestcase(req): if req.method == "POST": @@ -516,12 +554,13 @@ def showEditTestplan(request): else: interval_id = None if crontab_id != None: - runtime_year = datetime.now().year - runtime_month = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][5] - runtime_day = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][4] + # runtime_year = datetime.now().year + # runtime_month = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][5] + # runtime_day = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][4] runtime_hour = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][2] runtime_min = CrontabSchedule.objects.filter(id=crontab_id).values_list()[0][1] - run_time = str(runtime_year) + '-' + runtime_month + '-' + runtime_day + ' ' + runtime_hour + ':' + runtime_min + #run_time = str(runtime_year) + '-' + runtime_month + '-' + runtime_day + ' ' + runtime_hour + ':' + runtime_min + run_time = runtime_hour + ':' + runtime_min run_interval = None elif interval_id != None: interval_id = PeriodicTask.objects.filter(id=suit_info.task_id).first().interval_id @@ -536,7 +575,7 @@ def showEditTestplan(request): run_time = None else: run_interval = None - run_time = None + run_time = suit_info.run_time if suit_info.run_time else None suit_info1 = {'id1':id1, 'suit_name':suit_info.suit_name, 'url_host':suit_info.url_host, @@ -599,55 +638,58 @@ def saveEditTestplan(request): interfaces += interface_list[i] interfaces1 += name1 product_id = AutotestplatProduct.objects.filter(product_name=suit_info[4]).first().id - AutotestplatTestplan.objects.filter(id=suit_info[0]).update(suit_name=suit_info[1],interface_name=interfaces,url_host=suit_info[2], interface_name_display=interfaces1,product_id=product_id,task_progress=None,interface_num=L,run_time='') + crontab = suit_info[6] + run_time_value = crontab if crontab and crontab != '' else '' + AutotestplatTestplan.objects.filter(id=suit_info[0]).update(suit_name=suit_info[1],interface_name=interfaces,url_host=suit_info[2], interface_name_display=interfaces1,product_id=product_id,task_progress=None,interface_num=L,run_time=run_time_value) task = 'autotest.views_interfacetestplan.api_autotest_task' interval = suit_info[5] task_id = AutotestplatTestplan.objects.filter(id=suit_info[0]).first().task_id if interval == "每分钟1次": - PeriodicTask.objects.filter(id=task_id).update(interval_id=5) + PeriodicTask.objects.filter(id=task_id).update(interval_id=5,crontab_id=None) elif interval == "每小时6次": - PeriodicTask.objects.filter(id=task_id).update(interval_id=4) + PeriodicTask.objects.filter(id=task_id).update(interval_id=4,crontab_id=None) elif interval == "每天1次": - PeriodicTask.objects.filter(id=task_id).update(interval_id=3) + PeriodicTask.objects.filter(id=task_id).update(interval_id=3,crontab_id=None) elif interval == "每天10次": - PeriodicTask.objects.filter(id=task_id).update(interval_id=2) + PeriodicTask.objects.filter(id=task_id).update(interval_id=2,crontab_id=None) elif interval == "每天4次": - PeriodicTask.objects.filter(id=task_id).update(interval_id=1) + PeriodicTask.objects.filter(id=task_id).update(interval_id=1,crontab_id=None) else: PeriodicTask.objects.filter(id=task_id).update(interval_id=None) crontab = suit_info[6] if crontab != '' and crontab != None: - year = crontab.split('-')[0] - month = crontab.split('-')[1] - day = crontab.split('-')[2].split(' ')[0] - hour = crontab.split(':')[0].split(' ')[1] + # year = crontab.split('-')[0] + # month = crontab.split('-')[1] + # day = crontab.split('-')[2].split(' ')[0] + hour = crontab.split(':')[0] min = crontab.split(':')[1] task_id = AutotestplatTestplan.objects.filter(id=suit_info[0]).first().task_id crontab_exist = PeriodicTask.objects.filter(id=task_id).first() if task_id and crontab_exist and crontab_exist.crontab_id is not None: crontab_id = crontab_exist.crontab_id - CrontabSchedule.objects.filter(id=crontab_id).update(month_of_year=month, day_of_month=day, hour=hour, + CrontabSchedule.objects.filter(id=crontab_id).update(month_of_year='*', day_of_month='*', hour=hour, minute=min, day_of_week='*') elif task_id: - CrontabSchedule.objects.create(month_of_year=month, day_of_month=day, hour=hour, minute=min, + CrontabSchedule.objects.create(month_of_year='*', day_of_month='*', hour=hour, minute=min, day_of_week='*') crontab_id = CrontabSchedule.objects.order_by('-id').first().id - PeriodicTask.objects.filter(id=task_id).update(crontab_id=crontab_id) + PeriodicTask.objects.filter(id=task_id).update(crontab_id=crontab_id, interval_id=None) else: - CrontabSchedule.objects.create(month_of_year=month, day_of_month=day, hour=hour, minute=min, + CrontabSchedule.objects.create(month_of_year='*', day_of_month='*', hour=hour, minute=min, day_of_week='*') crontab_id = CrontabSchedule.objects.order_by('-id').first().id args = '[' + str(suit_info[0]) + ']' description = request.session.get('user', '') enabled = '1' - PeriodicTask.objects.create(name=suit_info[0], task=task, args=args, crontab_id=crontab_id, interval_id=None, + PeriodicTask.objects.create(name=str(suit_info[0]), task=task, args=args, crontab_id=crontab_id, interval_id=None, enabled=enabled, description=description) - max_task_id = PeriodicTask.objects.order_by('-id')[0].id + max_task_id = PeriodicTask.objects.filter(crontab_id=crontab_id).first().id AutotestplatTestplan.objects.filter(id=suit_info[0]).update(task_id=max_task_id) - PeriodicTask.objects.filter(id=max_task_id).update(crontab_id=crontab_id) + PeriodicTask.objects.filter(id=max_task_id).update(crontab_id=crontab_id, interval_id=None) else: task_id = AutotestplatTestplan.objects.filter(id=suit_info[0]).first().task_id - PeriodicTask.objects.filter(id=task_id).update(crontab_id=None) + if task_id: + PeriodicTask.objects.filter(id=task_id).update(crontab_id=None, interval_id=None) tmp_list = AutotestplatTestplanInterface.objects.filter(suit_id=suit_info[0]) for rec in tmp_list: interface_status1 = AutotestplatTestplanInterface.objects.filter(id=rec.id).update(suit_name=suit_info[1]) @@ -658,34 +700,52 @@ def saveEditTestplan(request): AutotestplatTestplanInterfaceOrder.objects.filter(suit_id=suit_info[0]).delete() suit_info_list = AutotestplatTestplan.objects.filter(id=suit_info[0])[0] id_list = [] - tmp_id_list = suit_info_list.interface_name.split(',') + tmp_id_list = suit_info_list.interface_name.split(',') if suit_info_list.interface_name else [] for rec in tmp_id_list: + if not rec or rec == '': + continue if (rec.startswith('testplan_interface') == True): tmp_id_insert = rec.split('_')[2] id_list.append(tmp_id_insert) - elif(rec!=''): + else: tmp_id = rec.split('_')[1] suit_id1 = suit_info[0] suit_name1 = suit_info[1] - interface_id1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().id - interface_name1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().name - url1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().url - assert_use_new1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().assert_use_new - assert_keywords_old1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().assert_keywords_old - assert_keywords_is_contain1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().assert_keywords_is_contain - body_format = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().body_format, - formated_dict = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().body, - update_cookie1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().update_cookie, - mode1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().mode, - head1 = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first().head, - para_suit_info = AutotestplatTestplanInterface(suit_id=suit_id1, suit_name=suit_name1,interface_id=interface_id1, - interface_name=interface_name1, url=url1, - body_format=body_format[0], body=formated_dict[0], - mode=mode1[0], - update_cookie=update_cookie1[0], head=head1[0], - assert_keywords_is_contain=assert_keywords_is_contain1, - assert_use_new=assert_use_new1, - assert_keywords_old=assert_keywords_old1) + interface_obj = AutotestplatInterfaceTestcase.objects.filter(id=tmp_id).first() + if not interface_obj: + continue + interface_id1 = interface_obj.id + interface_name1 = interface_obj.name + url1 = interface_obj.url + assert_use_new1 = interface_obj.assert_use_new + assert_keywords_old1 = interface_obj.assert_keywords_old + assert_keywords_is_contain1 = interface_obj.assert_keywords_is_contain + body_format = interface_obj.body_format, + formated_dict = interface_obj.body, + update_cookie1 = interface_obj.update_cookie, + mode1 = interface_obj.mode, + head1 = interface_obj.head, + try: + body_value = formated_dict[0] + if body_value: + # 尝试解析为 Python 对象 + body_obj = eval(body_value) if not body_value.startswith('{') and not body_value.startswith( + '[') else json.loads(body_value) + # 重新序列化为标准 JSON 格式 + if isinstance(body_obj, (dict, list)): + formated_dict = (json.dumps(body_obj, ensure_ascii=False),) + except Exception: + # 如果解析失败，保持原值 + pass + para_suit_info = AutotestplatTestplanInterface(suit_id=suit_id1, suit_name=suit_name1, + interface_id=interface_id1, + interface_name=interface_name1, url=url1, + body_format=body_format[0], body=formated_dict[0], + mode=mode1[0], + update_cookie=update_cookie1[0], head=head1[0], + assert_keywords_is_contain=assert_keywords_is_contain1, + assert_use_new=assert_use_new1, + assert_keywords_old=assert_keywords_old1) para_suit_info.save() max_id = AutotestplatTestplanInterface.objects.order_by('-id')[0].id id_list.append(max_id) @@ -716,7 +776,7 @@ def saveEditTestplan(request): interfaceid = interface[0] interfacename = interface[4] interface_name_list += 'testplan_interface_' + str(interfaceid) + '_' + interfacename + ',' - AutotestplatTestplan.objects.filter(id=suit_info[0]).update(interface_name=interface_name_list,task_progress=None,interface_num=i_num,run_time='') + AutotestplatTestplan.objects.filter(id=suit_info[0]).update(interface_name=interface_name_list,task_progress=None,interface_num=i_num,run_time=run_time_value) return HttpResponse('update_use_status success!') @@ -736,8 +796,25 @@ def updateTestplanInterface(request): interface_body = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().body interface_body_format = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().body_format interface_mode = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().mode - interface_assert_keywords_old = AutotestplatInterfaceTestcase.objects.filter(id=interface_id).first().assert_keywords_old - AutotestplatTestplanInterface.objects.filter(id=testplan_interface_id).filter(suit_id=testplan_id).filter(interface_id=interface_id).update(interface_name=interface_name,url=interface_url,head=interface_head,body=interface_body,body_format=interface_body_format,mode=interface_mode,assert_keywords_old=interface_assert_keywords_old) + interface_assert_keywords_old = AutotestplatInterfaceTestcase.objects.filter( + id=interface_id).first().assert_keywords_old + # 确保 body 是 JSON 字符串格式（双引号） + try: + if interface_body: + # 尝试解析为 Python 对象 + body_obj = eval(interface_body) if not interface_body.startswith( + '{') and not interface_body.startswith('[') else json.loads(interface_body) + # 重新序列化为标准 JSON 格式 + if isinstance(body_obj, (dict, list)): + interface_body = json.dumps(body_obj, ensure_ascii=False) + except Exception: + # 如果解析失败，保持原值 + pass + AutotestplatTestplanInterface.objects.filter(id=testplan_interface_id).filter(suit_id=testplan_id).filter( + interface_id=interface_id).update(interface_name=interface_name, url=interface_url, head=interface_head, + body=interface_body, body_format=interface_body_format, + mode=interface_mode, + assert_keywords_old=interface_assert_keywords_old) interface_exist=1 else: interface_name='' @@ -1048,71 +1125,130 @@ def startInterfaceTestplan(request): return HttpResponse( '【ERROR】：参数 ' + head[rec] + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + head[rec] + ' 的前置接口，以及确认Redis是否已启动') + + # === 新增：自动从 Redis 加载 Cookie 到请求头 === + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + try: + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + cookie_string = '; '.join([f"{k}={v}" for k, v in cookie_dict.items()]) + # 直接设置请求头的 Cookie 字段，不操作 session.cookies + head['Cookie'] = cookie_string + print_log(f'【自动注入Cookie】：已将 {len(cookie_dict)} 个Cookie注入请求头') + print_log(f'【DEBUG】Cookie值：{cookie_string}') + except Exception as e: + print_log(f'【WARNING】加载Cookie失败：{str(e)}') + # ================================================ + is_login_api = False n = 0 while (n < 5): body = eval(cur_interface.body) - for rec in body.keys(): - if (isinstance(body[rec], str)): - for rec1 in keyword_list1: - if (rec1 in body[rec]): - if ('captcha' not in rec1): - body[rec] = body[rec].replace(rec1, public_dict[ - rec1.replace('{', '').replace('}', '')]) - else: - is_login_api = True - yanzheng_url = public_dict[rec1.replace('{', '').replace('}', '')] - haha = request_get(yanzheng_url, {}, {}, 0) - with open(codefile, 'wb') as f: - f.write(haha.content) - yanzheng = getcaptcha() - body[rec] = yanzheng - print_log('【登录验证码】：', ','), print_log(yanzheng) - for rec5 in keyword_list5: - if (rec5 in str(body[rec])): - try: - var_name = public_dict[rec5.replace('{', '').replace('}', '')] - var_value = str(eval(var_name)) - body = str(body).replace(rec5, var_name) - body = str(body).replace(var_name, var_value) - body = ast.literal_eval(body) - except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse( - '【ERROR】：参数 ' + rec5 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec5 + ' 的前置接口，以及确认Redis是否已启动') - for rec2 in keyword_list2: - if (rec2 in body[rec]): - try: - parares = cache.get(rec2.replace('{', '').replace('}', '')) - if parares==None: - print_log('【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec2 + ' 的前置接口，以及确认Redis是否已启动') + if isinstance(body, dict): + for rec in body.keys(): + if (isinstance(body[rec], str)): + for rec1 in keyword_list1: + if (rec1 in body[rec]): + if ('captcha' not in rec1): + body[rec] = body[rec].replace(rec1, public_dict[ + rec1.replace('{', '').replace('}', '')]) else: - body[rec] = body[rec].replace(rec2, cache.get(rec2.replace('{', '').replace('}', '')).decode('utf-8')) + is_login_api = True + yanzheng_url = public_dict[rec1.replace('{', '').replace('}', '')] + haha = request_get(yanzheng_url, {}, {}, 0) + with open(codefile, 'wb') as f: + f.write(haha.content) + yanzheng = getcaptcha() + body[rec] = yanzheng + print_log('【登录验证码】：', ','), print_log(yanzheng) + for rec5 in keyword_list5: + if (rec5 in str(body[rec])): + try: + var_name = public_dict[rec5.replace('{', '').replace('}', '')] + var_value = str(eval(var_name)) + body = str(body).replace(rec5, var_name) + body = str(body).replace(var_name, var_value) + body = ast.literal_eval(body) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec5 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec5 + ' 的前置接口，以及确认 Redis 是否已启动') + for rec2 in keyword_list2: + if (rec2 in body[rec]): + try: + parares = cache.get(rec2.replace('{', '').replace('}', '')) + if parares == None: + print_log( + '【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec2 + ' 的前置接口，以及确认 Redis 是否已启动') + else: + body[rec] = body[rec].replace(rec2, cache.get( + rec2.replace('{', '').replace('}', '')).decode('utf-8')) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec2 + ' 的前置接口，以及确认 Redis 是否已启动') + for rec3 in keyword_list3: + if (rec3 in body[rec]): + para = copy.deepcopy(body) + body[rec] = sign_nb(para) + if ('select' in body[rec]): + try: + sql = body[rec] + cursor = connection.cursor() + cursor.execute(sql) + data = cursor.fetchall() + print(u'查询的结果为： ', data[0][0]) + body[rec] = data[0][0] except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse('【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec2 + ' 的前置接口，以及确认Redis是否已启动') - for rec3 in keyword_list3: - if (rec3 in body[rec]): - para = copy.deepcopy(body) - body[rec] = sign_nb(para) - if ('select' in body[rec]): + body[rec] = '【ERROR】：查询结果为空！' + elif isinstance(body, list): + body_str = str(body) + for rec5 in keyword_list5: + if (rec5 in body_str): try: - sql = body[rec] - cursor = connection.cursor() - cursor.execute(sql) - data = cursor.fetchall() - print(u'查询的结果为： ', data[0][0]) - body[rec] = data[0][0] + var_name = public_dict[rec5.replace('{', '').replace('}', '')] + var_value = str(eval(var_name)) + body_str = body_str.replace(rec5, var_value) except Exception: - body[rec] = '【ERROR】：查询结果为空！' + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec5 + ' 没有参数值，请确认系统参数设置是否正确') + for rec2 in keyword_list2: + if (rec2 in body_str): + try: + body_str = body_str.replace(rec2, cache.get( + rec2.replace('{', '').replace('}', '')).decode('utf-8')) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：参数 ' + rec2 + ' 没有参数值') + try: + body = ast.literal_eval(body_str) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：JSON 数组格式有误') + else: + return HttpResponse('【ERROR】：body 参数格式错误，应为字典或列表格式') mode = cur_interface.mode body_format = cur_interface.body_format starttime = datetime.now() response, cookie = interface_test_start(url, body, head, mode, body_format, True) endtime = datetime.now() response_time = (endtime - starttime).total_seconds() + try: + import requests.utils + cookie_dict = requests.utils.dict_from_cookiejar(session.cookies) + if cookie_dict: + cache.set('auto_cookies_all', json.dumps(cookie_dict)) + cache.expire('auto_cookies_all', 3600) + print_log(f'【自动保存Cookie】：已保存 {len(cookie_dict)} 个Cookie到Redis') + except Exception as e: + print_log(f'【WARNING】保存Cookie失败：{str(e)}') update_cookie = cur_interface.update_cookie public_resp = AutotestplatParameter.objects.filter(module_id=int(cur_id)) if (str(public_resp) != '[]'): @@ -1449,26 +1585,35 @@ def api_autotest_task(suit_id): global report_id test_time = time.strftime("%Y-%m-%d %H:%M:%S") report_id = str(datetime.now().strftime("%Y%m%d%H%M%S%f")) + print(f'【Celery 任务开始】suit_id={suit_id}, report_id={report_id}, 时间={test_time}') sql = "select t3.interface_id,t2.interface_name,t2.suit_id,t2.suit_name,t2.mode,t2.url,t2.body,t2.assert_keywords_old from autotestplat_testplan t1 INNER JOIN autotestplat_testplan_interface t2 on t1.id=t2.suit_id INNER JOIN autotestplat_testplan_interface_order t3 on t2.suit_id=t3.suit_id and t2.id=t3.interface_id where t2.suit_id="+str(suit_id)+" order by t1.product_id" cursor = connection.cursor() aa = cursor.execute(sql) interface_list = cursor.fetchmany(aa) - run_interface_num=0 + print(f'【Celery 任务】查询到 {len(interface_list)} 个接口') + run_interface_num = 0 for rec in interface_list: case_list = [] case_list.append(rec) - interfaceTestTask(case_list,test_time,response_time,report_id) - run_interface_num +=1 - AutotestplatTestplan.objects.filter(id=rec[2]).update(task_progress=run_interface_num,run_time=test_time) - interfacepass=AutotestplatTestplanInterfaceResult.objects.filter(report_id=report_id).filter(result=0).count() - interfaceall=len(interface_list) - testcase_pass_pers = '{:.0%}'.format(interfacepass / interfaceall) + try: + interfaceTestTask(case_list, test_time, response_time, report_id) + run_interface_num += 1 + print(f'【Celery 任务】接口 {run_interface_num} 执行成功') + except Exception as e: + print(f'【Celery 任务】接口执行失败：{str(e)}') + print(traceback.format_exc()) + AutotestplatTestplan.objects.filter(id=rec[2]).update(task_progress=run_interface_num, run_time=test_time) + interfacepass = AutotestplatTestplanInterfaceResult.objects.filter(report_id=report_id).filter(result=0).count() + interfaceall = len(interface_list) + print(f'【Celery 任务】通过数：{interfacepass}, 总数：{interfaceall}') + testcase_pass_pers = '{:.0%}'.format(interfacepass / interfaceall) if interfaceall > 0 else '0%' AutotestplatTestplanInterfaceResult.objects.filter(report_id=report_id).update(pass_pers=testcase_pass_pers) + print(f'【Celery 任务】报告已生成：{report_id}') cursor.close() except Exception: error_info = traceback.format_exc() - print(error_info) - return HttpResponse(error_info) + print(f'【Celery 任务异常】{error_info}') + return error_info def interfaceTestTask(case_list,test_time,response_time,report_id): res_flags = [] @@ -1569,62 +1714,131 @@ def interfaceTestTask(case_list,test_time,response_time,report_id): error_info = traceback.format_exc() print(error_info) return HttpResponse('【ERROR】：参数 '+head[rec]+' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 '+head[rec]+' 的前置接口，以及确认Redis是否已启动') + + # === 新增：自动从 Redis 加载 Cookie 到请求头 === + cached_cookies = cache.get('auto_cookies_all') + if cached_cookies: + try: + cookie_dict = json.loads(cached_cookies.decode('utf-8')) + if cookie_dict: + cookie_string = '; '.join([f"{k}={v}" for k, v in cookie_dict.items()]) + # 直接设置请求头的 Cookie 字段，不操作 session.cookies + head['Cookie'] = cookie_string + print_log(f'【自动注入Cookie】：已将 {len(cookie_dict)} 个Cookie注入请求头') + print_log(f'【DEBUG】Cookie值：{cookie_string}') + except Exception as e: + print_log(f'【WARNING】加载Cookie失败：{str(e)}') + # ================================================ + body = eval(cur_interface.body) - for rec in body.keys(): - if(isinstance(body[rec],str) or isinstance(body[rec],list) or isinstance(body[rec],dict)): - for rec1 in keyword_list1: - if(rec1 in body[rec]): - body[rec] = body[rec].replace(rec1, public_dict[rec1.replace('{','').replace('}','')]) - for rec5 in keyword_list5: - if (rec5 in str(body[rec])): - try: - var_name = public_dict[rec5.replace('{', '').replace('}', '')] - var_value = str(eval(var_name)) - body = str(body).replace(rec5, var_name) - body = str(body).replace(var_name, var_value) - body = ast.literal_eval(body) - except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse( - '【ERROR】：参数 ' + rec5 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec5 + ' 的前置接口，以及确认Redis是否已启动') - for rec2 in keyword_list2: - if(rec2 in body[rec]): - try: - parares = cache.get(rec2.replace('{', '').replace('}', '')) - if parares == None: - print_log('【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec2 + ' 的前置接口，以及确认Redis是否已启动') - else: - body[rec] = body[rec].replace(rec2,cache.get(rec2.replace('{', '').replace('}', '')).decode('utf-8')) - except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse('【ERROR】：参数 '+rec2+' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 '+rec2+' 的前置接口，以及确认Redis是否已启动') - for rec3 in keyword_list3: - if(rec3 in body[rec]): + print(f'【DEBUG】body 类型：{type(body)}, 内容：{str(body)[:200]}') + + # 处理字典类型的 body + if isinstance(body, dict): + for rec in body.keys(): + if (isinstance(body[rec], str) or isinstance(body[rec], list) or isinstance(body[rec], dict)): + for rec1 in keyword_list1: + if (rec1 in body[rec]): + body[rec] = body[rec].replace(rec1, public_dict[rec1.replace('{', '').replace('}', '')]) + for rec5 in keyword_list5: + if (rec5 in str(body[rec])): + try: + var_name = public_dict[rec5.replace('{', '').replace('}', '')] + var_value = str(eval(var_name)) + body = str(body).replace(rec5, var_name) + body = str(body).replace(var_name, var_value) + body = ast.literal_eval(body) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec5 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec5 + ' 的前置接口，以及确认 Redis 是否已启动') + for rec2 in keyword_list2: + if (rec2 in body[rec]): + try: + parares = cache.get(rec2.replace('{', '').replace('}', '')) + if parares == None: + print_log( + '【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec2 + ' 的前置接口，以及确认 Redis 是否已启动') + else: + body[rec] = body[rec].replace(rec2, cache.get( + rec2.replace('{', '').replace('}', '')).decode('utf-8')) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec2 + ' 的前置接口，以及确认 Redis 是否已启动') + for rec3 in keyword_list3: + if (rec3 in body[rec]): + try: + body[rec] = body[rec].replace(rec3, cache.get( + rec3.replace('{', '').replace('}', '')).decode('utf-8')) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse( + '【ERROR】：参数 ' + rec3 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec3 + ' 的前置接口，以及确认 Redis 是否已启动') + if ('select' in body[rec]): try: - body[rec] = body[rec].replace(rec3,cache.get(rec3.replace('{', '').replace('}', '')).decode('utf-8')) + sql = body[rec] + cursor = connection.cursor() + cursor.execute(sql) + data = cursor.fetchall() + print(u'查询的结果为： ', data[0][0]) + body[rec] = data[0][0] except Exception: - error_info = traceback.format_exc() - print(error_info) - return HttpResponse( - '【ERROR】：参数 ' + rec3 + ' 没有参数值，请确认系统参数设置是否正确，是否已执行返回 ' + rec3 + ' 的前置接口，以及确认Redis是否已启动') - if('select' in body[rec]): + body[rec] = '【ERROR】：查询结果为空！' + + # 处理列表类型的 body（JSON 数组） + elif isinstance(body, list): + print(f'【DEBUG】检测到 body 是列表类型，开始处理...') + body_str = str(body) + for rec5 in keyword_list5: + if (rec5 in body_str or '{{' + rec5[1:-1] + '}}' in body_str): try: - sql = body[rec] - cursor = connection.cursor() - cursor.execute(sql) - data = cursor.fetchall() - print(u'查询的结果为： ',data[0][0]) - body[rec] = data[0][0] + var_name = public_dict[rec5.replace('{', '').replace('}', '')] + var_value = str(eval(var_name)) + # 先替换双层大括号，再替换单层大括号 + body_str = body_str.replace('{{' + rec5[1:-1] + '}}', var_value) + body_str = body_str.replace(rec5, var_value) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：参数 ' + rec5 + ' 没有参数值') + for rec2 in keyword_list2: + if (rec2 in body_str or '{{' + rec2[1:-1] + '}}' in body_str): + try: + body_str = body_str.replace('{{' + rec2[1:-1] + '}}', cache.get( + rec2.replace('{', '').replace('}', '')).decode('utf-8')) + body_str = body_str.replace(rec2, cache.get( + rec2.replace('{', '').replace('}', '')).decode('utf-8')) except Exception: - body[rec] = '【ERROR】：查询结果为空！' + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：参数 ' + rec2 + ' 没有参数值') + try: + body = ast.literal_eval(body_str) + except Exception: + error_info = traceback.format_exc() + print(error_info) + return HttpResponse('【ERROR】：JSON 数组格式有误') + mode = cur_interface.mode body_format = cur_interface.body_format starttime = datetime.now() response,cookie = interface_test_start(url,body,head,mode,body_format,False) endtime = datetime.now() response_time = (endtime - starttime).total_seconds() + + # === 新增代码：无论是否配置正则，每次请求后都自动保存 Cookie === + if session.cookies: + for cookie_name, cookie_value in session.cookies.items(): + cache.set(f"Cookie_{cookie_name}", cookie_value) + print_log(f'【自动保存Cookie】：已更新 {cookie_name}') + # ========================================================= + + print( + f'【接口响应】URL={url[:100]}..., 响应时间={response_time}秒，状态码={response[:100] if response else "None"}...') update_cookie = cur_interface.update_cookie public_resp = AutotestplatParameter.objects.filter(module_id=int(case_id),type='res') if(str(public_resp) != '[]'): @@ -1730,18 +1944,50 @@ def interfaceTestTask(case_list,test_time,response_time,report_id): assert_keywords_old = data[0][0] except Exception: assert_keywords_old = '【ERROR】：查询结果为空！' - result1 = assert_test_old(response,assert_keywords_old,True) + result1 = assert_test_old(response, assert_keywords_old, True) suit_id1 = cur_interface.suit_id suit_name1 = cur_interface.suit_name product_id = cur_testplan.product_id product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name - result_info = AutotestplatTestplanInterfaceResult(report_id=report_id,product_id=product_id,product_name=product_name,suit_id=suit_id1,suit_name=suit_name1,suit_interface_id=case_id,interface_name=interface_name,url=url,body=body,mode=mode,assert_keywords_old=assert_keywords_old,response_time=response_time,task_mode="定时任务",response=response,result=result1,date_time=test_time) - result_info.save() + print(f'\n【保存测试结果】==================') + print(f' report_id: {report_id}') + print(f' interface_name: {interface_name}') + print(f' case_id: {case_id}') + print(f' result: {result1}') + print(f' product_id: {product_id}') + print(f' suit_id: {suit_id1}') + try: + result_info = AutotestplatTestplanInterfaceResult( + report_id=report_id, + product_id=product_id, + product_name=product_name, + suit_id=suit_id1, + suit_name=suit_name1, + suit_interface_id=case_id, + interface_name=interface_name, + url=url, + body=body, + mode=mode, + assert_keywords_old=assert_keywords_old, + response_time=str(response_time), + task_mode="定时任务", + response=response, + result=str(result1), + date_time=test_time + ) + result_info.save() + print(f'【保存成功】result_id={result_info.id}, report_id={report_id}') + print(f'==================\n') + except Exception as e: + print(f'【保存失败】{str(e)}') + print(f'错误详情：{traceback.format_exc()}') + print(f'==================\n') + raise return result1 except Exception: error_info = traceback.format_exc() - print(error_info) - return HttpResponse(error_info) + print(f'【接口执行异常】{error_info}') + return error_info diff --git a/autotest/views_jmeter.py b/autotest/views_jmeter.py index 62eb4ed8..6db82469 100644 --- a/autotest/views_jmeter.py +++ b/autotest/views_jmeter.py @@ -123,10 +123,10 @@ def body_thread2(jid,numbf,numxh,numsj): ' \n' return ss -def body_cookie(key,value): +def body_cookie(key,value,domain=''): ss = ' \n' \ ' '+value+'\n' \ - ' testone.0easy.com\n' \ + ' '+domain+'\n' \ ' \n' \ ' false\n' \ ' 0\n' \ @@ -201,12 +201,12 @@ def body_result(): ' false\n' \ ' true\n' \ ' true\n' \ - ' false\n' \ - ' false\n' \ + ' True\n' \ + ' True\n' \ ' false\n' \ ' false\n' \ - ' false\n' \ - ' false\n' \ + ' True\n' \ + ' True\n' \ ' false\n' \ ' false\n' \ ' 0\n' \ @@ -251,13 +251,13 @@ def body_report(): ' false\n' \ ' true\n' \ ' true\n' \ - ' false\n' \ - ' false\n' \ + ' true\n' \ + ' true\n' \ ' false\n' \ ' true\n' \ - ' false\n' \ - ' false\n' \ - ' false\n' \ + ' true\n' \ + ' true\n' \ + ' true\n' \ ' true\n' \ ' 0\n' \ ' true\n' \ @@ -399,3 +399,48 @@ def generate_jmx2(jname,jhost,jport,jpath,jscheme,jbody): File2.close() +def body_request_form(jid, jname, jhost, jport, jpath, jscheme, jbody, jhead, jassert, jcookie=''): + args_collection = '' + if isinstance(jbody, dict): + for k, v in jbody.items(): + key_str = str(k) + val_str = str(v) if v is not None else '' + args_collection += \ + ' \n' \ + ' true\n' \ + ' '+key_str+'\n' \ + ' '+val_str+'\n' \ + ' =\n' \ + ' true\n' \ + ' \n' + + cookie_xml = '' + if jcookie: + cookie_xml = body_httpcookie(jcookie) + + ss = ' \n' \ + ' false\n' \ + ' \n' \ + ' \n' + args_collection + \ + ' \n' \ + ' \n' \ + ' '+jhost+'\n' \ + ' '+jport+'\n' \ + ' \n' \ + ' \n' \ + ' '+jscheme+'\n' \ + ' UTF-8\n' \ + ' '+jpath+'\n' \ + ' POST\n' \ + ' false\n' \ + ' true\n' \ + ' false\n' \ + ' false\n' \ + ' Java\n' \ + ' false\n' \ + ' \n' \ + ' \n' \ + ' \n' + cookie_xml + body_httpheader(jhead) + body_reg() + body_assert(jassert) + ' \n' \ + + body_result() + ' \n' \ + ' \n' + return ss \ No newline at end of file diff --git a/autotest/views_performance.py b/autotest/views_performance.py index 8f4eb794..1c7f7332 100644 --- a/autotest/views_performance.py +++ b/autotest/views_performance.py @@ -1,4 +1,10 @@ -import os,re,json,traceback,copy,random,string,time,redis,ast,requests,codecs +import os,re,json,traceback,copy,random,string,time,redis,ast,requests,codecs,subprocess + +import domain + +from .views_jmeter import generate_jmx, body_request, body_request2, body_testplan, body_result, body_thread, body_thread2, body_head, body_cookie, body_httpcookie, generate_jmx2, body_request_form + +from django.db import connection from django.shortcuts import render from django.shortcuts import render_to_response from django.http import StreamingHttpResponse @@ -8,11 +14,10 @@ from urllib.parse import urlparse from django.core.cache import cache from django.conf import settings -from datetime import datetime +from datetime import datetime,timedelta from django.contrib.auth.decorators import login_required from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger from .models import * -from .views_jmeter import generate_jmx,body_request,body_request2,body_testplan,body_result,body_thread,body_thread2,body_head,body_cookie,generate_jmx2 current_dir = os.getcwd() jmxfile = os.path.join(current_dir, 'apache-jmeter-5.6.2/bin', 'apitest.jmx') @@ -64,7 +69,94 @@ def apiPerformance(request): return render_to_response("interface_performance.html", c) def report(request): - return render_to_response("output/index.html") + # 指向 JMeter 生成的 index.html 文件 + output_index_path = os.path.join(current_dir, 'autotest', 'static', 'output', 'index.html') + + if os.path.exists(output_index_path): + # 尝试多种编码方式读取，解决中文乱码问题 + content = None + # JMeter 在 Windows 上通常使用 GBK 编码，先尝试 UTF-8，失败后尝试 GBK + for encoding in ['utf-8', 'gbk', 'gb2312', 'gb18030']: + try: + with open(output_index_path, 'r', encoding=encoding) as f: + content = f.read() + print(f"【SUCCESS】使用 {encoding} 编码成功读取 JMeter 报告") + break + except UnicodeDecodeError: + continue + + if content is None: + return HttpResponse('

测试报告编码错误

无法正确读取报告文件

') + + content = translate_jmeter_report(content) + # 必须指定 content_type 为 text/html; charset=utf-8，确保浏览器使用 UTF-8 解码 + return HttpResponse(content, content_type='text/html; charset=utf-8') + else: + return HttpResponse('

测试报告不存在

请先执行性能测试

') + + +def translate_jmeter_report(content): + translations = { + 'Test and Report information': '测试和报告信息', + 'Source file': '源文件', + 'Start Time': '开始时间', + 'End Time': '结束时间', + 'Filter for display': '显示过滤器', + 'APDEX (Application Performance Index)': 'APDEX (应用性能指数)', + 'Apdex': 'Apdex指数', + 'T (Toleration threshold)': 'T (容忍阈值)', + 'F (Frustration threshold)': 'F (挫折阈值)', + 'Label': '标签', + 'Requests Summary': '请求摘要', + 'Statistics': '统计信息', + 'Requests': '请求', + 'Executions': '执行情况', + 'Response Times (ms)': '响应时间 (毫秒)', + 'Throughput': '吞吐量', + 'Network (KB/sec)': '网络 (KB/秒)', + '#Samples': '样本数', + 'FAIL': '失败', + 'Error %': '错误率', + 'Average': '平均值', + 'Min': '最小值', + 'Max': '最大值', + 'Median': '中位数', + '90th pct': '90百分位', + '95th pct': '95百分位', + '99th pct': '99百分位', + 'Transactions/s': '事务/秒', + 'Received': '接收', + 'Sent': '发送', + 'Total': '总计', + 'Response Time Percentiles': '响应时间百分位', + 'Response Time Distribution': '响应时间分布', + 'Active Threads Over Time': '活跃线程数随时间变化', + 'Time VS Threads': '时间 VS 线程数', + 'Bytes Throughput Over Time': '字节吞吐量随时间变化', + 'Response Times Over Time': '响应时间随时间变化', + 'Response Time Percentiles Over Time (successful requests only)': '响应时间百分位随时间变化 (仅成功请求)', + 'Synthetic Response Times Distribution': '合成响应时间分布', + 'Latencies Over Time': '延迟随时间变化', + 'Connect Time Over Time': '连接时间随时间变化', + 'Response Time Vs Request': '响应时间 VS 请求数', + 'Latencies Vs Request': '延迟 VS 请求数', + 'Hits Per Second': '每秒点击数', + 'Codes Per Second': '状态码/秒', + 'Total Transactions Per Second': '每秒总事务数', + 'Transactions Per Second': '每秒事务数', + 'Customs Graphs': '自定义图表', + 'Dashboard': '仪表板', + 'Charts': '图表', + 'Over Time': '随时间变化', + 'Throughput': '吞吐量', + 'Response Times': '响应时间', + } + + for english, chinese in translations.items(): + content = content.replace(f'>{english}<', f'>{chinese}<') + content = content.replace(f'"{english}"', f'"{chinese}"') + + return content def searchPerformanceInterface(request): if request.method == "POST": @@ -88,6 +180,7 @@ def searchPerformanceInterface(request): 'charger_list': charger_list,} return HttpResponse(json.dumps(interface_list2), content_type='application/json') + def generateJmeterFile(request): try: raw_data = request.body @@ -97,9 +190,66 @@ def generateJmeterFile(request): num_bf = raw_data['num_bf'] num_xh = raw_data['num_xh'] num_sj = raw_data['num_sj'] - num_sj = str(int(num_sj)*60) + num_sj = str(int(num_sj) * 60) mode = raw_data['mode'] deal_cookie = raw_data['deal_cookie'] + + print(f'【INFO】开始生成JMeter文件') + print(f'【INFO】登录接口列表: {id_list_login}') + print(f'【INFO】业务接口列表: {id_list_not_login}') + + login_cookie_dict = {} + + print(f'【INFO】开始从Redis获取Cookie用于性能测试') + try: + import redis as redis_client + r = redis_client.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True) + cached_cookies = r.get('auto_cookies_all') + print(f'【DEBUG】Redis中auto_cookies_all的原始值为: {cached_cookies}') + print(f'【DEBUG】Redis中auto_cookies_all的类型为: {type(cached_cookies)}') + + if cached_cookies: + try: + cookie_dict = None + + if isinstance(cached_cookies, str): + cookie_dict = json.loads(cached_cookies) + elif isinstance(cached_cookies, dict): + cookie_dict = cached_cookies + + print(f'【DEBUG】解析后的cookie_dict: {cookie_dict}') + + if cookie_dict and isinstance(cookie_dict, dict) and len(cookie_dict) > 0: + for key, value in cookie_dict.items(): + login_cookie_dict[key] = value + print(f'【SUCCESS】从Redis获取到Cookie: {key}={value}') + else: + print(f'【WARNING】cookie_dict为空或不是字典类型') + except Exception as e: + print(f'【ERROR】解析Redis Cookie失败: {e}') + + traceback.print_exc() + + try: + cookie_dict = ast.literal_eval(str(cached_cookies)) + if isinstance(cookie_dict, dict): + for key, value in cookie_dict.items(): + login_cookie_dict[key] = value + print(f'【SUCCESS】通过ast解析从Redis获取到Cookie: {key}={value}') + except Exception as ae: + print(f'【ERROR】ast解析也失败: {ae}') + else: + print(f'【WARNING】Redis中auto_cookies_all为空，请先在接口用例中执行登录接口') + except Exception as e: + print(f'【ERROR】从Redis获取Cookie失败: {e}') + + traceback.print_exc() + + if login_cookie_dict: + print(f'【SUCCESS】成功获取到 {len(login_cookie_dict)} 个Cookie: {login_cookie_dict}') + else: + print(f'【WARNING】未能从Redis获取到Cookie，后续接口将不带Cookie') + public_list = AutotestplatParameter.objects.filter() keyword_list = ["{" + rec.keywords + "}" for rec in public_list] public_list1 = AutotestplatParameter.objects.filter(Q(type='con')) @@ -134,15 +284,21 @@ def generateJmeterFile(request): public_dict.update(public_dict4) public_dict.update(public_dict5) body_list = [] - id='' - name='' + id = '' + name = '' + host = '' for id in id_list_login: interface_list = AutotestplatInterfaceTestcase.objects.filter(id=str(id)) for rec in interface_list: id = rec.id url = rec.url - url2= url.replace('?','//%') - parabody = rec.body + url2 = url.replace('?', '//%') + try: + parabody_dict = eval(rec.body) + if not isinstance(parabody_dict, dict): + parabody_dict = {} + except Exception: + parabody_dict = {} parsed_uri = urlparse(url2) name = rec.name head = rec.head @@ -152,48 +308,95 @@ def generateJmeterFile(request): if ("{" in url_host and "}" in url_host): end_index = url_host.find("}") key_url_host = url_host[:end_index + 1] - url_host = url_host.replace(key_url_host, public_dict[key_url_host.replace('{', '').replace('}', '').replace(' ', '')]) + url_host = url_host.replace(key_url_host, public_dict[ + key_url_host.replace('{', '').replace('}', '').replace(' ', '')]) except: return HttpResponse('【ERROR】：url_host参数 ' + url_host + ' 有误，请重新修改 ') - scheme = '{uri.scheme}'.format(uri=parsed_uri) - # domain = '{uri.netloc}'.format(uri=parsed_uri) - domain = url_host - path = '{uri.path}'.format(uri=parsed_uri) - path2 = path.replace('//%','?') - if url_host.startswith('http://') or url_host.startswith('https://'): - host = domain.split(':')[1].replace('//', '') - if len(domain.split(':')) > 2: - port = domain.split(':')[2] - else: - port = '' + if url_host.lower().startswith('https://'): + scheme = 'https' + clean_host = url_host[8:] + elif url_host.lower().startswith('http://'): + scheme = 'http' + clean_host = url_host[7:] else: - host = domain.split(':')[0] - if len(domain.split(':')) > 1: - port = domain.split(':')[1] - else: - port = '' + scheme = 'https' + clean_host = url_host + + if ':' in clean_host: + host = clean_host.split(':')[0] + port = clean_host.split(':')[1] + else: + host = clean_host + port = '443' if scheme == 'https' else '80' + path = '{uri.path}'.format(uri=parsed_uri) + path2 = path.replace('//%', '?') + head1 = eval(head) - head_list1 = [] - for item, value in head1.items(): - item1 = item - value1 = value - head1 = body_head(item1, value1) - head_list1.append(head1) - head1 = ''.join(head_list1) - body1 = body_request(str(id),name,host, port, path2, scheme, parabody,head1,assertkey,) + content_type = head1.get('Content-Type', '') + + has_content_type = any(k.lower() == 'content-type' for k in head1.keys()) + + if 'application/json' in content_type.lower(): + parabody_str = json.dumps(parabody_dict, ensure_ascii=False) + head_list1 = [body_head(k, v) for k, v in head1.items()] + body1 = body_request(str(id), name, host, port, path2, scheme, parabody_str, ''.join(head_list1), + assertkey) + else: + if not has_content_type: + head1['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8' + head_list1 = [body_head(k, v) for k, v in head1.items()] + body1 = body_request_form(str(id), name, host, port, path2, scheme, parabody_dict, + ''.join(head_list1), assertkey) body_list.append(body1) - islogin = True - start_interface_login(id) update_cookie = rec.update_cookie update_cookie = update_cookie.replace('{', '').replace('}', '') + if not host and id_list_not_login: + first_interface = AutotestplatInterfaceTestcase.objects.filter(id=str(id_list_not_login[0])).first() + if first_interface: + url_host = first_interface.url_host + try: + if ("{" in url_host and "}" in url_host): + end_index = url_host.find("}") + key_url_host = url_host[:end_index + 1] + url_host = url_host.replace(key_url_host, public_dict[ + key_url_host.replace('{', '').replace('}', '').replace(' ', '')]) + except: + pass + + if url_host.startswith('http://') or url_host.startswith('https://'): + url_host = url_host.split('://')[1] + + if ':' in url_host: + host = url_host.split(':')[0] + else: + host = url_host + print(f'【INFO】从业务接口获取到域名: {host}') + body1 = ''.join(body_list) - body_thread1='' + body_thread1 = '' try: - body_thread1 = body_thread(id,'1','1') + body_thread1 = body_thread(id, '1', '1') except Exception: pass - cookie='' + cookie_xml = '' + cookie_header_value = '' + if login_cookie_dict: + cookie_parts = [] + cookie_header_parts = [] + cookie_domain = host if host else '' + for key, value in login_cookie_dict.items(): + cookie_parts.append(body_cookie(key, value, cookie_domain)) + cookie_header_parts.append(f'{key}={value}') + print(f'【INFO】生成Cookie XML: {key}={value}, domain={cookie_domain}') + cookie_xml = ''.join(cookie_parts) + cookie_header_value = '; '.join(cookie_header_parts) + print(f'【INFO】拼接后的Cookie XML长度: {len(cookie_xml)}') + print(f'【INFO】拼接后的Cookie Header值: {cookie_header_value}') + print(f'【INFO】最终Cookie字典: {login_cookie_dict}') + else: + print(f'【ERROR】login_cookie_dict为空，无法生成Cookie XML，性能测试将会失败！') + body_list2 = [] for id2 in id_list_not_login: interface_list2 = AutotestplatInterfaceTestcase.objects.filter(id=str(id2)) @@ -204,7 +407,8 @@ def generateJmeterFile(request): if ("{" in url_host and "}" in url_host): end_index = url_host.find("}") key_url_host = url_host[:end_index + 1] - url_host = url_host.replace(key_url_host, public_dict[key_url_host.replace('{', '').replace('}', '').replace(' ', '')]) + url_host = url_host.replace(key_url_host, public_dict[ + key_url_host.replace('{', '').replace('}', '').replace(' ', '')]) except: return HttpResponse('【ERROR】：url_host参数 ' + url_host + ' 有误，请重新修改 ') url = rec.url @@ -215,7 +419,8 @@ def generateJmeterFile(request): url = url.replace(key_url, public_dict[key_url.replace('{', '').replace('}', '')]) body = eval(rec.body) for bodykey in body.keys(): - if(isinstance(body[bodykey],str) or isinstance(body[bodykey],list) or isinstance(body[bodykey],dict)): + if (isinstance(body[bodykey], str) or isinstance(body[bodykey], list) or isinstance(body[bodykey], + dict)): for rec1 in keyword_list1: if (rec1 in body[bodykey]): if ('captcha' not in rec1): @@ -229,7 +434,9 @@ def generateJmeterFile(request): f.write(haha.content) yanzheng = getcaptcha() body[bodykey] = yanzheng - print_detail('【验证码】：', ','), print_detail(yanzheng) + print_log('【验证码】：', ',') + print_log(yanzheng) + for rec5 in keyword_list5: if (rec5 in str(body[bodykey])): try: @@ -251,7 +458,8 @@ def generateJmeterFile(request): except Exception: error_info = traceback.format_exc() print(error_info) - return HttpResponse('【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认是否已执行返回 ' + rec2 + ' 的接口') + return HttpResponse( + '【ERROR】：参数 ' + rec2 + ' 没有参数值，请确认是否已执行返回 ' + rec2 + ' 的接口') for rec3 in keyword_list3: if (rec3 in body[bodykey]): try: @@ -281,6 +489,15 @@ def generateJmeterFile(request): domain = url_host path = '{uri.path}'.format(uri=parsed_uri) path2 = path.replace('//%', '?') + + if not scheme: + if url_host.lower().startswith('https://'): + scheme = 'https' + elif url_host.lower().startswith('http://'): + scheme = 'http' + else: + scheme = 'https' + if url_host.startswith('http://') or url_host.startswith('https://'): host = domain.split(':')[1].replace('//', '') if len(domain.split(':')) > 2: @@ -293,64 +510,98 @@ def generateJmeterFile(request): port = domain.split(':')[1] else: port = '' + + if not port: + port = '443' if scheme == 'https' else '80' + head2 = eval(head) + content_type = head2.get('Content-Type', '') head_list2 = [] - for item,value in head2.items(): + + has_content_type = any(k.lower() == 'content-type' for k in head2.keys()) + has_cookie_header = any(k.lower() == 'cookie' for k in head2.keys()) + + for item, value in head2.items(): item2 = item value2 = value if value2 == "{autotestplat}": - head2 = body_head(item2,str(cookie)) + if cookie_xml: + head2 = body_head(item2, cookie_xml) + print(f'【INFO】为接口 {id2} 添加Cookie到Header: {item2}={cookie_xml}') + else: + head2 = body_head(item2, '') else: head2 = body_head(item2, value2) head_list2.append(head2) - head2 = ''.join(head_list2) - if(cookie): - cookie = eval(cookie) - cookie_list = [] - cookie1 = body_cookie(cookie[1][0],cookie[1][1]) - cookie_list.append(cookie1) - cookie2 = body_cookie(cookie[2][0], cookie[2][1]) - cookie_list.append(cookie2) - cookie3 = body_cookie(cookie[3][0], cookie[3][1]) - cookie_list.append(cookie3) - for item, value in cookie[0].items(): - item2 = item - value2 = value - cookie = body_cookie(item2,value2) - cookie_list.append(cookie) - cookie = ''.join(cookie_list) + + if not has_content_type: + head_list2.append(body_head('Content-Type', 'application/x-www-form-urlencoded; charset=UTF-8')) + + if cookie_header_value and not has_cookie_header: + head_list2.append(body_head('Cookie', cookie_header_value)) + print(f'【INFO】为接口 {id2} 添加Cookie Header: {cookie_header_value}') + + head2_str = ''.join(head_list2) + + if 'application/json' in content_type.lower(): + body2 = body_request2(str(id2), name, host, port, path2, scheme, parabody, cookie_xml, head2_str, + assertkey) else: - cookie = '' - body2 = body_request2(str(id2), name,host, port, path2, scheme, parabody, cookie,head2,assertkey) + body2 = body_request_form(str(id2), name, host, port, path2, scheme, body, head2_str, assertkey, + cookie_xml) + body_list2.append(body2) islogin = False body2 = ''.join(body_list2) - body_thread22='' + body_thread22 = '' if num_xh == '0': num_xh = '-1' else: num_xh = num_xh try: - body_thread22 = body_thread2(id2,num_bf,num_xh,num_sj) + body_thread22 = body_thread2(id2, num_bf, num_xh, num_sj) except Exception: pass body3 = body_thread1 + str(body1) + '\n' + body_thread22 + str(body2) body4 = body_thread22 + str(body1) + str(body2) - if mode=='多用户': - generate_jmx(name,host, port, path2, scheme, body4,) + if mode == '多用户': + generate_jmx(name, host, port, path2, scheme, body4, ) else: - generate_jmx2(name,host, port, path2, scheme, body3,) + generate_jmx2(name, host, port, path2, scheme, body3, ) return HttpResponse("success") except Exception: traceback.print_exc() return HttpResponse("failed") + def prepareJmeter(request): try: - bin_file = "cd "+current_dir+"/apache-jmeter-5.6.2/bin && del testLogFile" - rm_report_file = "rmdir /s/q "+current_dir.replace('/','\\')+"\\autotest\\static\\output\\ " - os.system(bin_file) - os.system(rm_report_file) + bin_dir = os.path.join(current_dir, 'apache-jmeter-5.6.2', 'bin') + test_log_file = os.path.join(bin_dir, 'testLogFile') + output_dir = os.path.join(current_dir, 'autotest', 'static', 'output') + + # 删除测试日志文件（如果存在） + if os.path.exists(test_log_file): + try: + os.remove(test_log_file) + except: + pass + + # 创建空的测试日志文件，避免 JMeter 找不到文件 + with open(test_log_file, 'w', encoding='utf-8') as f: + pass + + # 删除旧的输出目录 + if os.path.exists(output_dir): + try: + import shutil + shutil.rmtree(output_dir) + except: + pass + + # 必须重新创建空的输出目录，否则 serve 视图找不到根目录会报错 + os.makedirs(output_dir) + return HttpResponse("success") except Exception: traceback.print_exc() @@ -358,8 +609,78 @@ def prepareJmeter(request): def startTestJmeter(request): try: - report_file = "cd "+current_dir+"/apache-jmeter-5.6.2/bin && jmeter -n -t "+current_dir+"/apache-jmeter-5.6.2/bin/apitest.jmx -l testLogFile -e -o "+current_dir+"/autotest/static/output" - os.system(report_file) + jmx_path = os.path.join(current_dir, 'apache-jmeter-5.6.2', 'bin', 'apitest.jmx') + logfile_path = os.path.join(current_dir, 'apache-jmeter-5.6.2', 'bin', 'testLogFile') + output_path = os.path.join(current_dir, 'autotest', 'static', 'output') + apache_jmeter_jar = os.path.join(current_dir, 'apache-jmeter-5.6.2', 'bin', 'ApacheJMeter.jar') + + if not os.path.exists(apache_jmeter_jar): + return HttpResponse("failed: JMeter核心jar包不存在") + + # --- 开始修改：显式指定 Java 路径 --- + java_path = None + + # 1. 优先尝试读取环境变量 JAVA_HOME + java_home = os.environ.get('JAVA_HOME') + if java_home: + candidate = os.path.join(java_home, 'bin', 'java.exe') + if os.path.exists(candidate): + java_path = candidate + print(f"从环境变量找到 Java: {java_path}") + + # 2. 如果环境变量读取失败，使用截图中的路径作为兜底 + if not java_path: + # 您截图中的实际路径 + candidate = r"C:\Program Files\Java\jdk1.8.0_202\bin\java.exe" + if os.path.exists(candidate): + java_path = candidate + print(f"使用硬编码路径找到 Java: {java_path}") + + # 3. 如果以上都没找到，报错提示 + if not java_path: + return HttpResponse("failed: 无法找到 Java 环境，请确认已安装 JDK 并配置了 JAVA_HOME。") + + if not os.path.exists(java_path): + return HttpResponse(f"failed: 找不到 Java 可执行文件，请检查路径: {java_path}") + # --- 结束修改 --- + + env = os.environ.copy() + env['JMETER_HOME'] = os.path.join(current_dir, 'apache-jmeter-5.6.2') + + jmeter_cmd = [ + java_path, + '-jar', apache_jmeter_jar, + '-n', + '-t', jmx_path, + '-l', logfile_path, + '-e', + '-o', output_path, + '-Jlanguage=zh_CN' + ] + + # 【关键修改】：使用 subprocess.Popen 异步启动 JMeter，不阻塞请求 + # 这样可以立即返回响应，前端可以继续轮询真实状态 + pid_file = os.path.join(current_dir, 'autotest', 'jmeter.pid') + + # 启动 JMeter 进程 + process = subprocess.Popen( + jmeter_cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + universal_newlines=True, + cwd=current_dir, + shell=False, + env=env, + creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == 'nt' else 0 # Windows 下创建新进程组 + ) + + # 保存进程 PID 到文件，便于后续查询状态 + with open(pid_file, 'w') as f: + f.write(str(process.pid)) + + print(f"JMeter 进程已启动，PID: {process.pid}") + + # 立即返回成功，不等待 JMeter 完成 return HttpResponse("success") except Exception: traceback.print_exc() @@ -381,6 +702,47 @@ def showProgress(request): return HttpResponse("success") +def getJmeterStatus(request): + """ + 检查JMeter进程的真实运行状态 + 返回: {"running": true/false, "progress": 当前进度秒数} + """ + try: + pid_file = os.path.join(current_dir, 'autotest', 'jmeter.pid') + + # 读取保存的PID + if not os.path.exists(pid_file): + return HttpResponse(json.dumps({"running": False, "progress": 0}), content_type='application/json') + + with open(pid_file, 'r') as f: + pid = int(f.read().strip()) + + # 检查进程是否还在运行 + import psutil + try: + process = psutil.Process(pid) + if process.is_running() and process.status() != psutil.STATUS_ZOMBIE: + # 进程还在运行，计算已经运行的时间 + create_time = process.create_time() + import time + elapsed_seconds = int(time.time() - create_time) + return HttpResponse(json.dumps({ + "running": True, + "progress": elapsed_seconds, + "pid": pid + }), content_type='application/json') + else: + # 进程已结束 + return HttpResponse(json.dumps({"running": False, "progress": 0}), content_type='application/json') + except psutil.NoSuchProcess: + # 进程不存在 + return HttpResponse(json.dumps({"running": False, "progress": 0}), content_type='application/json') + except Exception as e: + traceback.print_exc() + # 如果psutil不可用或其他错误，返回未运行状态 + return HttpResponse(json.dumps({"running": False, "progress": 0, "error": str(e)}), content_type='application/json') + + def start_interface_login(id1): public_list = AutotestplatParameter.objects.all() keyword_list = ["{"+rec.keywords+"}" for rec in public_list] @@ -488,7 +850,7 @@ def start_interface_login(id1): if('select' in body[rec]): try: sql = body[rec] - cursor = connection.cursor() + cursor = conne.cursor() cursor.execute(sql) data = cursor.fetchall() print(u'查询的结果为： ',data[0][0]) @@ -635,14 +997,23 @@ def interface_test_start(url,body,head,mode,body_format,is_out): return 1 def request_post(url, body, head, body_format): - if (body_format == 'JSON'): - body = json.dumps(body) - else: - keys = body.keys() - for rec in keys: - if ('[{' in str(body[rec]) and '}]' in str(body[rec])): - body = json.dumps(body) - break + try: + if (body_format == 'JSON' or body_format == 'json'): + # 检测是否为 JSON 数组 + if isinstance(body, list): + body = json.dumps(body, ensure_ascii=False) # 序列化 JSON 数组 + elif isinstance(body, dict): + body = json.dumps(body, ensure_ascii=False) # 序列化 JSON 对象 + else: + # FORM 格式，不需要 json.dumps + keys = body.keys() + for rec in keys: + if ('[{' in str(body[rec]) and '}]' in str(body[rec])): + body = json.dumps(body, ensure_ascii=False) + break + except Exception as e: + print_log(f'【ERROR】请求体序列化失败：{e}') + print_log(f'body_format: {body_format}, body type: {type(body)}') r = session.post(url, body, headers=head) return r diff --git a/autotest/views_product.py b/autotest/views_product.py index 493d5ee2..c5a9451c 100644 --- a/autotest/views_product.py +++ b/autotest/views_product.py @@ -14,14 +14,46 @@ from .views_user import * from django.contrib.auth import get_user_model +# def productView(request): +# user_name = request.session.get('user', '') +# product_all = AutotestplatProduct.objects.filter(delete_flag='N') +# product_id = AuthUser.objects.filter(username=user_name).first().last_name +# product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name +# c = csrf(request) +# c.update({"product_name":product_name,"product_alls":product_all}) +# return render_to_response("product.html",c) + def productView(request): user_name = request.session.get('user', '') + + # 检查用户是否登录 + if not user_name: + # 用户未登录，重定向到登录页面或返回错误 + return redirect('/autotest/login/') + + # 获取用户对象并进行空值检查 + user = AuthUser.objects.filter(username=user_name).first() + if user is None: + # 用户不存在，返回错误 + return HttpResponse('用户不存在') + product_all = AutotestplatProduct.objects.filter(delete_flag='N') - product_id = AuthUser.objects.filter(username=user_name).first().last_name - product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name + product_id = user.last_name + + # 检查 product_id 是否存在 + if not product_id: + # 用户没有关联产品，可能需要处理 + product_name = "未关联产品" + else: + product = AutotestplatProduct.objects.filter(id=product_id).first() + if product is None: + product_name = "产品不存在" + else: + product_name = product.product_name + c = csrf(request) - c.update({"product_name":product_name,"product_alls":product_all}) - return render_to_response("product.html",c) + c.update({"product_name": product_name, "product_alls": product_all}) + return render_to_response("product.html", c) @csrf_exempt def loadProduct(request): @@ -42,8 +74,7 @@ def addProduct(request): product_introduction = request.POST.get('product_introduction') charger = request.session.get('user', '') delete_flag = 'N' - product_name_exist = AutotestplatProduct.objects.filter(product_name=product_name).first() - if product_name_exist: + if AutotestplatProduct.objects.filter(product_name=product_name,delete_flag='N').first() is not None: return HttpResponse(f'产品”{product_name}“已存在，请重新填写') AutotestplatProduct.objects.create(product_name=product_name, product_type=product_type, product_introduction=product_introduction,charger=charger,delete_flag=delete_flag) return HttpResponse('200') diff --git a/autotest/views_user.py b/autotest/views_user.py index 528d00c5..62ea4f82 100644 --- a/autotest/views_user.py +++ b/autotest/views_user.py @@ -43,14 +43,46 @@ def lastLogin(request): def logout(request): return HttpResponse('logout') +# def userView(request): +# user_name = request.session.get('user', '') +# product_all = AutotestplatProduct.objects.filter(delete_flag='N') +# product_id = AuthUser.objects.filter(username=user_name).first().last_name +# product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name +# c = csrf(request) +# c.update({"product_name":product_name,"product_alls":product_all}) +# return render_to_response("user.html",c) + def userView(request): user_name = request.session.get('user', '') + + # 检查用户是否登录 + if not user_name: + # 用户未登录，重定向到登录页面或返回错误 + return redirect('/autotest/login/') + + # 获取用户对象并进行空值检查 + user = AuthUser.objects.filter(username=user_name).first() + if user is None: + # 用户不存在，返回错误 + return HttpResponse('用户不存在') + product_all = AutotestplatProduct.objects.filter(delete_flag='N') - product_id = AuthUser.objects.filter(username=user_name).first().last_name - product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name + product_id = user.last_name + + # 检查 product_id 是否存在 + if not product_id: + # 用户没有关联产品，可能需要处理 + product_name = "未关联产品" + else: + product = AutotestplatProduct.objects.filter(id=product_id).first() + if product is None: + product_name = "产品不存在" + else: + product_name = product.product_name + c = csrf(request) - c.update({"product_name":product_name,"product_alls":product_all}) - return render_to_response("user.html",c) + c.update({"product_name": product_name, "product_alls": product_all}) + return render_to_response("user.html", c) @csrf_exempt def loadUser(request): diff --git a/autotest/views_webtestcase.py b/autotest/views_webtestcase.py index 1531cfe0..94e16875 100644 --- a/autotest/views_webtestcase.py +++ b/autotest/views_webtestcase.py @@ -6,7 +6,7 @@ import time,json,pymysql,os from datetime import datetime from django.views.decorators.csrf import csrf_exempt -from django.shortcuts import render +from django.shortcuts import render, redirect from django.http import JsonResponse, HttpResponse from django.db.models import Count from django.db.models import Q @@ -28,14 +28,47 @@ current_dir = os.getcwd() +# def getWebView(request): +# user_name = request.session.get('user', '') +# product_all = AutotestplatProduct.objects.filter(delete_flag='N') +# product_id = AuthUser.objects.filter(username=user_name).first().last_name +# product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name +# c = csrf(request) +# c.update({"product_name":product_name,"product_alls":product_all}) +# return render_to_response("web_testcase.html",c) + def getWebView(request): user_name = request.session.get('user', '') + + # 检查用户是否登录 + if not user_name: + # 用户未登录，重定向到登录页面或返回错误 + return redirect('/autotest/login/') + + # 获取用户对象并进行空值检查 + user = AuthUser.objects.filter(username=user_name).first() + if user is None: + # 用户不存在，返回错误 + return HttpResponse('用户不存在') + product_all = AutotestplatProduct.objects.filter(delete_flag='N') - product_id = AuthUser.objects.filter(username=user_name).first().last_name - product_name = AutotestplatProduct.objects.filter(id=product_id).first().product_name + product_id = user.last_name + + # 检查 product_id 是否存在 + if not product_id: + # 用户没有关联产品，可能需要处理 + product_name = "未关联产品" + else: + product = AutotestplatProduct.objects.filter(id=product_id).first() + if product is None: + product_name = "产品不存在" + else: + product_name = product.product_name + c = csrf(request) - c.update({"product_name":product_name,"product_alls":product_all}) - return render_to_response("web_testcase.html",c) + c.update({"product_name": product_name, "product_alls": product_all}) + return render_to_response("web_testcase.html", c) + @csrf_exempt def loadWebTestcaseTable(request): @@ -340,7 +373,12 @@ def runWebtestcase(request,web_testcase_code): testcase_template_playwright(user_product_id, web_testcase_code) else: testcase_template_selenium(user_product_id, web_testcase_code) - return HttpResponse("运行完成。 该用例为：" + str(web_testcase_code) + " " + "。" + "") + # return HttpResponse("运行完成。 该用例为：" + str(web_testcase_code) + " " + "。" + "") + return HttpResponse(json.dumps({ + 'status': 'success', + 'message': f'运行完成。该用例为：{web_testcase_code}', + 'web_testcase_code': web_testcase_code + }), content_type='application/json') @csrf_exempt def runAllTestcase(request): @@ -361,7 +399,12 @@ def runAllTestcase(request): testcase_template_playwright(user_product_id, web_testcase_code) else: testcase_template_selenium(user_product_id, web_testcase_code) - return HttpResponse("运行完成。 该用例为：" + str(web_testcase_code) + " " + "。" + "") + # return HttpResponse("运行完成。 该用例为：" + str(web_testcase_code) + " " + "。" + "") + return HttpResponse(json.dumps({ + 'status': 'success', + 'message': f'运行完成。该用例为：{web_testcase_code}', + 'web_testcase_code': web_testcase_code + }), content_type='application/json') def testcase_template_selenium(user_product_id,web_testcase_code): time.sleep(1) @@ -369,12 +412,15 @@ def testcase_template_selenium(user_product_id,web_testcase_code): print("autotest product_id %s start..." % user_product_id ) chrome_driver_path=r""+current_dir+"\chromedriver.exe" driver = webdriver.Chrome(executable_path=chrome_driver_path) + # 设置全局隐式等待时间 + driver.implicitly_wait(10) + driver.maximize_window() if(web_testcase_code==''): sql = "SELECT id,web_testcase_findmethod,web_testcase_evelement,web_testcase_optmethod,web_testcase_testdata,web_testcase_assertdata,`web_testcase_stepresult`,web_testcase_code from autotestplat_web_testcase where autotestplat_web_testcase.product_id=" + str(user_product_id) + " ORDER BY web_testcase_code_order ASC " caseResultInit(user_product_id, '未执行') else: sql = "SELECT id,web_testcase_findmethod,web_testcase_evelement,web_testcase_optmethod,web_testcase_testdata,web_testcase_assertdata,`web_testcase_stepresult`,web_testcase_code from autotestplat_web_testcase where autotestplat_web_testcase.web_testcase_code=" + str(web_testcase_code) + " ORDER BY id ASC " - coon = pymysql.connect(user='root', passwd='test123456', db='autotestplat', port=3306, host='127.0.0.1',charset='utf8') + coon = pymysql.connect(user='root', passwd='Liuzhichao@123', db='autotestplat', port=3306, host='127.0.0.1',charset='utf8') cursor = coon.cursor() aa = cursor.execute(sql) info = cursor.fetchmany(aa) @@ -481,7 +527,7 @@ def testcase_template_cypress(user_product_id,web_testcase_code): else: sql = "SELECT id,web_testcase_findmethod,web_testcase_evelement,web_testcase_optmethod,web_testcase_testdata,web_testcase_assertdata,`web_testcase_stepresult`,web_testcase_code from autotestplat_web_testcase where autotestplat_web_testcase.web_testcase_code=" + str( web_testcase_code) + " ORDER BY id ASC " - coon = pymysql.connect(user='root', passwd='test123456', db='autotestplat', port=3306, host='127.0.0.1', + coon = pymysql.connect(user='root', passwd='Liuzhichao@123', db='autotestplat', port=3306, host='127.0.0.1', charset='utf8') cursor = coon.cursor() aa = cursor.execute(sql) @@ -590,7 +636,7 @@ def testcase_template_playwright(user_product_id,web_testcase_code): else: sql = "SELECT id,web_testcase_findmethod,web_testcase_evelement,web_testcase_optmethod,web_testcase_testdata,web_testcase_assertdata,`web_testcase_stepresult`,web_testcase_code from autotestplat_web_testcase where autotestplat_web_testcase.web_testcase_code=" + str( web_testcase_code) + " ORDER BY id ASC " - coon = pymysql.connect(user='root', passwd='test123456', db='autotestplat', port=3306, host='127.0.0.1', + coon = pymysql.connect(user='root', passwd='Liuzhichao@123', db='autotestplat', port=3306, host='127.0.0.1', charset='utf8') cursor = coon.cursor() aa = cursor.execute(sql) @@ -690,7 +736,7 @@ def casestepResult(web_testcase_code, result): now = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())) sql = "UPDATE autotestplat_web_testcase set autotestplat_web_testcase.web_testcase_stepresult=%s,autotestplat_web_testcase.run_time=%s where autotestplat_web_testcase.id=%s;" param = (result, now, web_testcase_code) - coon = pymysql.connect(user='root', passwd='test123456', db='autotestplat', port=3306, host='127.0.0.1',charset='utf8') + coon = pymysql.connect(user='root', passwd='Liuzhichao@123', db='autotestplat', port=3306, host='127.0.0.1',charset='utf8') cursor = coon.cursor() cursor.execute(sql, param) coon.commit() @@ -703,7 +749,7 @@ def caseResult(web_testcase_code, result): sql = "UPDATE autotestplat_web_testcase set autotestplat_web_testcase.web_testcase_result=%s,autotestplat_web_testcase.run_time=%s where autotestplat_web_testcase.web_testcase_code=%s;" param = (result, now, web_testcase_code) print('web autotest result is ' + result.decode()) - coon = pymysql.connect(user='root', passwd='test123456', db='autotestplat', port=3306, host='127.0.0.1',charset='utf8') + coon = pymysql.connect(user='root', passwd='Liuzhichao@123', db='autotestplat', port=3306, host='127.0.0.1',charset='utf8') cursor = coon.cursor() cursor.execute(sql, param) coon.commit() @@ -716,7 +762,7 @@ def caseResultInit(user_product_id,result): sql = "UPDATE autotestplat_web_testcase set autotestplat_web_testcase.web_testcase_result=%s,autotestplat_web_testcase.web_testcase_stepresult=%s,autotestplat_web_testcase.run_time=%s where autotestplat_web_testcase.product_id=%s" param = (result, result, now, user_product_id) print('web autotest result init is ' + result.decode()) - coon = pymysql.connect(user='root', passwd='test123456', db='autotestplat', port=3306, host='127.0.0.1',charset='utf8') + coon = pymysql.connect(user='root', passwd='Liuzhichao@123', db='autotestplat', port=3306, host='127.0.0.1',charset='utf8') cursor = coon.cursor() cursor.execute(sql, param) coon.commit() diff --git a/chromedriver.exe b/chromedriver.exe index 9eedfd8d..f8d40efd 100644 Binary files a/chromedriver.exe and b/chromedriver.exe differ diff --git a/requirements.txt b/requirements.txt index 9852421a..1f4d8ff3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,24 +1,62 @@ -setuptools==28.8.0 -i https://pypi.tuna.tsinghua.edu.cn/simple -kombu==3.0.37 -i https://pypi.tuna.tsinghua.edu.cn/simple -celery==3.1.26.post2 -i https://pypi.tuna.tsinghua.edu.cn/simple -django==2.1.3 -i https://pypi.tuna.tsinghua.edu.cn/simple -django-celery==3.2.2 -i https://pypi.tuna.tsinghua.edu.cn/simple -redis==2.10.6 -i https://pypi.tuna.tsinghua.edu.cn/simple -celery-with-redis==3.0 -i https://pypi.tuna.tsinghua.edu.cn/simple -django-redis==4.0.0 -i https://pypi.tuna.tsinghua.edu.cn/simple -PyMySQL==1.0.2 -i https://pypi.tuna.tsinghua.edu.cn/simple -requests==2.15.1 -i https://pypi.tuna.tsinghua.edu.cn/simple -urllib3==1.22 -i https://pypi.tuna.tsinghua.edu.cn/simple -chardet==3.0.4 -i https://pypi.tuna.tsinghua.edu.cn/simple -certifi==2023.7.22 -i https://pypi.tuna.tsinghua.edu.cn/simple -idna==2.6 -i https://pypi.tuna.tsinghua.edu.cn/simple -adbutils==1.2.11 -i https://pypi.tuna.tsinghua.edu.cn/simple -django-cors-headers==3.0.0 -i https://pypi.tuna.tsinghua.edu.cn/simple -mysqlclient==1.4.6 -i https://pypi.tuna.tsinghua.edu.cn/simple -paramiko==3.3.1 -i https://pypi.tuna.tsinghua.edu.cn/simple -psutil==5.9.5 -i https://pypi.tuna.tsinghua.edu.cn/simple -pywin32==305 -i https://pypi.tuna.tsinghua.edu.cn/simple -faker==14.2.1 -i https://pypi.tuna.tsinghua.edu.cn/simple -uiautomator2==2.16.25 -i https://pypi.tuna.tsinghua.edu.cn/simple -weditor==0.6.4 -i https://pypi.tuna.tsinghua.edu.cn/simple -selenium==3.141.0 -i https://pypi.tuna.tsinghua.edu.cn/simple +adbutils==1.2.11 +amqp==1.4.9 +anyjson==0.3.3 +apkutils2==1.0.0 +Appium-Python-Client==1.3.0 +bcrypt==4.0.1 +billiard==3.3.0.23 +cached-property==1.5.2 +celery==3.1.26.post2 +celery-with-redis==3.0 +certifi==2023.7.22 +cffi==1.15.1 +chardet==3.0.4 +cigam==0.0.3 +colorama==0.4.5 +construct==2.10.70 +cryptography==40.0.2 +dataclasses==0.8 +decorator==5.1.1 +Deprecated==1.2.18 +deprecation==2.1.0 +Django==2.1.3 +django-celery==3.2.2 +django-cors-headers==3.0.0 +django-redis==4.10.0 +domain==0.1.4 +facebook-wda==1.5.4 +Faker==14.2.1 +filelock==3.4.1 +idna==2.6 +kombu==3.0.37 +logzero==1.7.0 +lxml==5.4.0 +packaging==21.3 +paramiko==3.3.1 +pdfkit==1.0.0 +Pillow==8.4.0 +progress==1.6.1 +psutil==5.9.5 +py==1.11.0 +pycparser==2.21 +pyelftools==0.32 +PyMySQL==1.0.2 +PyNaCl==1.5.0 +pyparsing==3.0.7 +python-dateutil==2.9.0.post0 +pytz==2026.1.post1 +pywin32==305 +PyYAML==6.0.1 +redis==2.10.6 +requests==2.15.1 +retry==0.9.2 +selenium==3.141.0 +six==1.17.0 +tornado==6.1 +typing_extensions==4.1.1 +uiautomator2==2.16.25 +urllib3==1.22 +weditor==0.6.4 +whichcraft==0.6.1 +wrapt==1.16.0 +xmltodict==0.15.0 diff --git a/start_beat.bat b/start_beat.bat index 9ba54fdd..f92131cf 100644 --- a/start_beat.bat +++ b/start_beat.bat @@ -1,4 +1,13 @@ @echo off -del /q /s "E:\Autotestplat\celerybeat.pid" +REM 切换到项目根目录 +cd /d D:\AutoTestProjects\Autotestplat + +REM 删除旧的 pid 文件（如果存在） +if exist "celerybeat.pid" ( + echo [INFO] Removing old celerybeat.pid file... + del /f /q "celerybeat.pid" + echo [INFO] Old pid file removed. +) + python manage.py celery beat pause \ No newline at end of file diff --git a/start_service.bat b/start_service.bat index d65cfa2a..19721a2e 100644 --- a/start_service.bat +++ b/start_service.bat @@ -1 +1 @@ -python manage.py runserver 0.0.0.0:80 \ No newline at end of file +python manage.py runserver 0.0.0.0:8000 \ No newline at end of file