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

How to use _read_from_backing_file method in autotest

Best Python code snippet using autotest_python

base_job.py

Source:base_job.py

...130    state is consistently maintained.131    """132    @with_backing_lock133    def wrapped_method(self, *args, **dargs):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_file...

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.