Merge pull request wuranxu#3 from wuranxu/feature/test_report

Feature/test report

app/dao/__init__.py

@@ -7,5 +7,7 @@
 from app.models.environment import Environment
 from app.models.gconfig import GConfig
 from app.models.constructor import Constructor
+from app.models.report import PityReport
+from app.models.result import PityTestResult
 
 Base.metadata.create_all(engine)

app/dao/test_case/TestReport.py

@@ -0,0 +1,50 @@
+from datetime import datetime
+
+from sqlalchemy import select
+
+from app.models import async_session
+from app.models.report import PityReport
+from app.utils.logger import Log
+
+
+class TestReportDao(object):
+    log = Log("TestReportDao")
+
+    @staticmethod
+    async def start(executor: int, env: int, mode: int = 0) -> int:
+        """
+        生成buildId，开始执行任务，任务完成后通过回调方法更新报告
+        :return: 返回report_id
+        """
+        try:
+            async with async_session() as session:
+                async with session.begin():
+                    report = PityReport(executor, env, mode=mode)
+                    session.add(report)
+                    await session.flush()
+                    return report.id
+        except Exception as e:
+            TestReportDao.log.error(f"新增报告失败, error: {e}")
+            raise Exception("新增报告失败")
+
+    @staticmethod
+    async def end(report_id: int, success_count: int, failed_count: int,
+                  error_count: int, skipped_count: int, status: int) -> None:
+        try:
+            async with async_session() as session:
+                async with session.begin():
+                    sql = select(PityReport).where(PityReport.id == report_id)
+                    data = await session.execute(sql)
+                    report = data.scalars().first()
+                    if report is None:
+                        raise Exception("获取报告失败")
+                    report.status = status
+                    report.success_count = success_count
+                    report.failed_count = failed_count
+                    report.error_count = error_count
+                    report.skipped_count = skipped_count
+                    report.finished_at = datetime.now()
+                    await session.flush()
+        except Exception as e:
+            TestReportDao.log.error(f"更新报告失败, error: {e}")
+            raise Exception("更新报告失败")

app/dao/test_case/TestResult.py

@@ -0,0 +1,29 @@
+from datetime import datetime
+
+from app.models import async_session
+from app.models.result import PityTestResult
+from app.utils.logger import Log
+
+
+class TestResultDao(object):
+    log = Log("TestResultDao")
+
+    @staticmethod
+    async def insert(report_id: int, case_id: int, status: int,
+                     case_log: str, start_at: datetime, finished_at: datetime,
+                     url: str, body: str, request_method: str, cost: str,
+                     asserts: str, response_headers: str, response: str,
+                     status_code: int, cookies: str, retry: int = None, ) -> None:
+        try:
+            async with async_session() as session:
+                async with session.begin():
+                    report = PityTestResult(report_id, case_id, status,
+                                            case_log, start_at, finished_at,
+                                            url, body, request_method, cost,
+                                            asserts, response_headers, response,
+                                            status_code, cookies, retry)
+                    session.add(report)
+                    await session.flush()
+        except Exception as e:
+            TestResultDao.log.error(f"新增测试结果失败, error: {e}")
+            raise Exception("新增测试结果失败")

app/middleware/AsyncHttpClient.py

@@ -37,6 +37,15 @@ async def get_resp(resp):
             data = await resp.text()
         return data
 
+    @staticmethod
+    def get_request_data(body):
+        request_body = body
+        if isinstance(body, bytes):
+            request_body = request_body.decode()
+        if isinstance(request_body, str) or request_body is None:
+            return request_body
+        return json.dumps(request_body, ensure_ascii=False)
+
     @staticmethod
     async def collect(status, request_data, status_code=200, response=None, response_header=None,
                       request_header=None, cookies=None, elapsed=None, msg="success"):
@@ -58,9 +67,7 @@ async def collect(status, request_data, status_code=200, response=None, response
         cookies = {k: v for k, v in cookies.items()} if cookies is not None else {}
         return {
             "status": status, "response": response, "status_code": status_code,
-            "request_data": request_data if isinstance(request_data, str) or request_data is None else json.dumps(
-                request_data,
-                ensure_ascii=False),
+            "request_data": AsyncRequest.get_request_data(request_data),
             "response_header": response_header, "request_header": request_header,
             "msg": msg, "elapsed": elapsed, "cookies": cookies,
         }

app/models/report.py

@@ -0,0 +1,53 @@
+from datetime import datetime
+
+from sqlalchemy import INT, Column, DATETIME
+from sqlalchemy.dialects.mysql import SMALLINT
+
+from app.models import Base
+
+
+class PityReport(Base):
+    __tablename__ = 'pity_report'
+    id = Column(INT, primary_key=True)
+    # 执行人 0则为CPU
+    executor = Column(INT, index=True)
+
+    # 环境
+    env = Column(INT, nullable=False)
+
+    # 测试集合id，预留字段
+    plan_id = Column(INT, index=True, nullable=True)
+    # 开始时间
+    start_at = Column(DATETIME, nullable=False)
+    # 结束时间
+    finished_at = Column(DATETIME)
+    # 成功数量
+    success_count = Column(INT, nullable=False, default=0)
+    error_count = Column(INT, nullable=False, default=0)
+    failed_count = Column(INT, nullable=False, default=0)
+    skipped_count = Column(INT, nullable=False, default=0)
+
+    # 执行状态
+    status = Column(SMALLINT, nullable=False, comment="0: pending, 1: running, 2: stopped, 3: finished", index=True)
+
+    # case执行模式
+    mode = Column(SMALLINT, default=0, comment="0: 手动, 1: 自动, 2: 测试集, 3: pipeline, 4: 其他")
+
+    deleted_at = Column(DATETIME, index=True)
+
+    def __init__(self, executor: int, env: int, success_count: int = 0, failed_count: int = 0,
+                 error_count: int = 0, skipped_count: int = 0, status: int = 0, mode: int = 0,
+                 plan_id: int = None, finished_at: datetime = None):
+        self.executor = executor
+        self.env = env
+        self.start_at = datetime.now()
+        self.success_count = success_count
+        self.failed_count = failed_count
+        self.error_count = error_count
+        self.skipped_count = skipped_count
+        self.status = status
+        self.mode = mode
+        self.status = status
+        self.plan_id = plan_id
+        self.finished_at = finished_at
+        self.deleted_at = None

app/models/result.py

@@ -0,0 +1,76 @@
+from datetime import datetime
+
+from sqlalchemy import INT, Column, DATETIME, String
+from sqlalchemy import SMALLINT
+from sqlalchemy import TEXT
+
+from app.models import Base
+
+
+class PityTestResult(Base):
+    __tablename__ = 'pity_test_result'
+    id = Column(INT, primary_key=True)
+
+    # 报告id
+    report_id = Column(INT, index=True)
+
+    # case_id
+    case_id = Column(INT, index=True)
+
+    status = Column(SMALLINT, comment="对应状态 0: 成功 1: 失败 2: 出错 3: 跳过")
+
+    # 开始时间
+    start_at = Column(DATETIME, nullable=False)
+    # 结束时间
+    finished_at = Column(DATETIME, nullable=False)
+
+    case_log = Column(TEXT)
+
+    # 重试次数，预留字段
+    retry = Column(INT, default=0)
+
+    # http状态码
+    status_code = Column(INT)
+
+    url = Column(TEXT, nullable=False)
+
+    body = Column(TEXT)
+
+    request_method = Column(String(12), nullable=True)
+
+    cost = Column(String(12), nullable=False)
+
+    asserts = Column(TEXT)
+
+    response_headers = Column(TEXT)
+
+    response = Column(TEXT)
+
+    cookies = Column(TEXT)
+
+    deleted_at = Column(DATETIME, index=True)
+
+    def __init__(self, report_id: int, case_id: int, status: int,
+                 case_log: str, start_at: datetime, finished_at: datetime,
+                 url: str, body: str, request_method: str, cost: str,
+                 asserts: str, response_headers: str, response: str,
+                 status_code: int, cookies: str, retry: int = None,
+                 ):
+        self.report_id = report_id
+        self.case_id = case_id
+        self.status = status
+        self.case_log = case_log
+        self.start_at = start_at
+        self.finished_at = finished_at
+        self.status = status
+        self.retry = retry
+        self.status_code = status_code
+        self.url = url
+        self.request_method = request_method
+        self.body = body
+        self.cost = cost
+        self.response = response
+        self.response_headers = response_headers
+        self.asserts = asserts
+        self.cookies = cookies
+        self.deleted_at = None

app/routers/request/http.py

@@ -59,6 +59,12 @@ async def execute_case(case_id: List[int], user_info=Depends(Permission())):
     return dict(code=0, data=data, msg="操作成功")
 
 
+@router.post("/run/multiple")
+async def execute_as_report(case_id: List[int], user_info=Depends(Permission())):
+    report_id = await Executor.run_multiple(user_info['id'], 1, case_id)
+    return dict(code=0, data=report_id, msg="操作成功")
+
+
 async def run_single(case_id: int, data: Dict[int, tuple]):
     executor = Executor()
     data[case_id] = await executor.run(case_id)

app/utils/executor.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import re
 from datetime import datetime
@@ -6,6 +7,8 @@
 from app.dao.config.GConfigDao import GConfigDao
 from app.dao.test_case.TestCaseAssertsDao import TestCaseAssertsDao
 from app.dao.test_case.TestCaseDao import TestCaseDao
+from app.dao.test_case.TestReport import TestReportDao
+from app.dao.test_case.TestResult import TestResultDao
 from app.middleware.AsyncHttpClient import AsyncRequest
 from app.models.constructor import Constructor
 from app.models.test_case import TestCase
@@ -126,21 +129,24 @@ async def execute_constructors(self, logger, path, params, req_params, construct
 
     async def execute_constructor(self, logger, index, path, params, req_params, constructor: Constructor):
         if constructor.type == 0:
-            self.append(logger, f"当前路径: {path}, 第{index + 1}条构造方法")
-            # 说明是case
-            executor = Executor(logger)
             data = json.loads(constructor.constructor_json)
-            new_param = data.get("params")
-            if new_param:
-                temp = json.loads(new_param)
-                req_params.update(temp)
             case_id = data.get("case_id")
             testcase, _ = await TestCaseDao.async_query_test_case(case_id)
-            result, err = await executor.run(case_id, params, req_params, f"{path}->{testcase.name}")
-            if err:
-                raise Exception(f"{path}->{testcase.name} 第{index + 1}个构造方法执行失败: {err}")
-            params[constructor.value] = result
-            await self.parse_params(logger, testcase, params)
+            try:
+                self.append(logger, f"当前路径: {path}, 第{index + 1}条构造方法")
+                # 说明是case
+                executor = Executor(logger)
+                new_param = data.get("params")
+                if new_param:
+                    temp = json.loads(new_param)
+                    req_params.update(temp)
+                result, err = await executor.run(case_id, params, req_params, f"{path}->{testcase.name}")
+                if err:
+                    raise Exception(err)
+                params[constructor.value] = result
+                await self.parse_params(logger, testcase, params)
+            except Exception as e:
+                raise Exception(f"{path}->{testcase.name} 第{index + 1}个构造方法执行失败: {e}")
 
     @case_log
     async def run(self, case_id: int, params_pool: dict = None, request_param: dict = None, path="主case"):
@@ -222,6 +228,39 @@ async def run(self, case_id: int, params_pool: dict = None, request_param: dict
             Executor.log.error(f"执行用例失败: {str(e)}")
             return result, f"执行用例失败: {str(e)}"
 
+    @staticmethod
+    async def run_single(data, report_id, case_id, params_pool: dict = None, request_param: dict = None,
+                         path="主case"):
+        start_at = datetime.now()
+        executor = Executor()
+        result, err = await executor.run(case_id, params_pool, request_param, path)
+        finished_at = datetime.now()
+        cost = "{}s".format((finished_at - start_at).seconds)
+        if err is not None:
+            status = 2
+        else:
+            if result.get("status"):
+                status = 0
+            else:
+                status = 1
+        asserts = json.dumps(result.get("asserts"), ensure_ascii=False)
+        url = result.get("url")
+        case_logs = result.get("logs")
+        body = result.get("request_data")
+        status_code = result.get("status_code")
+        request_method = result.get("request_method")
+        response = result.get("response")
+        if not isinstance(response, str):
+            response = json.dumps(response, ensure_ascii=False)
+        response_headers = json.dumps(result.get("response_header"), ensure_ascii=False)
+        cookies = json.dumps(result.get("cookies"), ensure_ascii=False)
+        data[case_id] = status
+        await TestResultDao.insert(report_id, case_id, status,
+                                   case_logs, start_at, finished_at,
+                                   url, body, request_method, cost,
+                                   asserts, response_headers, response,
+                                   status_code, cookies, 0)
+
     @case_log
     def replace_body(self, req_params, body):
         """根据传入的构造参数进行参数替换"""
@@ -321,3 +360,24 @@ def parse_variable(self, response_info, string: str):
         except Exception as e:
             return None, f"获取变量失败: {str(e)}"
         return json.dumps(result, ensure_ascii=False), None
+
+    @staticmethod
+    async def run_multiple(executor: int, env: int, case_list: List[int]):
+        # step1 新增测试报告数据
+        report_id = await TestReportDao.start(executor, env)
+
+        # step2 开始执行用例
+        result_data = dict()
+        await asyncio.gather(*(Executor.run_single(result_data, report_id, c) for c in case_list))
+        ok, fail, skip, error = 0, 0, 0, 0
+        for case_id, status in result_data.items():
+            if status == 0:
+                ok += 1
+            elif status == 1:
+                fail += 1
+            elif status == 2:
+                error += 1
+            else:
+                skip += 1
+        await TestReportDao.end(report_id, ok, fail, error, skip, 3)
+        return report_id