How to use _record_attributes method in autotest

Best Python code snippet using autotest_python

_cli_methods.py

Source:_cli_methods.py

...232        """233        self.__update_attributes(method)234        return self.__wrap_method(method)235236    def _record_attributes(self, **attributes):237        """238        Record attributes to be passed when wrapping the next method239        """240        for key in attributes:241            self.__attributes[key] = attributes[key]242243    def __update_attributes(self, method):244        """245        Passes all the recorded attributes to the wrapped method246        """247        for key in self.__attributes:248            self._methods_dict[method.__name__].attributes[key] = self.__attributes[key]249        250        self.__attributes = {}251252    def __wrap_method(self, method):253        """254        records the given method in the dictionary and redirects it to the compiled method255        """256        self._record_method(method)257258        @cli_parser.copy_argspec(method)259        def redirected(*args, **kwargs):260            linked_methods = self._methods_dict.compiled(args[0])261            if method.__name__ not in linked_methods:262                raise cli_exceptions.InitializationException("Cannot access CLI operations and settings before an instance is constructed")263            return linked_methods[method.__name__](*(args[1:]), **kwargs)264    265        return redirected266267    268class OperationDecorator(MethodDecorator):269    """270    Wraps a method as an Operation271    """272    def _record_method(self, method):273        self._methods_dict[method.__name__].setExecution(method, "Operation")274275    def __call__(self): return super().__call__276277class SettingDecorator(MethodDecorator):278    """279    Wraps a method as a Setting280    """281    def _record_method(self, method):282        self._methods_dict[method.__name__].setExecution(method, "Setting")283284    def __call__(self, initial_value=None, updates_value=True):285        self._record_attributes(**{"initial_value":initial_value, "updates_value":updates_value})286        return super().__call__287288class DelegateDecorator(MethodDecorator):289    """290    Wraps a method as a Setting291    """292    def _record_method(self, method):293        self._methods_dict[method.__name__].setExecution(method, "Delegate")294295    def __call__(self, reuse=True):296        self._record_attributes(**{"reuse":reuse})297        return super().__call__298299class ValidationDecorator(MethodDecorator):300    """301    Wraps a method as a Validation for it's corresponding Operaion or Setting302    """303    def _record_method(self, method):304        self._methods_dict[method.__name__].addValidation(method)305306    def __call__(self): return super().__call__307
...

main.py

Source:main.py

1from __future__ import annotations2import datetime as dt3import csv4import logging5import configparser6from dataclasses import dataclass7from constants import * 8from generators import * 9from interfaces import *10from utils import Singleton, create_file_path11from storage import RecordRepository, ArrayStorage, MySQLStorage12class RecordFactory(RecordFactoryInterface):13	def __init__(self, config: dict) -> None:14		self._builder = RecordBuilder(config)15	@property16	def builder(self) -> RecordBuilder:17		return self._builder18	@builder.setter19	def builder(self, builder: RecordBuilder) -> None:20		self._builder = builder21	def create_history_record(self) -> None:22		self._builder.produce_id()23		self._builder.produce_side()24		self._builder.produce_instrument()25		self._builder.produce_status()26		self._builder.produce_pxinit()27		self._builder.produce_pxfill()28		self._builder.produce_volumeinit()29		self._builder.produce_volumefill()30		self._builder.produce_note()31		self._builder.produce_tags()32		self._builder.produce_date()33		record = self._builder.collect_record()34		return record35class RecordBuilder(RecordBuilderInterface):36	def __init__(self, config):37		self._total_record_counter = 138		self._record_attributes = {}39		self.record_model = RecordModel()40		self.id_obj = IdGenerator(*config["IDSettings"].values())41		self.side_obj = SideGenerator(*config["SideSettings"].values())42		self.instrument_obj = InstrumentGenerator(*config["InstrumentSettings"].values())43		self.status_obj = StatusGenerator(*config["StatusSettings"].values())44		self.pxinit_obj = PXInitGenerator()45		self.pxfill_obj = PXFillGenerator(*config["PXFillSettings"].values())46		self.volume_init_obj = VolumeInitGenerator(*config["VolumeInitSettings"].values())47		self.volume_fill_obj = VolumeFillGenerator(*config["VolumeFillSettings"].values())48		self.date_obj = DateGenerator(*config["DateSettings"].values())49		self.note_obj = NoteGenerator(*config["NoteSettings"].values())50		self.tag_obj = TagGenerator(*config["TagSettings"].values())51	def produce_id(self):52		if self._is_a_new_order_record("ID"):53			id_ = self.id_obj.generate_value()54			self._record_attributes["ID"] = id_55	def produce_side(self):56		if self._is_a_new_order_record("SIDE"):57			side = self.side_obj.generate_value()58			self._record_attributes["SIDE"] = side59	def produce_instrument(self):60		if self._is_a_new_order_record("INSTRUMENT"):61			instrument = self.instrument_obj.generate_value()62			self._record_attributes["INSTRUMENT"] = instrument63	def produce_status(self):64		if self._is_a_new_order_record("STATUS"):65			status_on_broker = self.status_obj.generate_value()66			self._record_attributes["STATUS"] = status_on_broker67	def produce_pxinit(self):68		if self._is_a_new_order_record("PX_INIT"):69			init_price = self.pxinit_obj.generate_value()70			self._record_attributes["PX_INIT"] = init_price71	def produce_pxfill(self):72		if self._is_a_new_order_record("PX_FILL"):73			fill_price = self.pxfill_obj.generate_value()74			self._record_attributes["PX_FILL"] = fill_price75	def produce_volumeinit(self):76		if self._is_a_new_order_record("VOLUME_INIT"):77			init_volume = self.volume_init_obj.generate_value()78			self._record_attributes["VOLUME_INIT"] = init_volume79	def produce_volumefill(self):80		if self._is_a_new_order_record("VOLUME_FILL"):81			fill_volume = self.volume_fill_obj.generate_value()82			self._record_attributes["VOLUME_FILL"] = fill_volume83	def produce_date(self):84		date = self.date_obj.generate_value()85		self._record_attributes["DATE"] = date86	def produce_note(self):87		if self._is_a_new_order_record("NOTE"):88			note = self.note_obj.generate_value()89			self._record_attributes["NOTE"] = note90	def produce_tags(self):91		if self._is_a_new_order_record("TAGS"):92			tags = self.tag_obj.generate_value()93			self._record_attributes["TAGS"] = tags94	def collect_record(self):95		self._total_record_counter += 196		self.record_model.record = self._record_attributes97		self.record_model.parameter_mapping()98		self.record_model.convert_history_record_to_order_record()99		record = RecordDTO(*self.record_model.record.values())100		return record101	def _is_a_new_order_record(self, attribute_name, record_counters={}):102		try:103			record_counters[attribute_name] += 1104		except KeyError as ex:105			record_counters[attribute_name] = 1106		number_of_records_for_order = self._define_number_of_records_for_order()107		if record_counters[attribute_name] == 1:108			return True109		elif record_counters[attribute_name] == number_of_records_for_order:110			record_counters[attribute_name] = 0111			return False112	def _define_number_of_records_for_order(self):113		if self._total_record_counter <= MAX_LIMIT_RECORDS_FOR_FIRST_SEGMENT:114			number_of_records = NUMBER_OF_RECORDS_FOR_FIRST_SEGMENT115			is_first_segment = True116		elif self._total_record_counter <= MAX_LIMIT_RECORDS_FOR_SECOND_SEGMENT:117			number_of_records = NUMBER_OF_RECORDS_FOR_SECOND_SEGMENT118			is_first_segment = False119		else:120			number_of_records = NUMBER_OF_RECORDS_FOR_THIRD_SEGMENT121			is_first_segment = False122		return number_of_records123@dataclass124class RecordDTO:125	id_: hex  = 0126	side: str = "NULL"127	instrument: str = "NULL"128	status: str = "NULL"129	px_init: float = 0.0130	px_fill: float = 0.0131	volume_init: int = 0132	volume_fill: int = 0133	note: str = "NULL"134	tags: str = "NULL"135	date: str = "NULL"136class RecordModel:137	def __init__(self):138		self._total_record_counter = 0139		self._record_number = -1140	@property141	def record(self):142		return self._record143	@record.setter144	def record(self, record: dict):145		self._record = record146	def parameter_mapping(self):147		mapped_record = {}148		for key in ORDER_ATTRIBUTES:149			try:150				mapped_record[key] = hex(self._record[key]) if key == "ID" else self._record[key]151			except KeyError as ex:152				mapped_record[key] = "NULL"153		self._record = mapped_record154	def convert_history_record_to_order_record(self):155		self._total_record_counter += 1156		self._record_number += 1157		number_of_records_for_order, is_first_segment = self._define_number_of_records_for_order()158		159		if is_first_segment:160			self._record["STATUS"] = self._record["STATUS"] if self._record_number == 1 else STATUSES[self._record_number+1]161		else:162			self._record["STATUS"] = self._record["STATUS"] if self._record_number == 2 else STATUSES[self._record_number]163		self._record["VOLUME_FILL"] = 0 if self._record["STATUS"] in STATUSES[:2] else self._record["VOLUME_FILL"]164		self._record["PX_FILL"] = 0 if self._record["STATUS"] in STATUSES[:2] else self._record["PX_FILL"]165		if self._record_number == number_of_records_for_order-1:166			self._record_number = -1167	def _define_number_of_records_for_order(self):168		if self._total_record_counter <= MAX_LIMIT_RECORDS_FOR_FIRST_SEGMENT:169			number_of_records = NUMBER_OF_RECORDS_FOR_FIRST_SEGMENT170			is_first_segment = True171		elif self._total_record_counter <= MAX_LIMIT_RECORDS_FOR_SECOND_SEGMENT:172			number_of_records = NUMBER_OF_RECORDS_FOR_SECOND_SEGMENT173			is_first_segment = False174		else:175			number_of_records = NUMBER_OF_RECORDS_FOR_THIRD_SEGMENT176			is_first_segment = False177		return number_of_records, is_first_segment178class Config(metaclass=Singleton):179	def __init__(self):180		self._path_to_config = 'settings/config.ini'181		self.parameters_set = {}182	def setup(self):183		config = configparser.ConfigParser()184		config.read(self._path_to_config)185		self.parameters_set = {}186		try:187			for section in config:188				if section != "DEFAULT":189					self.parameters_set[section] = {}190				for field in config[section]:191					if section == "LoggingSettings":192						self.parameters_set[section][field] = config[section][field].lower()193					elif section in ("DateSettings", "PXFillSettings", ):194						if field == "start_date":195							dt.datetime.strptime(config[section][field], DATE_FORMAT_FOR_DATE_ATTRIBUTE)196							self.parameters_set[section][field] = config[section][field]197						else:198							self.parameters_set[section][field] = float(config[section][field])199					elif section in ("Path", "MySQLSettings", ):200						self.parameters_set[section][field] = config[section][field]201					else:202						self.parameters_set[section][field] = int(config[section][field])203			if not self.parameters_set:204				raise FileNotFoundError("Config file not found")205		except (ValueError, FileNotFoundError, ) as ex:206			print("Incorrect parameters in the config file or the file is missing at all. " +207				f"Path: {path_to_config}. Ex: {ex}")208			os._exit(0)209		return self.parameters_set210def setup():211	config = Config()212	parameters_set = config.setup()213	logging_setup(214		parameters_set["Path"]["path_to_log"], 215		*parameters_set["LoggingSettings"].values(),216	)217	return parameters_set218def logging_setup(path_to_log, log_level, log_filemode):219	if not log_level in TRUE_LOG_LEVELS:220		log_level = 'DEBUG'221	if not log_filemode in TRUE_FILE_MODES:222		log_filemode = 'a'223	path = create_file_path(path_to_log)224	logging.basicConfig(filename=path, 225		level=log_level,226		filemode=log_filemode, 227		format=MESSAGE_FORMAT_FOR_LOGGER,228		datefmt=DATE_FORMAT_FOR_LOGGER)229def workflow(parameters_set):230	factory = RecordFactory(parameters_set)231	storage = ArrayStorage(parameters_set)232	repo = RecordRepository(storage)233	for i in range(7200):234		record = factory.create_history_record()235		repo.create(record)236	print(*repo.show_all())237if __name__ == "__main__":238	parameters_set = setup()239	logging.debug("Config and logger setup was successful. " + 240		f"Number of sections from config: {len(parameters_set.keys())}")241	workflow(parameters_set)...

attribute_manager.py

Source:attribute_manager.py

1from collections import defaultdict2from sentinels import NOTHING3class AttributeManager(object):4    def __init__(self, forge):5        super(AttributeManager, self).__init__()6        self.forge = forge7        self._record_attributes = defaultdict(dict)8        self._replay_attributes = defaultdict(dict)9    def set_attribute(self, mock, attr, value):10        d = self._replay_attributes if self.forge.is_replaying() else self._record_attributes11        d[mock.__forge__.id][attr] = value12    def get_attribute(self, mock, attr):13        returned = self._get_attribute(mock, attr, self._replay_attributes)14        if returned is NOTHING:15            returned = self._get_attribute(mock, attr, self._record_attributes)16        return returned17    def _get_attribute(self, mock, attr, attr_dict):18        return attr_dict[mock.__forge__.id].get(attr, NOTHING)19    def reset_replay_attributes(self):20        self._replay_attributes.clear()21    def has_attribute(self, mock, attr):...

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.