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
from apps.user.middleware.rolecontrol import RoleControl
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 = []

    # 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 = TestCase.objects.filter(task=task, delete=False)
            file_name = task.title+'-全部测试用例.xls'
        elif export_type == '1':
            cases = TestCase.objects.filter(~Q(state=0), task=task, delete=False)
            file_name = task.title+'-已执行测试用例.xls'
        elif export_type == '2':
            cases = TestCase.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 = [['被测软件ID：', task.plan.software.id],
                      ['被测软件名：', task.plan.software.name],
                      ['被测软件版本', task.plan.version],
                      ['编号', '名称', '前置条件', '优先级', '测试环境', '测试类型', '测试步骤', '预期结果', '实际结果', '状态',
                       '编写人', '评审员', '执行人', '备注']]

        for case in cases:
            state = '未执行'
            if case.state == 1:
                state = '执行成功'
            elif case.state == 2:
                state = '执行失败'
            cases_info.append([case.id_in_task, case.name, case.requisite, case.priority, case.environment,
                               case.type, case.process, case.expected_result, case.actual_result, state,
                               case.writer, case.assessor, case.executor, 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)