Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use generate_sas_token method in lisa

Best Python code snippet using lisa_python

getblob.py

Source:getblob.py

...72                    try:73                        if astr == '' and len(ext) == 1:74                            astr = blob.name75                            fobj['file_name'] = blob.name76                            # sasFileUrl = self.generate_sas_token(fobj)77                           78                       79                        # elif (blob.name.find(ext[0]+'.output.json') > 0) and len(ext) > 1 :80                        elif (blob.name.find(ext[0]+'.output.json') > 0) and len(ext) > 1 :81                            # ocr_str = burl+blob.name 82                            # create output file sas token83                            fobj['file_name'] = blob.name                            84                            output_str = self.generate_sas_token(fobj)85                            # create original file sas token86                            fobj['file_name']=astr87                            sasFileUrl = self.generate_sas_token(fobj)88                            pobj = {"container":container_name ,"file_name":astr, "file":sasFileUrl,"ocr":ocr_str, "output":output_str ,"label":lbl_str, "ocr_dt":'', "label_dt":'',"category":'Invoice',"invoice_date":datetime.utcnow(), "status":status}89                            # astr = blob.name90                            lbl_str=''91                            ocr_str=''92                            output_str=''93                           94                            ar_lst.append(pobj)95                           96                            # pobj = {"container":container_name ,"file_name":astr, "file":sasFileUrl,"ocr":ocr_str, "output":output_str ,"label":lbl_str, "ocr_dt":'', "label_dt":'',"category":'Invoice',"invoice_date":datetime.utcnow(), "status":status}97                                                        98                            # ar_lst.append(pobj)99                                                            100                        # elif output_str != '' and astr != blob.name and len(ext) == 1:101                        elif astr !='' and astr != blob.name and len(ext) == 1:102                                                   103                            # fobj['file_name']=astr104                            # sasFileUrl = self.generate_sas_token(fobj)105                            # pobj = {"container":container_name ,"file_name":astr, "file":sasFileUrl,"ocr":ocr_str, "output":output_str ,"label":lbl_str, "ocr_dt":'', "label_dt":'',"category":'Invoice',"invoice_date":datetime.utcnow(), "status":status}106                            astr = blob.name107                            lbl_str=''108                            ocr_str=''109                            output_str=''110                            111                            # ar_lst.append(pobj)112                                    113                    except Exception as ex:114                        print(ex)115                        continue116           # print(ar_lst)    117        except Exception as ex:118            print("exception :")119            print(ex)120        return JsonResponse({"result": ar_lst})121    def get(self, request):122        return JsonResponse({"key": "Hey"})123       124        #return render(request, 'view_a.html')125    126    def generate_sas_token(self, fobj):127        # create zulu timestamp128        # print(dt_now_utc)129        exdt = (dt_now_utc.isoformat()).split('.')[0]130        # print(exdt)131        sas = generate_blob_sas(account_name=AZURE_ACC_NAME,132                            account_key=AZURE_PRIMARY_KEY,133                            container_name=fobj['container'],134                            blob_name=fobj['file_name'],135                            permission=BlobSasPermissions(read=True),136                            expiry= exdt+'Z')137    138        sas_url ='https://'+AZURE_ACC_NAME+'.blob.core.windows.net/'+fobj['container']+'/'+urllib.parse.quote(fobj['file_name'])+'?'+sas139        140        # sas_url = urllib.parse.quote(sas_url)141        # print(sas_url)142        return sas_url143    144    def getDocDetails(self, dobj):145        146        # print(dobj['output'])147        ocr_dt=''148        # code for read json file149        # if dobj['container'] != 'manual':150        #     ocr_dt = pd.read_json(dobj['ocr'])151        #     dobj['ocr_dt'] = ocr_dt.to_json()152        #     label_dt = pd.read_json(dobj['label'])153        #     dobj['label_dt'] = label_dt.to_json()154        # else:     155       156        # for ubnatu server , uncomment for deployment157        ocr_dt = pd.read_json(dobj['output'])158        # for local windows encoding159        #ocr_dt = pd.read_json(dobj['output'])160        head_val = ocr_dt['headers']161        for hvl in head_val:162            print(isinstance(hvl, dict))163            if isinstance(hvl, dict) == True:164               hvl['text'] = ftfy.fix_encoding(hvl['text'])165                166               167        hconfig = self.getHeaderConfig()168        # convert to dictionary169        hconf_obj = hconfig.to_dict(orient='record')170        table = ocr_dt['tables']171        for pno in table['page']:172            173            for pg in table['page'][pno]:174                for ind, tdt in enumerate(table['page'][pno][pg]):175                    176                    if ind == 0:               177                        for ix, thr in enumerate(tdt):178                             179                            # match text in config ui translation180                            if str(isinstance(thr, str)) == 'True' :181                                tdt[ix] = {	182                                    "columnIndex": ix, "elements": '' ,183                                    "rowIndex": ind,184                                    "text": thr,185                                    "header": 'small',186                                    "ui_label":''187                                }                                188                            elif str(isinstance(thr, str)) == 'False' and thr['text'] != '':189                              190                                hcl = list(filter(lambda x: x['JSON Lable(order tree)'] == thr['text'], hconf_obj))191                                192                                thr['header'] =  'small'193                                thr['ui_label'] = ''194                                if len(hcl) != 0:195                                    thr['header'] =  hcl[0]['Text Size']196                                    thr['ui_label'] = hcl[0]['UI Lable']197                    elif tdt != "":198                        199                        for jv, tvl in enumerate(tdt):200                            201                            if tvl != "":202                                tvl['text'] = ftfy.fix_encoding(tvl['text'])203                               204                                205        # print(table)206        dobj['output_dt'] = ocr_dt.to_json()207        # dobj['output_dt'] = ftfy.fix_encoding(dobj['output_dt'])208        # print(dobj['output_dt'])209        return dobj210    def getHeadsize(self, vl, th):211        212        return vl213    214    def getHeaderConfig(self):215        216        # generate sas url217        fobj = {}218        fobj['container'] = constant.CONFIG_CONTAINER219        fobj['file_name'] = constant.CONFIG_FILE_NAME220        221        config_sas_url = self.generate_sas_token(fobj)222        223        head_config = pd.read_excel(config_sas_url,usecols=['Table headers(Unique element)', 'Translation', 'JSON Lable(order tree)', 'UI Lable','Text Size'], engine='openpyxl')224        # print(head_config.columns.ravel())225        # print(head_config['Translation'])226        return head_config227def view_b(request):...

SASgenerator-DPS.py

Source:SASgenerator-DPS.py

...9from hashlib import sha25610from time import time11from urllib.parse import quote_plus, urlencode12from hmac import HMAC13def generate_sas_token(uri, key, policy_name, expiry=3600):14    # Time to live is set such that we can specify a duration (expiry) in [s] since15    # the token creation. Then, the UNIX timestamp of the expiry date will be created.16    ttl = time() + expiry17    sign_key = "%s\n%d" % ((quote_plus(uri)), int(ttl))18    signature = b64encode(19        HMAC(b64decode(key), sign_key.encode('utf-8'), sha256).digest())20    rawtoken = {21        'sr':  uri,22        'sig': signature,23        'se': str(int(ttl))24    }25    if policy_name is not None:26        rawtoken['skn'] = policy_name27    return 'SharedAccessSignature ' + urlencode(rawtoken)28def generate_device_key(registrationId, masterKey):29    '''Creates a device derived key by hashing the device unique registrationId with the DPS key.'''30    deviceKey = b64encode(HMAC(b64decode(masterKey),31                               registrationId.encode('utf-8'),32                               sha256)33                          .digest())34    return deviceKey.decode('UTF-8')35if __name__ == "__main__":36    # ------------- SELECT MODE -------------37    select = 238    '''39    Select 1: create a SAS token for a device in an Enrollment Group40    Select 2: create a SAS token for a device in an Individual Enrollment41    '''42    if select == 1:43        # Device derived key generation for DPS Enrollment Groups44        # ------------- INSERT DATA -------------45        dps_scopeId = '***'46        device_registration_id = '***'47        dps_key = '***'48        # ----------- INSERT OPTIONS ------------49        # Expiry time in [s] from token creation50        expiry = 365*12*30*24*60*60  # [s]51        # ---------------------------------------52        53        # Default policy for device registration54        policy = 'registration'55        # uri: [dps_scope_id]/registrations/[device_registration_id]56        uri = dps_scopeId + '/registrations/' + device_registration_id57        58        # Device derived key generation59        device_derived_key = generate_device_key(device_registration_id, dps_key)60        print('Device "' + device_registration_id + '" derived key: ' + device_derived_key)61        # Device SAS token generation62        device_sas_token = generate_sas_token(uri, device_derived_key, policy, expiry)63        print('Device SAS token: ' + device_sas_token)64    if select == 2:65        # Individual Enrollment SAS token generation66        # ------------- INSERT DATA -------------67        dps_scopeId = '***'68        device_registration_id = '***' # Coincides with Individual Enrollment ID69        dps_key = '***'70        # ----------- INSERT OPTIONS ------------71        # Expiry time in [s] from token creation72        expiry = 365*12*30*24*60*6073        # ---------------------------------------74        # uri: [dps_scope_id]/registrations/[device_registration_id]75        uri = dps_scopeId + '/registrations/' + device_registration_id76        # Default policy for device registration77        policy = 'registration'78        device_sas_token = generate_sas_token(uri, dps_key, policy, expiry)...

dps_sas_token_generation.py

Source:dps_sas_token_generation.py

2from hashlib import sha2563from time import time4from urllib import parse5from hmac import HMAC6def generate_sas_token(uri, key, policy_name, expiry=3600):7    ttl = time() + expiry8    sign_key = "%s\n%d" % ((parse.quote_plus(uri)), int(ttl))9    print(sign_key)10    signature = b64encode(HMAC(b64decode(key), sign_key.encode('utf-8'), sha256).digest())11    rawtoken = {12        'sr' : uri,13        'sig': signature,14        'se' : str(int(ttl))15    }16    if policy_name is not None:17        rawtoken['skn'] = policy_name18    print ('SharedAccessSignature ' + parse.urlencode(rawtoken))19    return 'SharedAccessSignature ' + parse.urlencode(rawtoken)20#Fill Scope ID, Registration ID and Provisioning SAS Key below to generate SAS token for DPS application.21#SAS token expiry time will be 7200 minutes.Exapmle - generate_sas_token("0ne... 2F/registrations/azurea...regID1", "RfD ... ==" , None, expiry=7200)...

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.