KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use list_group_users method in tempest

Best Python code snippet using tempest_python

transfers_utils.py

Source:transfers_utils.py

1# coding=utf-82import time3import re4import pandas as pd5import logging6from cms_support.utils.constants import CteRucio7import xlsxwriter8class ExcelGenerator:9    def __init__(self, dict_results, target):10        self.dict_results = dict_results11        self.target = target12    @staticmethod13    def get_column_id(num_rows, num_columns_ahead=0, num_rows_ahead=0):14        letter_column = chr(64 + num_columns_ahead)15        structure_id = "{}{}:{}{}".format(letter_column, num_rows_ahead + 1, letter_column,16                                          num_rows + num_rows_ahead + 1)17        return structure_id18    @staticmethod19    def get_sub_table(dict_grouped_by_id, list_elements):20        list_group = []21        for group_id, data in dict_grouped_by_id.items():22            new_row = []23            dict_data = dict((i, data.count(i)) for i in data)24            if dict_data:25                for element_value in list_elements:26                    if element_value in dict_data.keys():27                        new_row.append(dict_data[element_value])28                    else:29                        new_row.append(None)30                list_group.append([group_id] + new_row)31        return list_group32    @staticmethod33    def write_lfn_txt(lfns_file_name, lfns):34        text = ""35        for error, list_lfns in lfns.items():36            text += "*" * 30 + "\n"37            text += error.capitalize() + "\n"38            text += "*" * 30 + "\n"39            for lfn in list_lfns:40                text += lfn + "\n"41        f = open(lfns_file_name + ".txt", "a")42        f.write(text)43        f.close()44    @staticmethod45    def get_user(url_pfn):46        user = ""47        raw_users = re.findall("/user/(.*)/", str(url_pfn))48        if raw_users:49            user = raw_users[0].split("/")[0].strip("")50            user = user.split(".")[0]  # e.g. gbakas.9c1d054d2d278c14ddc228476ff7559c10393d8d51        if len(raw_users) > 2:52            raise Exception("MULTIPLE USERS ON PFN")53        return user54    def results_to_csv(self, write_lfns=False):55        src_dst_info = ["rse", "type", "se", "url", "lfn", "protocol"]56        expanded_src_dst_info = []57        [expanded_src_dst_info.append(direction + "_" + info) for direction in ["src", "dst"] for info in src_dst_info]58        file_info = ['transfer_id', 'checksum_adler', 'file_size', 'transfer_link',59                     'submitted_at', 'started_at', 'transferred_at', 'purged_reason']60        columns = expanded_src_dst_info + file_info61        ############  ITERATE OVER ALL SRMs HOSTS ############62        for storage_element, se_value in self.dict_results.items():63            time_analysis = round(time.time())64            file_name = '{}_{}'.format(time_analysis, self.target)65            writer = pd.ExcelWriter(file_name + ".xlsx", engine='xlsxwriter')66            ############ GET DATA ORIGIN AND DESTINATION ############67            for direction, direction_value in se_value.items():68                list_errors, list_groups, list_users, list_other_endpoint = [], [], [], []69                users, endpoints, lfns = {}, {}, {}70                group_id = 171                if "s" == direction[0]:72                    other_direction = CteRucio.REF_SE_DST73                    other_url_direction = CteRucio.REF_PFN_DST74                    same_url_direction = CteRucio.REF_PFN_SRC75                    same_lfn = CteRucio.REF_LFN_SRC76                else:77                    other_direction = CteRucio.REF_SE_SRC78                    other_url_direction = CteRucio.REF_PFN_SRC79                    same_url_direction = CteRucio.REF_PFN_DST80                    same_lfn = CteRucio.REF_LFN_DST81                ############  ITERATE OVER ALL ERROR GROUPS ############82                for error_key, error_value in direction_value.items():83                    users.update({group_id: []})84                    endpoints.update({group_id: []})85                    lfns.update({error_key: []})86                    failed_transfers = 087                    ############  ITERATE OVER ALL ERRORS ############88                    for single_error in error_value:89                        # ADD USER IN LIST90                        user_site = self.get_user(single_error[same_url_direction])91                        user_other = self.get_user(single_error[other_url_direction])92                        if user_site:93                            users[group_id] += [user_site] * single_error[CteRucio.REF_NUM_ERRORS]94                            if user_other and user_site != user_other:95                                logging.error("Different users {} vs {}".format(user_site, user_other))96                            if user_site not in list_users:97                                list_users.append(user_site)98                        # ADD ENDPOINT IN LIST99                        other_endpoint = single_error[other_direction]100                        endpoints[group_id] += [other_endpoint] * single_error[CteRucio.REF_NUM_ERRORS]101                        if other_endpoint not in list_other_endpoint:102                            list_other_endpoint.append(other_endpoint)103                        # ADD LIST LFNs104                        if write_lfns and single_error[same_lfn] and single_error[same_lfn] not in lfns[error_key]:105                            lfns[error_key].append(single_error[same_lfn])106                        # ADD ALL THE ERROR INFORMATION107                        values_columns = [single_error[elem] for elem in columns]108                        values_columns.append(user_site)109                        values_columns.append(group_id)110                        # Row errors table111                        list_errors.append(values_columns)112                        # Count total of failed transfers for each group113                        failed_transfers += single_error[CteRucio.REF_NUM_ERRORS]114                    # Row table (legend) group errors115                    list_groups.append([group_id, error_key, len(error_value), failed_transfers])116                    group_id += 1117                # WRITE TXT WITH LFNs118                if write_lfns:119                    lfns_file_name = file_name + "_LFNs_{}".format(direction)120                    self.write_lfn_txt(lfns_file_name, lfns)121                # DF ERRORS122                columns_errors = columns + [CteRucio.REF_USER, "group_id"]123                num_columns_error = len(columns_errors)124                df = pd.DataFrame(list_errors, columns=columns_errors)125                df.to_excel(writer, sheet_name=direction, index=False)126                column_id_error = self.get_column_id(len(list_errors), num_columns_error)127                # DF LEGEND GROUPS128                columns_groups = ["group_id", "error_ref", "num_diff_errors", "total_errors"]129                start_column = num_columns_error + CteRucio.SEPARATION_COLUMNS130                df_group = pd.DataFrame(list_groups, columns=columns_groups)131                df_group.to_excel(writer, sheet_name=direction, startcol=start_column, index=False)132                column_id_group = self.get_column_id(len(list_groups), start_column + 1)133                # DF USERS134                list_group_users = self.get_sub_table(users, list_users)135                columns_users = ["group_id"] + list_users136                start_column = num_columns_error + CteRucio.SEPARATION_COLUMNS137                start_row_users = len(list_groups) + CteRucio.SEPARATION_ROWS138                if list_group_users:139                    df_users = pd.DataFrame(list_group_users, columns=columns_users)140                    df_users.to_excel(writer, sheet_name=direction, startcol=start_column, startrow=start_row_users,141                                      index=False)142                # DF ENDPOINTS143                list_group_endpoints = self.get_sub_table(endpoints, list_other_endpoint)144                columns_endpoints = ["group_id"] + list_other_endpoint145                start_column = num_columns_error + CteRucio.SEPARATION_COLUMNS146                start_row = start_row_users + len(list_group_users) + CteRucio.SEPARATION_ROWS147                if list_group_endpoints:148                    df_endpoint = pd.DataFrame(list_group_endpoints, columns=columns_endpoints)149                    df_endpoint.to_excel(writer, sheet_name=direction, startcol=start_column, startrow=start_row,150                                         index=False)151                # COLOR SHEET152                worksheet = writer.sheets[direction]153                worksheet.conditional_format(column_id_error, {'type': '3_color_scale'})154                worksheet.conditional_format(column_id_group, {'type': '3_color_scale'})...

user.py

Source:user.py

...71    else:72        print("Remove user from group failure")73# Query users in a user group74# GET /v3/groups/{group_id}/users75def list_group_users(group_id):76    groups = conn.identity.list_group_users(group_id)77    for group in groups:78        print(group)79# Query the user group to which a user belongs80# GET /v3/users/{user_id}/groups81def list_user_groups(user_id):82    groups = conn.identity.list_user_groups(user_id)83    for group in groups:84        print(group)85if __name__ == "__main__":86    group_id = "**********"87    user_id = "**********"88    password_attr = {89        "user": {90            "password": "**********",91            "original_password": "**********"92        }93    }94    get_user_list()95    get_user_detail(user_id)96    create_user()97    modify_user(user_id)98    change_password(user_id, **password_attr)99    remove_user_from_group(group_id, user_id)100    list_group_users(group_id)101    list_user_groups(user_id)...

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.