LOG.debug("LockHandler %%r:\n%r", locker)
LOG.debug("LockHandler %%s:\n%s", str(locker))
+ # -------------------------------------------------------------------------
+ def test_simple_lockfile(self):
+
+ LOG.info("Testing creation and removing of a simple lockfile.")
+
+ from webhooks.lock_handler import LockHandler
+
+ locker = LockHandler(
+ appname='test_base_object',
+ verbose=self.verbose,
+ lockdir=self.lock_dir,
+ )
+ LOG.debug("Creating lockfile %r ...", self.lock_file)
+ locker.create_lockfile(self.lock_basename)
+ LOG.debug("Removing lockfile %r ...", self.lock_file)
+ locker.remove_lockfile(self.lock_basename)
+
+ # -------------------------------------------------------------------------
+ def test_lockobject(self):
+
+ LOG.info("Testing lock object on creation of a simple lockfile.")
+
+ from webhooks.lock_handler import LockHandler
+
+ locker = LockHandler(
+ appname='test_lock',
+ verbose=self.verbose,
+ lockdir=self.lock_dir,
+ )
+ try:
+ LOG.debug("Creating lockfile %r ...", self.lock_file)
+ lock = locker.create_lockfile(self.lock_basename)
+ LOG.debug("PbLock object %%r: %r", lock)
+ LOG.debug("PbLock object %%s: %s", str(lock))
+ finally:
+ LOG.debug("Removing lockfile %r ...", self.lock_file)
+ locker.remove_lockfile(self.lock_basename)
+
+ # -------------------------------------------------------------------------
+ def test_refresh_lockobject(self):
+
+ LOG.info("Testing refreshing of a lock object.")
+
+ from webhooks.lock_handler import LockHandler
+
+ locker = LockHandler(
+ appname='test_lock',
+ verbose=self.verbose,
+ lockdir=self.lock_dir,
+ )
+ try:
+ LOG.debug("Creating lockfile %r ...", self.lock_file)
+ lock = locker.create_lockfile(self.lock_basename)
+ LOG.debug("Current ctime: %s" % (lock.ctime.isoformat(' ')))
+ LOG.debug("Current mtime: %s" % (lock.mtime.isoformat(' ')))
+ fstat1 = os.stat(lock.lockfile)
+ mtime1 = fstat1.st_mtime
+ LOG.debug("Sleeping two seconds ...")
+ time.sleep(2)
+ lock.refresh()
+ LOG.debug("New mtime: %s" % (lock.mtime.isoformat(' ')))
+ fstat2 = os.stat(lock.lockfile)
+ mtime2 = fstat2.st_mtime
+ tdiff = mtime2 - mtime1
+ LOG.debug("Got a time difference between mtimes of %0.3f seconds." % (tdiff))
+ self.assertGreater(mtime2, mtime1)
+ finally:
+ LOG.debug("Removing lockfile %r ...", self.lock_file)
+ locker.remove_lockfile(self.lock_basename)
+
# =============================================================================
suite.addTest(TestLockHandler('test_import', verbose))
suite.addTest(TestLockHandler('test_could_not_occupy_lockfile_error', verbose))
suite.addTest(TestLockHandler('test_object', verbose))
+ suite.addTest(TestLockHandler('test_simple_lockfile', verbose))
+ suite.addTest(TestLockHandler('test_lockobject', verbose))
+ suite.addTest(TestLockHandler('test_refresh_lockobject', verbose))
runner = unittest.TextTestRunner(verbosity=verbose)