From: Frank Brehm Date: Fri, 9 Mar 2018 14:55:53 +0000 (+0100) Subject: Neue Test-Methoden X-Git-Tag: 0.8.4^2~8 X-Git-Url: https://git.uhu-banane.org/?a=commitdiff_plain;h=2395bb15b3ce15c8e2c95bf61a441b4b73befc8b;p=pixelpark%2Fpuppetmaster-webhooks.git Neue Test-Methoden --- diff --git a/test/test_lock.py b/test/test_lock.py index fb13dfa..8d7fe83 100755 --- a/test/test_lock.py +++ b/test/test_lock.py @@ -46,6 +46,29 @@ class TestLockHandler(WebHooksTestcase): log.debug("Removing %r ...", self.lock_file) os.remove(self.lock_file) + # ------------------------------------------------------------------------- + def create_lockfile(self, content): + + (fd, filename) = tempfile.mkstemp() + + LOG.debug("Created temporary file %r, writing in it.", filename) + content = to_utf8(str(content)) + os.write(fd, content) + os.close(fd) + + LOG.debug("Created test lockfile: %r.", filename) + + return filename + + # ------------------------------------------------------------------------- + def remove_lockfile(self, filename): + + if os.path.exists(filename): + LOG.debug("Removing test lockfile %r ...", filename) + os.remove(filename) + else: + LOG.debug("Lockfile %r doesn't exists.", filename) + # ------------------------------------------------------------------------- def test_import(self): @@ -170,6 +193,182 @@ class TestLockHandler(WebHooksTestcase): LOG.debug("Removing lockfile %r ...", self.lock_file) locker.remove_lockfile(self.lock_basename) + # ------------------------------------------------------------------------- + def test_invalid_dir(self): + + LOG.info("Testing creation lockfile in an invalid lock directory.") + + from webhooks.lock_handler import LockHandler + from webhooks.lock_handler import LockdirNotExistsError + from webhooks.lock_handler import LockdirNotWriteableError + + ldir = '/etc/passwd' + locker = LockHandler( + appname='test_lock', + verbose=self.verbose, + lockdir=ldir, + ) + with self.assertRaises(LockdirNotExistsError) as cm: + locker.create_lockfile(self.lock_basename) + e = cm.exception + LOG.debug( + "%s raised as expected on lockdir = %r: %s.", + 'LockdirNotExistsError', ldir, e) + del locker + + if os.getegid(): + ldir = '/var' + locker = LockHandler( + appname='test_lock', + verbose=self.verbose, + lockdir=ldir, + ) + with self.assertRaises(LockdirNotWriteableError) as cm: + locker.create_lockfile(self.lock_basename) + e = cm.exception + LOG.debug( + "%s raised as expected on lockdir = %r: %s.", + 'LockdirNotWriteableError', ldir, e) + + # ------------------------------------------------------------------------- + def test_valid_lockfile(self): + + LOG.info("Testing fail on creation lockfile with a valid PID.") + + from webhooks.lock_handler import LockHandler + from webhooks.errors import CouldntOccupyLockfileError + + content = "%d\n" % (os.getpid()) + + locker = LockHandler( + appname='test_lock', + verbose=self.verbose, + lockdir=self.lock_dir, + ) + + lockfile = self.create_lockfile(content) + result = None + + try: + with self.assertRaises(CouldntOccupyLockfileError) as cm: + result = locker.create_lockfile( + lockfile, + delay_start=0.2, + delay_increase=0.4, + max_delay=5 + ) + e = cm.exception + LOG.debug( + "%s raised as expected on an valid lockfile: %s", + e.__class__.__name__, e) + self.assertEqual(lockfile, e.lockfile) + if result: + self.fail("PbLockHandler shouldn't be able to create the lockfile.") + finally: + self.remove_lockfile(lockfile) + + # ------------------------------------------------------------------------- + def test_invalid_lockfile1(self): + + LOG.info("Testing creation lockfile with an invalid previous lockfile #1.") + + from webhooks.lock_handler import LockHandler + from webhooks.errors import CouldntOccupyLockfileError + + locker = LockHandler( + appname='test_lock', + verbose=self.verbose, + lockdir=self.lock_dir, + ) + + content = "\n\n" + lockfile = self.create_lockfile(content) + result = None + + try: + with self.assertRaises(CouldntOccupyLockfileError) as cm: + result = locker.create_lockfile( # noqa + lockfile, + delay_start=0.2, + delay_increase=0.4, + max_delay=5 + ) + e = cm.exception + LOG.debug( + "%s raised as expected on an invalid lockfile (empty lines): %s", + e.__class__.__name__, e) + + finally: + self.remove_lockfile(lockfile) + + # ------------------------------------------------------------------------- + def test_invalid_lockfile2(self): + + LOG.info("Testing creation lockfile with an invalid previous lockfile #2.") + + from webhooks.lock_handler import LockHandler + from webhooks.errors import CouldntOccupyLockfileError + + locker = LockHandler( + appname='test_lock', + verbose=self.verbose, + lockdir=self.lock_dir, + ) + + content = "Bli bla blub\n\n" + lockfile = self.create_lockfile(content) + result = None + + try: + with self.assertRaises(CouldntOccupyLockfileError) as cm: + result = locker.create_lockfile( + lockfile, + delay_start=0.2, + delay_increase=0.4, + max_delay=5 + ) + e = cm.exception + LOG.debug( + "%s raised as expected on an invalid lockfile (non-numeric): %s", + e.__class__.__name__, e) + + locker.remove_lockfile(lockfile) + if result: + self.fail("LockHandler should not be able to create the lockfile.") + + finally: + self.remove_lockfile(lockfile) + + # ------------------------------------------------------------------------- + def test_invalid_lockfile3(self): + + LOG.info("Testing creation lockfile with an invalid previous lockfile #3.") + + from webhooks.lock_handler import LockHandler + + locker = LockHandler( + appname='test_lock', + verbose=self.verbose, + lockdir=self.lock_dir, + ) + + content = "123456\n\n" + lockfile = self.create_lockfile(content) + result = None + + try: + result = locker.create_lockfile( + lockfile, + delay_start=0.2, + delay_increase=0.4, + max_delay=5 + ) + locker.remove_lockfile(lockfile) + if not result: + self.fail("PbLockHandler should be able to create the lockfile.") + finally: + self.remove_lockfile(lockfile) + # ============================================================================= @@ -193,6 +392,11 @@ if __name__ == '__main__': suite.addTest(TestLockHandler('test_simple_lockfile', verbose)) suite.addTest(TestLockHandler('test_lockobject', verbose)) suite.addTest(TestLockHandler('test_refresh_lockobject', verbose)) + suite.addTest(TestLockHandler('test_invalid_dir', verbose)) + suite.addTest(TestLockHandler('test_valid_lockfile', verbose)) + suite.addTest(TestLockHandler('test_invalid_lockfile1', verbose)) + suite.addTest(TestLockHandler('test_invalid_lockfile2', verbose)) + suite.addTest(TestLockHandler('test_invalid_lockfile3', verbose)) runner = unittest.TextTestRunner(verbosity=verbose)