KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _aquire_lock method in autotest

Best Python code snippet using autotest_python

logger.py

Source:logger.py

...33        #     return logging.handlers.TimedRotatingFileHandler._open(self)34        with _lock:35            while True:36                try:37                    self._aquire_lock()38                    return logging.handlers.TimedRotatingFileHandler._open(self)39                except (IOError, BlockingIOError):40                    self._release_lock()41                    time.sleep(random.random())42                finally:43                    self._release_lock()4445    def _aquire_lock(self):46        try:47            self._lockf = open(self.baseFilename + '_rotating_lock', 'a')48        except PermissionError:49            # name = './{}_rotating_lock'.format(os.path.basename(self.baseFilename))50            # self._lockf = open(name, 'a')51            pass52        # fcntl.flock(self._lockf, fcntl.LOCK_EX | fcntl.LOCK_NB)5354    def _release_lock(self):55        if not self._lockf.closed:56            # fcntl.lockf(self._lockf, fcntl.LOCK_UN)57            self._lockf.close()5859    def is_same_file(self, file1, file2):60        """check is files are same by comparing inodes"""61        return os.fstat(file1.fileno()).st_ino == os.fstat(file2.fileno()).st_ino6263    def doRollover(self):64        """65        do a rollover; in this case, a date/time stamp is appended to the filename66        when the rollover happens.  However, you want the file to be named for the67        start of the interval, not the current time.  If there is a backup count,68        then we have to get a list of matching filenames, sort them and remove69        the one with the oldest suffix.70        """71        with _lock:72            return self._inner_rollover()7374    def _inner_rollover(self):75        try:76            self._aquire_lock()77        except (IOError, BlockingIOError):78            # cant aquire lock, return79            self._release_lock()80            return8182        # get the time that this sequence started at and make it a TimeTuple83        t = self.rolloverAt - self.interval84        if self.utc:85            timeTuple = time.gmtime(t)86        else:87            timeTuple = time.localtime(t)88        dfn = self.baseFilename + "." + time.strftime(self.suffix, timeTuple)8990        # check if file is same
...

__init__.py

Source:__init__.py

...14            return logging.handlers.TimedRotatingFileHandler._open(self)15        with _lock:16            while True:17                try:18                    self._aquire_lock()19                    return logging.handlers.TimedRotatingFileHandler._open(self)20                except (IOError, BlockingIOError):21                    self._release_lock()22                    time.sleep(random.random())23                finally:24                    self._release_lock()25    def _aquire_lock(self):26        try:27            self._lockf = open(self.baseFilename + '_rotating_lock', 'a')28        except PermissionError:29            name = './{}_rotating_lock'.format(os.path.basename(self.baseFilename))30            self._lockf = open(name, 'a')31        fcntl.flock(self._lockf, fcntl.LOCK_EX | fcntl.LOCK_NB)32    def _release_lock(self):33        if not self._lockf.closed:34            fcntl.lockf(self._lockf, fcntl.LOCK_UN)35            self._lockf.close()36    def is_same_file(self, file1, file2):37        """check is files are same by comparing inodes"""38        return os.fstat(file1.fileno()).st_ino == os.fstat(file2.fileno()).st_ino39    def doRollover(self):40        """41        do a rollover; in this case, a date/time stamp is appended to the filename42        when the rollover happens.  However, you want the file to be named for the43        start of the interval, not the current time.  If there is a backup count,44        then we have to get a list of matching filenames, sort them and remove45        the one with the oldest suffix.46        """47        with _lock:48            return self._innerDoRollover()49    def _innerDoRollover(self):50        try:51            self._aquire_lock()52        except (IOError, BlockingIOError):53            # cant aquire lock, return54            self._release_lock()55            return56        # get the time that this sequence started at and make it a TimeTuple57        t = self.rolloverAt - self.interval58        if self.utc:59            timeTuple = time.gmtime(t)60        else:61            timeTuple = time.localtime(t)62        dfn = self.baseFilename + "." + time.strftime(self.suffix, timeTuple)63        # check if file is same64        try:65            if self.stream:...

log_handlers.py

Source:log_handlers.py

...11        pass12    def write(self, *args, **kwargs):13        pass14class TimedRotatingFileHandlerSafe(logging.handlers.TimedRotatingFileHandler):15    def _aquire_lock(self):16        self._lockf = open(join(tempfile.gettempdir(),17                                self.baseFilename + '_log_rotating_lock'), 'a')18        fcntl.flock(self._lockf,fcntl.LOCK_EX|fcntl.LOCK_NB)19    def _release_lock(self):20        self._lockf.close()21    def doRollover(self):22        """23        do a rollover; in this case, a date/time stamp is appended to the filename24        when the rollover happens.  However, you want the file to be named for the25        start of the interval, not the current time.  If there is a backup count,26        then we have to get a list of matching filenames, sort them and remove27        the one with the oldest suffix.28        """29        try:30            self._aquire_lock()31        except IOError:32            # cant aquire lock, return33            return34        if self.stream:35            self.stream.close()36        # get the time that this sequence started at and make it a TimeTuple37        t = self.rolloverAt - self.interval38        if self.utc:39            timeTuple = time.gmtime(t)40        else:41            timeTuple = time.localtime(t)42        dfn = self.baseFilename + "." + time.strftime(self.suffix, timeTuple)43        if not os.path.exists(dfn):44            os.rename(self.baseFilename, dfn)...

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.