How to use report_test_name method in Testify

Best Python code snippet using Testify_python

run_magellan.py

Source:run_magellan.py

1import pandas as pd2import numpy as np3np.random.seed(42)4import random5random.seed(42)6import os7import time8import glob9import py_entitymatching as em10from sklearn.linear_model import LogisticRegression11from sklearn.naive_bayes import GaussianNB12from sklearn.ensemble import RandomForestClassifier13from sklearn.svm import LinearSVC14from sklearn.tree import DecisionTreeClassifier15from sklearn.model_selection import RandomizedSearchCV16from sklearn.model_selection import PredefinedSplit17from sklearn.metrics import classification_report18import xgboost as xgb19classifiers = {'NaiveBayes':  {'clf':GaussianNB(),20                            'params':{}},21                   'XGBoost': {'clf':xgb.XGBClassifier(random_state=42, n_jobs=4),22                                'params':{"learning_rate": [0.1, 0.01, 0.001],23                           "gamma" : [0.01, 0.1, 0.3, 0.5, 1, 1.5, 2],24                           "max_depth": [2, 4, 7, 10],25                           "colsample_bytree": [0.3, 0.6, 0.8, 1.0],26                           "subsample": [0.2, 0.4, 0.5, 0.6, 0.7],27                           "reg_alpha": [0, 0.5, 1],28                           "reg_lambda": [1, 1.5, 2, 3, 4.5],29                           "min_child_weight": [1, 3, 5, 7],30                           "n_estimators": [100]}},31                   'RandomForest':  {'clf':RandomForestClassifier(random_state=42, n_jobs=4),32                                'params':{'n_estimators': [100],33                                 'max_features': ['sqrt', 'log2', None],34                                 'max_depth': [2,4,7,10],35                                 'min_samples_split': [2, 5, 10, 20],36                                 'min_samples_leaf': [1, 2, 4, 8],37                                 'class_weight':[None, 'balanced_subsample']38                                 }},39                   'DecisionTree':  {'clf':DecisionTreeClassifier(random_state=42),40                                'params':{'max_features': ['sqrt', 'log2', None],41                                 'max_depth': [2,4,7,10],42                                 'min_samples_split': [2, 5, 10, 20],43                                 'min_samples_leaf': [1, 2, 4, 8],44                                 'class_weight':[None, 'balanced']45                                 }},46                   'LinearSVC':  {'clf':LinearSVC(random_state=42, dual=False),47                      'params':{'C': [0.0001 ,0.001, 0.01, 0.1, 1, 10, 100, 1000],48                      'class_weight':[None, 'balanced']}},49                   'LogisticRegression': {'clf':LogisticRegression(random_state=42, solver='liblinear'),50                        'params':{'C': [0.0001 ,0.001, 0.01, 0.1, 1, 10, 100, 1000],51                        'class_weight':[None, 'balanced']}},52                   }53def run_magellan(train_set, valid_set, test_set, feature_combinations, classifiers, experiment_name,54                 write_test_set_for_inspection=False):55    train_path = os.path.dirname(train_set)56    train_file = os.path.basename(train_set)57    test_path = os.path.dirname(test_set)58    test_file = os.path.basename(test_set)59    report_train_name = train_file.replace('.csv', '')60    report_test_name = test_file.replace('.csv', '')61    train_set_left = train_file.replace('pairs', 'left')62    train_set_right = train_file.replace('pairs', 'right')63    test_set_left = test_file.replace('pairs', 'left')64    test_set_right = test_file.replace('pairs', 'right')65    os.makedirs(os.path.dirname('../../../reports/magellan/{}/'.format(experiment_name)),66                exist_ok=True)67    try:68        os.remove('../../../reports/magellan/{}/{}_{}.csv'.format(experiment_name, report_train_name, report_test_name))69    except OSError:70        pass71    with open('../../../reports/magellan/{}/{}_{}.csv'.format(experiment_name, report_train_name, report_test_name),72              "w") as f:73        f.write(74            'feature#####model#####mean_train_score#####std_train_score#####mean_valid_score#####std_valid_score#####precision_test#####recall_test#####f1_test#####best_params#####train_time#####prediction_time#####feature_importance#####experiment_name#####train_set#####test_set\n')75    for run in range(1, 4):76        for feature_combination in feature_combinations:77            A_t = em.read_csv_metadata(train_path + '/' + train_set_left, key='mag_id')78            B_t = em.read_csv_metadata(train_path + '/' + train_set_right, key='mag_id')79            # Load the pre-labeled data80            S_t = em.read_csv_metadata(train_set,81                                       key='_id',82                                       ltable=A_t, rtable=B_t,83                                       fk_ltable='ltable_mag_id', fk_rtable='rtable_mag_id')84            A_gs = em.read_csv_metadata(test_path + '/' + test_set_left, key='mag_id')85            B_gs = em.read_csv_metadata(test_path + '/' + test_set_right, key='mag_id')86            # Load the pre-labeled data87            S_gs = em.read_csv_metadata(test_set,88                                        key='_id',89                                        ltable=A_gs, rtable=B_gs,90                                        fk_ltable='ltable_mag_id', fk_rtable='rtable_mag_id')91            A_t.fillna('', inplace=True)92            A_gs.fillna('', inplace=True)93            B_t.fillna('', inplace=True)94            B_gs.fillna('', inplace=True)95            S_t.fillna('', inplace=True)96            S_gs.fillna('', inplace=True)97            ## DIRTY FIX, CLEAN UP!98            if 'name' in A_t.columns:99                A_t["price"] = A_t["price"].replace(r'^\s*$', np.nan, regex=True)100                A_t["price"] = A_t["price"].astype('float64')101                A_gs["price"] = A_gs["price"].replace(r'^\s*$', np.nan, regex=True)102                A_gs["price"] = A_gs["price"].astype('float64')103                B_t["price"] = B_t["price"].replace(r'^\s*$', np.nan, regex=True)104                B_t["price"] = B_t["price"].astype('float64')105                B_gs["price"] = B_gs["price"].replace(r'^\s*$', np.nan, regex=True)106                B_gs["price"] = B_gs["price"].astype('float64')107                S_t["ltable_price"] = S_t["ltable_price"].replace(r'^\s*$', np.nan, regex=True)108                S_t["ltable_price"] = S_t["ltable_price"].astype('float64')109                S_t["rtable_price"] = S_t["rtable_price"].replace(r'^\s*$', np.nan, regex=True)110                S_t["rtable_price"] = S_t["rtable_price"].astype('float64')111                S_gs["ltable_price"] = S_gs["ltable_price"].replace(r'^\s*$', np.nan, regex=True)112                S_gs["ltable_price"] = S_gs["ltable_price"].astype('float64')113                S_gs["rtable_price"] = S_gs["rtable_price"].replace(r'^\s*$', np.nan, regex=True)114                S_gs["rtable_price"] = S_gs["rtable_price"].astype('float64')115            atypes1 = em.get_attr_types(A_t)116            atypes2 = em.get_attr_types(B_t)117            match_c = em.get_attr_corres(A_t, B_t)118            match_c['corres'] = []119            # select attributes to compare120            for feature in feature_combination:121                match_c['corres'].append((feature, feature))122            tok = em.get_tokenizers_for_matching()123            sim = em.get_sim_funs_for_matching()124            F_t = em.get_features(A_t, B_t, atypes1, atypes2, match_c, tok, sim)125            H_t = em.extract_feature_vecs(S_t,126                                          feature_table=F_t,127                                          attrs_after=['label', 'pair_id'],128                                          show_progress=False)129            H_gs = em.extract_feature_vecs(S_gs,130                                           feature_table=F_t,131                                           attrs_after='label',132                                           show_progress=False)133            H_t = H_t.fillna(-1)134            H_gs = H_gs.fillna(-1)135            validation_ids_df = pd.read_csv(valid_set)136            val_df = H_t[H_t['pair_id'].isin(validation_ids_df['pair_id'].values)]137            train_only_df = H_t[~H_t['pair_id'].isin(validation_ids_df['pair_id'].values)]138            train_only_df = train_only_df.drop(columns='pair_id')139            val_df = val_df.drop(columns='pair_id')140            train_only_df = train_only_df.sample(frac=1, random_state=42)141            pos_neg = H_t['label'].value_counts()142            pos_neg = round(pos_neg[0] / pos_neg[1])143            train_ind = []144            val_ind = []145            for i in range(len(train_only_df) - 1):146                train_ind.append(-1)147            for i in range(len(val_df) - 1):148                val_ind.append(0)149            ps = PredefinedSplit(test_fold=np.concatenate((train_ind, val_ind)))150            train_df = pd.concat([train_only_df, val_df])151            for k, v in classifiers.items():152                classifier = v['clf']153                if 'random_state' in classifier.get_params().keys():154                    classifier = classifier.set_params(**{'random_state': run})155                # add pos_neg ratio to XGBoost params156                if k == 'XGBoost':157                    v['params']['scale_pos_weight']: [1, pos_neg]158                model = RandomizedSearchCV(cv=ps, estimator=classifier, param_distributions=v['params'],159                                           random_state=42, n_jobs=4, scoring='f1', n_iter=500, pre_dispatch=8,160                                           return_train_score=True)161                feats_train = train_df.drop(['_id', 'ltable_mag_id', 'rtable_mag_id', 'label'], axis=1)162                labels_train = train_df['label']163                feats_gs = H_gs.drop(['_id', 'ltable_mag_id', 'rtable_mag_id', 'label'], axis=1)164                labels_gs = H_gs['label']165                try:166                    model.fit(feats_train, labels_train)167                except ValueError:168                    set_trace()169                parameters = model.best_params_170                score_names = ['mean_train_score', 'std_train_score', 'mean_test_score', 'std_test_score']171                scores = {}172                score_string = ''173                for name in score_names:174                    scores[name] = model.cv_results_[name][model.best_index_]175                    score_string = score_string + name + ': ' + str(scores[name]) + ' '176                feature_names = list(feats_train.columns)177                if k == 'LogisticRegression' or k == 'LinearSVC':178                    most_important_features = model.best_estimator_.coef_179                    word_importance = zip(feature_names, most_important_features[0].tolist())180                    word_importance = sorted(word_importance, key=lambda importance: importance[1], reverse=True)181                if k == 'RandomForest' or k == 'DecisionTree':182                    most_important_features = model.best_estimator_.feature_importances_183                    word_importance = zip(feature_names, most_important_features.tolist())184                    word_importance = sorted(word_importance, key=lambda importance: importance[1], reverse=True)185                if k == 'NaiveBayes':186                    word_importance = ''187                if k == 'XGBoost':188                    most_important_features = model.best_estimator_.feature_importances_189                    word_importance = zip(feature_names, most_important_features.tolist())190                    word_importance = sorted(word_importance, key=lambda importance: importance[1], reverse=True)191                if k == 'LogisticRegression':192                    learner = LogisticRegression(random_state=run, solver='liblinear', **parameters)193                elif k == 'NaiveBayes':194                    learner = GaussianNB()195                elif k == 'DecisionTree':196                    learner = DecisionTreeClassifier(random_state=run, **parameters)197                elif k == 'LinearSVC':198                    learner = LinearSVC(random_state=run, dual=False, **parameters)199                elif k == 'RandomForest':200                    learner = RandomForestClassifier(random_state=run, n_jobs=4, **parameters)201                elif k == 'XGBoost':202                    learner = xgb.XGBClassifier(random_state=run, n_jobs=4, **parameters)203                else:204                    print('Learner is not a valid option')205                    break206                model = learner207                feats_train = train_only_df.sample(frac=1, random_state=42)208                feats_train = train_only_df.drop(['_id', 'ltable_mag_id', 'rtable_mag_id', 'label'], axis=1)209                labels_train = train_only_df['label']210                start = time.time()211                model.fit(feats_train, labels_train)212                end = time.time()213                train_time = end - start214                start = time.time()215                preds_gs = model.predict(feats_gs)216                end = time.time()217                pred_time = end - start218                gs_report = classification_report(labels_gs, preds_gs, output_dict=True)219                feature_report = '+'.join(feature_combination)220                if write_test_set_for_inspection:221                    out_path = '../../../data/processed/wdc-lspc/inspection/{}/magellan/'.format(experiment_name)222                    os.makedirs(os.path.dirname(out_path), exist_ok=True)223                    file_name = '_'.join([os.path.basename(train_set), os.path.basename(test_set), k, feature_report])224                    file_name = file_name.replace('.csv', '')225                    file_name += f'_{run}.pkl.gz'226                    test_inspection_df = S_gs.copy()227                    if k == 'LinearSVC':228                        proba_gs = model.decision_function(feats_gs).tolist()229                    else:230                        proba_gs = model.predict_proba(feats_gs).tolist()231                    test_inspection_df['pred'] = preds_gs232                    test_inspection_df['Class Prob'] = proba_gs233                    test_inspection_df.to_pickle(out_path + file_name, compression='gzip')234                with open('../../../reports/magellan/{}/{}_{}.csv'.format(experiment_name, report_train_name,235                                                                          report_test_name), "a") as f:236                    f.write(feature_report + '#####' + k + '#####' + str(237                        scores['mean_train_score']) + '#####' + str(scores['std_train_score'])238                            + '#####' + str(scores['mean_test_score']) + '#####' + str(239                        scores['std_test_score']) + '#####' + str(gs_report['1']['precision']) + '#####' + str(240                        gs_report['1']['recall']) + '#####' + str(gs_report['1']['f1-score'])241                            + '#####' + str(parameters) + '#####' + str(train_time) + '#####' + str(pred_time)242                            + '#####' + str(word_importance[243                                            0:100]) + '#####' + experiment_name + '#####' + report_train_name + '#####' + report_test_name + '\n')244if __name__ == '__main__':245    # learning-curve experiment246    feature_combinations = [['title'], ['title', 'description'], ['title', 'description', 'brand'],247                            ['title', 'description', 'brand', 'specTableContent']]248    experiment_name = 'learning-curve'249    for file in glob.glob('../../../data/processed/wdc-lspc/magellan/learning-curve/formatted/*'):250        if 'train_' in file and 'pairs' in file and 'metadata' not in file:251            valid = file.replace('train_', 'valid_')252            test_cat = '_'.join(os.path.basename(file).split('_')[:2])253            test = '../../../data/processed/wdc-lspc/magellan/learning-curve/formatted/{}_gs_magellan_pairs_formatted.csv'.format(254                test_cat)255            run_magellan(file, valid, test, feature_combinations, classifiers, experiment_name,...

run_wordcooc.py

Source:run_wordcooc.py

1import pandas as pd2import numpy as np3np.random.seed(42)4import random5random.seed(42)6import scipy7import os8import time9import glob10import json11from sklearn.linear_model import LogisticRegression12from sklearn.naive_bayes import BernoulliNB13from sklearn.ensemble import RandomForestClassifier14from sklearn.svm import LinearSVC15from sklearn.tree import DecisionTreeClassifier16from sklearn.model_selection import RandomizedSearchCV17from sklearn.model_selection import PredefinedSplit18from sklearn.metrics import classification_report19import xgboost as xgb20classifiers = {'NaiveBayes':  {'clf':BernoulliNB(),21                            'params':{}},22                   'XGBoost': {'clf':xgb.XGBClassifier(random_state=42, n_jobs=4),23                                'params':{"learning_rate": [0.1, 0.01, 0.001],24                           "gamma" : [0.01, 0.1, 0.3, 0.5, 1, 1.5, 2],25                           "max_depth": [2, 4, 7, 10],26                           "colsample_bytree": [0.3, 0.6, 0.8, 1.0],27                           "subsample": [0.2, 0.4, 0.5, 0.6, 0.7],28                           "reg_alpha": [0, 0.5, 1],29                           "reg_lambda": [1, 1.5, 2, 3, 4.5],30                           "min_child_weight": [1, 3, 5, 7],31                           "n_estimators": [100]}},32                   'RandomForest':  {'clf':RandomForestClassifier(random_state=42, n_jobs=4),33                                'params':{'n_estimators': [100],34                                 'max_features': ['sqrt', 'log2', None],35                                 'max_depth': [2,4,7,10],36                                 'min_samples_split': [2, 5, 10, 20],37                                 'min_samples_leaf': [1, 2, 4, 8],38                                 'class_weight':[None, 'balanced_subsample']39                                 }},40                   'DecisionTree':  {'clf':DecisionTreeClassifier(random_state=42),41                                'params':{'max_features': ['sqrt', 'log2', None],42                                 'max_depth': [2,4,7,10],43                                 'min_samples_split': [2, 5, 10, 20],44                                 'min_samples_leaf': [1, 2, 4, 8],45                                 'class_weight':[None, 'balanced']46                                 }},47                   'LinearSVC':  {'clf':LinearSVC(random_state=42, dual=False),48                      'params':{'C': [0.0001 ,0.001, 0.01, 0.1, 1, 10, 100, 1000],49                      'class_weight':[None, 'balanced']}},50                   'LogisticRegression': {'clf':LogisticRegression(random_state=42, solver='liblinear'),51                        'params':{'C': [0.0001 ,0.001, 0.01, 0.1, 1, 10, 100, 1000],52                        'class_weight':[None, 'balanced']}},53                   }54def run_wordcooc(train_set, valid_set, test_set, feature_combinations, classifiers, experiment_name,55                 write_test_set_for_inspection=False):56    train_path = os.path.dirname(train_set)57    train_file = os.path.basename(train_set)58    test_path = os.path.dirname(test_set)59    test_file = os.path.basename(test_set)60    report_train_name = train_file.replace('.pkl.gz', '')61    report_test_name = test_file.replace('.pkl.gz', '')62    os.makedirs(os.path.dirname('../../../reports/wordcooc/{}/'.format(experiment_name)),63                exist_ok=True)64    try:65        os.remove('../../../reports/wordcooc/{}/{}_{}.csv'.format(experiment_name, report_train_name, report_test_name))66    except OSError:67        pass68    with open('../../../reports/wordcooc/{}/{}_{}.csv'.format(experiment_name, report_train_name, report_test_name),69              "w") as f:70        f.write(71            'feature#####model#####mean_train_score#####std_train_score#####mean_valid_score#####std_valid_score#####precision_test#####recall_test#####f1_test#####best_params#####train_time#####prediction_time#####feature_importance#####experiment_name#####train_set#####test_set\n')72    for run in range(1, 4):73        for feature_combination in feature_combinations:74            train_original_df = pd.read_pickle(train_set, compression='gzip')75            gs_df = pd.read_pickle(test_set, compression='gzip')76            feature_file_name = train_file.replace('.pkl.gz', '_words.json')77            with open(train_path + '/feature-names/' + feature_file_name) as json_data:78                words = json.load(json_data)79            validation_ids_df = pd.read_pickle(valid_set, compression='gzip')80            val_df = train_original_df[train_original_df['pair_id'].isin(validation_ids_df['pair_id'].values)]81            train_only_df = train_original_df[~train_original_df['pair_id'].isin(validation_ids_df['pair_id'].values)]82            train_only_df = train_only_df.sample(frac=1, random_state=42)83            pos_neg = train_original_df['label'].value_counts()84            pos_neg = round(pos_neg[0] / pos_neg[1])85            train_ind = []86            val_ind = []87            for i in range(len(train_only_df) - 1):88                train_ind.append(-1)89            for i in range(len(val_df) - 1):90                val_ind.append(0)91            ps = PredefinedSplit(test_fold=np.concatenate((train_ind, val_ind)))92            train_df = pd.concat([train_only_df, val_df])93            for k, v in classifiers.items():94                classifier = v['clf']95                if 'random_state' in classifier.get_params().keys():96                    classifier = classifier.set_params(**{'random_state': run})97                # add pos_neg ratio to XGBoost params98                if k == 'XGBoost':99                    v['params']['scale_pos_weight']: [1, pos_neg]100                model = RandomizedSearchCV(cv=ps, estimator=classifier, param_distributions=v['params'],101                                           random_state=42, n_jobs=4, scoring='f1', n_iter=500, pre_dispatch=8,102                                           return_train_score=True)103                feats_train = scipy.sparse.vstack(train_df[feature_combination + '_wordcooc'])104                labels_train = train_df['label']105                feats_gs = scipy.sparse.vstack(gs_df[feature_combination + '_wordcooc'])106                labels_gs = gs_df['label']107                model.fit(feats_train, labels_train)108                parameters = model.best_params_109                score_names = ['mean_train_score', 'std_train_score', 'mean_test_score', 'std_test_score']110                scores = {}111                score_string = ''112                for name in score_names:113                    scores[name] = model.cv_results_[name][model.best_index_]114                    score_string = score_string + name + ': ' + str(scores[name]) + ' '115                if k == 'LogisticRegression' or k == 'LinearSVC':116                    most_important_features = model.best_estimator_.coef_117                    word_importance = zip(words[feature_combination], most_important_features[0].tolist())118                    word_importance = sorted(word_importance, key=lambda importance: importance[1], reverse=True)119                if k == 'RandomForest' or k == 'DecisionTree':120                    most_important_features = model.best_estimator_.feature_importances_121                    word_importance = zip(words[feature_combination], most_important_features.tolist())122                    word_importance = sorted(word_importance, key=lambda importance: importance[1], reverse=True)123                if k == 'NaiveBayes':124                    word_importance = ''125                if k == 'XGBoost':126                    most_important_features = model.best_estimator_.feature_importances_127                    word_importance = zip(words[feature_combination], most_important_features.tolist())128                    word_importance = sorted(word_importance, key=lambda importance: importance[1], reverse=True)129                if k == 'LogisticRegression':130                    learner = LogisticRegression(random_state=run, solver='liblinear', **parameters)131                elif k == 'NaiveBayes':132                    learner = BernoulliNB()133                elif k == 'DecisionTree':134                    learner = DecisionTreeClassifier(random_state=run, **parameters)135                elif k == 'LinearSVC':136                    learner = LinearSVC(random_state=run, dual=False, **parameters)137                elif k == 'RandomForest':138                    learner = RandomForestClassifier(random_state=run, n_jobs=4, **parameters)139                elif k == 'XGBoost':140                    learner = xgb.XGBClassifier(random_state=run, n_jobs=4, **parameters)141                else:142                    print('Learner is not a valid option')143                    break144                model = learner145                feats_train = scipy.sparse.vstack(train_only_df[feature_combination + '_wordcooc'])146                labels_train = train_only_df['label']147                start = time.time()148                model.fit(feats_train, labels_train)149                end = time.time()150                train_time = end - start151                start = time.time()152                preds_gs = model.predict(feats_gs)153                end = time.time()154                pred_time = end - start155                gs_report = classification_report(labels_gs, preds_gs, output_dict=True)156                if write_test_set_for_inspection:157                    out_path = '../../../data/processed/wdc-lspc/inspection/{}/wordcooc/'.format(experiment_name)158                    os.makedirs(os.path.dirname(out_path), exist_ok=True)159                    file_name = '_'.join(160                        [os.path.basename(train_set), os.path.basename(test_set), k, feature_combination])161                    file_name = file_name.replace('.csv', '')162                    file_name += f'_{run}.pkl.gz'163                    test_inspection_df = gs_df.copy()164                    if k == 'LinearSVC':165                        proba_gs = model.decision_function(feats_gs).tolist()166                    else:167                        proba_gs = model.predict_proba(feats_gs).tolist()168                    test_inspection_df['pred'] = preds_gs169                    test_inspection_df['Class Prob'] = proba_gs170                    test_inspection_df.to_pickle(out_path + file_name, compression='gzip')171                with open('../../../reports/wordcooc/{}/{}_{}.csv'.format(experiment_name, report_train_name,172                                                                          report_test_name), "a") as f:173                    f.write(feature_combination + '#####' + k + '#####' + str(174                        scores['mean_train_score']) + '#####' + str(scores['std_train_score'])175                            + '#####' + str(scores['mean_test_score']) + '#####' + str(176                        scores['std_test_score']) + '#####' + str(gs_report['1']['precision']) + '#####' + str(177                        gs_report['1']['recall']) + '#####' + str(gs_report['1']['f1-score'])178                            + '#####' + str(parameters) + '#####' + str(train_time) + '#####' + str(pred_time)179                            + '#####' + str(word_importance[180                                            0:100]) + '#####' + experiment_name + '#####' + report_train_name + '#####' + report_test_name + '\n')181if __name__ == '__main__':182    # learning-curve experiment183    feature_combinations = ['title', 'title+description', 'title+description+brand',184                            'title+description+brand+specTableContent']185    experiment_name = 'learning-curve'186    for file in glob.glob('../../../data/processed/wdc-lspc/wordcooc/learning-curve/*'):187        if 'train_' in file and '_gs' not in file:188            valid = file.replace('train_', 'valid_')189            test_cat = '_'.join(os.path.basename(file).split('_')[:2])190            test = os.path.basename(file)191            test = test.replace('.pkl.gz', '_{}_gs.pkl.gz'.format(test_cat))192            test = '../../../data/processed/wdc-lspc/wordcooc/learning-curve/{}'.format(test)193            run_wordcooc(file, valid, test, feature_combinations, classifiers, experiment_name,...

test_logger.py

Source:test_logger.py

...30        self.results = []31        self.test_case_classes = set()32    def test_start(self, result):33        self.test_case_classes.add((result['method']['module'], result['method']['class']))34        self.report_test_name(result['method'])35    def test_complete(self, result):36        self.report_test_result(result)37        self.results.append(result)38        if not result['success']:39            self.report_failure(result)40    def fixture_start(self, result):41        self.test_case_classes.add((result['method']['module'], result['method']['class']))42    def class_teardown_complete(self, result):43        if not result['success']:44            self.report_test_name(result['method'])45            self.report_test_result(result)46            self.results.append(result)47    def report(self):48        # All the TestCases have been run - now collate results by status and log them49        results_by_status = collections.defaultdict(list)50        for result in self.results:51            if result['success']:52                results_by_status['successful'].append(result)53            elif result['failure'] or result['error']:54                results_by_status['failed'].append(result)55            elif result['interrupted']:56                results_by_status['interrupted'].append(result)57            else:58                results_by_status['unknown'].append(result)59        if self.options.summary_mode:60            self.report_failures(results_by_status['failed'])61        self.report_stats(len(self.test_case_classes), **results_by_status)62        if len(self.results) == 0:63            return False64        else:65            return (66                (67                    len(results_by_status['failed']) +68                    len(results_by_status['interrupted']) +69                    len(results_by_status['unknown'])70                ) == 071            )72    def report_test_name(self, test_method):73        pass74    def report_test_result(self, result):75        pass76    def report_failures(self, failed_results):77        if failed_results:78            self.heading('FAILURES', 'The following tests are expected to pass.')79            for result in failed_results:80                self.failure(result)81        else:82            # throwing this in so that someone looking at the bottom of the83            # output won't have to scroll up to figure out whether failures84            # were expected or not.85            self.heading('FAILURES', 'None!')86    def report_failure(self, result):87        pass88    def report_stats(self, test_case_count, all_results, failed_results, unknown_results):89        pass90    def _format_test_method_name(self, test_method):91        """Take a test method as input and return a string for output"""92        if test_method['module'] != '__main__':93            return "%s %s.%s" % (test_method['module'], test_method['class'], test_method['name'])94        else:95            return "%s.%s" % (test_method['class'], test_method['name'])96class TextTestLogger(TestLoggerBase):97    def __init__(self, options, stream=sys.stdout):98        super(TextTestLogger, self).__init__(options, stream)99        # Checking for color support isn't as fun as we might hope.  We're100        # going to use the command 'tput colors' to get a list of colors101        # supported by the shell. But of course we if this fails terribly,102        # we'll want to just fall back to no colors103        self.use_color = False104        # if TERM is not present in environ, tput prints to stderr105        # if tput's stderr is a pipe, it lies.106        if sys.stdin.isatty() and 'TERM' in os.environ:107            try:108                output = subprocess.check_output(('tput', 'colors'))109                if int(output.strip()) >= 8:110                    self.use_color = True111            except Exception as e:112                if self.options.verbosity >= VERBOSITY_VERBOSE:113                    self.writeln("Failed to find color support: %r" % e)114    def write(self, message):115        """Write a message to the output stream, no trailing newline"""116        if six.PY2:117            self.stream.write(message.encode('UTF-8') if isinstance(message, six.text_type) else message)118        else:119            self.stream.write(message.decode('UTF-8') if isinstance(message, bytes) else message)120        self.stream.flush()121    def writeln(self, message):122        """Write a message and append a newline"""123        self.write(message)124        self.write('\n')125    BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(30, 38)126    def _colorize(self, message, color=CYAN):127        if not color or not self.use_color:128            return message129        else:130            start_color = '\033[1;%sm' % color131            end_color = '\033[m'132            return start_color + message + end_color133    def test_discovery_failure(self, exc):134        self.writeln(self._colorize("DISCOVERY FAILURE!", self.MAGENTA))135        self.writeln("There was a problem importing one or more tests:")136        self.writeln(str(exc))137    def report_test_name(self, test_method):138        if self.options.verbosity >= VERBOSITY_VERBOSE:139            self.write("%s ... " % self._format_test_method_name(test_method))140    def report_test_result(self, result):141        if self.options.verbosity > VERBOSITY_SILENT:142            if result['success']:143                if result['previous_run']:144                    status = "flaky"145                else:146                    status = "success"147            elif result['failure']:148                status = "fail"149            elif result['error']:150                status = "error"151            elif result['interrupted']:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.