# -*- coding: utf-8 -*-
|
"""
|
@File : CI.py
|
@Time : 2023/3/14 11:25
|
@Author : geekbing
|
@LastEditTime : -
|
@LastEditors : -
|
@Description : 持续集成CI视图
|
"""
|
|
import datetime
|
import json
|
import re
|
import time
|
|
from ast import literal_eval
|
from typing import Dict
|
|
import xmltodict
|
from django.http import HttpResponse
|
from django.core.exceptions import ObjectDoesNotExist
|
from django.conf import settings
|
from django_celery_beat.models import PeriodicTask
|
from drf_yasg.utils import swagger_auto_schema
|
|
from lunarlink import models
|
from lunarlink.utils import loader, qy_message
|
from lunarlink.utils.decorator import request_log
|
from django.utils.decorators import method_decorator
|
from rest_framework import status
|
from rest_framework.response import Response
|
from rest_framework.viewsets import GenericViewSet
|
from lunarlink.serializers import CISerializer, CIReportSerializer
|
from lunarlink.utils.loader import save_summary
|
from lunarlink.utils import response
|
|
|
def summary2junit(summary: Dict) -> Dict:
|
"""初始化JUnit的数据结构
|
:param summary:
|
:return:
|
"""
|
res = {
|
"testsuites": {
|
"testsuite": {
|
"errors": 0,
|
"failures": 0,
|
"hostname": "",
|
"name": "",
|
"skipped": 0,
|
"tests": 0,
|
"time": "0",
|
"timestamp": "20210524T18:04:50.941913",
|
"testcase": [],
|
}
|
}
|
}
|
|
time_info = summary.get("time")
|
res["testsuites"]["testsuite"]["time"] = time_info.get("duration")
|
start_at: str = time_info.get("start_at")
|
timestamp = datetime.datetime.fromtimestamp(int(float(start_at))).strftime(
|
"%Y-%m-%dT%H:%M:%S.%f"
|
)
|
res["testsuites"]["testsuite"]["timestamp"] = timestamp
|
|
details = summary.get("details", [])
|
res["testsuites"]["testsuite"]["tests"] = len(details)
|
for detail in details:
|
test_case = build_testcase(detail, res)
|
res["testsuites"]["testsuite"]["testcase"].append(test_case)
|
|
return res
|
|
|
def build_testcase(detail, res):
|
"""
|
构建Junit Testcase
|
:param detail:
|
:param res:
|
:return:
|
"""
|
test_case = {"classname": "", "file": "", "line": "", "name": "", "time": ""}
|
name = detail.get("name")
|
test_case["classname"] = name # 对应junit的Suite
|
records = detail.get("records")
|
test_case["line"] = len(records)
|
test_case["time"] = detail["time"]["duration"]
|
result = detail.get("success")
|
step_names = []
|
for index, record in enumerate(records):
|
step_names.append(f"{index}-{record['name']}")
|
test_case["name"] = "\n".join(step_names) # 对应junit的每个case的Name
|
|
# 记录错误case的详细信息
|
if result is False:
|
case_error, failure_details = build_failure_detail(records)
|
if case_error:
|
res["testsuites"]["testsuite"]["errors"] += 1
|
else:
|
res["testsuites"]["testsuite"]["failures"] += 1
|
|
failure = {"message": "断言或者抽取失败", "#text": "\n".join(failure_details)}
|
test_case["failure"] = failure
|
|
return test_case
|
|
|
def build_failure_detail(records):
|
"""记录错误case的详细信息
|
:param records:
|
:return:
|
"""
|
case_error = False
|
failure_details = []
|
for index, record in enumerate(records):
|
step_status = record.get("status")
|
if step_status == "failure":
|
failure_details.append(
|
f"{index}-{record['name']}\n{record.get('attachment')}\n{'*' * 68}"
|
)
|
elif step_status == "error":
|
case_error = True
|
return case_error, failure_details
|
|
|
class CIView(GenericViewSet):
|
authentication_classes = []
|
serializer_class = CISerializer
|
pagination_class = None
|
|
@swagger_auto_schema(operation_summary="gitlab-ci触发自动化用例运行")
|
@method_decorator(request_log(level="INFO"))
|
def run_ci_tests(self, request):
|
"""
|
gitlab-ci发送请求
|
|
测试平台解析参数，定时任务中的ci_project_ic和ci_env同时匹配时，返回触发执行任务
|
"""
|
ser = CISerializer(data=request.data)
|
if ser.is_valid():
|
task_name = "lunarlink.tasks.schedule_debug_suite"
|
ci_project_id: int = ser.validated_data.get("ci_project_id")
|
ci_env: str = ser.validated_data.get("env")
|
|
query = PeriodicTask.objects.filter(enabled=1, task=task_name)
|
pk_kwargs_list = query.values("pk", "kwargs", "description")
|
enabled_task_ids = []
|
project = None
|
for pk_kwargs in pk_kwargs_list:
|
pk: int = pk_kwargs["pk"]
|
kwargs: dict = json.loads(pk_kwargs["kwargs"])
|
ci_project_ids: list = literal_eval(
|
kwargs.get("ci_project_ids") or "[]"
|
)
|
if isinstance(ci_project_ids, int):
|
ci_project_ids = [ci_project_ids]
|
|
# 定时任务中的ci_project_id和ci_env同时匹配
|
if (
|
ci_project_id in ci_project_ids
|
and ci_env
|
and ci_env == kwargs.get("ci_env")
|
):
|
enabled_task_ids.append(pk)
|
project = pk_kwargs["description"]
|
|
# 没有匹配用例，直接返回
|
if not enabled_task_ids:
|
timestamp = datetime.datetime.fromtimestamp(int(time.time())).strftime(
|
"%Y-%m-%dT%H:%M:%S.%f"
|
)
|
not_found_case_res = {
|
"testsuites": {
|
"testsuite": {
|
"errors": 0,
|
"failures": 0,
|
"hostname": "",
|
"name": "没有找到匹配的用例",
|
"skipped": 0,
|
"tests": 0,
|
"time": "0",
|
"timestamp": timestamp,
|
"testcase": [],
|
}
|
}
|
}
|
|
xml_data = xmltodict.unparse(not_found_case_res)
|
return HttpResponse(xml_data, content_type="text/xml")
|
|
test_sets = []
|
suite_list = []
|
config_list = []
|
config = None
|
webhook_set = set()
|
override_config_body = None
|
for task_id in enabled_task_ids:
|
task_obj = query.filter(id=task_id).first()
|
# 如果task中存在重载配置，就覆盖用例中的配置
|
override_config = json.loads(task_obj.kwargs).get("config")
|
if (
|
override_config_body is None
|
and override_config
|
and override_config != "请选择"
|
):
|
try:
|
override_config_body = literal_eval(
|
models.Config.objects.get(
|
name=override_config, project__id=project
|
).body
|
)
|
except ObjectDoesNotExist:
|
return Response(response.CONFIG_NOT_EXISTS)
|
|
if task_obj:
|
# 判断webhook是否合法
|
case_ids = task_obj.args
|
url = json.loads(task_obj.kwargs).get("webhook")
|
url_pattern = r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"
|
if re.match(url_pattern, url):
|
webhook_set.add(url)
|
else:
|
continue
|
# 反查出一个task中包含的所有用例
|
suite = list(
|
models.Case.objects.filter(pk__in=literal_eval(case_ids))
|
.order_by("id")
|
.values("id", "name")
|
)
|
for case in suite:
|
case_step_list = (
|
models.CaseStep.objects.filter(case__id=case["id"])
|
.order_by("step")
|
.values("body")
|
)
|
testcase_list = []
|
for case_step in case_step_list:
|
body = literal_eval(case_step["body"])
|
if body["request"].get("url"):
|
testcase_list.append(body)
|
elif config is None and body["request"].get("base_url"):
|
# 当前步骤时配置
|
# 如果task中存在重载配置，就覆盖用例中的配置
|
if override_config_body:
|
config = override_config_body
|
else:
|
try:
|
config = literal_eval(
|
models.Config.objects.get(
|
name=body["name"], project__id=project
|
).body
|
)
|
except ObjectDoesNotExist:
|
return Response(response.CONFIG_NOT_EXISTS)
|
config_list.append(config)
|
test_sets.append(testcase_list)
|
config = None
|
suite_list.extend(suite)
|
override_config_body = None
|
# 同步运行用例
|
summary, _ = loader.debug_suite(
|
suite=test_sets,
|
project=project,
|
obj=suite_list,
|
config=config_list,
|
save=False,
|
)
|
ci_project_namespace = ser.validated_data["ci_project_namespace"]
|
ci_project_name = ser.validated_data["ci_project_name"]
|
ci_job_id = ser.validated_data["ci_job_id"]
|
summary["name"] = f"{ci_project_namespace}_{ci_project_name}_job{ci_job_id}"
|
|
report_id = save_summary(
|
name=summary.get("name"),
|
summary=summary,
|
project=project,
|
report_type=4,
|
user=ser.validated_data["start_job_user"],
|
ci_metadata=ser.validated_data,
|
)
|
junit_results = summary2junit(summary)
|
xml_data = xmltodict.unparse(junit_results)
|
summary["task_name"] = "gitlab-ci_" + summary.get("name")
|
summary["report_id"] = report_id
|
for webhook in webhook_set:
|
# TODO: 还需要优化企微发送，加入ci集成参数
|
qy_message.send_message(
|
summary=summary,
|
webhook=webhook,
|
ci_job_url=ser.validated_data["ci_job_url"],
|
ci_pipeline_url=ser.validated_data["ci_pipeline_url"],
|
case_count=junit_results["testsuites"]["testsuite"]["tests"],
|
)
|
return HttpResponse(xml_data, content_type="text/xml")
|
else:
|
return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
@swagger_auto_schema(
|
query_serializer=CIReportSerializer(),
|
operation_summary="获取gitlab-ci运行的报告url",
|
)
|
def get_ci_report_url(self, request):
|
"""获取gitlab-ci运行的报告url"""
|
ser = CIReportSerializer(data=request.query_params)
|
if ser.is_valid():
|
ci_job_id = ser.validated_data["ci_job_id"]
|
report_obj = models.Report.objects.filter(ci_job_id=ci_job_id).first()
|
if report_obj:
|
report_url = f"{settings.BASE_REPORT_URL}/{report_obj.id}/"
|
else:
|
return Response(data=f"查找的ci_job_id: {ci_job_id}不存在")
|
return Response(data=report_url)
|
else:
|
return Response(data=ser.errors, status=status.HTTP_400_BAD_REQUEST)
|
