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']:...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!