import datetime from rest_framework.response import Response from rest_framework.views import APIView from django.http import HttpResponse from TestLaboratory.settings import PLAN_ROOT, SOFTWARE_ROOT, HTTP_HEAD, TASK_ROOT, CASE_TEMPLATE_ROOT, CASE_FILE_ROOT from apps.file.models import FileManager from apps.log.models import get_log, gen_log from apps.task.models import TestTask from apps.user.middleware.rolecontrol import RoleControl from apps.user.models import User from utils.util_add_id import get_id from utils.util_file.util_fileio import write_file import logging logger = logging.getLogger('django') class TaskView(APIView): # 登录权限验证 authentication_classes = [] # delete task @staticmethod @RoleControl def delete(request, task_id, *args, **kwargs): task = TestTask.objects.filter(id=task_id, delete=False) if not task: logger.error("测试任务已删除或不存在") return HttpResponse(status=404, content='测试任务已删除或不存在') task = task[0] task_sids = task.statement_file.split('&&') if task_sids[0]: for statement_id in task_sids: statement = FileManager.objects.get(id=statement_id) statement.delete = True statement.save() task.delete = True task.save() executor, action, method = get_log(request) gen_log(action, "测试任务", task.title, method, executor) return HttpResponse("删除完成") # view task @staticmethod @RoleControl def get(request, task_id, *args, **kwargs): task = TestTask.objects.filter(id=task_id, delete=False) if not task: logger.error("测试计划已删除或不存在") return HttpResponse(status=404, content='测试计划已删除或不存在') task = task[0] version = task.plan.software.version_set.filter(number=task.plan.version, delete=False) if not version: logger.error("测试任务对应软件版本不存在或已删除") return HttpResponse(status=404, content='测试任务对应软件版本不存在或已删除') version = version[0] version_fid = version.version_file version_file = FileManager.objects.filter(id=version_fid) if not version_file: logger.error("无法定位测试任务对应的软件实体") return HttpResponse(status=404, content='无法定位测试任务对应的软件实体') version_file = version_file[0] plan_sids = task.plan.statement_file.split('&&') plan_file_names = [] plan_file_urls = [] if plan_sids[0]: for plan_statement_id in plan_sids: plan_statement = FileManager.objects.get(id=plan_statement_id) plan_file_urls.append(PLAN_ROOT + plan_statement.path) plan_file_names.append(plan_statement.path.split('/')[-1]) task_sids = task.statement_file.split('&&') task_file_names = [] task_file_urls = [] if task_sids[0]: for task_statement_id in task_sids: task_statement = FileManager.objects.get(id=task_statement_id) task_file_urls.append(TASK_ROOT + task_statement.path) task_file_names.append(task_statement.path.split('/')[-1]) case_file_info = {} if task.case_file: case_file = FileManager.objects.get(id=task.case_file) case_file_info = {'file_name': case_file.path.split('/')[-1], 'file_url': HTTP_HEAD + CASE_FILE_ROOT + case_file.path} executor, action, method = get_log(request) gen_log(action, "测试任务", task.title, method, executor) return Response({ 'software_id': task.plan.software.id, 'software_name': task.plan.software.name, 'version': task.plan.version, 'version_file': {'file_name': version_file.path.split('/')[-1], 'file_url': HTTP_HEAD + SOFTWARE_ROOT + version_file.path}, 'plan': { 'id': task.plan.id, 'title': task.plan.title, 'description': task.plan.description, 'state': task.plan.state, 'statements': [{'file_id': file_sid, 'file_name': file_name, 'file_url': HTTP_HEAD + file_url} for file_sid, file_name, file_url in zip(plan_sids, plan_file_names, plan_file_urls)], 'creator': {'executor_id': task.plan.creator.id, 'executor_name': task.plan.creator.username}, }, 'task': { 'id': task.id, 'title': task.title, 'description': task.description, 'state': task.state, 'statements': [{'file_id': file_sid, 'file_name': file_name, 'file_url': HTTP_HEAD + file_url} for file_sid, file_name, file_url in zip(task_sids, task_file_names, task_file_urls)], 'case_template': HTTP_HEAD + CASE_TEMPLATE_ROOT, 'case_file': case_file_info, 'executor': {'id': task.executor.id, 'name': task.executor.username}, 'case_all': task.case_all, 'case_not_execute': task.case_not_execute, 'case_success': task.case_success, 'case_fail': task.case_fail, }, 'create_time': task.create_time, 'update_time': task.update_time, }) class TaskEdit(APIView): # 登录权限验证 authentication_classes = [] # edit task @staticmethod @RoleControl def post(request, task_id, *args, **kwargs): title = request.POST.get('title') state = request.POST.get('state') description = request.POST.get('description') files = request.FILES.getlist('files') edit_state_only = request.POST.get('edit_state_only') task = TestTask.objects.filter(id=task_id, delete=False) if not task: logger.error("测试任务已删除或不存在") return HttpResponse(status=404, content='测试任务已删除或不存在') task = task[0] if edit_state_only == '1': task.state = int(state) task.save() return HttpResponse(status=200, content='编辑已保存') if title: task.title = title if state: task.state = int(state) if description: task.description = description fids = '' for file in files: create_time = update_time = datetime.datetime.now() try: write_file(file, TASK_ROOT + "/" + task_id + "/" + str(create_time)) except: logger.error("文件上传失败") return HttpResponse(status=500, content="文件上传失败") id_file = get_id(FileManager, "TestLaboratory_V1_File_1") try: file_new = FileManager.objects.create(id=id_file, category="plan_statement", path=task_id + "/" + str( create_time) + "/" + file.name, create_time=create_time, update_time=update_time) except: logger.error("文件路径写入数据库失败") return HttpResponse(status=500, content="文件路径写入数据库失败") fids += id_file + '&&' fids = fids[:-2] task_sids = task.statement_file.split('&&') print(task_sids) if task_sids is not None and task_sids[0] != '': for statement_id in task_sids: statement = FileManager.objects.get(id=statement_id) statement.delete = True statement.save() task.statement_file = fids task.save() executor, action, method = get_log(request) gen_log(action, "测试任务", task.title, method, executor) return HttpResponse("编辑已保存")