测试平台系列(113) 简单打造工作台.md

米洛 2022/6/13 测试平台接口测试FastApiPythonReact

大家好~我是米洛！
我正在从0到1打造一个开源的接口测试平台, 也在编写一套与之对应的教程，希望大家多多支持。
欢迎关注我的公众号米洛的测开日记，获取最新文章教程!

# 回顾

上一节我们编写了多环境配置相关的代码，相信大家都已经会部署多套环境的代码了。

这一节我们来调整下辣个丑丑的工作台。

可以看到，内容很空泛，作为一个首页，着实有点寒碜。既然它是我们的门面，那当然不能继续让它丑下去。所以这节我们就来狠狠整顿一下。

# 参考

在我们毫无头绪的时候，不妨看看一些专业的平台。比如antd pro的工作台，我们可以看看有哪些能借鉴的内容。

上面的内容我们可以借鉴的有红线的3个部分，我们简单概括下就是:

快速导航
用户榜单用来查看用户的一些排名啊啥的
项目

这里的项目我们会以测试计划为维度，体现各个测试计划的数据。什么意思呢？当我们认为某些核心测试集很值得关注的时候，我们可以关注该内容，这样工作台会给我们最直观的数据显示，比如通过率等。

那，话不多说，甘蔗!我们来各个击破。

# 排名部分

我们需要3个数据，用户参与了哪些项目/用户创建了多少case/总共有多少人。

app/crud/project/ProjectDao.py中新增以下方法

    @staticmethod
    async def query_user_project(user_id: int) -> int:
        """
        created by woody at 2022-02-13 12:05
        查询用户有多少项目
        :param user_id: 用户id
        :return: 返回项目数量
        """
        ans = set()
        async with async_session() as session:
            async with session.begin():
                # 先选出未被删除的用户
                project_sql = select(Project).where(Project.deleted_at == 0)
                projects = await session.execute(project_sql)
                project_list = []
                # 将数据放入列表，把owner等于该用户的放入列表
                for r in projects.scalars().all():
                    project_list.append(r.id)
                    if r.owner == user_id:
                        ans.add(r.id)
                # 接着查询项目角色表有该用户的角色，把角色的项目id放入列表
                # 由于是set，所以不会重复
                query = await session.execute(
                    select(ProjectRole).where(ProjectRole.deleted_at == 0, ProjectRole.user_id == user_id))
                for q in query.scalars().all():
                    ans.add(q.project_id)
        return len(ans)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

一个用户有多少项目，有2种情况:

有多少个项目的owner是他
项目角色表里有多少个用户是他，并且还要做到项目去重

由于我的SQL是真的不太好，所以就采用了代码的形式直接编写了。

这样我们就能查出用户在哪些项目id里面了。

新增app/crud/test_case/TestCaseDao.py里面的查询用户排名方法

    @staticmethod
    @RedisHelper.cache("rank")
    async def query_user_case_list() -> Dict[str, List]:
        """
        created by woody at 2022-10-13 12:59
        查询用户case数量和排名
        :return:
        """
        ans = dict()
        async with async_session() as session:
            async with session.begin():
                sql = select(TestCase.create_user, func.count(TestCase.id)) \
                    .outerjoin(User, and_(User.deleted_at == 0, TestCase.create_user == User.id)).where(
                    TestCase.deleted_at == 0).group_by(TestCase.create_user).order_by(
                    desc(func.count(TestCase.id)))
                query = await session.execute(sql)
                for i, q in enumerate(query.all()):
                    user, count = q
                    ans[str(user)] = [count, i + 1]
        return ans

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

这个操作是查询出用户有多少条用例，并根据用户的case数量排序，最后的ans存储的是一个映射关系, userid => 用例数量和排名。

由于这块数据算高频数据，所以我们放入redis，大概30秒的过期时间。（后续新增case的时候会影响到数量，所以还得去删除缓存。）

# 编写app/routers/workspace/workspace.py

from fastapi import APIRouter, Depends

from app.crud.project.ProjectDao import ProjectDao
from app.crud.test_case.TestCaseDao import TestCaseDao
from app.handler.fatcory import PityResponse
from app.routers import Permission

router = APIRouter(prefix="/workspace")


@router.get("/", description="获取工作台用户统计数据")
async def query_user_statistics(user_info=Depends(Permission())):
    user_id = user_info['id']
    count = await ProjectDao.query_user_project(user_id)
    rank = await TestCaseDao.query_user_case_list()
    case_count, user_rank = rank[str(user_id)]
    return PityResponse.success(dict(project_count=count, case_count=case_count,
                                     user_rank=user_rank, total_user=len(rank)))