BytePush: A Quality Assurance Platform for Intelligent Alerts and Process Automation

Background

Fast value delivery while ensuring quality is a constant goal of the intelligent QA team. Efficient value delivery aligns with agile development and continuous delivery. Existing quality‑assurance platforms have shifted testing left, with code coverage, branch inspection, performance testing, etc., moved to the RD self‑test stage; after submission, QA performs risk evaluation. Currently, RD and QA jointly define entry/exit criteria, and QA collects metrics at each gate, requiring one QA and one RD to follow up throughout each iteration.

Goal

As a risk control (process, people, system) spanning the entire product‑research lifecycle, BytePush uses measurable standards to discover and expose quality and process issues via various cards, achieves data convergence through quality operations, continuously analyzes and upgrades standards, and forms a closed loop.

Difficulties

Creating a universal solution that supports business‑customized card content when data source platforms allow business‑defined configuration.

Achieving business‑customized push strategies when push policies, targets, and methods differ across businesses.

Pain Points

The biggest pain point lies in the code layer:

Initial impression of the robot: just call an open API, parse, push a card – done.

Result: only satisfies current business, cannot support diversified or customized demands, cannot timely meet business changes.

When supporting multi‑direction businesses, large amounts of redundant code appear, making requirement changes difficult and modifications costly.

Based on the initial impression, many people rush to write cards but later abandon them.

To solve the above pain points:

Database usage transformation

Before: frequent raw SQL writing.

After: BytePush encapsulates aiomysql into an ORM layer; only need to pass parameters according to rules and call methods.

def search_branch_message(branch_name, rid, rd_name, version, src):
    sql = "SELECT * from branch_message WHERE branch_name = %s and rid = %s rd_name = %s and version = %s and src = %s"
    args = (branch_name, rid, rd_name, version, src)
    rows = db_base.search_mysql_args(sql, args)
    return len(rows)

'''
增删改
增删改传递参数：
mode：必填，"add" or "update" or "delete"
model：必填，数据库表名
data：必填，要操做的数据
'''
def insert_by_params(mode=None, model=None, data=None):
    s = DBORM.Select()
    task = DBORM.insert_task_pool(mode=mode, models=model, data=data)
    result = s.insert_many(task)
    s.close()
    return result
'''
查询  
查询传递参数：
need_typee：必填，表格名称
dimension_type：选填，维度信息（人维度、需求维度、版本维度）
dimension_param：选填，查询参数，同时也表示要在卡片上显示的内容对应的字段
'''
def get_search_result(self, need_type=None, dimension_type=None, dimension_param=None):
    em = exception_message(mode="group")
    tasks = []
    for need_type_ in need_type.keys():
        try:
            if hasattr(em, need_type_):
                exp_message = getattr(em, need_type_)
                if dimension_type and dimension_type in ["rid", "manager"]:
                    self.select_params[dimension_type] = dimension_param
                tasks += exp_message(self.select_params)
        except AttributeError:
            print('没有这个项目')
    result = self.s.select_many(tasks)
    return result

Card generation transformation

Before: a single bug‑fetching interface was repeatedly used across many cards.

After: establish a metrics factory; for example, produce a version‑merge card by selecting indicators such as meego defect (bug_message), slardar memory leak (mom_message), slardar crash (crash_message) and assembling them; produce an open_bug timeout card by selecting only bug_message.

# 从指标工厂挑选指标：dict中的key对应指标工厂中的方法名，value即为需要显示的字段
need_exceptions_rid = {
    "h_owner_schedule_notice": ["node_name"],
    "bug_message": ["priority_name"],  # 表示根据优先级聚合
    "crash_message": ["crash_type"],   # 表示根据类型聚合
    "mom_message": ["issue_level"],    # 表示根据优先级聚合
}
# 指标工厂
class exception_message(Mode_Message):
    def __init__(self, mode):
        Mode_Message.__init__(self, mode)
        self.mode = mode
    def bug_message(self, args=None):
        result = self.mode_message("bug_message", args)
        return result
    def crash_message(self, args=None):
        result = self.mode_message("crash_message", args)
        return result
    def mom_message(self, args=None):
        result = self.mode_message("mom_message", args)
        return result
    def code_coverage_message(self, args=None):
        result = self.mode_message("code_coverage_message", args)
        return result
class Mode_Message(Base_Message):
    def __init__(self, mode=None):
        Base_Message.__init__(self)
        self.mode = mode
    def mode_message(self, model, args):
        if self.mode == "single":
            result = self.just_show_search_by_params(model, args)
            return result
        elif self.mode == "group":
            task = self.many_together_contain(model, args)
            return task
class Base_Message(object):
    @staticmethod
    def just_show_search_by_params(model=None, message_select_args=None):
        s = DBORM.Select()
        task = DBORM.task_pool(models=[model], message_select_args=message_select_args)
        result = s.select_many(task)
        return result
    @staticmethod
    def many_together_contain(model=None, message_select_args=None):
        task = DBORM.task_pool(models=[model], message_select_args=message_select_args)
        return task

Support custom configuration parameters

Before: code redundancy was huge due to ad‑hoc parsing.

After: build a field factory; for generic fields only pass the needed key:value; for user‑defined custom fields pass the corresponding dict.

# condition_params_dict：通用字段
# defined_condition_params： 自定义字段
class MeegoPayload(CommonInterface):
    def __init__(self, src, condition_params_dict, SearchType, defined_condition_params=None):
        CommonInterface.__init__(self, src)
        self.SearchType = SearchType
        self.defined_condition_params = defined_condition_params
        self.condition_params = condition_params_dict.keys()
        if "planning_version" in self.condition_params:
            self.planning_version_value_list = condition_params_dict["planning_version"]  # 需求计划版本
        if "actual_online_version" in self.condition_params:
            self.actual_online_version_value_list = condition_params_dict["actual_online_version"]  # 实际上车版本
        if "business" in self.condition_params:
            self.bussiness_value_list = condition_params_dict["business"]  # 拉取配置好的业务
        if "stage" in self.condition_params:
            self.view_value_list = condition_params_dict["stage"]  # 配置好视图
        if "work_item_status" in self.condition_params:
            self.state_value_list = condition_params_dict["work_item_status"]  # 缺陷状态
        if "resolve_version" in self.condition_params:
            self.resolve_version_value_list = condition_params_dict["resolve_version"]  # 缺陷解决版本
        if "linked_story" in self.condition_params:
            self.linked_story_value_list = condition_params_dict["linked_story"]  # 缺陷关联需求
        if "priority" in self.condition_params:
            self.priority_value_list = condition_params_dict["priority"]  # 缺陷优先级
        if "discovery_version" in self.condition_params:
            self.discovery_version_value_list = condition_params_dict["discovery_version"]  # 缺陷发现版本

Implement shell project

Upper layer: obtain configuration, public parameters, version, business line.

Middle layer: data collection, gathering various atomic data.

Lower layer: data consumption, push card dimensions.

'''
统一的卡片流程
'''
def process_slave(self, start_time, card_config):
    # ---获取配置---
    need_params = card_config["need_params"]
    trigger_mode = card_config["trigger_mode"]
    send_mode = card_config["send_mode"]
    params = {"stage": ["all_story"]}
    if self.mode not in trigger_mode:
        print("配置中不支持这个触发方式")
        return
    # 这里拿到的是某个类型的方法的形参
    need_func, params_process, need_type = self.process_master(need_params)
    # ---数据采集---
    version = self.get_version_by_config()
    params[need_params["version_type"]] = self.mv.get_work_items_version(version)
    if self.business and len(self.business) == 0:
        get_bussinesses = self.mv.get_bussinesses()
        params["business"] = get_bussinesses
        self.insert_gather_result(need_func, params_process, params)
    else:
        params["business"] = self.business
        self.insert_gather_result(need_func, params_process, params)
    # ---数据消费---
    # 数据消费查询条件
    select_params_1 = {
        "task_id": start_time
    }
    # 推送list到群
    if "chat" in send_mode:
        self.list_to_chat(select_params_1, card_config, need_type)
    # 推送给个人
    if "single" in send_mode:
        self.send_person_chat(select_params_1, card_config, need_type)

Solution

The solution consists of three stages: automation, platformization, and intelligence.

First stage (Automation)

Goals: automatically identify project shape, current version, current stage; automatically identify current process quality status; establish an initial rule model (pre‑defined rules) for tool‑based judgment.

Achieved by: … (details omitted for brevity).

Second stage (Platformization)

Upper‑level strategy dashboard: provides process metrics and baseline values; bound cards expose risk points.

Lower‑level strategy dashboard: each business configures card push strategies (time, method, target).

Third stage (Intelligence)

Goals: automatically predict risk based on business status; auto‑set thresholds per project shape; auto‑allocate manpower according to version demand and estimated effort; auto‑identify whether a project can proceed to the next phase, stage, or version.

Achieved by: version‑level statistics with manual calibration; evolving from fixed rules to data‑driven learning; when sufficient data accumulates, introduce model algorithms for intelligent judgment.

Technical Architecture

(Diagram omitted)

Deployment Strategy

Push cards: normal and urgent pushes.

Push targets: personal, group, leader.

Generate quality reports: weekly, monthly, import into metrics platform.

Effect Evaluation

Continuous analysis of core data outputs quality reports, forming a closed loop of “process planning – formulation – inspection – improvement”.

Benefits

Manpower & metrics: BytePush has been integrated into 16+ projects, saving 0.5 person‑power per project for developing and maintaining robot cards; saving 1–2 person‑power per project for full‑time iteration follow‑up.

Schedule, owner fill‑out ratio reaches 100%, turnover timeliness 95%, enabling measure‑efficiency data calibration.

Bug timely‑resolution ratio up 10% MoM.

Release‑day on‑time ratio up 50% MoM.

Overall crash < 1‰ achieved, timely‑resolution ratio up 50% MoM.

Release‑day demand quality check no longer requires QA follow‑up, greatly freeing QA resources.

Usage Frequency

Schedule reminders, flow reminders, owner reminders cards: average 32 uses per day.

Open_bug timeout cards: average 32 uses per day.

Requirement execution progress: average 10 uses per day.

Requirement execution progress pre‑ and post‑release: average >1000 uses per day.

Gray‑scale monitoring cards: average 16 uses per day.

Release data cards: average 16 uses per day.

Best Practices Based on Template Card

Example: Power‑intelligent learning lamp.

Across the product‑research process, give reminders at key nodes, continuously track data, and continually achieve data convergence.

Process: daily timed personal & group pushes for schedule reminders, completion reminders, owner fill‑out reminders.

During development: daily early‑morning rebase.

Each morning: push各业务需求执行进展.

When feature branch submits MR: trigger pipeline‑based requirement quality check.

Bug data: two days before release‑day regression, push timed unsolved bug alerts (group list, personal urgent).

From two days before release‑day to release‑day: daily morning & evening push feature merge cards and unmerged branch lists.

After release‑day to gray‑scale day: daily morning & evening push version merge cards.

During gray‑scale: daily morning & evening push gray‑scale monitoring cards.

During version: daily morning & evening push online monitoring cards.

Quality report: continuously iterate and polish standards.

Applicability

Currently the best practice for the learning lamp also supports other project version shapes: shift‑based, version‑based, and non‑version‑based.

Future Plans

Formulate various quality‑operation templates, e.g., test model – efficiency chapter, etc.

Open the shell project, providing custom data collection and custom output capabilities.

Leverage machine‑learning algorithms and historical data to achieve business intelligent early warning.

Ongoing Recruitment

👉「社招」测试开发（高级）工程师

👉「2022届秋招提前批」测试开发工程师