Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use case_handler method in locust
Best Python code snippet using locust
asset_model_exporter.py
Source:asset_model_exporter.py 
...39# - a case_handler function40# = a shape_filter dictionary to filter the describe response to match the CFN shape41def walk_dict_filter(dictionary, case_handler, shape_filter):42 if isinstance(dictionary, dict):43 return {k[0].upper() + k[1:]: walk_dict_filter(case_handler(k,v), case_handler, shape_filter) for k, v in dictionary.items() if k[0].upper() + k[1:] in shape_filter}44 elif isinstance(dictionary, list):45 return [walk_dict_filter(item, case_handler, shape_filter) for item in dictionary]46 else:47 return dictionary48#Recursive function to do a second pass walk of the CFN shape created in the first pass. This is to replace IDs with LogicalIDs49def walk_dict(dictionary, case_handler, **kwargs):50 if isinstance(dictionary, dict):51 return {k: walk_dict(case_handler(k,v, **kwargs), case_handler, **kwargs) for k, v in dictionary.items()}52 elif isinstance(dictionary, list):53 return [walk_dict(item, case_handler, **kwargs) for item in dictionary]54 else:55 return dictionary56#Function to handle special cases on the first pass where the api response does not match the CFN shape57def case_handler_1st_pass(k,v):58 if k == 'type' and isinstance(v, dict):59 if 'measurement' in v:60 return {'TypeName': 'Measurement'}61 if 'transform' in v:62 return {'TypeName': 'Transform', 'Transform':v['transform']}63 if 'attribute' in v:64 return {'TypeName': 'Attribute', 'Attribute':v['attribute']}65 if 'metric' in v:...
logger.py
Source:logger.py
1import logging2import logging.handlers3import os4import zipfile5import time6logger_level = {7 "INFO": logging.INFO,8 "WARNING": logging.WARNING,9 "DEBUG": logging.DEBUG,10 "ERROR": logging.ERROR,11 "CRITICAL": logging.CRITICAL12}13def _zip_directory(dirpath, output_file):14 """15 å°ç®æ æ件夹å缩æzipå
并ä¸è¾åºå°output_file16 并ä¸å°æ件å é¤17 """18 zip = zipfile.ZipFile(output_file, "w", zipfile.ZIP_DEFLATED)19 for path, dirnames, filenames in os.walk(dirpath, topdown=False):20 target_path = path.replace(dirpath, '')21 for filename in filenames:22 if os.path.join(path, filename) == output_file:23 # skip self24 continue25 zip.write(os.path.join(path, filename), os.path.join(target_path, filename))26 os.remove(os.path.join(path, filename))27 if target_path != '':28 os.rmdir(path)29 zip.close()30def _check_and_create_directory(filename):31 dir_name = os.path.dirname(filename)32 if dir_name == '':33 return34 if not os.path.exists(dir_name):35 os.makedirs(dir_name)36class LoggerManager:37 def __init__(self):38 # ç¨äºè®°å½loggerçé
置信æ¯39 self.logger_info = dict()40 def register(self, logger_name, filename=None, console=True,41 default_level=logging.INFO, **kwargs):42 """43 注ålogger44 """45 log_format = kwargs.get("format", None)46 zip_logger = kwargs.get("zip", False)47 # å¦æ设置äºfile countï¼åé»è®¤ä¸ä¸ªæ件大å°ä¸º1M48 file_size_limit = kwargs.get("size_limit", 1024*1024)49 max_files = kwargs.get("max_files", None)50 file_mode = kwargs.get("mode", "w")51 if log_format is None:52 log_format = "[%(asctime)s][%(name)s]-<thread:%(thread)s>-(line:%(lineno)s), [%(levelname)s]: %(message)s"53 # æ¥å¿å¯¹æµè¯ç¨ä¾å¼æ¾54 for_test = kwargs.get("for_test", False)55 is_test = kwargs.get("is_test", False)56 # è·åæ°çlogger å®ä¾57 logger = logging.getLogger(logger_name)58 self.logger_info[logger_name] = dict()59 self.logger_info[logger_name]['timestamp'] = time.localtime()60 self.logger_info[logger_name]['for_test'] = for_test61 self.logger_info[logger_name]['is_test'] = is_test62 if filename:63 _check_and_create_directory(filename)64 self.logger_info[logger_name]['file_path'] = os.path.dirname(filename)65 self.logger_info[logger_name]['file_name'] = os.path.basename(filename)66 self.logger_info[logger_name]['zip'] = zip_logger67 if max_files:68 file_handler = \69 logging.handlers.RotatingFileHandler(70 filename=filename,71 mode=file_mode,72 maxBytes=file_size_limit,73 backupCount=max_files)74 else:75 file_handler = logging.FileHandler(filename, mode=file_mode)76 file_handler.setFormatter(logging.Formatter(fmt=log_format))77 logger.addHandler(file_handler)78 if is_test:79 for llogger, lvalue in self.logger_info.items():80 # éè¦æ¯ä¸ä¸ªæ³¨åäºæ¥å¿æ件ç模åæè½åæµè¯ç¨ä¾è¾åºæ¥å¿81 if lvalue['for_test'] and "file_name" in lvalue:82 logger_filename = os.path.join(83 os.path.dirname(filename), f"{llogger}.log")84 case_handler = logging.FileHandler(logger_filename, mode="w")85 case_handler.setFormatter(logging.Formatter(fmt=log_format))86 lvalue['case_handler'] = case_handler87 lvalue['logger'].addHandler(case_handler)88 if console:89 stream_handler = logging.StreamHandler()90 stream_handler.setFormatter(logging.Formatter(fmt=log_format))91 logger.addHandler(stream_handler)92 logger.setLevel(default_level)93 self.logger_info[logger_name]['logger'] = logger94 return logger95 def unregister(self, logger_name):96 """97 å é¤æ³¨åçloggerï¼åæ¶å°éè¦æå
çloggeræ件æå
98 """99 if logger_name in logging.Logger.manager.loggerDict:100 logging.Logger.manager.loggerDict.pop(logger_name)101 # å¦æ注åäºæµè¯ç¨ä¾çhandlerï¼å移é¤102 if self.logger_info[logger_name]['is_test']:103 for llogger, lvalue in self.logger_info.items():104 if 'case_handler' in lvalue:105 lvalue['logger'].removeHandler(lvalue['case_handler'])106 lvalue.pop("case_handler")107 self._achieve_files(logger_name)108 self.logger_info.pop(logger_name)109 def _achieve_files(self, logger_name):110 if self.logger_info[logger_name]['zip']:111 current = time.localtime()112 output_file = \113 "achieved_logs_%d_%d_%d_%d_%d_%d.zip" % (114 current.tm_year, current.tm_mon, current.tm_mday,115 current.tm_hour, current.tm_min, current.tm_sec116 )117 _zip_directory(118 self.logger_info[logger_name]['file_path'],119 os.path.join(self.logger_info[logger_name]['file_path'], output_file))120 def get_logger(self, logger_name):121 if logger_name in self.logger_info:122 return self.logger_info[logger_name]["logger"]123 raise NameError(f"No log named {logger_name}")124logger = LoggerManager()125if __name__=="__main__":126 log1 = logger.register(r"Module", "./logtest/test1.log", for_test=True)127 case_logger = logger.register("Test_Case", "./testcase/democase.log", is_test=True)128 case_logger.info("This is from Test log")129 log1.info("This is from Module Log")...
try_except_finally.py
Source:try_except_finally.py
1from LearningPy.concepts.playing_with_exceptions.Notification import Notification2def test_rail_replica():3 case_handler = None4 try:5 case_handler = Notification("Chirag is here!!")6 raise7 except Exception:8 print("\n\nUPDATE TEST FAILED")9 raise10 finally:11 print("Creating Case Processor Dumps...")12 case_handler.get_message()13 print("UPDATE TEST COMPLETE")14if __name__ == '__main__':...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run locust automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
