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

How to use nested_task method in locust

Best Python code snippet using locust

utils.py

Source:utils.py

1# Copyright 2020 VMware, Inc.  All rights reserved. -- VMware Confidential  #2__author__ = 'jradhakrishna'3import json4import time5import requests6import urllib37import getpass8urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)9import re10class Utils:11    def __init__(self, args):12        self.hostname = args[0]13        self.username = args[1]14        self.password = args[2]15        self.header = {'Content-Type': 'application/json'}16        self.token_url = 'https://'+ self.hostname +'/v1/tokens'17        self.get_token()18    19    def get_token(self):20        payload = {"username": self.username,"password": self.password}21        response = self.post_request(payload=payload,url=self.token_url)22        token = response['accessToken']23        self.header['Authorization'] = 'Bearer ' + token24    25    def get_request(self,url):26        self.get_token()27        time.sleep(5)28        response = requests.get(url, headers=self.header,verify=False)29        if(response.status_code == 200 or response.status_code == 202):30            data = json.loads(response.text)31        else:32            print ("Error reaching the server.")33            exit(1)34        return data35    def post_request(self,payload,url):36        response = requests.post(url, headers=self.header, json=payload,verify=False)37        if(response.status_code == 200 or response.status_code == 202):38            data = json.loads(response.text)39            return data40        else:41            print ("Error reaching the server.")42            print (response.text)43            exit(1)44    def post_request_raw(self,payload,url):45        response = requests.post(url, headers=self.header, json=payload,verify=False)46        if(response.status_code in [200, 202]):47            return response48        else:49            print ("Error reaching the server.")50            print (response.text)51            exit(1)52    53    def patch_request(self,payload,url):54        response = requests.patch(url, headers=self.header, json=payload,verify=False)55        if(response.status_code == 202):56            data = json.loads(response.text)57            return data58        elif(response.status_code == 200):59            return60        else:61            print ("Error reaching the server from patch.")62            print (response.text)63            exit(1)64    def get_poll_request(self, url, expected_status):65        response = self.get_request(url)66        while response['status'] in ['In Progress', 'IN_PROGRESS', 'Pending']:67            time.sleep(5)68            response = self.get_request(url)69        if response['status'] == expected_status:70            return response71        else:72            print('\033[91m Operation failed \033[00m', end='\n\n')73            exit(1)74    def poll_on_id(self,url,task):75        key = ''76        if(task):77            key = 'status'78        else:79            key = 'executionStatus'80        status = self.get_request(url)[key]81        while(status in ['In Progress','IN_PROGRESS','Pending']):82            response = self.get_request(url)83            status = response[key]84            time.sleep(10)85        if(task):86            return status87        if(status == 'COMPLETED'):88            return response['resultStatus']89        else:90            print ('Operation failed')91            exit(1)92    def poll_on_queries(self,url):93        response = self.get_request(url)94        status = response['queryInfo']['status']95        while(status in ['In Progress','IN_PROGRESS','Pending']):96            response = self.get_request(url)97            status = response['queryInfo']['status']98            time.sleep(10)99        if(status == 'COMPLETED'):100            return response['result']101        else:102            print ('Operation failed')103            exit(1)104    105    def delete_request(self,payload,url):106        response = requests.delete(url,json=payload,headers=self.header,verify=False)107        if(response.status_code == 202):108            data = json.loads(response.text)109            return data110        else:111            print ("Error reaching the server.")112            print (response.text)113            exit(1)114    115    def read_input(self, file):116        with open(file) as json_file:117            data = json.load(json_file)118        return data119    def print_validation_errors(self, url):120        validation_response = self.get_request(url)121        if "validationChecks" in validation_response:122            failed_tasks = list(123                filter(lambda x: x["resultStatus"] == "FAILED", validation_response["validationChecks"]))124            for failed_task in failed_tasks:125                self.printRed(failed_task['description'] + ' ' + 'failed')126                if "errorResponse" in failed_task and "message" in failed_task["errorResponse"]:127                    self.printRed(failed_task["errorResponse"]["message"])128                if "nestedValidationChecks" in failed_task:129                    for nested_task in failed_task["nestedValidationChecks"]:130                        if "errorResponse" in nested_task and "message" in nested_task["errorResponse"]:131                            self.printRed(nested_task["errorResponse"]["message"])132    def password_check(self, pwd, cannotbe = None):133        #rule: minlen = 8, maxlen = 32, at least 1 number, 1 upper, 1 lower, 1 special char134        minlen = 8135        maxlen = 32136        hasnum = True137        hasupper = True138        haslower =True139        hasspecial = True140        res = True141        if cannotbe is not None and pwd == cannotbe:142            self.print_error ("Password cannot be same as you have inputed")143            return False144        if len(pwd) < minlen:145            self.print_error ("Length should be at least {}".format(minlen))146            res = False147        elif len(pwd) > maxlen:148            self.print_error ("Length should be less than {}".format(maxlen))149            res = False150        if hasnum and not any(c.isdigit() for c in pwd):151            self.print_error ("Password should have at least one number")152            res = False153        if hasupper and not any(c.isupper() for c in pwd):154            self.print_error ("Password should have at least one upper case letter")155            res = False156        if haslower and not any(c.islower() for c in pwd):157            self.print_error ("Password should have at least one lower case letter")158            res = False159        if hasspecial and len(re.compile('[0-9 a-z A-Z]').sub('', pwd)) == 0:160            self.print_error ("Password should contain at least one special character")161            res = False162        return res163    def print_error(self, msg):164        RED = '\033[1;31m'165        TAIL = '\033[0m'166        head = RED + "Error:" + TAIL167        print("{}{}".format(head, msg))168    def valid_input(self, inputinfo, defaultvalue = None, validfunc = None, ext_args = None, is_password = False):169        while(True):170            if is_password:171                inputstr = getpass.getpass(inputinfo)172            else:173                inputstr = input(inputinfo)174            if len(str(inputstr).strip()) == 0 and defaultvalue is not None:175                return defaultvalue176            if validfunc is not None:177                checkresult = validfunc(inputstr) if ext_args is None else validfunc(inputstr, ext_args)178                if checkresult:179                    return inputstr180                else:181                    self.printRed('Unable to validate the input')182            else:183                break184        return inputstr185    def printRed(self, message):186        print("\033[91m {}\033[00m".format(message))187    def printGreen(self, message):188        print("\033[92m {}\033[00m".format(message))189    def printYellow(self, message):190        print("\033[93m {}\033[00m".format(message))191    def printCyan(self, message):192        print("\033[96m {}\033[00m".format(message))193    def printBold(self, message):194        print("\033[95m {}\033[00m".format(message))195    def valid_pwd_match(self, inputstr, ext_args):196        res = inputstr == ext_args197        if not res:198            self.print_error("Password doesn't match")...

test_task.py

Source:test_task.py

...123    @data(None, "xp1", Path("xp1"))124    @dir_with_env("xp1", symlink=True)125    def test_decorator_nested_current(self, env_specifier, dir_with_env, env):126        @enostask()127        def nested_task(env=None):128            self.assertEqual(dir_with_env.joinpath("xp1"), env.env_name)129            self.assertEqual("bar", env["foo"])130        @enostask()131        def top_task(env=None):132            env["foo"] = "bar"133            nested_task(env=env)134        top_task(env=env_specifier)135    # test with absolute path136    @data(True, False)137    def test_decorator_nested_current_absolute_path(self, symlink):138        @enostask()139        def nested_task(env=None):140            self.assertEqual(Path("xp1").resolve(), env.env_name)141            self.assertEqual("bar", env["foo"])142        @enostask()143        def top_task(env=None):144            env["foo"] = "bar"145            nested_task(env=env)146        with dummy_env("xp1", symlink=symlink) as _:147            # abs path here148            top_task(env=Path("xp1").resolve())149    @data("xp1", Path("xp1"))150    @dir_with_env("xp1", symlink=False)151    def test_decorator_nested_nocurrent(self, env_specifier, dir_with_env, env):152        @enostask()153        def nested_task(env=None):154            self.assertEqual(dir_with_env.joinpath("xp1"), env.env_name)155            self.assertEqual("bar", env["foo"])156        @enostask()157        def top_task(env=None):158            env["foo"] = "bar"159            nested_task(env=env)160        # specifying an env as a string (relative path)...

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.