Neue Test-Methoden

diff --git a/test/test_lock.py b/test/test_lock.py
index fb13dfaa7e787db8314d249728a894bf97959f62..8d7fe83d041e1d80c137ad42fcacd7d934b3b638 100755 (executable)
--- 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)