Best Python code snippet using autotest_python
base_job.py
Source:base_job.py
...134 self._read_from_backing_file()135 try:136 return method(self, *args, **dargs)137 finally:138 self._write_to_backing_file()139 wrapped_method.__name__ = method.__name__140 wrapped_method.__doc__ = method.__doc__141 return wrapped_method142class job_state(object):143 """A class for managing explicit job and user state, optionally persistent.144 The class allows you to save state by name (like a dictionary). Any state145 stored in this class should be picklable and deep copyable. While this is146 not enforced it is recommended that only valid python identifiers be used147 as names. Additionally, the namespace 'stateful_property' is used for148 storing the valued associated with properties constructed using the149 property_factory method.150 """151 NO_DEFAULT = object()152 PICKLE_PROTOCOL = 2 # highest protocol available in python 2.4153 def __init__(self):154 """Initialize the job state."""155 self._state = {}156 self._backing_file = None157 self._backing_file_initialized = False158 self._backing_file_lock = None159 def _lock_backing_file(self):160 """Acquire a lock on the backing file."""161 if self._backing_file:162 self._backing_file_lock = open(self._backing_file, 'a')163 fcntl.flock(self._backing_file_lock, fcntl.LOCK_EX)164 def _unlock_backing_file(self):165 """Release a lock on the backing file."""166 if self._backing_file_lock:167 fcntl.flock(self._backing_file_lock, fcntl.LOCK_UN)168 self._backing_file_lock.close()169 self._backing_file_lock = None170 def read_from_file(self, file_path, merge=True):171 """Read in any state from the file at file_path.172 When merge=True, any state specified only in-memory will be preserved.173 Any state specified on-disk will be set in-memory, even if an in-memory174 setting already exists.175 @param file_path: The path where the state should be read from. It must176 exist but it can be empty.177 @param merge: If true, merge the on-disk state with the in-memory178 state. If false, replace the in-memory state with the on-disk179 state.180 @warning: This method is intentionally concurrency-unsafe. It makes no181 attempt to control concurrent access to the file at file_path.182 """183 # we can assume that the file exists184 if os.path.getsize(file_path) == 0:185 on_disk_state = {}186 else:187 on_disk_state = pickle.load(open(file_path))188 if merge:189 # merge the on-disk state with the in-memory state190 for namespace, namespace_dict in on_disk_state.iteritems():191 in_memory_namespace = self._state.setdefault(namespace, {})192 for name, value in namespace_dict.iteritems():193 if name in in_memory_namespace:194 if in_memory_namespace[name] != value:195 logging.info('Persistent value of %s.%s from %s '196 'overridding existing in-memory '197 'value', namespace, name, file_path)198 in_memory_namespace[name] = value199 else:200 logging.debug('Value of %s.%s is unchanged, '201 'skipping import', namespace, name)202 else:203 logging.debug('Importing %s.%s from state file %s',204 namespace, name, file_path)205 in_memory_namespace[name] = value206 else:207 # just replace the in-memory state with the on-disk state208 self._state = on_disk_state209 # lock the backing file before we refresh it210 with_backing_lock(self.__class__._write_to_backing_file)(self)211 def write_to_file(self, file_path):212 """Write out the current state to the given path.213 @param file_path: The path where the state should be written out to.214 Must be writable.215 @warning: This method is intentionally concurrency-unsafe. It makes no216 attempt to control concurrent access to the file at file_path.217 """218 outfile = open(file_path, 'w')219 try:220 pickle.dump(self._state, outfile, self.PICKLE_PROTOCOL)221 finally:222 outfile.close()223 def _read_from_backing_file(self):224 """Refresh the current state from the backing file.225 If the backing file has never been read before (indicated by checking226 self._backing_file_initialized) it will merge the file with the227 in-memory state, rather than overwriting it.228 """229 if self._backing_file:230 merge_backing_file = not self._backing_file_initialized231 self.read_from_file(self._backing_file, merge=merge_backing_file)232 self._backing_file_initialized = True233 def _write_to_backing_file(self):234 """Flush the current state to the backing file."""235 if self._backing_file:236 self.write_to_file(self._backing_file)237 @with_backing_file238 def _synchronize_backing_file(self):239 """Synchronizes the contents of the in-memory and on-disk state."""240 # state is implicitly synchronized in _with_backing_file methods241 pass242 def set_backing_file(self, file_path):243 """Change the path used as the backing file for the persistent state.244 When a new backing file is specified if a file already exists then245 its contents will be added into the current state, with conflicts246 between the file and memory being resolved in favor of the file247 contents. The file will then be kept in sync with the (combined)...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!