How to use _aquire_lock method in autotest

Best Python code snippet using autotest_python

logger.py

Source:logger.py Github

copy

Full Screen

...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 ...

Full Screen

Full Screen

__init__.py

Source:__init__.py Github

copy

Full Screen

...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:...

Full Screen

Full Screen

log_handlers.py

Source:log_handlers.py Github

copy

Full Screen

...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)...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful