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

How to use get_pkey method in lisa

Best Python code snippet using lisa_python

ufm_pkeys.py

Source:ufm_pkeys.py

1#!/usr/bin/python32"""3@copyright:4    Copyright (C) Mellanox Technologies Ltd. 2014-2020.  ALL RIGHTS RESERVED.5    This software product is a proprietary product of Mellanox Technologies6    Ltd. (the "Company") and all right, title, and interest in and to the7    software product, including all associated intellectual property rights,8    are and shall remain exclusively with the Company.9    This software product is governed by the End User License Agreement10    provided with the software product.11@author: Anan Al-Aghbar12@date:   Oct 3, 202113"""14import os15import sys16from http import HTTPStatus17try:18    from utils.utils import Utils19    from utils.ufm_rest_client import UfmRestClient, HTTPMethods20    from utils.args_parser import ArgsParser21    from utils.config_parser import ConfigParser22    from utils.logger import Logger, LOG_LEVELS23except ModuleNotFoundError as e:24    print("Error occurred while importing python modules, "25          "Please make sure that you exported your repository to PYTHONPATH by running: "26          f'export PYTHONPATH="${{PYTHONPATH}}:{os.path.dirname(os.getcwd())}"')27class UfmPkeysConstants:28    PKEYS_API_URL = "resources/pkeys"29    API_GUIDS = "guids"30    API_PKEY = "pkey"31    API_INDEX0 = "index0"32    API_IP_OVER_IB = "ip_over_ib"33    API_MEMBERSHIP = "membership"34    API_MEMBERSHIPS = "memberships"35    PKEYS_OPERATIONS = {36        "get_pkeys": "get_pkeys",37        "get_pkey": "get_pkey",38        "set_pkey": "set_pkey",39        "delete_pkey": "delete_pkey"40    }41    args_list = [42        {43            "name": f'--{PKEYS_OPERATIONS.get("set_pkey")}',44            "help": "Option to set a Pkey network",45            "no_value": True46        },47        {48            "name": f'--{API_PKEY}',49            "help": "Network Pkey [Hexadecimal string between '0x0'-'0x7fff' exclusive]"50        },51        {52            "name": f'--{API_GUIDS}',53            "help": "The List of port GUIDs(comma seprated), Each GUID is a hexadecimal string with a minimum length of "54                    "16 characters and maximum length of 20 characters,"55                    " e.g.043f720300dd1d3c,0c42a103007aca90,etc... "56        },57        {58            "name": f'--{API_MEMBERSHIPS}',59            "help": "List of âfullâ or âlimitedâ comma-separated strings. "60                    "It must be the same length as the GUIDs list. "61                    "Each value by an index represents a GUID membership."62                    " e.g. ['full', 'limited', etc...]"63                    "This parameter conflicts with the âmembershipâ parameter. "64                    "You must select either a list of memberships or just one membership for all GUIDs."65        },66        {67            "name": f'--{API_MEMBERSHIP}',68            "help": "'full' | 'limited'. âfullâ- members with full membership can communicate with all hosts (members)"69                    " within the network/partition"70                    "âlimitedâ - members with limited membership "71                    "cannot communicate with other members with limited membership. "72                    "However, communication is allowed between every other combination of membership types."73                    "[Default = 'full']"74                    "This parameter will be ignored in case the âmembershipsâ parameter has been set"75        },76        {77            "name": f'--{API_INDEX0}',78            "help": "If true, the API will store the PKey at index 0 of the PKey table of the GUID.[Default = False]"79        },80        {81            "name": f'--{API_IP_OVER_IB}',82            "help": "If true, PKey is a member in a multicast group that uses IP over InfiniBand.[Default = True]"83        },84        {85            "name": f'--{PKEYS_OPERATIONS.get("get_pkeys")}',86            "help": "Option to get all existing pkeys data",87            "no_value": True88        },89        {90            "name": f'--{PKEYS_OPERATIONS.get("get_pkey")}',91            "help": "Option to get specific Pkey data"92        },93        {94            "name": f'--{PKEYS_OPERATIONS.get("delete_pkey")}',95            "help": "Option to delete specific Pkey"96        }97    ]98class UfmPkeysConfigParser(ConfigParser):99    def __init__(self, args):100        super().__init__(args)101class UfmPkeysManagement:102    @staticmethod103    def set_pkey(pkey,ports_guids_list,memberships= None,104                    default_membership = "full", index0 = False, ip_over_ib = True):105        payload = {106            UfmPkeysConstants.API_GUIDS:ports_guids_list,107            UfmPkeysConstants.API_PKEY: pkey,108            UfmPkeysConstants.API_INDEX0 : index0,109            UfmPkeysConstants.API_IP_OVER_IB : ip_over_ib110        }111        if memberships and len(memberships) :112            payload[UfmPkeysConstants.API_MEMBERSHIPS] = memberships113        else:114            payload[UfmPkeysConstants.API_MEMBERSHIP] = default_membership115        response = ufm_rest_client.send_request(UfmPkeysConstants.PKEYS_API_URL, HTTPMethods.PUT, payload=payload)116        if response and response.status_code == HTTPStatus.OK:117            Logger.log_message(f'The pkey: {pkey} has been set successfully')118        else:119            Logger.log_message(f'The pkey: {pkey} hasn`t been set', LOG_LEVELS.ERROR)120        return response121    @staticmethod122    def get_pkeys(pkey=None):123        if pkey:124            url = f'{UfmPkeysConstants.PKEYS_API_URL}/{pkey}'125        else:126            url = UfmPkeysConstants.PKEYS_API_URL127        url = f'{url}?guids_data=true'128        response = ufm_rest_client.send_request(url)129        if response and response.status_code == HTTPStatus.OK:130            Logger.log_message(response.json())131        else:132            Logger.log_message(response, LOG_LEVELS.ERROR)133        return response134    @staticmethod135    def delete_pkey(pkey):136        url = f'{UfmPkeysConstants.PKEYS_API_URL}/{pkey}'137        response = ufm_rest_client.send_request(url, HTTPMethods.DELETE)138        if response:139            if response.status_code == HTTPStatus.OK:140                Logger.log_message(f'The Pkey: {pkey} has been removed successfully')141            elif response.status_code == HTTPStatus.NOT_FOUND:142                Logger.log_message(f'The Pkey: {pkey} is not found', LOG_LEVELS.ERROR)143        else:144            Logger.log_message(f'The Pkey: {pkey} hasn`t been removed successfully', LOG_LEVELS.ERROR)145        return response146if __name__ == "__main__":147    # init app args148    args = ArgsParser.parse_args("UFM Pkeys Management", UfmPkeysConstants.args_list)149    # init app config parser & load config files150    config_parser = UfmPkeysConfigParser(args)151    # init logs configs152    logs_file_name = config_parser.get_logs_file_name()153    logs_level = config_parser.get_logs_level()154    Logger.init_logs_config(logs_file_name, logs_level)155    # init ufm rest client156    ufm_rest_client = UfmRestClient(host = config_parser.get_ufm_host(),157                                    client_token=config_parser.get_ufm_access_token())158    args_dict = args.__dict__159    if args_dict.get(UfmPkeysConstants.PKEYS_OPERATIONS.get("set_pkey")):160        try:161            pkey = config_parser.get_config_value(args_dict.get(UfmPkeysConstants.API_PKEY),162                                                  None,None)163            guids = config_parser.safe_get_list(args_dict.get(UfmPkeysConstants.API_GUIDS),164                                                   None, None)165        except Exception as e:166            Logger.log_message("create_pkey operation requires at least the following parameters: --pkey,--guids", LOG_LEVELS.ERROR)167            sys.exit(0)168        index0 = config_parser.safe_get_bool(args_dict.get(UfmPkeysConstants.API_INDEX0),169                                             None, None, False)170        ip_over_ib = config_parser.safe_get_bool(args_dict.get(UfmPkeysConstants.API_IP_OVER_IB),171                                   None, None, True)172        memberships = config_parser.safe_get_list(args_dict.get(UfmPkeysConstants.API_MEMBERSHIPS),173                                    None, None, [])174        membership = config_parser.get_config_value(args_dict.get(UfmPkeysConstants.API_MEMBERSHIP),175                                   None, None, "full")176        UfmPkeysManagement.set_pkey(pkey, guids, memberships=memberships,177                                       default_membership=membership,index0=bool(index0),178                                       ip_over_ib=bool(ip_over_ib))179    elif args_dict.get(UfmPkeysConstants.PKEYS_OPERATIONS.get("get_pkeys")):180        UfmPkeysManagement.get_pkeys()181    elif args_dict.get(UfmPkeysConstants.PKEYS_OPERATIONS.get("get_pkey")):182        pkey = args_dict.get(UfmPkeysConstants.PKEYS_OPERATIONS.get("get_pkey"))183        UfmPkeysManagement.get_pkeys(pkey)184    elif args_dict.get(UfmPkeysConstants.PKEYS_OPERATIONS.get("delete_pkey")):185        pkey = args_dict.get(UfmPkeysConstants.PKEYS_OPERATIONS.get("delete_pkey"))186        UfmPkeysManagement.delete_pkey(pkey)187    else:188        message = "You must provide one of the following operations: "+ \189                  ''.join(['--{0} | '.format(item) for key,item in UfmPkeysConstants.PKEYS_OPERATIONS.items()])...

engine.py

Source:engine.py

...50        çækey51        """52        return 'FIRST|MEMCACHE' + "|" + cls.__module__ + "." + cls.__name__ + '|' + str(pkey)    53    54    def get_pkey(self):55        """56        æ ¹æ®åç±»å¯¹è±¡è·åkey57        """58        return self.__class__.generate_cache_key(str(self.uid))59        60    @classmethod61    def get(cls, pkey):62        """63        cache_key:cache key64        pkey:uid å¯ä¸æ ç¤º65        """66        data = None67        level = 068        cache_key = cls.generate_cache_key(pkey)69        # ä»åå¨å±è·ådumpsåçå¯¹è±¡æ°æ®  å¯¹åºè¿éç½®æä»¶ç ç°å¨å°±éç½®äº memcached  è¿å¯ä»¥æå¶ä»ç mysql redisçç70        for engine_name in engins:71            #æ ¹æ®éç½®çå¼æ è·åä¸ä¸ªå¯¹è±¡ æ¯å¦memcacheçå®ä¾72            engine_obj = app[engine_name]73            level += 174            data = engine_obj.get_data(cls, cache_key)75            #å¦æåå°äº å°±ä¸å éå76            if data is not None:77                break78        #å¦æå®å¨åä¸å°å°±æ¯æ²¡ææ°æ®79        if data is None:80            return None81        82        # è·åå°dumpsæ°æ®ï¼è½¬æ¢æä¸ºå¯¹è±¡å®ä¾  è¿æ¶åå¯¹è±¡æ¯ åå¸ç±»åç 83        #å¦æä»åºå±æ°æ®åºä¸è¯»ååºæ¥ç  åç¼åå°æ´ä¸å±cacheä¸84        obj = pickle.loads(data)85        if level > 1:86            top_engine_obj = app[engins.keys()[0]]87            top_engine_obj.put_data(cls, cache_key, data, False)88            89        return obj90            91    def put(self):92        cls = self.__class__93        data = pickle.dumps(self)94        pkey = self.get_pkey()95        print pkey96        for engine_name in engins:97            engine_obj = app[engine_name]98            engine_obj.put_data(cls, pkey, data, False)99    100    def put_only_bottom(self):101        """ç´æ¥åå°åå¨çæåºä¸å±102        """103        cls = self.__class__104        data = self.dumps()105        pkey = self.get_pkey()106        engine_obj = app[engins.keys()[-1]]107        engine_obj.put_data(cls, pkey, data, self.need_insert)108    def do_delete(self):109        cls = self.__class__110        pkey = self.get_pkey()111        for engine_name in engins:112            engine_obj = app[engine_name]...

rsa_rw.py

Source:rsa_rw.py

...13def mulinv(b, n):14    g, x, _ = egcd(b, n)15    if g == 1:16        return x % n17def get_pkey(file):18	f = open(file,'r')19	buff = f.read()20	f.close()21	return RSA.importKey(buff)22pkey = get_pkey('oscar.manas_pubkeyRSA_RW.pem')23n = pkey.n24e = pkey.e25print 'n:', pkey.n26print 'e:', pkey.e27folder = 'keys/'28for file in os.listdir(folder):29	pkey2 = get_pkey(folder + file)30	n2 = pkey2.n31	p = gcd(n,n2)32	if p > 1:33		print file34		break35q = n / p36phi = (p-1)*(q-1)37d = mulinv(e,phi)38print 'd:', d39privkey = RSA.construct((n,e,d))40f = open('oscar.manas_privkeyRSA_RW.pem','w')41f.write(privkey.exportKey())42f.close()43# openssl rsautl -decrypt -in oscar.manas_RSA_RW.enc -out oscar.manas_RSA_RW.dec -inkey oscar.manas_privkeyRSA_RW.pem

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.