精准测试中应用的算法模型

在 APK 测试的过程中，采用自动化脚本，可以降低我们人工测试的成本，随着 APK 不断的迭代，自动化脚本也会不断的增多，大大的增加了自动化脚本的维护成本。

精准测试，就是在开发人员提交 apk 后，找出本次提交和上一次提交之间的代码差异，根据 Android 代码覆盖度采集的数据，训练出一个模型，可以寻找到自动化脚本和代码之间的一种关系，推荐出 diff 代码所需要执行自动化脚本的一个推荐顺序。

整体模型流程

这里只介绍模型的整体方案，不介绍代码和用例之间是如何建立的关系。（用例和代码产生的关系在其他文章中介绍）主要分为五个部分：

• 数据文件的接收服务

• 数据文件处理

• 模型训练

• 模型预测

• 模型，数据文件存储模式

这里只介绍 数据文件处理，模型训练，模型预测三部分核心功能。数据文件接收服务和存储模式只是业务的逻辑，不重点介绍。

数据文件处理

数据处理比较麻烦，小伙伴们可以慢慢看~~~

这边统计的数据是一个 csv 文件：

贴上数据处理代码：

# caseId
caseId_dict = cvsData['caseId'].unique().tolist()
cvsData['caseId_value'] = cvsData['caseId'].apply(lambda x:caseId_dict.index(x) + 1)

# className
className_dict = cvsData['className'].unique().tolist()
cvsData['className_value'] = cvsData['className'].apply(lambda x: className_dict.index(x) + 1)
cvsData['className_value'] = cvsData['className_value'].apply(lambda x: df_min_max(x, cvsData['className_value']))

# packageName
cvsData['packageName'] = cvsData['className'].str.rsplit("/", 1).str[0]
packageName_dict = cvsData['packageName'].unique().tolist()
cvsData['packageName_value'] = cvsData['packageName'].apply(lambda x: packageName_dict.index(x) + 1)
cvsData['packageName_value'] = cvsData['packageName_value'].apply(lambda x: df_min_max(x, cvsData['packageName_value']))

# funcName
funcName_dict = cvsData['funcName'].unique().tolist()
cvsData['funcName_value'] = cvsData['funcName'].apply(lambda x: funcName_dict.index(x) + 1)
cvsData['funcName_value'] = cvsData['funcName_value'].apply(lambda x: df_min_max(x, cvsData['funcName_value']))

# 方法参数
cvsData['funParameter'] = cvsData['funcDesc'].str.rsplit(")", 1).str[0]
cvsData['funParameter'] = cvsData['funParameter'].apply(lambda x: spliteFunParameter(x))
dict_Parameter.update(Statistics(removeNull(list(cvsData['funParameter']))))
dict_Parameter.update({"unk": 0})
cvsData['funParameter_value'] = cvsData['funParameter'].apply(lambda x: LineStatistics(x, dict_Parameter))

# 方法返回值 字典
cvsData['funReturn'] = cvsData['funcDesc'].str.rsplit(")", 1).str[1]
funReturn_dict = cvsData['funReturn'].unique().tolist()
cvsData['funReturn_value'] = cvsData['funReturn'].apply(lambda x: funReturn_dict.index(x) + 1)
cvsData['funReturn_value'] = cvsData['funReturn_value'].apply(lambda x: df_min_max(x, cvsData['funReturn_value']))

#lineCode 分词
# cvsData['lineCode_word'] = cvsData['lineCode'].apply(lambda x: PkusegParticiple.participle(x))
# jieba 分词
cvsData['lineCode_word'] = cvsData['lineCode'].apply(lambda x: JieBaParticiple.participle(x))

• 我们要对每一列进行处理，把每一列转换为大于等于 1 的数字，并且相同列值对应的数值相同，并进行归一化，减少某数值过大，影响最后结果。数字 0 表示没有出现过的类型。
• 这里把全路径的 className 拆分为两个部分，className 和 PackageName，我们在训练的时候 PackageName 也可以做为一个特征。

• 方法参数的处理需要把每一个参数对应的的类型组合成一个没有重复的 int 类型的定长数组，在把每个方法的参数在这个数组中表示。增加一个占位标识没有出现过的类型。

• 最后对 linecode 进行分词,，这里使用了 pkuseg 分词和 jieba 分词 (或者是 jieba_fast)

pkuseg 分词

优势： 分词细致，准确。

缺点： 速度较慢

# pkuseg 模型初始化
def __init__():    
    global seg, stopwords    
    seg = pkuseg.pkuseg(model_name="default", user_dict=common.word_dict, postag=False)  # 以默认配置加载模型    
    stopwords = stopwordslist(common.stop_words)  # 这里加载停用词的路径

# lineCode 分词， 返回分词后的结果
def participle(linecode):
    global seg
    text = seg.cut(linecode.lower().strip())  # 进行分词
    return movestopwords(text)


# 创建停用词list
def stopwordslist(filepath):
    stopwords = [line.strip() for line in open(filepath, 'r', encoding='UTF-8').readlines()]
    return stopwords


# 对句子去除停用词
def movestopwords(sentence):
    global stopwords
    santi_words =[x for x in sentence if len(x) > 1 and x not in stopwords]
    return santi_words

init 方法需要提前初始化，要不然这里速度会比较慢。

jieba 分词（或者使用 jieba_fast）

优势： 分词速度快。

缺点： 分词结果比较粗糙，有一些特殊符号的分词不标准。

# pkuseg 模型预加载
def __init__():    
    global stopwords    
    # 加载用户自定义词典       
    jieba.load_userdict(common.word_dict)    
    stopwords = stopwordslist(common.stop_words)  # 这里加载停用词的路径


# lineCode 分词， 返回分词后的结果
def participle(linecode):
    # jieba.add_word(x) # 只添加一个单词
    words = jieba.cut(linecode)
    word_list = movestopwords(words)  # 去除停用词
    return word_list


# 创建停用词list
def stopwordslist(filepath):
    stopword = [line.strip() for line in open(filepath, 'r', encoding='UTF-8').readlines()]
    return stopword


# 对句子去除停用词
def movestopwords(sentence):
    global stopwords
    santi_words =[x for x in sentence if len(x) >1 and x not in stopwords]
    return santi_words

模型训练

在训练之前，我们还需要对数据进行处理。我们需要处理为两部分的数据，一个特征向量，一个分类标签的向量。

LDA 模型

我们需要把上面分词得到的数组，进行语义分析，我们这里预设一个话题数为 8，最后每一行的代码就表示为一个 8 维的词向量空间。

def wordToVector(topic, arr):    
    cntVector = CountVectorizer(analyzer='char')    
    cntTf = cntVector.fit_transform(arr)    
    return LDAPlay(topic, cntTf)


def LDAPlay(topic, cntTf):
    lda = LatentDirichletAllocation(topic, 50., 0)
    docres = lda.fit_transform(cntTf)
    np.set_printoptions(threshold=np.inf)
    return docres

word2vec 模型

我们也可以使用 word2vec 进行语义分析，我们这边预设每个词语使用一个 50 维词向量空间表示，最后整行的词向量为每个词向量相加求平均来表示：

def word2vecPlay(linecodeArr):    
    result = change(linecodeArr)    
    model = word2vec.Word2Vec(result, hs=1, min_count=1, window=3, size=50)        docres = []    
    for item in result:        
        if isinstance(item, str):            
            data = np.array(model[strChangeArr(item)])        
        else:            
            data = np.array(model[item])        
            docres.append(np.mean(data, axis=0))    
     return docres

特征向量

特征向量由 类名，包名，方法名，方法返回值，方法参数，每一行代码，对应的归一化的值，或者转化为向量的值，组合一个新的二位向量：

# 特征数据抽取 合并
def dataMerge(cvSurface, docresLDA):    
    result = []    
    # 添加特征值  className, packageName, funName,  funDesc，LDA 处理后的代码集合
    for i in range(0, len(cvSurface)):        
        lineresult = list(            
            cvSurface.loc[i, ['className_value', 'packageName_value', 'funcName_value', 'funReturn_value']])        
        funParameter_value = cvSurface['funParameter_value'][i]        
        if isinstance(funParameter_value, str):
            lineresult.extend(strChangeArr(funParameter_value))        
        else:            
            lineresult.extend(funParameter_value)  
        lineresult.extend(docresLDA[i])        
        result.append(lineresult)    
     return result

生成新的分类标签

我们在处理数据的时候，难免会碰到相同的代码片段对应不同的用例的情况，或者说，一个用例下对应多行代码，多行代码对应不同的用例，我们怎么分类呢。。。。

我们需要根据这种情况，重新进行一个标签的组合分类，相当于变相的多标签分类。

• 首先我们要把当前一行内容分属于那些 caseID 进行一个统计，生成一个数组。
• 在把相同数组归为一类，重新进行分类。

# 每一个特征的 MD5 对应的 caseId 字典集合
md5recode = {}
# 每一个特征的 MD5 集合
newLarber = []
for i in range(len(XGBTData)):    
    md5 = md5_convert(str(XGBTData[i]))  # 当前数据的 md5
    newLarber.append(md5)    
    caseId_valueTrue = cvsData['caseId_value'][i]    
    if md5 in md5recode.keys():        
        yuan = md5recode[md5]        
        yuan.append(caseId_valueTrue)        
        yuan = list(set(yuan))        
        md5recode.update({md5: yuan})    
    else:        
        md5recode.update({md5: [caseId_valueTrue]})
# 每一行特征 MD5 值存入 pandas 表对象中，生成列 AllFeraterMD5
cvsData['AllFeraterMD5'] = newLarber
# 生成列 md5_value ，根据每一行的 md5 值，我们赋值 给 md5 对应的 
caseIdcvsData['md5_value'] = cvsData['AllFeraterMD5'].apply(lambda x: str(md5recode[x]))
new_Larber_dict = cvsData['md5_value'].unique().tolist()
# 生成列 new_Larber_value， 我们相同 md5_value 值组成一个 新的 larber
cvsData['new_Larber_value'] = cvsData['md5_value'].apply(lambda x: new_Larber_dict.index(x))

接下来，我们终于可以把数据放到模型中进行训练了。

xgboost 分类模型

# 获取当前 集合的 larber
labels = CvsData['new_Larber_value']
XGBTData = np.array(XGBTData)
labels = np.array(labels)
num_class = len(CvsData['new_Larber_value'].drop_duplicates())
# n_estimators 适合的树数量。 subsample 训练每棵树时，使用的数据占全部训练集的比例。 colsample_bytree 训练每棵树时，使用的特征占全部特征的比例
classifier = XGBClassifier(max_depth=2, learning_rate=0.25, n_estimators=40, subsample=0.9, colsample_bytree=0.9,objective='multi:softprob', num_class=num_class)
classifier.fit(XGBTData, labels, eval_set=[(XGBTData, labels)], eval_metric='merror', early_stopping_rounds=20, verbose=True)
# classifier.fit(XGBTData, labels)
evals_result = classifier.evals_result()
print(evals_result)

深度神经网络分类模型

model = keras.Sequential([    
    keras.layers.Flatten(input_shape=(dimensionLen, 1)),    
    keras.layers.Dense(128, activation='relu'),    
    keras.layers.Dense(128, activation='relu'), 
    keras.layers.Dense(num_class, activation='softmax')
])

model.compile(optimizer='adam',
              loss='sparse_categorical_crossentropy',
              metrics=['accuracy'],
              run_eagerly=True)

model.fit(XGBTData, labels, epochs=10)

test_loss, test_acc = model.evaluate(TextXGBTData, TextLabels, verbose=2)

print('test_acc :', test_acc)


predictions = model.predict(TextXGBTData)

result = ResultArr.result(cvsDatas, predictions)
print(result)

多标签分类训练模型（MultiLabel）

上面的两种分类其实是伪多标签分类，就是变相的把应该多标签分类的问题转换为单标签分类的问题，等待分类完毕后，我们在转换为原来的 caseID。下面我们来介绍真正的多标签分类。

还是上面的步骤，在数据处理的时候，我们不生成 new_Larber_value 这一列数据，我们还是需要统计每一行代码所属的原先标签进行一个统计，把这个数组作为一个 laber

mlb = MultiLabelBinarizer()
XGBTData = np.array(XGBTData)
labels = np.array(cvsData['md5_value'])
labels = mlb.fit_transform(labels)

TextXGBTData = np.array(TextXGBTData)
TextLabels = np.array(TextDiffData['md5_value'])
TextLabels = mlb.fit_transform(TextLabels)

classif = OneVsRestClassifier(SVC(kernel='linear'))
print('开始训练')
classif.fit(XGBTData, labels)
print('结束训练')

模型预测

对需要预测的数据也需要进行处理。

处理思路

不能按照当前的数据进行处理，那特征向量的值和训练数据的特征向量的值对应不上，按照上面相同规则把数据分开，但是归一化的值，我们需要从训练数据中进行查询赋值，如果训练数据没有的时候，我们把当前的特征赋值为 0

# className
diffCsv['className_value'] = diffCsv['className'].apply(    lambda x: dictClass.get(x, 0))

# packageName
diffCsv['packageName'] = diffCsv['className'].str.rsplit("/", 1).str[0]
diffCsv['packageName_value'] = diffCsv['packageName'].apply(
        lambda x: dictPackage.get(x, 0))

# funcName
diffCsv['funcName_value'] = diffCsv['funcName'].apply(lambda x: dictFuncName.get(x, 0))

# 方法参数
diffCsv['funParameter'] = diffCsv['funcDesc'].str.rsplit(")", 1).str[0]
diffCsv['funParameter'] = diffCsv['funParameter'].apply(lambda x: DataProcessingPD.spliteFunParameter(x))
diffCsv['funParameter_value'] = diffCsv['funParameter'].apply(lambda x: DataProcessingPD.LineStatistics(x, dict_Parameter))

# 方法返回值
diffCsv['funReturn'] = diffCsv['funcDesc'].str.rsplit(")", 1).str[1]
diffCsv['funReturn_value'] = diffCsv['funReturn'].apply(
        lambda x: dictFunReturn.get(x, 0))

# lineCode 分词
# diffCsv['lineCode_word'] = diffCsv['lineCode'].apply(lambda x: PkusegParticiple.participle(x))
# jieba 分词
diffCsv['lineCode_word'] = diffCsv['lineCode'].apply(lambda x: JieBaParticiple.participle(x))

xgboost 分类模型预测

test_y = model.predict_proba(textXGBTData)

其中，每一个代码片段都预测出固定分类的一个一维向量，和我们最后要推荐的优先级别不符，需要对这个结果进行处理

处理方式

• 把所有代码片段得到的结果概率累加取平均值
• 保存一个 {位置：均值} 形式的一个字典
• 每一个位置对应回原先的 caseID
• 循环把相同 caseID 的概率相加
• 按照大小进行排序

最后结果就是我们要推荐的结果。所有模型推荐的结果都是这么处理的。下面就不再说了。

深度神经网络分类模型预测

predictions = model.predict(TextXGBTData)

多标签分类训练模型预测（MultiLabel）

predictions = classif.predict(TextXGBTData)

结果展示