LimsBackend/apps/task/views/caseview.py

import datetime
import os

from django.db.models import Q
from django.http import HttpResponse, HttpResponseBadRequest
from rest_framework.response import Response
from rest_framework.views import APIView

from TestLaboratory.settings import CASE_FILE_ROOT, CASE_FILE_EXPORT, HTTP_HEAD
from apps.file.models import FileManager
from apps.log.models import get_log, gen_log
from apps.task.models import TestTask, TestCase, BeidouCase
from apps.user.middleware.rolecontrol import RoleControl
from apps.user.models import User
from util_xlsx import load_xlsx, write_to_xlsx
from utils.util_add_id import get_id
from utils.util_file.util_fileio import write_file

import logging
logger = logging.getLogger('django')


class CaseView(APIView):
    # 登录权限验证
    authentication_classes = []

    @staticmethod
    def convert_to_string(word_list):
        for i in range(len(word_list)):
            if word_list[i] is None:
                word_list[i] = ''
            elif not isinstance(word_list[i], str):
                word_list[i] = str(word_list[i])
        return word_list

    # get testcase
    @staticmethod
    # @RoleControl
    def get(request, task_id, *args, **kwargs):
        export_type = request.GET.get('type')

        task = TestTask.objects.filter(id=task_id, delete=False)
        if not task:
            logger.error("测试任务已删除或不存在")
            return HttpResponse(status=404, content='测试任务已删除或不存在')
        task = task[0]

        cases = None
        file_name = None
        if export_type == '0':
            cases = BeidouCase.objects.filter(task=task, delete=False)
            file_name = task.title+'-全部测试用例.xls'
        elif export_type == '1':
            cases = BeidouCase.objects.filter(~Q(state=0), task=task, delete=False)
            file_name = task.title+'-已执行测试用例.xls'
        elif export_type == '2':
            cases = BeidouCase.objects.filter(state=2, task=task, delete=False)
            file_name = task.title+'-不合格测试用例.xls'

        if not cases:
            logger.error("无符合要求的测试用例")
            return HttpResponse(status=404, content='无符合要求的测试用例')

        update_time = datetime.datetime.now()
        file_dir = CASE_FILE_EXPORT + task_id + "/" + str(update_time)
        if not os.path.exists(file_dir):
            os.makedirs(file_dir)
        file_path = file_dir + "/" + file_name

        cases_info = [
            ['测试项目', '版本', '测试内容', '定位精度', '定时精度', '伪距测量精度', '载波测量精度', '测速精度',
             '捕获灵敏度', '接收灵敏度', '跟踪灵敏度', '功率范围', '自主完好性', '系统完好性', '首次捕获时间',
             '失锁重捕时间', '定位测速更新率', '跟踪通道数', '最大速度', '最大加速度', '执行结果', '测试人员', '备注']]

        for case in cases:
            power = ''
            if case.power_range_low and case.power_range_high:
                power = str(case.power_range_low) + '-' + str(case.power_range_high)
            state = '成功'
            if case.state == 2:
                state = '失败'
            executor_name = User.objects.filter(id=case.executor_id)[0].username

            cases_info.append(CaseView.convert_to_string(
                [case.software.name, case.version, case.title, case.positioning_accuracy, case.timing_accuracy,
                 case.pseudo_range_measurement_accuracy, case.carrier_measurement_accuracy,
                 case.speed_measurement_accuracy, case.sensitivity_capture, case.receiving_sensitivity,
                 case.sensitivity_track, power, case.self_integrity, case.system_integrity, case.capture_time,
                 case.recapture_time, case.update_rate, case.trace_channels, case.max_velocity, case.max_acceleration,
                 state, executor_name, case.remark]
            ))

        sio = write_to_xlsx(file_path, cases_info)
        sio.seek(0)
        response = HttpResponse(sio.getvalue(), content_type='application/vnd.ms-excel')
        # response['Content-Disposition'] = 'attachment; filename*=UTF-8''{}'.format(escape_uri_path(file_name))
        response['Content-Disposition'] = 'attachment; filename="{0}"'.format(file_name).encode('utf-8', 'ISO-8859-1')
        response.write(sio.getvalue())
        # executor, action, method = get_log(request)
        # gen_log(action, "测试任务", task.title, method, executor)
        return response


class CaseExecute(APIView):
    # 登录权限验证
    authentication_classes = []

    # execute task
    @staticmethod
    @RoleControl
    def post(request, task_id, *args, **kwargs):
        executor, action, method = get_log(request)
        task = TestTask.objects.filter(id=task_id, delete=False)
        if not task:
            logger.error("测试任务已删除或不存在")
            return HttpResponseBadRequest(status=404, content='测试任务已删除或不存在')
        task = task[0]

        file = request.FILES.get('file')

        create_time = update_time = datetime.datetime.now()
        write_file(file, CASE_FILE_ROOT + task_id + "/" + str(create_time))

        return_info = load_xlsx(
            CASE_FILE_ROOT + task_id + "/" + str(create_time) + '/' + file.name,
            task)
        if return_info['code'] != 200:
            logger.error(str(return_info['info']))
            return Response({'status': return_info['code'],
                             'error': str(return_info['info'])})

        case_info_dict_list = []
        case_infos = return_info['info']

        total_num = len(case_infos)
        not_execute_num = success_num = fail_num = 0
        id_error = []
        for i in range(len(case_infos)):
            case_info_list = case_infos[i]

            cases_same_id = TestCase.objects.filter(id=task.id,
                                                    id_in_task=str(
                                                        case_info_list[0]),
                                                    delete=False)
            if cases_same_id:
                case_name = cases_same_id[0].name
                if case_name != case_info_list[1]:
                    id_error.append('第{}行id为{}的测试用例名称与前版本不同，请检查错误'
                                    .format(i + 3, cases_same_id[0].id_in_task))
            state = 0
            if case_info_list[9] == '未执行':
                not_execute_num += 1
            if case_info_list[9] == '执行成功':
                state = 1
                success_num += 1
            elif case_info_list[9] == '执行失败':
                state = 2
                fail_num += 1
            case_info = {
                'id_in_task': str(case_info_list[0]),
                'name': case_info_list[1],
                'requisite': case_info_list[2],
                'priority': case_info_list[3],
                'environment': case_info_list[4],
                'type': case_info_list[5],
                'process': case_info_list[6],
                'expected_result': case_info_list[7],
                'actual_result': case_info_list[8],
                'state': state,
                'writer': case_info_list[10],
                'assessor': case_info_list[11],
                'executor': case_info_list[12],
                'task': task,
                'remark': case_info_list[13],
                'software': task.plan.software,
                'version': task.plan.version
            }
            case_info_dict_list.append(case_info)
        if id_error:
            logger.error("id错误")
            return Response({'status': 50001, 'error': id_error})
        # 在创建测试用例阶段，每次创建都为新建，新建之前删除task下的所有测试用例
        if task.state == 0:
            cases_before = task.testcase_set.filter(delete=False)
            for case_before in cases_before:
                case_before.delete = True
                case_before.save()
            for case_info in case_info_dict_list:
                case_info['id'] = get_id(TestCase, "TestLaboratory_V1_Case_1")
                case_info['create_time'] = create_time
                case_info['update_time'] = update_time
                case_info['executor_id_id'] = executor.id
                TestCase.objects.create(**case_info)

        elif task.state == 1:
            for i in range(len(case_info_dict_list)):
                case_info = case_info_dict_list[i]
                case = TestCase.objects.filter(delete=False, task=task,
                                               id_in_task=case_info[
                                                   'id_in_task'])
                if not case:
                    logger.error(
                        'id为{}的测试用例不存在'.format(case_info['id_in_task']))
                    return Response({'status': 50001,
                                     'error': 'id为{}的测试用例不存在'.format(
                                         case_info['id_in_task'])})
                elif len(case) > 1:
                    logger.error(
                        'id为{}的测试用例存在多个'.format(case_info['id_in_task']))
                    return Response({'status': 50001,
                                     'error': 'id为{}的测试用例存在多个'.format(
                                         case_info['id_in_task'])})
                case.update(**case_info)
                case[0].save()

        file_id = get_id(FileManager, "TestLaboratory_V1_File_1")
        try:
            file_new = FileManager.objects.create(id=file_id,
                                                  category="case_file",
                                                  path=task_id + "/" + str(
                                                      create_time) + "/" + file.name,
                                                  create_time=create_time,
                                                  update_time=update_time)
        except:
            logger.error("文件路径写入数据库失败")
            return Response({'status': 50001, 'error': "文件路径写入数据库失败"})

        task.case_file = file_id
        task.case_all = len(task.testcase_set.filter(delete=False))
        task.case_not_execute = len(
            task.testcase_set.filter(delete=False, state=0))
        task.case_success = len(task.testcase_set.filter(delete=False, state=1))
        task.case_fail = len(task.testcase_set.filter(delete=False, state=2))
        print(task.case_all, task.case_not_execute, task.case_success,
              task.case_fail)

        task.save()
        executionInfo = [
            {
                'operationName': 'filename',
                'content': file.name
            },
            {
                'operationName': 'fileurl',
                'content': HTTP_HEAD + CASE_FILE_ROOT + file_new.path
            },
            {
                'operationName': 'state',
                'content': task.state
            },
            {
                'operationName': 'updateTime',
                'content': update_time
            },
            {
                'operationName': 'totalNum',
                'content': task.case_all
            },
            {
                'operationName': 'notExecuteNum',
                'content': task.case_not_execute
            },
            {
                'operationName': 'successNum',
                'content': task.case_success
            },
            {
                'operationName': 'failNum',
                'content': task.case_fail
            }
        ]


        gen_log(action, "测试任务", task.title, method, executor)

        return Response(executionInfo)