# -*- coding: utf-8 -*-
|
"""
|
@File : report.py
|
@Time : 2023/2/13 09:42
|
@Author : geekbing
|
@LastEditTime : -
|
@LastEditors : -
|
@Description : 测试报告视图
|
"""
|
import json
|
import re
|
from ast import literal_eval
|
from shlex import quote
|
from typing import Dict
|
|
from django.core.exceptions import ObjectDoesNotExist
|
from django.db import transaction
|
from django.shortcuts import render
|
from django.utils.decorators import method_decorator
|
from django.utils import timezone
|
from rest_framework.permissions import AllowAny
|
from rest_framework.response import Response
|
from rest_framework.viewsets import GenericViewSet
|
|
from backend.utils import pagination
|
from lunarlink import models, serializers
|
from lunarlink.utils import response
|
from lunarlink.utils.convert2hrp import Hrp
|
from lunarlink.utils.decorator import request_log
|
|
|
class ConvertRequest:
|
@classmethod
|
def _to_curl(cls, request, compressed: bool = False, verify: bool = True):
|
"""Return string with curl command by provided request object
|
|
:param request:
|
:param compressed: If `True` then `--compressed` argument will be added to result
|
:param verify:
|
:return:
|
"""
|
|
parts = [
|
("curl", None),
|
("-X", request.method),
|
]
|
parts += [
|
(None, request.url),
|
]
|
|
for k, v in sorted(request.headers.items()):
|
parts += [("-H", "{0}: {1}".format(k, v))]
|
|
if request.body:
|
body = request.body
|
if isinstance(body, bytes):
|
body = body.decode("utf-8")
|
if isinstance(body, dict):
|
body = json.dumps(body)
|
parts += [("-d", body)]
|
|
if compressed:
|
parts += [("--compressed", None)]
|
|
if not verify:
|
parts += [("--insecure", None)]
|
|
flat_parts = []
|
for k, v in parts:
|
if k:
|
flat_parts.append(quote(k))
|
if v:
|
flat_parts.append(quote(v))
|
|
if k == "-H":
|
flat_parts.append(" \\\n")
|
|
return " ".join(flat_parts)
|
|
@classmethod
|
def _make_fake_req(cls, request_meta_dict):
|
class RequestMeta:
|
...
|
|
req = RequestMeta()
|
setattr(req, "method", request_meta_dict["method"])
|
setattr(req, "url", request_meta_dict["url"])
|
setattr(req, "headers", request_meta_dict["headers"])
|
body = request_meta_dict.get("body") or request_meta_dict.get("data")
|
setattr(req, "body", body)
|
return req
|
|
@classmethod
|
def to_curl(cls, req: Dict) -> str:
|
_req = cls._make_fake_req(req)
|
return cls._to_curl(_req, compressed=True, verify=False)
|
|
@classmethod
|
def to_hrp(cls, req: Dict) -> Dict:
|
hrp = Hrp(faster_req_json=req)
|
return hrp.get_testcase().dict()
|
|
@classmethod
|
def generate_curl(cls, report_details, convert_type=("curl",)):
|
for detail in report_details:
|
for record in detail["records"]:
|
meta_data = record["meta_data"]
|
for t in convert_type:
|
req = meta_data["request"]
|
method_name = f"to_{t}"
|
method = getattr(ConvertRequest, method_name)
|
record["meta_data"][t] = method(req)
|
|
|
class ReportView(GenericViewSet):
|
"""报告视图"""
|
|
queryset = models.Report.objects
|
serializer_class = serializers.ReportSerializer
|
pagination_class = pagination.MyPageNumberPagination
|
|
def get_authenticators(self):
|
# 查看报告详情不需要鉴权
|
pattern = re.compile(r"/api/lunarlink/reports/\d+")
|
if (
|
self.request.method == "GET"
|
and re.search(pattern, self.request.path) is not None
|
):
|
return [] # 不需要任何鉴权
|
return super().get_authenticators() # 默认所有鉴权
|
|
def get_permissions(self):
|
# 如果是look方法，不需要任何权限
|
if self.action == "look":
|
return [AllowAny()]
|
return super().get_permissions()
|
|
@method_decorator(request_log(level="DEBUG"))
|
def list(self, request):
|
"""获取测试报告列表"""
|
|
project = request.query_params.get("project")
|
search = request.query_params.get("search")
|
report_type = request.query_params.get("reportType")
|
report_status = request.query_params.get("reportStatus")
|
only_me = request.query_params.get("onlyMe")
|
|
queryset = (
|
self.get_queryset().filter(project__id=project).order_by("-update_time")
|
)
|
|
# 前端传过来是小写的字符串，不是python的True
|
if only_me == "true":
|
queryset = queryset.filter(creator=request.user)
|
|
if search != "":
|
queryset = queryset.filter(name__contains=search)
|
|
if report_type != "":
|
queryset = queryset.filter(type=report_type)
|
|
if report_status != "":
|
queryset = queryset.filter(status=report_status)
|
|
page_report = self.paginate_queryset(queryset)
|
serializer = self.get_serializer(page_report, many=True)
|
return self.get_paginated_response(serializer.data)
|
|
@method_decorator(request_log(level="INFO"))
|
def look(self, request, pk):
|
"""
|
查看报告
|
|
查看报告详情
|
"""
|
try:
|
report = models.Report.objects.get(id=pk)
|
report_detail = models.ReportDetail.objects.get(report__id=pk)
|
except ObjectDoesNotExist:
|
return Response(response.REPORT_NOT_EXISTS)
|
|
summary = json.loads(report.summary)
|
summary["details"] = literal_eval(report_detail.summary_detail)
|
ConvertRequest.generate_curl(summary["details"], convert_type=("curl",))
|
summary["html_report_name"] = report.name
|
|
return render(request, template_name="report_template.html", context=summary)
|
|
@method_decorator(request_log(level="INFO"))
|
def destroy(self, request, pk):
|
"""
|
单个删除
|
|
pk: id
|
"""
|
report_obj = models.Report.objects.filter(id=pk).first()
|
if not report_obj:
|
return Response(response.REPORT_NOT_EXISTS)
|
report_obj.is_deleted = True
|
report_obj.updater = request.user.id
|
report_obj.update_time = timezone.now()
|
report_obj.save() # 这将触发 pre_save 信号，并执行 delete_related_report_detail 处理器函数
|
|
return Response(response.REPORT_DEL_SUCCESS)
|
|
@method_decorator(request_log(level="INFO"))
|
def bulk_destroy(self, request):
|
"""
|
批量删除报告
|
|
[{id:int}]
|
"""
|
ids = [content["id"] for content in request.data]
|
# TODO: 如果有大量的数据，考虑性能问题
|
objs = list(models.Report.objects.filter(id__in=ids)) # 将QuerySet转换为list
|
if not objs:
|
return Response(response.REPORT_NOT_EXISTS)
|
|
try:
|
with transaction.atomic():
|
models.Report.objects.filter(id__in=ids).update(
|
is_deleted=True,
|
update_time=timezone.now(),
|
updater=request.user.id,
|
)
|
models.ReportDetail.objects.filter(report__in=objs).update(
|
is_deleted=True
|
)
|
except Exception as e:
|
return Response({"error": str(e)}, status=400)
|
|
return Response(response.REPORT_DEL_SUCCESS)
|
|
# TODO: 下载报告 待实现
|
@method_decorator(request_log(level="INFO"))
|
def download(self, request, **kwargs):
|
"""下载报告"""
|
pass
|
