效能度量专栏文章质量保障系统的落地实践 (三) CI 管理设计 - 集成设计

亦攸 · 2024年05月24日 · 最后由亦攸回复于 2024年06月06日 · 2891 次阅读

往期文章

专栏文章质量保障系统的落地实践 (一) 概览篇
 专栏文章质量保障系统的落地实践 (二) 项目管理设计 - 基础信息与缺陷信息设计
 专栏文章质量保障系统的落地实践 (二) 项目管理设计 - 代码信息设计
 专栏文章质量保障系统的落地实践 (三) CI 管理设计 - 基础设计

前言

上期文章中介绍了 CI 管理的基础设计，概要包括如何定义校验粒度，以及如何提取校验粒度的方法，那么在这篇文章开头，补充一些这种方案的好处：1、首先，文本解析的方式需要自动化脚本有一定的编写规范，而编写规范又常常是较难推进的，一个团队内多名成员，如何保证每次的代码提交编写规范都能满足期望？可以使用 code review、代码规范工具等方式被动推进编写规范，而使用这种方案最直接的好处是，要想自己的数据被完整统计，就需要遵循定义的规范，这是一个主动行为，不是被动的，后置的行为；2、其次，这种方案的适配性很高，与自动化框架不产生耦合，适配任意自动化框架编写的脚本；3、若是使用执行结果进行校验粒度的提取，碰上判断分支的情况，往往会出现统计数据偏少的情况，而文本解析则可避免这些问题；4、编写代码简单直接，容易上手。
那么劣势在哪呢？
最大的劣势在于变量的替换结果无法读取，如 sql 语句：select * from xxx.table where id="${recordId}"这类的动态代码，由于文本解析无法侵入执行过程，所以这种方式拿不到具体填充的 recordId。
补充完上述情况后，我们开始后续的设计与讨论

脚本执行

在讨论如何集成之前，我们需要有一个基本概念—任何代码编写的自动化框架，均有一个执行入口。而当我们使用代码编译器的时候，往往点击执行按钮即可执行编写的代码，这个原理是什么？
那我们举一个 python 的例子

def print_(name):
  print(f"name is {name}.")

print_("亦攸")

想要执行这个文件，可以使用编译器的执行按钮，如 PYCHARM，而另一种更底层一些的方式是 python3 xxx.py。
以此类推，用 java 编写的代码，想要启动 java 代码，使用：javac MyProgram.java java MyProgram 的方式。
可见无论使用什么语言编写的自动化框架，都有一个最底层的执行入口。

如何集成

考虑到将自动化指标与自动化执行集成起来，需要考虑哪些情况？
1、首先我们需要知道某一次执行的结果，是成功，是失败？
2、其次是这次执行中的校验粒度如何等等信息，因为每隔一段时间提交情况都会发生变更。
3、后续统计自动化脚本完善趋势时，需要依托每一次执行的结果。
为了达到以上的要求，我们可以定义任务概念，以任务承载以上的数据。
有了承载主体后，接下来我们来聊聊集成设计，这里需要分成两种情况：1、公司基础建设较为完善，有成熟的自动化执行平台；2、需要我们自己搭建执行环境。
第一种情况，可以通过自动化执行平台的接口触发对应的脚本：

而第二种情况，需要我们自己去触发自动化脚本执行，那么就需要使用到脚本执行小节的内容了：

由此可见，有没有成熟的自动化执行平台其实并不会影响我们构建自动化质量保障体系，有无成熟的执行平台的差别在于是否需要自行触发自动化脚本。

数据设计

接下来我们看看数据表结构设计：

# CI管理-任务管理-任务信息
class CITaskInfo(BaseModel):
    department_id = models.IntegerField(verbose_name=u"归属组织节点")
    name = models.CharField(max_length=150, verbose_name=u"任务名称")
    app_code = models.CharField(max_length=150, verbose_name=u"应用code")
    status = models.IntegerField(default=CITaskStatusEnum.INIT.value, verbose_name=u"任务执行状态")
    env = models.IntegerField(verbose_name=u"归属环境(测试/预发/生产)")
    trigger_type = models.IntegerField(verbose_name=u"触发类型")
    error = models.TextField(max_length=250, null=True, blank=True, verbose_name=u"异常原因")
    report_url = models.CharField(max_length=250, null=True, blank=True, verbose_name=u"执行报告")

    class Meta:
        db_table = "CI_Task_Info"
        verbose_name = "CI管理-CI任务信息"
        verbose_name_plural = verbose_name

# CI管理-任务管理-执行记录信息表
class CITaskExecutionInfo(BaseModel):
    department_id = models.IntegerField(verbose_name=u"归属组织节点")
    task_id = models.IntegerField(verbose_name=u"task_id")
    app_code = models.CharField(max_length=50, verbose_name=u"应用code")
    domain = models.CharField(max_length=150, verbose_name=u"接口域名")
    path = models.CharField(max_length=255, verbose_name=u"接口路径")
    owner_name = models.CharField(max_length=11, verbose_name=u"负责人姓名")
    owner_phone = models.CharField(max_length=11, verbose_name=u"负责人手机号")
    success = models.BooleanField(default=False, verbose_name=u"是否通过")
    trigger_type = models.IntegerField(verbose_name=u"触发类型")
    env = models.IntegerField(verbose_name=u"执行环境")

    class Meta:
        db_table = "CI_Task_Execution_Info"
        verbose_name = "CI管理-CI任务执行记录"
        verbose_name_plural = verbose_name

# CI管理-任务管理-api信息表
class CITaskApiInfo(BaseModel):
    department_id = models.IntegerField(verbose_name=u"归属组织节点")
    task_id = models.IntegerField(verbose_name=u"task_id")
    app_code = models.CharField(max_length=50, verbose_name=u"服务code")
    api_name = models.CharField(max_length=150, verbose_name=u"api名称")
    api_analysis_prefix = models.CharField(max_length=50, verbose_name=u"api解析前缀")
    api_analysis_postfix = models.CharField(max_length=50, verbose_name=u"api解析后缀")
    domain = models.CharField(max_length=150, verbose_name=u"域名")
    path = models.CharField(max_length=255, verbose_name=u"接口路径")
    owner_name = models.CharField(max_length=11, verbose_name=u"负责人姓名")
    owner_phone = models.CharField(max_length=11, verbose_name=u"负责人手机号")
    env = models.IntegerField(verbose_name=u"归属环境(测试/预发/生产)")
    trigger_type = models.IntegerField(verbose_name=u"任务触发方式")

    class Meta:
        db_table = "CI_Task_Api_Info"
        verbose_name = "CI管理-CI任务api信息"
        verbose_name_plural = verbose_name

# CI管理-任务管理-校验粒度信息表
class CITaskKeywordInfo(BaseModel):
    department_id = models.IntegerField(verbose_name=u"归属组织节点")
    task_id = models.IntegerField(verbose_name=u"task_id")
    app_code = models.CharField(max_length=50, verbose_name=u"服务code")
    api_name = models.CharField(max_length=150, verbose_name=u"api名称")
    api_analysis_prefix = models.CharField(max_length=50, verbose_name=u"api解析前缀")
    api_analysis_postfix = models.CharField(max_length=50, verbose_name=u"api解析后缀")
    keyword_name = models.CharField(max_length=150, verbose_name=u"keyword名称")
    keyword_analysis_prefix = models.CharField(max_length=50, verbose_name=u"keyword解析前缀")
    keyword_analysis_postfix = models.CharField(max_length=50, verbose_name=u"keyword解析后缀")
    domain = models.CharField(max_length=150, verbose_name=u"域名")
    path = models.CharField(max_length=255, verbose_name=u"接口路径")
    owner_name = models.CharField(max_length=11, verbose_name=u"负责人姓名")
    owner_phone = models.CharField(max_length=11, verbose_name=u"负责人手机号")
    sentence_count = models.FloatField(verbose_name=u"校验粒度")
    env = models.IntegerField(verbose_name=u"归属环境(测试/预发/生产)")
    trigger_type = models.IntegerField(verbose_name=u"任务触发方式")

    class Meta:
        db_table = "CI_Task_Keyword_Info"
        verbose_name = "CI管理-CI任务keyword信息"
        verbose_name_plural = verbose_name

以上的数据表涉及数据均可通过解析文件逐步提取出来，通过 CITaskKeywordInfo 聚合每一位成员在每次任务下的具体数据：

# CI管理-任务管理-归属人总览信息表
class CITaskMemberSummaryInfo(BaseModel):
    department_id = models.IntegerField(verbose_name=u"归属组织节点")
    task_id = models.IntegerField(verbose_name=u"task_id")
    app_code = models.CharField(max_length=50, verbose_name=u"服务code")
    owner_name = models.CharField(max_length=11, verbose_name=u"负责人姓名")
    owner_phone = models.CharField(max_length=11, verbose_name=u"负责人手机号")
    env = models.IntegerField(verbose_name=u"归属环境(测试/预发/生产)")
    total_api_count = models.IntegerField(verbose_name=u"归属接口总量")
    covered_api_count = models.IntegerField(verbose_name=u"覆盖归属接口总量")
    covered_api_rate = models.FloatField(verbose_name=u"归属接口覆盖率")
    total_sentence_count = models.IntegerField(verbose_name=u"校验粒度总量")
    average_sentence_count = models.FloatField(verbose_name=u"平均粒度(校验粒度总量/归属接口总量)")
    trigger_type = models.IntegerField(verbose_name=u"触发类型")

    class Meta:
        db_table = "CI_Task_Member_Summary_Info"
        verbose_name = "CI管理-CI任务归属人总览信息"
        verbose_name_plural = verbose_name

# 获取校验粒度记录
    member_granularity_records = CITaskKeywordInfo.objects.filter(
        department_id=department_id, task_id=ci_task_id, env=env, date_delete=None). \
        order_by("owner_phone", "dimension")
    for member_granularity_info in member_granularity_records:
        owner_name = member_granularity_info.owner_name
        owner_phone = member_granularity_info.owner_phone
        app_code = member_granularity_info.app_code
        dimension = member_granularity_info.dimension

        # 以app_code、归属人、统计维度聚合数据
        combine_key = f"{app_code} {owner_name} {owner_phone} {dimension}"
        member_granularity_map[combine_key].append(member_granularity_info)

    for combine_key, date_sequence in member_granularity_map.items():
        app_code, owner_name, owner_phone, dimension = combine_key.split()

        # 计算平均校验粒度
        total_owner_dimension_api_count = len(date_sequence)
        total_owner_dimension_owner_sentence = sentence_pass_count = 0
        for date_item in date_sequence:
            sentence_count = date_item.sentence_count
            total_owner_dimension_owner_sentence += sentence_count

        if total_owner_dimension_api_count == 0:
            average_sentence_count = 0
        else:
            average_sentence_count = round(total_owner_dimension_owner_sentence /
                                           total_owner_dimension_api_count, 2)
      .......